""" 会议记录数据库 - SQLite 存储 """ import os import json import sqlite3 from datetime import datetime from typing import List, Optional, Dict from contextlib import contextmanager from sqlalchemy import create_engine, Column, String, Float, Integer, DateTime, JSON, Text from sqlalchemy.orm import sessionmaker, declarative_base from sqlalchemy.pool import StaticPool Base = declarative_base() class Meeting(Base): """会议表""" __tablename__ = "meetings" meeting_id = Column(String, primary_key=True) title = Column(String, nullable=False) date = Column(DateTime, default=datetime.now) duration = Column(Float, default=0) # 秒 audio_path = Column(String, nullable=True) # 音频文件路径 transcript_path = Column(String, nullable=True) # 转录文件路径 speaker_count = Column(Integer, default=0) status = Column(String, default="pending") # pending, processing, completed, failed segments_json = Column(Text, nullable=True) # 存储片段 JSON brief_summary = Column(Text, nullable=True) # 简要摘要 detailed_summary = Column(Text, nullable=True) # 详细摘要 created_at = Column(DateTime, default=datetime.now) updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now) class Database: """数据库管理器""" def __init__(self, db_path: str = None): if db_path is None: # 优先使用环境变量（Tauri 打包后使用），否则使用项目根目录 env_dir = os.environ.get("MEETING_DATA_DIR") if env_dir: data_dir = os.path.join(env_dir, "data") else: project_root = os.path.dirname(os.path.abspath(__file__)) data_dir = os.path.join(project_root, "data") db_path = os.path.join(data_dir, "meetings.db") # 确保目录存在 os.makedirs(os.path.dirname(db_path), exist_ok=True) self.db_path = db_path self.engine = create_engine( f"sqlite:///{db_path}", connect_args={"check_same_thread": False}, poolclass=StaticPool ) Base.metadata.create_all(self.engine) self.Session = sessionmaker(bind=self.engine) @contextmanager def get_session(self): """获取数据库会话""" session = self.Session() try: yield session session.commit() except Exception: session.rollback() raise finally: session.close() def create_meeting( self, meeting_id: str, title: str, duration: float = 0, audio_path: str = None, status: str = "pending" ) -> Meeting: """创建会议记录""" meeting = Meeting( meeting_id=meeting_id, title=title, duration=duration, audio_path=audio_path, status=status ) with self.get_session() as session: session.add(meeting) return meeting def update_meeting( self, meeting_id: str, segments: List[Dict] = None, status: str = None, transcript_path: str = None, speaker_count: int = None, duration: float = None, brief_summary: str = None, detailed_summary: str = None, title: str = None ): """更新会议记录""" with self.get_session() as session: meeting = session.query(Meeting).filter(Meeting.meeting_id == meeting_id).first() if meeting: if title is not None: meeting.title = title if segments is not None: meeting.segments_json = json.dumps(segments, ensure_ascii=False) if status is not None: meeting.status = status if transcript_path is not None: meeting.transcript_path = transcript_path if speaker_count is not None: meeting.speaker_count = speaker_count if duration is not None: meeting.duration = duration if brief_summary is not None: meeting.brief_summary = brief_summary if detailed_summary is not None: meeting.detailed_summary = detailed_summary def get_meeting(self, meeting_id: str) -> Optional[Dict]: """获取单个会议记录""" with self.get_session() as session: meeting = session.query(Meeting).filter(Meeting.meeting_id == meeting_id).first() if meeting: return self._meeting_to_dict(meeting) return None def get_all_meetings(self, limit: int = 100) -> List[Dict]: """获取所有会议记录""" with self.get_session() as session: meetings = session.query(Meeting).order_by(Meeting.created_at.desc()).limit(limit).all() return [self._meeting_to_dict(m) for m in meetings] def search_meetings(self, keyword: str) -> List[Dict]: """搜索会议记录""" with self.get_session() as session: meetings = session.query(Meeting).filter( Meeting.title.like(f"%{keyword}%") ).order_by(Meeting.created_at.desc()).all() return [self._meeting_to_dict(m) for m in meetings] def delete_meeting(self, meeting_id: str): """删除会议记录""" with self.get_session() as session: meeting = session.query(Meeting).filter(Meeting.meeting_id == meeting_id).first() if meeting: session.delete(meeting) def _meeting_to_dict(self, meeting: Meeting) -> Dict: """转换为字典""" segments = [] if meeting.segments_json: try: segments = json.loads(meeting.segments_json) except json.JSONDecodeError: segments = [] return { "meeting_id": meeting.meeting_id, "title": meeting.title, "date": meeting.date.isoformat() if meeting.date else None, "duration": meeting.duration, "audio_path": meeting.audio_path, "transcript_path": meeting.transcript_path, "speaker_count": meeting.speaker_count, "status": meeting.status, "segments": segments, "brief_summary": meeting.brief_summary, "detailed_summary": meeting.detailed_summary, "created_at": meeting.created_at.isoformat() if meeting.created_at else None, "updated_at": meeting.updated_at.isoformat() if meeting.updated_at else None } def get_meetings_by_date(self, date: str) -> List[Dict]: """按日期获取会议""" with self.get_session() as session: meetings = session.query(Meeting).filter( Meeting.date >= datetime.fromisoformat(date), Meeting.date < datetime.fromisoformat(date + "T23:59:59") ).order_by(Meeting.date.desc()).all() return [self._meeting_to_dict(m) for m in meetings] """数据库单例实例""" _db_instance = None def init_db(db_path: str = None) -> Database: """初始化数据库（单例模式）""" global _db_instance if _db_instance is None: _db_instance = Database(db_path) return _db_instance if __name__ == "__main__": # 测试 db = init_db(":memory:") # 创建测试数据 meeting_id = "test_001" db.create_meeting( meeting_id=meeting_id, title="测试会议", duration=300 ) db.update_meeting( meeting_id=meeting_id, segments=[ {"start": 0, "end": 10, "speaker": 0, "text": "测试文本1"}, {"start": 10, "end": 20, "speaker": 1, "text": "测试文本2"} ], status="completed", speaker_count=2 ) # 查询 meeting = db.get_meeting(meeting_id) print(f"会议: {meeting['title']}") print(f"状态: {meeting['status']}") print(f"片段数: {len(meeting['segments'])}")