| """ |
| Speech Script Repository |
| 数据访问层，处理演讲稿的数据库操作 |
| """ |
|
|
| import time |
| from typing import List, Optional, Dict, Any |
| from sqlalchemy.orm import Session |
| from sqlalchemy import desc, and_ |
|
|
| from ..database.models import SpeechScript |
| from ..database.database import get_db, SessionLocal |
|
|
|
|
| class SpeechScriptRepository: |
| """演讲稿数据访问层""" |
|
|
| def __init__(self, db: Session = None): |
| self.db = db |
| self._should_close_db = db is None |
| if self.db is None: |
| self.db = SessionLocal() |
| |
| async def save_speech_script( |
| self, |
| project_id: str, |
| slide_index: int, |
| language: str, |
| slide_title: str, |
| script_content: str, |
| generation_params: Dict[str, Any], |
| estimated_duration: Optional[str] = None, |
| speaker_notes: Optional[str] = None |
| ) -> SpeechScript: |
| """保存演讲稿到数据库，如果已存在则覆盖""" |
|
|
| max_retries = 3 |
| retry_delay = 0.5 |
|
|
| for attempt in range(max_retries): |
| try: |
| |
| existing_script = self.db.query(SpeechScript).filter( |
| and_( |
| SpeechScript.project_id == project_id, |
| SpeechScript.slide_index == slide_index, |
| SpeechScript.language == (language or "zh"), |
| ) |
| ).first() |
|
|
| if existing_script: |
| |
| existing_script.slide_title = slide_title |
| existing_script.script_content = script_content |
| existing_script.language = (language or "zh") |
| existing_script.estimated_duration = estimated_duration |
| existing_script.speaker_notes = speaker_notes |
| existing_script.generation_type = generation_params.get('generation_type', 'single') |
| existing_script.tone = generation_params.get('tone', 'conversational') |
| existing_script.target_audience = generation_params.get('target_audience', 'general_public') |
| existing_script.custom_audience = generation_params.get('custom_audience') |
| existing_script.language_complexity = generation_params.get('language_complexity', 'moderate') |
| existing_script.speaking_pace = generation_params.get('speaking_pace', 'normal') |
| existing_script.custom_style_prompt = generation_params.get('custom_style_prompt') |
| existing_script.include_transitions = generation_params.get('include_transitions', True) |
| existing_script.include_timing_notes = generation_params.get('include_timing_notes', False) |
| existing_script.updated_at = time.time() |
|
|
| self.db.commit() |
| self.db.refresh(existing_script) |
| return existing_script |
| else: |
| |
| speech_script = SpeechScript( |
| project_id=project_id, |
| slide_index=slide_index, |
| language=(language or "zh"), |
| slide_title=slide_title, |
| script_content=script_content, |
| estimated_duration=estimated_duration, |
| speaker_notes=speaker_notes, |
| generation_type=generation_params.get('generation_type', 'single'), |
| tone=generation_params.get('tone', 'conversational'), |
| target_audience=generation_params.get('target_audience', 'general_public'), |
| custom_audience=generation_params.get('custom_audience'), |
| language_complexity=generation_params.get('language_complexity', 'moderate'), |
| speaking_pace=generation_params.get('speaking_pace', 'normal'), |
| custom_style_prompt=generation_params.get('custom_style_prompt'), |
| include_transitions=generation_params.get('include_transitions', True), |
| include_timing_notes=generation_params.get('include_timing_notes', False), |
| created_at=time.time(), |
| updated_at=time.time() |
| ) |
|
|
| self.db.add(speech_script) |
| self.db.commit() |
| self.db.refresh(speech_script) |
| return speech_script |
|
|
| except Exception as e: |
| self.db.rollback() |
| if "database is locked" in str(e) and attempt < max_retries - 1: |
| import asyncio |
| await asyncio.sleep(retry_delay * (attempt + 1)) |
| continue |
| else: |
| raise |
| |
| async def get_speech_scripts_by_project( |
| self, |
| project_id: str, |
| language: str = "zh", |
| limit: Optional[int] = None |
| ) -> List[SpeechScript]: |
| """获取项目的所有演讲稿""" |
| |
| query = self.db.query(SpeechScript).filter( |
| and_( |
| SpeechScript.project_id == project_id, |
| SpeechScript.language == (language or "zh"), |
| ) |
| ).order_by( |
| SpeechScript.slide_index.asc(), |
| SpeechScript.created_at.desc() |
| ) |
| |
| if limit: |
| query = query.limit(limit) |
| |
| return query.all() |
| |
| async def get_speech_scripts_by_slide( |
| self, |
| project_id: str, |
| slide_index: int, |
| language: str = "zh", |
| limit: Optional[int] = None |
| ) -> List[SpeechScript]: |
| """获取特定幻灯片的演讲稿历史""" |
| |
| query = self.db.query(SpeechScript).filter( |
| and_( |
| SpeechScript.project_id == project_id, |
| SpeechScript.slide_index == slide_index, |
| SpeechScript.language == (language or "zh"), |
| ) |
| ).order_by(desc(SpeechScript.created_at)) |
| |
| if limit: |
| query = query.limit(limit) |
| |
| return query.all() |
| |
| async def get_current_speech_scripts_by_project( |
| self, |
| project_id: str, |
| language: str = "zh", |
| ) -> List[SpeechScript]: |
| """获取项目每个幻灯片的当前演讲稿（每页只有一个）""" |
|
|
| return self.db.query(SpeechScript).filter( |
| and_( |
| SpeechScript.project_id == project_id, |
| SpeechScript.language == (language or "zh"), |
| ) |
| ).order_by(SpeechScript.slide_index.asc()).all() |
|
|
| async def get_speech_script_by_slide( |
| self, |
| project_id: str, |
| slide_index: int, |
| language: str = "zh", |
| ) -> Optional[SpeechScript]: |
| """获取特定幻灯片的演讲稿""" |
|
|
| return self.db.query(SpeechScript).filter( |
| and_( |
| SpeechScript.project_id == project_id, |
| SpeechScript.slide_index == slide_index, |
| SpeechScript.language == (language or "zh"), |
| ) |
| ).first() |
| |
| async def get_speech_script_by_id(self, script_id: int) -> Optional[SpeechScript]: |
| """根据ID获取演讲稿""" |
| return self.db.query(SpeechScript).filter(SpeechScript.id == script_id).first() |
| |
| async def delete_speech_script(self, script_id: int) -> bool: |
| """删除演讲稿""" |
| script = await self.get_speech_script_by_id(script_id) |
| if script: |
| self.db.delete(script) |
| self.db.commit() |
| return True |
| return False |
| |
| async def delete_speech_scripts_by_project(self, project_id: str) -> int: |
| """删除项目的所有演讲稿""" |
| count = self.db.query(SpeechScript).filter( |
| SpeechScript.project_id == project_id |
| ).count() |
| |
| self.db.query(SpeechScript).filter( |
| SpeechScript.project_id == project_id |
| ).delete() |
| |
| self.db.commit() |
| return count |
| |
| async def get_speech_scripts_grouped_by_slide( |
| self, |
| project_id: str, |
| language: str = "zh", |
| ) -> Dict[int, List[SpeechScript]]: |
| """获取按幻灯片分组的演讲稿""" |
| |
| scripts = await self.get_speech_scripts_by_project(project_id, language=language) |
| grouped = {} |
| |
| for script in scripts: |
| if script.slide_index not in grouped: |
| grouped[script.slide_index] = [] |
| grouped[script.slide_index].append(script) |
| |
| return grouped |
|
|
| def close(self): |
| """关闭数据库连接""" |
| if self._should_close_db and self.db: |
| self.db.close() |
|
|
