vent_ai_platform/app/services/document_service.py

from __future__ import annotations

import shutil
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path

from sqlalchemy.orm import Session

from app.core.config import settings
from app.core.errors import APIError
from app.db.models.document import BusinessDocument, DocumentExtraction
from app.db.repositories import document as document_repo
from app.db.session import SessionLocal
from app.schemas.document import (
    BusinessDocumentCreate,
    BusinessDocumentUpdate,
    DocumentExtractionCreate,
    DocumentExtractionUpdate,
)
from app.services.extraction_accuracy import judge_extraction_edit


@dataclass
class ExtractionEditTask:
    extraction_id: str
    edit_count: int
    field_name: str
    original_value: str
    edited_value: str
    evidence_text: str
    document_title: str


def create_document(
    db: Session,
    payload: BusinessDocumentCreate,
) -> BusinessDocument:
    data = payload.model_dump(exclude={"metadata"})
    data["metadata_json"] = payload.metadata
    return document_repo.create_document(db, data)


def update_document(
    db: Session,
    document_id: str,
    payload: BusinessDocumentUpdate,
) -> BusinessDocument:
    document = document_repo.get_document(db, document_id)
    if document is None:
        raise APIError(404, "document_not_found", "业务资料不存在")
    updates = payload.model_dump(exclude_unset=True)
    if "metadata" in updates:
        updates["metadata_json"] = updates.pop("metadata")
    return document_repo.update_document(db, document, updates)


def get_document_detail(
    db: Session,
    document_id: str,
) -> tuple[BusinessDocument, list[DocumentExtraction]]:
    document = document_repo.get_document(db, document_id)
    if document is None:
        raise APIError(404, "document_not_found", "业务资料不存在")
    return document, document_repo.list_extractions(db, document_id)


def add_extraction(
    db: Session,
    payload: DocumentExtractionCreate,
) -> DocumentExtraction:
    if document_repo.get_document(db, payload.document_id) is None:
        raise APIError(404, "document_not_found", "业务资料不存在")
    data = payload.model_dump(exclude={"metadata"})
    data["metadata_json"] = payload.metadata
    return document_repo.create_extraction(db, data)


def update_extraction(
    db: Session,
    document_id: str,
    extraction_id: str,
    payload: DocumentExtractionUpdate,
) -> tuple[DocumentExtraction, ExtractionEditTask | None]:
    extraction = document_repo.get_extraction(db, extraction_id)
    if extraction is None or extraction.document_id != document_id:
        raise APIError(404, "document_extraction_not_found", "结构化字段不存在")
    updates = payload.model_dump(exclude_unset=True)
    new_value = updates.get("normalized_value")
    if new_value is None:
        new_value = updates.get("field_value")
    previous_value = extraction.normalized_value or extraction.field_value or ""
    task: ExtractionEditTask | None = None
    if new_value is not None and str(new_value) != str(previous_value):
        document = document_repo.get_document(db, document_id)
        if not extraction.original_value:
            updates["original_value"] = previous_value
        updates["edited"] = True
        updates["edit_count"] = int(extraction.edit_count or 0) + 1
        reverted_to_original = bool(extraction.edited) and (
            str(new_value).strip() == str(extraction.original_value or "").strip()
        )
        if reverted_to_original:
            updates["edited"] = False
            updates["extraction_correct"] = True
            updates["edit_judgement_status"] = "completed"
            updates["edit_reason"] = "用户最终恢复为原始抽取值，字段按正确计算。"
            updates["edit_judgement_source"] = "reverted"
            updates["edit_judgement_confidence"] = 1.0
            updates["edit_judged_at"] = datetime.now(timezone.utc)
        else:
            updates["edit_judgement_status"] = "pending"
            updates["edit_reason"] = ""
            updates["edit_judgement_source"] = ""
            updates["edit_judgement_confidence"] = None
            updates["edit_judged_at"] = None
            task = ExtractionEditTask(
                extraction_id=extraction.id,
                edit_count=int(updates["edit_count"]),
                field_name=extraction.field_name or extraction.field_key,
                original_value=previous_value,
                edited_value=str(new_value),
                evidence_text=extraction.evidence_text or "",
                document_title=document.title if document else "",
            )
    updates["verified"] = True
    updates["verified_by"] = payload.verified_by or "前端用户"
    saved = document_repo.update_extraction(db, extraction, updates)
    return saved, task


async def process_extraction_edit_judgement(task: ExtractionEditTask) -> None:
    judgement = await judge_extraction_edit(
        field_name=task.field_name,
        original_value=task.original_value,
        edited_value=task.edited_value,
        evidence_text=task.evidence_text,
        document_title=task.document_title,
        extraction_id=task.extraction_id,
    )
    with SessionLocal() as db:
        extraction = document_repo.get_extraction(db, task.extraction_id)
        if extraction is None or int(extraction.edit_count or 0) != task.edit_count:
            return
        current_value = extraction.normalized_value or extraction.field_value or ""
        if str(current_value) != task.edited_value:
            return
        document_repo.update_extraction(
            db,
            extraction,
            {
                "extraction_correct": bool(extraction.extraction_correct)
                and not judgement.is_extraction_error,
                "edit_reason": judgement.reason,
                "edit_judgement_source": judgement.source,
                "edit_judgement_confidence": judgement.confidence,
                "edit_judgement_status": "completed",
                "edit_judged_at": datetime.now(timezone.utc),
            },
        )


def delete_document(db: Session, document_id: str) -> None:
    document = document_repo.get_document(db, document_id)
    if document is None:
        raise APIError(404, "document_not_found", "业务资料不存在")
    upload_root = settings.upload_path.resolve()
    stored_file = Path(document.file_path)
    if stored_file.exists():
        resolved_file = stored_file.resolve()
        if resolved_file.is_relative_to(upload_root):
            resolved_file.unlink(missing_ok=True)
    parsed_dir = upload_root / "parsed" / document.id
    if parsed_dir.exists():
        resolved_parsed = parsed_dir.resolve()
        if resolved_parsed.is_relative_to(upload_root):
            shutil.rmtree(resolved_parsed)
    document_repo.delete_document(db, document)