ax-backend-v1/app/services/document_service.py

"""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