from __future__ import annotations import hashlib import json import os import re import traceback from datetime import datetime from pathlib import Path from tempfile import NamedTemporaryFile from typing import Any from uuid import uuid4 import magic from fastapi import BackgroundTasks, Depends, FastAPI, File, Form, HTTPException, UploadFile from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse, Response from pydantic import BaseModel, Field from sqlalchemy import func from sqlalchemy.orm import Session from app.db import Base, engine, get_db from app.config import settings from app.extractor import ( RuleBasedLaborExtractor, build_case_elements_table, load_case_elements_schema, merge_dispute_template_fields, refresh_derived_element_fields, ) from app.migrate import ensure_mysql_schema from app.models import Case, CaseElementsVersion, CaseFile, ProcessingTask from app.services.document_parser import DocumentParser from app.services.hybrid_extractor import HybridExtractor from app.services.complexity_classifier import classify_complexity from app.services.portrait_generator import generate_portrait from app.services.risk_predictor import assess_risk app = FastAPI(title="Labor Arbitration Backend", version="0.1.0") app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # 开发期轻量迁移(避免你改模型后 MySQL 表不同步导致 500) ensure_mysql_schema(engine) Base.metadata.create_all(bind=engine) extractor = RuleBasedLaborExtractor() hybrid_extractor = HybridExtractor() _ollama_claims = None if settings.use_ollama: try: from app.anj import OllamaClaimsExtractor _ollama_claims = OllamaClaimsExtractor(settings.ollama_base_url, settings.ollama_model_name) except Exception: _ollama_claims = None @app.exception_handler(Exception) async def global_exception_handler(request, exc: Exception): return JSONResponse( status_code=500, content={"detail": "服务器内部错误", "error": str(exc), "trace": traceback.format_exc()[:4000]}, ) def _upload_dir() -> Path: p = Path(os.getenv("UPLOAD_DIR", "uploads")) p.mkdir(parents=True, exist_ok=True) return p # 入库全文上限(MySQL LONGTEXT 足够;此处防止极端大文件占满内存) _MAX_STORED_APPLICATION_TEXT = 200_000 # 接口返回给前端的最大长度(避免 JSON 过大) _MAX_API_APPLICATION_TEXT = 50_000 def _sync_case_elements_table(elements: dict[str, Any], *, rebuild_hierarchy: bool = True) -> None: """扁平要素定稿后再生成分组表;可选按扁平字段重算案由层级树(避免 PUT 仅改层级时被覆盖)。""" elements["case_elements_table"] = build_case_elements_table(elements, load_case_elements_schema()) if not rebuild_hierarchy: return from app.hierarchy_extract import build_elements_hierarchy_for_cause, load_hierarchy_templates cause = elements.get("primary_cause_type") tmpl = load_hierarchy_templates() if cause and tmpl.get(cause): h = build_elements_hierarchy_for_cause(cause, elements) if h is not None: elements["elements_hierarchy"] = h def _normalize_case_name(name: str) -> str: return (name or "").replace("\u3000", " ").strip() def _material_fingerprint_from_file_contents(contents: list[bytes]) -> str: if not contents: return hashlib.sha256(b"").hexdigest() digests = sorted(hashlib.sha256(c).hexdigest() for c in contents) return hashlib.sha256("|".join(digests).encode("utf-8")).hexdigest() def _normalized_text_fingerprint(merged_stored: str) -> str: """空白规范化后哈希,减轻解析换行/空格差异导致的重复漏判。""" slice_ = (merged_stored or "")[:_MAX_STORED_APPLICATION_TEXT] norm = re.sub(r"[\s\u3000]+", " ", slice_.strip()) return hashlib.sha256(norm.encode("utf-8", errors="ignore")).hexdigest() def _merged_text_from_prepared(prepared: list[tuple[UploadFile, bytes, str]]) -> str: chunks: list[str] = [] for f, content, _ in prepared: suffix = Path(f.filename).suffix.lower() or ".txt" tmp = NamedTemporaryFile(delete=False, suffix=suffix) try: tmp.write(content) tmp.close() chunks.append(DocumentParser.parse_file(tmp.name)) finally: try: Path(tmp.name).unlink(missing_ok=True) except OSError: pass return "\n\n".join(t for t in chunks if t.strip()) def _find_existing_duplicate_case( db: Session, norm_name: str, fingerprint: str, merged_stored: str ) -> Case | None: """ 去重顺序: 1) 同名 + 材料指纹一致,或同名 + 规范化正文哈希一致(兼容旧行无指纹) 2) 任意案件材料指纹一致(材料相同即视为重复,不强制名称一致) 3) 最近若干条有正文的案件中,规范化正文哈希一致(无指纹旧数据) """ text_h = _normalized_text_fingerprint(merged_stored) if norm_name: cands_name = db.query(Case).filter(func.trim(Case.case_name) == norm_name).all() for c in cands_name: mf = (c.material_fingerprint or "").strip() if mf and mf == fingerprint: return c app_raw = c.application_text or "" if app_raw and _normalized_text_fingerprint(app_raw) == text_h: return c row = db.query(Case).filter(Case.material_fingerprint == fingerprint).filter(Case.material_fingerprint.isnot(None)).first() if row: return row # 仅「同名 + 正文哈希一致」才命中,避免不同案件名称因正文规范化后相同而误命中旧案(导致前端一直显示首次抽取结果) recent = ( db.query(Case) .filter(Case.application_text.isnot(None)) .filter(Case.application_text != "") .order_by(Case.id.desc()) .limit(400) .all() ) for c in recent: if _normalize_case_name(c.case_name or "") != norm_name: continue if _normalized_text_fingerprint(c.application_text or "") == text_h: return c return None def _allowed(content_type: str, filename: str) -> bool: ext = Path(filename).suffix.lower() if ext in {".pdf", ".docx", ".txt"}: return True if content_type in { "application/pdf", "application/vnd.openxmlformats-officedocument.wordprocessingml.document", "text/plain", }: return True return False def _jaccard(a: set[str], b: set[str]) -> float: if not a or not b: return 0.0 inter = len(a & b) union = len(a | b) return inter / union if union else 0.0 def _case_tag_set(elements: dict[str, Any]) -> set[str]: tags = set() for k in ["case_cause", "employer_nature", "worker_position", "applicant_name", "respondent_name"]: v = elements.get(k) if v: tags.add(str(v)) for law in (elements.get("law_refs") or []): tags.add(str(law)) claims = elements.get("claims") or {} for item in (claims.get("items") or []): if item: tags.add(str(item)[:20]) for ct in elements.get("claim_types") or []: if ct: tags.add(str(ct)) return tags def _elements_brief(elements: dict[str, Any] | None) -> dict[str, Any]: el = elements or {} claims = el.get("claims") or {} return { "case_number": el.get("case_number"), "arbitration_org": el.get("arbitration_org"), "applicant_name": el.get("applicant_name"), "respondent_name": el.get("respondent_name"), "employer_nature": el.get("employer_nature"), "worker_position": el.get("worker_position"), "case_cause": el.get("case_cause"), "claim_types": el.get("claim_types"), "entry_date": el.get("entry_date"), "leave_date": el.get("leave_date"), "month_salary": el.get("month_salary"), "claims_amount_total": claims.get("amount_total"), "claims_items_preview": (claims.get("items") or [])[:5], "law_refs": (el.get("law_refs") or [])[:8], } def _portrait_scores(portrait: dict[str, Any] | None) -> dict[str, Any]: p = portrait or {} scores = p.get("scores") or {} return { "legal": scores.get("legal"), "fact": scores.get("fact"), "risk": scores.get("risk"), "risk_level": p.get("risk_level"), } _SIM_WEIGHT_LEGAL = 0.45 _SIM_WEIGHT_FACT = 0.35 _SIM_WEIGHT_RISK = 0.20 def _subscore(p: dict[str, Any] | None, dim: str, name: str) -> float | None: if not p: return None d = (p.get(dim) or {}).get("subscores") or {} v = d.get(name) try: return float(v) if v is not None else None except (TypeError, ValueError): return None def _sim_0_100(a: float | None, b: float | None) -> float | None: if a is None or b is None: return None return max(0.0, 100.0 - abs(float(a) - float(b))) def _weighted_similarity_from_portraits(base_p: dict[str, Any] | None, sim_p: dict[str, Any] | None) -> dict[str, Any]: """ 计算法律/事实/风险三维度加权相似度(0-100 越高越相似),并返回 breakdown。 - 法律维度:争议焦点、法律适用/法条引用(用画像 subscores 近似) - 事实维度:证据完备度、事实清晰度 - 风险维度:诉求支持可能性、矛盾激化程度 """ legal_sim = None ls1 = _sim_0_100(_subscore(base_p, "legal_dimension", "争议焦点明确度"), _subscore(sim_p, "legal_dimension", "争议焦点明确度")) ls2 = _sim_0_100(_subscore(base_p, "legal_dimension", "法律条款引用充分度"), _subscore(sim_p, "legal_dimension", "法律条款引用充分度")) legal_parts = [x for x in (ls1, ls2) if x is not None] if legal_parts: legal_sim = sum(legal_parts) / len(legal_parts) fact_sim = None fs1 = _sim_0_100(_subscore(base_p, "fact_dimension", "证据完备度(近似)"), _subscore(sim_p, "fact_dimension", "证据完备度(近似)")) fs2 = _sim_0_100(_subscore(base_p, "fact_dimension", "事实描述清晰度"), _subscore(sim_p, "fact_dimension", "事实描述清晰度")) fact_parts = [x for x in (fs1, fs2) if x is not None] if fact_parts: fact_sim = sum(fact_parts) / len(fact_parts) risk_sim = None rs1 = _sim_0_100( _subscore(base_p, "risk_dimension", "诉求支持可能性(规则近似)"), _subscore(sim_p, "risk_dimension", "诉求支持可能性(规则近似)"), ) rs2 = _sim_0_100( _subscore(base_p, "risk_dimension", "矛盾激化程度(规则近似)"), _subscore(sim_p, "risk_dimension", "矛盾激化程度(规则近似)"), ) risk_parts = [x for x in (rs1, rs2) if x is not None] if risk_parts: risk_sim = sum(risk_parts) / len(risk_parts) breakdown: list[dict[str, Any]] = [ {"dimension": "法律维度", "weight": _SIM_WEIGHT_LEGAL, "similarity": legal_sim}, {"dimension": "事实维度", "weight": _SIM_WEIGHT_FACT, "similarity": fact_sim}, {"dimension": "风险维度", "weight": _SIM_WEIGHT_RISK, "similarity": risk_sim}, ] total_w = 0.0 total = 0.0 for r in breakdown: sim = r.get("similarity") if sim is None: continue w = float(r["weight"]) total_w += w total += w * float(sim) overall = (total / total_w) if total_w > 0 else None return {"overall": overall, "breakdown": breakdown} def _comparison_with_current(base_el: dict[str, Any], sim_el: dict[str, Any]) -> list[dict[str, Any]]: rows: list[tuple[str, str, str]] = [ ("案件编号", "case_number", "case_number"), ("仲裁机构", "arbitration_org", "arbitration_org"), ("案由类型", "case_cause", "case_cause"), ("申请人", "applicant_name", "applicant_name"), ("被申请人", "respondent_name", "respondent_name"), ("单位性质", "employer_nature", "employer_nature"), ("岗位", "worker_position", "worker_position"), ("入职日期", "entry_date", "entry_date"), ("离职日期", "leave_date", "leave_date"), ("月工资标准", "month_salary", "month_salary"), ] out: list[dict[str, Any]] = [] for label, key, _ in rows: a, b = base_el.get(key), sim_el.get(key) out.append( { "dimension": label, "current": a, "similar": b, "aligned": a is not None and b is not None and str(a) == str(b), } ) bc = (base_el.get("claims") or {}).get("amount_total") sc = (sim_el.get("claims") or {}).get("amount_total") amt_aligned = False if bc is not None and sc is not None: try: amt_aligned = float(bc) == float(sc) except (TypeError, ValueError): amt_aligned = str(bc) == str(sc) out.append( { "dimension": "请求金额合计", "current": bc, "similar": sc, "aligned": amt_aligned, } ) return out def _flatten_hierarchy_values(elements: dict[str, Any] | None) -> dict[str, Any]: """ 将 elements.elements_hierarchy 扁平化为 { \"一层 / 二层 / 三层\": value },仅保留有内容的叶子。 """ el = elements or {} root = el.get("elements_hierarchy") or {} schema = load_case_elements_schema() field_labels: dict[str, str] = schema.get("field_labels") or {} out: dict[tuple[str, ...], Any] = {} _COMMON_SEG_CN = { "name": "姓名", "gender": "性别", "birthday": "出生日期", "birthdate": "出生日期", "age": "年龄", "phone": "联系电话", "mobile": "联系电话", "tel": "联系电话", "residence": "住所地", "address": "地址", "identity_number": "身份证号", "id_number": "身份证号", "idcard": "身份证号", "nationality": "国籍/民族", "nation": "民族", "company": "单位名称", "company_name": "单位名称", "legal_representative": "法定代表人", "representative": "代表人", "position": "职务/岗位", } def cn_seg(seg: str) -> str: s = (seg or "").strip() if not s: return s # 若是已是中文则直接返回 if re.search(r"[\u4e00-\u9fff]", s): return s if s in _COMMON_SEG_CN: return _COMMON_SEG_CN[s] # 扁平字段英文 key -> 中文 label(来自 schema) return field_labels.get(s, s) def walk(node: Any, path: list[str]) -> None: if isinstance(node, dict): for k, v in node.items(): walk(v, path + [cn_seg(str(k))]) return if node is None: return if isinstance(node, str) and not node.strip(): return if not path: return out[tuple(path)] = node walk(root, []) return out def _hierarchy_comparison_with_current(base_el: dict[str, Any], sim_el: dict[str, Any]) -> list[dict[str, Any]]: a = _flatten_hierarchy_values(base_el) b = _flatten_hierarchy_values(sim_el) keys = sorted(set(a.keys()) | set(b.keys())) out: list[dict[str, Any]] = [] for k in keys: va = a.get(k) vb = b.get(k) if va is None and vb is None: continue aligned = va is not None and vb is not None and str(va) == str(vb) out.append({"path_parts": list(k), "current": va, "similar": vb, "aligned": aligned}) return out def _build_similar_case_detail(base_el: dict[str, Any], base_portrait: dict[str, Any] | None, row: Case, score: float) -> dict[str, Any]: sim_el = row.elements or {} full_text = row.application_text or "" api_text = full_text[:_MAX_API_APPLICATION_TEXT] if full_text else "" wsim = _weighted_similarity_from_portraits(base_portrait or {}, row.portrait or {}) return { "case_id": row.id, "case_name": row.case_name, "score": round(float(score), 6), "similarity_percent": round(float(score) * 100, 2), "weighted_similarity_percent": (round(float(wsim["overall"]), 2) if wsim.get("overall") is not None else None), "weighted_similarity_breakdown": wsim.get("breakdown") or [], "application_text": api_text, "application_truncated": len(full_text) > len(api_text), "application_preview": (full_text[:320] + "…") if len(full_text) > 320 else full_text, "ruling_result": row.ruling_result or "暂无裁决结论(可在数据库为该案件录入 ruling_result,或后续对接裁决书解析模块)", "arbitration_org": row.arbitration_org or "未录入仲裁机构(可在数据库为该案件录入 arbitration_org)", "elements_brief": _elements_brief(sim_el), "portrait_scores": _portrait_scores(row.portrait or {}), "comparison_with_current": _comparison_with_current(base_el, sim_el), "hierarchy_comparison_with_current": _hierarchy_comparison_with_current(base_el, sim_el), } # ----------------------------- # Pydantic Models(按接口定义) # ----------------------------- class TaskResponse(BaseModel): task_id: str status: str progress: int message: str | None = None class ElementsResponse(BaseModel): case_id: int elements: dict[str, Any] version: int class ElementsUpdateRequest(BaseModel): updated_by: str | None = None patch: dict[str, Any] = Field(default_factory=dict, description="部分字段更新:只传要改的字段") class PortraitResponse(BaseModel): case_id: int portrait: dict[str, Any] complexity_level: dict[str, Any] risk_assessment: dict[str, Any] class SimilarRequest(BaseModel): top_k: int = 10 class SimilarCaseDetailResponse(BaseModel): case_id: int case_name: str score: float similarity_percent: float application_text: str = "" application_truncated: bool = False application_preview: str = "" ruling_result: str = "" arbitration_org: str = "" elements_brief: dict[str, Any] = Field(default_factory=dict) portrait_scores: dict[str, Any] = Field(default_factory=dict) comparison_with_current: list[dict[str, Any]] = Field(default_factory=list) hierarchy_comparison_with_current: list[dict[str, Any]] = Field(default_factory=list) weighted_similarity_percent: float | None = None weighted_similarity_breakdown: list[dict[str, Any]] = Field(default_factory=list) class SimilarResponse(BaseModel): case_id: int items: list[SimilarCaseDetailResponse] class MetricsResponse(BaseModel): metrics: dict[str, Any] class UploadCompatResponse(BaseModel): """ 兼容旧前端(React 版)用的同步返回格式: - extracted_elements: 规则/模型抽取结果 - case_profile: 画像数据(含 complexity_level/risk_assessment 等) - similar_cases: 相似案件列表(Jaccard + 类案详情) """ case_id: int case_name: str extracted_elements: dict[str, Any] case_profile: dict[str, Any] similar_cases: list[dict[str, Any]] current_application_text: str = "" current_application_truncated: bool = False current_elements_brief: dict[str, Any] = Field(default_factory=dict) reused_existing: bool = Field(False, description="True 表示命中已有案件,未新建 cases 行") def _build_upload_compat_response(db: Session, case: Case, *, reused_existing: bool) -> UploadCompatResponse: elements = case.elements or {} base_tags = _case_tag_set(elements) rows = db.query(Case).filter(Case.id != case.id).all() scored: list[tuple[float, Case]] = [] for row in rows: s = _jaccard(base_tags, _case_tag_set(row.elements or {})) scored.append((s, row)) scored.sort(key=lambda x: x[0], reverse=True) similar_cases = [_build_similar_case_detail(elements, case.portrait or {}, r, s) for s, r in scored[:10] if s > 0] full_text = case.application_text or "" cur_api_text = full_text[:_MAX_API_APPLICATION_TEXT] return UploadCompatResponse( case_id=case.id, case_name=case.case_name, extracted_elements=elements, case_profile=case.portrait or {}, similar_cases=similar_cases, current_application_text=cur_api_text, current_application_truncated=len(full_text) > len(cur_api_text), current_elements_brief=_elements_brief(elements), reused_existing=reused_existing, ) @app.get("/health") def health_check(): return {"status": "ok"} @app.post("/api/cases/extract") async def extract_case(file: UploadFile = File(...)): """ 上传单文件(.txt / .pdf / .docx),解析全文后使用混合抽取器返回 JSON 要素。 """ suffix = Path(file.filename or "").suffix.lower() if suffix not in (".txt", ".pdf", ".docx"): raise HTTPException(status_code=400, detail="仅支持 .txt、.pdf、.docx 格式") content = await file.read() if not content: raise HTTPException(status_code=400, detail="空文件") tmp_path: str | None = None try: tmp = NamedTemporaryFile(delete=False, suffix=suffix) tmp_path = tmp.name tmp.write(content) tmp.close() text = DocumentParser.parse_file(tmp_path) except Exception as e: raise HTTPException(status_code=400, detail=f"文件解析失败:{e}") from e finally: if tmp_path: try: Path(tmp_path).unlink(missing_ok=True) except OSError: pass if not (text or "").strip(): raise HTTPException(status_code=400, detail="未解析到有效文本") return hybrid_extractor.extract(text) @app.post("/api/cases/upload", response_model=UploadCompatResponse) async def upload_case_compat( case_name: str = Form(...), files: list[UploadFile] = File(...), extractor_mode: str = Form("rules"), db: Session = Depends(get_db), ): """ 兼容接口:旧前端会调用 /api/cases/upload 并期待立即返回抽取结果。 新版推荐走:/api/cases/{case_id}/files + /api/tasks/{task_id} 轮询。 extractor_mode: rules / bert / ollama / hybrid """ if not files: raise HTTPException(status_code=400, detail="未上传文件") upload_dir = _upload_dir() norm_name = _normalize_case_name(case_name) if not norm_name: raise HTTPException(status_code=400, detail="案件名称不能为空") prepared: list[tuple[UploadFile, bytes, str]] = [] for f in files: content = await f.read() guessed = magic.from_buffer(content[:2048], mime=True) if content else "" if not _allowed(guessed, f.filename): raise HTTPException(status_code=400, detail=f"不支持的文件类型:{f.filename}({guessed})") prepared.append((f, content, guessed)) merged_full = _merged_text_from_prepared(prepared) if not merged_full.strip(): raise HTTPException(status_code=400, detail="未读取到有效文本") merged_stored = merged_full[:_MAX_STORED_APPLICATION_TEXT] fingerprint = _material_fingerprint_from_file_contents([b for _, b, _ in prepared]) text_h = _normalized_text_fingerprint(merged_stored) case = _find_existing_duplicate_case(db, norm_name, fingerprint, merged_stored) # 补充:同名 + 正文一致即视为重复 if not case and norm_name: cands = db.query(Case).filter(func.trim(Case.case_name) == norm_name).filter(Case.application_text.isnot(None)).filter(Case.application_text != "").all() for c in cands: if _normalized_text_fingerprint(c.application_text or "") == text_h: case = c break if case: # 案件已存在:直接返回已有数据,跳过模型抽取 case.case_name = norm_name # 清理旧文件并保存新文件 old_files = db.query(CaseFile).filter(CaseFile.case_id == case.id).all() for cf in old_files: try: Path(cf.storage_path).unlink(missing_ok=True) except OSError: pass db.query(CaseFile).filter(CaseFile.case_id == case.id).delete(synchronize_session=False) db.commit() for f, content, guessed in prepared: suffix = Path(f.filename).suffix.lower() storage = upload_dir / f"{case.id}_{uuid4().hex}{suffix}" storage.write_bytes(content) db.add(CaseFile(case_id=case.id, filename=f.filename, content_type=guessed, storage_path=str(storage), size_bytes=len(content))) db.commit() case.material_fingerprint = fingerprint case.application_text = merged_stored db.commit() return _build_upload_compat_response(db, case, reused_existing=False) # 新案件:走完整抽取流程 case = Case(case_name=norm_name, elements={}, portrait={}, material_fingerprint=fingerprint) db.add(case) db.commit() db.refresh(case) for f, content, guessed in prepared: suffix = Path(f.filename).suffix.lower() storage = upload_dir / f"{case.id}_{uuid4().hex}{suffix}" storage.write_bytes(content) db.add(CaseFile(case_id=case.id, filename=f.filename, content_type=guessed, storage_path=str(storage), size_bytes=len(content))) db.commit() merged = merged_full hybrid_extractor.mode = extractor_mode if extractor_mode == "rules": elements = extractor.extract(merged) else: elements = hybrid_extractor.extract(merged) if _ollama_claims is not None and extractor_mode in ("ollama", "hybrid"): try: elements["claims"] = _ollama_claims.extract_claims(merged) elements.update(merge_dispute_template_fields(merged, elements)) except Exception: pass try: patch = _ollama_claims.extract_dispute_template_fields(merged) for k, v in (patch or {}).items(): if v is None: continue if isinstance(v, str) and not str(v).strip(): continue elements[k] = v if elements.get("tmpl_primary_cause"): elements["primary_cause_type"] = elements["tmpl_primary_cause"] except Exception: pass refresh_derived_element_fields(elements, merged) portrait = generate_portrait(elements, raw_text=merged, evidence_count=len(files)) complexity = classify_complexity(elements, evidence_count=len(files)) risk = assess_risk(elements) latest_version = ( db.query(CaseElementsVersion) .filter(CaseElementsVersion.case_id == case.id) .order_by(CaseElementsVersion.version.desc()) .first() ) next_ver = (latest_version.version + 1) if latest_version else 1 db.add(CaseElementsVersion(case_id=case.id, version=next_ver, elements=elements, updated_by="system")) case.elements = elements case.portrait = {**portrait, "complexity_level": complexity, "risk_assessment": risk} case.application_text = merged_stored case.material_fingerprint = fingerprint _update_case_index_fields(case, elements, portrait, risk.get("level")) db.commit() return _build_upload_compat_response(db, case, reused_existing=False) def _create_or_get_case(db: Session, case_id: int) -> Case: case = db.query(Case).filter(Case.id == case_id).first() if case: return case # 若前端先创建 case_id 再上传文件,这里兜底自动创建 case = Case(id=case_id, case_name=f"案件{case_id}", elements={}, portrait={}) db.add(case) db.commit() db.refresh(case) return case def _update_case_index_fields(case: Case, elements: dict[str, Any], portrait: dict[str, Any], risk_level: str | None) -> None: case.applicant_name = elements.get("applicant_name") case.respondent_name = elements.get("respondent_name") case.case_cause = elements.get("case_cause") case.risk_level = risk_level or portrait.get("risk_level") def _process_case_task(task_id: str, case_id: int) -> None: """ 后台任务:解析文件 -> 合并文本 -> 规则抽取 -> 写入 cases.elements + 版本历史 -> 画像/复杂度/风险 """ from app.db import SessionLocal db = SessionLocal() try: task = db.query(ProcessingTask).filter(ProcessingTask.id == task_id).first() if not task: return task.status = "RUNNING" task.progress = 5 task.message = "解析文件中" db.commit() case = db.query(Case).filter(Case.id == case_id).first() if not case: task.status = "FAILED" task.error = "案件不存在" task.progress = 100 db.commit() return files = db.query(CaseFile).filter(CaseFile.case_id == case_id).all() texts: list[str] = [] for f in files: try: texts.append(DocumentParser.parse_file(f.storage_path)) except Exception: continue merged = "\n\n".join([t for t in texts if t.strip()]) task.progress = 45 task.message = "要素抽取中" db.commit() elements = extractor.extract(merged) if _ollama_claims is not None: try: elements["claims"] = _ollama_claims.extract_claims(merged) elements.update(merge_dispute_template_fields(merged, elements)) except Exception: pass try: patch = _ollama_claims.extract_dispute_template_fields(merged) for k, v in (patch or {}).items(): if v is None: continue if isinstance(v, str) and not str(v).strip(): continue elements[k] = v if elements.get("tmpl_primary_cause"): elements["primary_cause_type"] = elements["tmpl_primary_cause"] except Exception: pass refresh_derived_element_fields(elements, merged) task.progress = 70 task.message = "构建画像中" db.commit() portrait = generate_portrait(elements, raw_text=merged, evidence_count=len(files)) complexity = classify_complexity(elements, evidence_count=len(files)) risk = assess_risk(elements) # 保存 elements 版本 latest_version = ( db.query(CaseElementsVersion) .filter(CaseElementsVersion.case_id == case_id) .order_by(CaseElementsVersion.version.desc()) .first() ) next_ver = (latest_version.version + 1) if latest_version else 1 db.add(CaseElementsVersion(case_id=case_id, version=next_ver, elements=elements, updated_by="system")) merged_stored = merged[:_MAX_STORED_APPLICATION_TEXT] case.elements = elements case.application_text = merged_stored case.portrait = { **portrait, "complexity_level": complexity, "risk_assessment": risk, } _update_case_index_fields(case, elements, portrait, risk.get("level")) contents: list[bytes] = [] for cf in files: try: contents.append(Path(cf.storage_path).read_bytes()) except OSError: continue if contents: case.material_fingerprint = _material_fingerprint_from_file_contents(contents) db.commit() task.status = "SUCCEEDED" task.progress = 100 task.message = "处理完成" db.commit() except Exception as e: try: task = db.query(ProcessingTask).filter(ProcessingTask.id == task_id).first() if task: task.status = "FAILED" task.progress = 100 task.error = f"{e}\n{traceback.format_exc()}" db.commit() finally: pass finally: db.close() @app.post("/api/cases/{case_id}/files", response_model=TaskResponse) async def upload_case_files( case_id: int, background: BackgroundTasks, files: list[UploadFile] = File(...), db: Session = Depends(get_db), ): case = _create_or_get_case(db, case_id) if not files: raise HTTPException(status_code=400, detail="未上传文件") upload_dir = _upload_dir() saved = 0 for f in files: content = await f.read() guessed = magic.from_buffer(content[:2048], mime=True) if content else "" if not _allowed(guessed, f.filename): raise HTTPException(status_code=400, detail=f"不支持的文件类型:{f.filename}({guessed})") suffix = Path(f.filename).suffix.lower() storage = upload_dir / f"{case_id}_{uuid4().hex}{suffix}" storage.write_bytes(content) db.add( CaseFile( case_id=case.id, filename=f.filename, content_type=guessed, storage_path=str(storage), size_bytes=len(content), ) ) saved += 1 db.commit() task_id = uuid4().hex task = ProcessingTask(id=task_id, case_id=case.id, status="PENDING", progress=0, message=f"已接收{saved}个文件") db.add(task) db.commit() background.add_task(_process_case_task, task_id, case.id) return TaskResponse(task_id=task_id, status=task.status, progress=task.progress, message=task.message) @app.get("/api/tasks/{task_id}", response_model=TaskResponse) def get_task(task_id: str, db: Session = Depends(get_db)): task = db.query(ProcessingTask).filter(ProcessingTask.id == task_id).first() if not task: raise HTTPException(status_code=404, detail="任务不存在") return TaskResponse(task_id=task.id, status=task.status, progress=task.progress, message=task.message or task.error) @app.delete("/api/cases/{case_id}", status_code=204) def delete_case(case_id: int, db: Session = Depends(get_db)): """ 删除案件及其关联数据。此前未暴露该接口,且 processing_tasks 外键指向 cases, 直接在库里删 cases 行可能被外键拒绝。 """ case = db.query(Case).filter(Case.id == case_id).first() if not case: raise HTTPException(status_code=404, detail="案件不存在") files = db.query(CaseFile).filter(CaseFile.case_id == case_id).all() for cf in files: try: Path(cf.storage_path).unlink(missing_ok=True) except OSError: pass # 必须先删子表行:MySQL 外键默认 RESTRICT,否则会报 1451(case_elements_versions 等) db.query(CaseElementsVersion).filter(CaseElementsVersion.case_id == case_id).delete(synchronize_session=False) db.query(CaseFile).filter(CaseFile.case_id == case_id).delete(synchronize_session=False) db.query(ProcessingTask).filter(ProcessingTask.case_id == case_id).delete(synchronize_session=False) db.query(Case).filter(Case.id == case_id).delete(synchronize_session=False) db.commit() return Response(status_code=204) @app.get("/api/cases/{case_id}/elements", response_model=ElementsResponse) def get_elements(case_id: int, db: Session = Depends(get_db)): case = db.query(Case).filter(Case.id == case_id).first() if not case: raise HTTPException(status_code=404, detail="案件不存在") latest_version = ( db.query(CaseElementsVersion) .filter(CaseElementsVersion.case_id == case_id) .order_by(CaseElementsVersion.version.desc()) .first() ) ver = latest_version.version if latest_version else 0 return ElementsResponse(case_id=case.id, elements=case.elements or {}, version=ver) @app.put("/api/cases/{case_id}/elements", response_model=ElementsResponse) def update_elements(case_id: int, payload: ElementsUpdateRequest, db: Session = Depends(get_db)): case = db.query(Case).filter(Case.id == case_id).first() if not case: raise HTTPException(status_code=404, detail="案件不存在") patch = payload.patch or {} current = dict(case.elements or {}) current.update(patch) # PATCH 含 elements_hierarchy 时保留用户编辑的层级,不根据扁平字段重算覆盖 _sync_case_elements_table(current, rebuild_hierarchy="elements_hierarchy" not in patch) latest_version = ( db.query(CaseElementsVersion) .filter(CaseElementsVersion.case_id == case_id) .order_by(CaseElementsVersion.version.desc()) .first() ) next_ver = (latest_version.version + 1) if latest_version else 1 db.add(CaseElementsVersion(case_id=case_id, version=next_ver, elements=current, updated_by=payload.updated_by)) # 更新画像缓存(编辑后即时刷新) files_cnt = db.query(CaseFile).filter(CaseFile.case_id == case_id).count() portrait = generate_portrait(current, raw_text="", evidence_count=files_cnt) complexity = classify_complexity(current, evidence_count=files_cnt) risk = assess_risk(current) case.elements = current case.portrait = {**portrait, "complexity_level": complexity, "risk_assessment": risk} _update_case_index_fields(case, current, portrait, risk.get("level")) db.commit() return ElementsResponse(case_id=case.id, elements=case.elements or {}, version=next_ver) @app.get("/api/cases/{case_id}/portrait", response_model=PortraitResponse) def get_portrait(case_id: int, db: Session = Depends(get_db)): case = db.query(Case).filter(Case.id == case_id).first() if not case: raise HTTPException(status_code=404, detail="案件不存在") portrait = case.portrait or {} complexity = portrait.get("complexity_level") or {} risk = portrait.get("risk_assessment") or {} return PortraitResponse(case_id=case.id, portrait=portrait, complexity_level=complexity, risk_assessment=risk) @app.post("/api/cases/{case_id}/similar", response_model=SimilarResponse) def similar_cases(case_id: int, payload: SimilarRequest, db: Session = Depends(get_db)): base = db.query(Case).filter(Case.id == case_id).first() if not base: raise HTTPException(status_code=404, detail="案件不存在") base_el = base.elements or {} base_portrait = base.portrait or {} base_tags = _case_tag_set(base_el) rows = db.query(Case).filter(Case.id != case_id).all() scored: list[tuple[float, Case]] = [] for row in rows: s = _jaccard(base_tags, _case_tag_set(row.elements or {})) scored.append((s, row)) scored.sort(key=lambda x: x[0], reverse=True) top_k = max(1, min(50, payload.top_k)) items = [ SimilarCaseDetailResponse(**_build_similar_case_detail(base_el, base_portrait, r, s)) for s, r in scored[:top_k] if s > 0 ] return SimilarResponse(case_id=case_id, items=items) @app.get("/api/evaluation/metrics", response_model=MetricsResponse) def get_eval_metrics(db: Session = Depends(get_db)): """ 评估页指标: 1) 要素抽取:Precision/Recall/F1(demo 样例;可替换为真实标注集) 2) 画像完整性/有效性:基于数据库历史案件的 portrait 字段统计 """ # 兼容以 `uvicorn app.main:app` 启动时的模块路径:此时并不存在顶层包名 backend from tools.evaluate_extractor import demo_samples, run_eval # type: ignore extraction_report = run_eval(demo_samples()) rows = db.query(Case).all() total = len(rows) with_portrait = 0 with_scores = 0 with_subscores = 0 with_keywords = 0 with_tags = 0 with_risk_assessment = 0 with_complexity = 0 for c in rows: p = c.portrait or {} if p: with_portrait += 1 scores = (p.get("scores") or {}) if isinstance(p, dict) else {} if isinstance(scores, dict) and any(scores.get(k) is not None for k in ("legal", "fact", "risk")): with_scores += 1 # subscores:三个维度任一包含 subscores 即认为完整 ok_sub = False for dim in ("legal_dimension", "fact_dimension", "risk_dimension"): d = (p.get(dim) or {}) if isinstance(p, dict) else {} subs = d.get("subscores") if isinstance(d, dict) else None if isinstance(subs, dict) and len(subs) > 0: ok_sub = True if ok_sub: with_subscores += 1 kws = p.get("keywords") if isinstance(p, dict) else None if isinstance(kws, list) and len(kws) > 0: with_keywords += 1 tags = p.get("tags") if isinstance(p, dict) else None if isinstance(tags, list) and len(tags) > 0: with_tags += 1 if isinstance(p.get("risk_assessment"), dict): with_risk_assessment += 1 if isinstance(p.get("complexity_level"), dict): with_complexity += 1 def ratio(x: int) -> float: return round((x / total) if total else 0.0, 4) portrait_report = { "total_cases": total, "has_portrait": {"count": with_portrait, "ratio": ratio(with_portrait)}, "has_scores": {"count": with_scores, "ratio": ratio(with_scores)}, "has_subscores": {"count": with_subscores, "ratio": ratio(with_subscores)}, "has_keywords": {"count": with_keywords, "ratio": ratio(with_keywords)}, "has_tags": {"count": with_tags, "ratio": ratio(with_tags)}, "has_risk_assessment": {"count": with_risk_assessment, "ratio": ratio(with_risk_assessment)}, "has_complexity_level": {"count": with_complexity, "ratio": ratio(with_complexity)}, } metrics = { "extraction_prf": extraction_report, "portrait_quality": portrait_report, "weights": {"legal": _SIM_WEIGHT_LEGAL, "fact": _SIM_WEIGHT_FACT, "risk": _SIM_WEIGHT_RISK}, } return MetricsResponse(metrics=metrics)