"""document_service.py — 文档 CRUD(基于 SQLite Block 存储)""" import tempfile from datetime import date, datetime, timezone from pathlib import Path import httpx from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from app.config import settings from app.core.exceptions import DocumentNotFoundError, DocumentParseError from app.models.document import Document from app.schemas.document import CreateDocumentRequest from app.services.content_db import ContentDB from app.services.word_parser import parse_word_to_blocks # ------------------------------------------------------------------ # # Word 文档下载 # ------------------------------------------------------------------ # async def _download_word(file_url: str) -> Path: """下载 Word 文档到临时文件""" suffix = Path(file_url.split("?")[0]).suffix.lower() or ".docx" if suffix not in (".doc", ".docx"): raise DocumentParseError(f"不支持的文件格式: {suffix}") try: async with httpx.AsyncClient(timeout=60) as client: resp = await client.get(file_url) resp.raise_for_status() except httpx.HTTPError as exc: raise DocumentParseError(f"文件下载失败: {exc}") from exc with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as tmp: tmp_path = Path(tmp.name) tmp_path.write_bytes(resp.content) return tmp_path # ------------------------------------------------------------------ # # 服务类 # ------------------------------------------------------------------ # class DocumentService: def __init__(self, db: AsyncSession) -> None: self.db = db async def create_document(self, data: CreateDocumentRequest) -> Document: """创建文档:下载 Word → 解析为 Blocks → 存入 SQLite""" # 1. 下载 Word 文档 tmp_path = await _download_word(data.file_url) try: # 2. 解析为 Blocks blocks = parse_word_to_blocks(tmp_path) # 3. 创建 SQLite 数据库 user_id = data.user_id or "default-user" today = date.today().strftime("%Y-%m-%d") db_dir = Path(settings.temp_dir) / user_id / "sqlite" db_dir.mkdir(parents=True, exist_ok=True) # 生成文档 ID import uuid doc_id = f"doc-{uuid.uuid4().hex[:12]}" db_path = db_dir / f"{doc_id}.db" # 4. 写入 Blocks with ContentDB(str(db_path)) as content_db: content_db.create_tables() content_db.insert_blocks(blocks) # 5. 创建文档记录 doc = Document( id=doc_id, content_db_path=str(db_path), session_id=data.session_id, created_by=data.user_id, ) self.db.add(doc) await self.db.commit() await self.db.refresh(doc) return doc finally: # 6. 清理临时文件 tmp_path.unlink(missing_ok=True) async def get_document(self, document_id: str, include_blocks: bool = False) -> Document: """获取文档详情""" result = await self.db.execute( select(Document).where(Document.id == document_id) ) doc = result.scalar_one_or_none() if doc is None: raise DocumentNotFoundError(document_id) # 可选:加载 blocks if include_blocks: with ContentDB(doc.content_db_path) as content_db: blocks = content_db.get_blocks() # 动态添加 blocks 属性 doc.blocks = blocks return doc async def list_documents( self, user_id: str, page: int = 1, page_size: int = 20, session_id: str | None = None, sort_by: str = "updated_at", sort_order: str = "desc", ) -> tuple[list[Document], int]: """获取文档列表""" query = select(Document).where(Document.created_by == user_id) if session_id: query = query.where(Document.session_id == session_id) sort_col = getattr(Document, sort_by, Document.updated_at) query = query.order_by(sort_col.asc() if sort_order == "asc" else sort_col.desc()) count_q = select(func.count()).select_from(query.subquery()) total: int = (await self.db.execute(count_q)).scalar_one() offset = (page - 1) * page_size result = await self.db.execute(query.offset(offset).limit(page_size)) return list(result.scalars().all()), total async def delete_document(self, document_id: str) -> None: """删除文档(同时删除 SQLite 文件)""" doc = await self.get_document(document_id) # 删除 SQLite 文件 db_path = Path(doc.content_db_path) if db_path.exists(): db_path.unlink() # 删除数据库记录 await self.db.delete(doc) await self.db.commit() async def delete_documents_by_session(self, session_id: str) -> int: """删除指定 sessionId 的所有文档(同时删除 SQLite 文件)""" result = await self.db.execute( select(Document).where(Document.session_id == session_id) ) docs = list(result.scalars().all()) for doc in docs: # 删除 SQLite 文件 db_path = Path(doc.content_db_path) if db_path.exists(): db_path.unlink() await self.db.delete(doc) await self.db.commit() return len(docs) async def update_document_timestamp(self, document_id: str) -> Document: """更新文档的 updated_at 时间戳""" doc = await self.get_document(document_id) doc.updated_at = datetime.now(timezone.utc) await self.db.commit() await self.db.refresh(doc) return doc