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)