import os import shutil from uuid import uuid4 from typing import Dict, Any, List from fastapi import FastAPI, UploadFile, File, Form from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import HTMLResponse from fastapi.staticfiles import StaticFiles from backend.db import ( init_case_db, upsert_case_management, list_case_management, delete_case_management, store_case_record, update_case_materials, fetch_case_management, fetch_case_record, parse_case_description ) from backend.text_utils import save_uploads, list_files_by_ext from backend.services import ( extract_application_text, extract_transcript_text, process_case_text_with_evidence, build_case_summary_text ) from tools.documents_extractor import DocumentReader from application_extractor.rectify_OCR_result import RectifyClient_application from transcript_extractor.rectify_transcript import RectifyClient_transcript from law_rag.run import law_rag_run CASE_CACHE: Dict[str, Dict[str, Any]] = {} def build_file_items(file_paths: List[str], uploads_dir: str) -> List[Dict[str, str]]: items = [] for path in file_paths: rel_path = os.path.relpath(path, uploads_dir).replace("\\", "/") items.append({"name": os.path.basename(path), "url": f"/uploads/{rel_path}"}) return items def create_app() -> FastAPI: app = FastAPI() app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"] ) root_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) frontend_dir = os.path.join(root_dir, "frontend") uploads_dir = os.path.join(root_dir, "uploads") os.makedirs(uploads_dir, exist_ok=True) if os.path.exists(frontend_dir): app.mount("/assets", StaticFiles(directory=frontend_dir), name="assets") @app.get("/", response_class=HTMLResponse) def serve_index(): with open(os.path.join(frontend_dir, "index.html"), "r", encoding="utf-8") as f: return f.read() app.mount("/uploads", StaticFiles(directory=uploads_dir), name="uploads") @app.post("/api/arbitration/submit") async def arbitration_submit( case_id: str = Form(...), case_title: str = Form(""), applicationFile: List[UploadFile] = File(default_factory=list), transcriptFile: List[UploadFile] = File(default_factory=list), evidenceWage: List[UploadFile] = File(default_factory=list), evidenceTerminate: List[UploadFile] = File(default_factory=list), evidenceAttendance: List[UploadFile] = File(default_factory=list), evidenceLabor: List[UploadFile] = File(default_factory=list), evidenceBank: List[UploadFile] = File(default_factory=list), evidenceList: List[UploadFile] = File(default_factory=list), evidenceWitness: List[UploadFile] = File(default_factory=list), evidenceOther: List[UploadFile] = File(default_factory=list) ): init_case_db() case_dir = os.path.join(uploads_dir, case_id) application_dir = os.path.join(case_dir, "申请书") transcript_dir = os.path.join(case_dir, "庭审笔录") evidence_dir = os.path.join(case_dir, "证据") application_paths = await save_uploads(applicationFile, application_dir) transcript_paths = await save_uploads(transcriptFile, transcript_dir) evidence_map = { "工资单": evidenceWage, "解除劳动关系相关材料": evidenceTerminate, "考勤表": evidenceAttendance, "劳动关系证明材料": evidenceLabor, "银行流水": evidenceBank, "证据清单": evidenceList, "证人证言": evidenceWitness, "其他文字材料": evidenceOther } evidence_files = {} for category, files in evidence_map.items(): file_paths = await save_uploads(files, os.path.join(evidence_dir, category)) evidence_files[category] = build_file_items(file_paths, uploads_dir) application_text = extract_application_text(application_dir) transcript_text = extract_transcript_text(transcript_dir) CASE_CACHE[case_id] = { "case_title": case_title, "application_text": application_text, "transcript_text": transcript_text, "application_result": application_text, "transcript_result": transcript_text, "application_files": build_file_items(application_paths, uploads_dir), "transcript_files": build_file_items(transcript_paths, uploads_dir), "evidence_files": evidence_files, "evidence_dir": evidence_dir, "uploads_dir": uploads_dir } upsert_case_management(case_id, case_title or case_id, "", "草稿", "材料提交") update_case_materials( case_id, { "application_result": application_text, "transcript_result": transcript_text, "application_files": build_file_items(application_paths, uploads_dir), "transcript_files": build_file_items(transcript_paths, uploads_dir), "evidence_files": evidence_files } ) return { "case_id": case_id, "application_result": application_text, "transcript_result": transcript_text, "application_files": build_file_items(application_paths, uploads_dir), "transcript_files": build_file_items(transcript_paths, uploads_dir), "evidence_files": evidence_files } @app.post("/api/arbitration/judgement") def arbitration_judgement(payload: Dict[str, Any]): case_id = payload.get("case_id", "") application_result = payload.get("application_result", "") transcript_result = payload.get("transcript_result", "") cache = CASE_CACHE.get(case_id, {}) evidence_dir = cache.get("evidence_dir", "") result = process_case_text_with_evidence(case_id, application_result, transcript_result, evidence_dir) cache.update(result) CASE_CACHE[case_id] = cache upsert_case_management(case_id, cache.get("case_title", case_id), "", "草稿", "裁决中") return { "case_id": case_id, "final_decision": result["final_judgement"].get("final_decision", ""), "final_judgement": result["final_judgement"], "similar_cases": result["similar_cases"], "law_results": result["law_results"] } @app.get("/api/arbitration/case") def arbitration_case(case_id: str = ""): init_case_db() management = fetch_case_management(case_id) materials = parse_case_description(management.get("description", "")).get("materials", {}) record = fetch_case_record(case_id) case_dir = os.path.join(uploads_dir, case_id) application_dir = os.path.join(case_dir, "申请书") transcript_dir = os.path.join(case_dir, "庭审笔录") evidence_dir = os.path.join(case_dir, "证据") application_files = list_files_by_ext(application_dir, [".pdf", ".doc", ".docx", ".png", ".jpg", ".jpeg", ".bmp", ".gif"]) transcript_files = list_files_by_ext(transcript_dir, [".pdf", ".doc", ".docx", ".png", ".jpg", ".jpeg", ".bmp", ".gif"]) evidence_files = {} if os.path.exists(evidence_dir): for name in os.listdir(evidence_dir): full_path = os.path.join(evidence_dir, name) if os.path.isdir(full_path): files = list_files_by_ext(full_path, [".pdf", ".doc", ".docx", ".png", ".jpg", ".jpeg", ".bmp", ".gif"]) evidence_files[name] = build_file_items(files, uploads_dir) data = { "case_id": case_id, "title": management.get("title", ""), "description": management.get("description", ""), "status": management.get("status", ""), "stage": management.get("stage", ""), "application_result": materials.get("application_result", ""), "transcript_result": materials.get("transcript_result", ""), "application_files": materials.get("application_files") or build_file_items(application_files, uploads_dir), "transcript_files": materials.get("transcript_files") or build_file_items(transcript_files, uploads_dir), "evidence_files": materials.get("evidence_files") or evidence_files, "law_results": record.get("law_results", {}), "final_judgement": record.get("final_judgement", {}), "final_decision": record.get("final_judgement", {}).get("final_decision", "") } return {"case_id": case_id, "case": data} @app.post("/api/arbitration/confirm") def arbitration_confirm(payload: Dict[str, Any]): case_id = payload.get("case_id", "") final_decision = payload.get("final_decision", "") cache = CASE_CACHE.get(case_id, {}) final_judgement = cache.get("final_judgement", {}) if isinstance(final_judgement, dict): final_judgement["final_decision"] = final_decision init_case_db() store_case_record( case_id, build_case_summary_text(cache.get("case_profile", {}), cache.get("dispute_points", [])), cache.get("case_profile", {}), cache.get("dispute_points", []), cache.get("law_results", {}), cache.get("evidence_results", {}), final_judgement, cache.get("embedding", []) ) upsert_case_management(case_id, cache.get("case_title", case_id), "", "已完成", "裁决完成") return {"case_id": case_id, "status": "stored"} @app.get("/api/cases") def list_cases(): init_case_db() return {"cases": list_case_management()} @app.post("/api/cases") def create_case(payload: Dict[str, Any]): init_case_db() case_id = payload.get("case_id", "") title = payload.get("title", "") description = payload.get("description", "") status = payload.get("status", "草稿") stage = payload.get("stage", "案件管理") return {"case": upsert_case_management(case_id, title, description, status, stage)} @app.put("/api/cases") def update_case(payload: Dict[str, Any]): init_case_db() case_id = payload.get("case_id", "") title = payload.get("title", "") description = payload.get("description", "") status = payload.get("status", "草稿") stage = payload.get("stage", "案件管理") return {"case": upsert_case_management(case_id, title, description, status, stage)} @app.delete("/api/cases") def delete_case(payload: Dict[str, Any]): init_case_db() case_id = payload.get("case_id", "") delete_case_management(case_id) if case_id in CASE_CACHE: CASE_CACHE.pop(case_id, None) case_dir = os.path.join(uploads_dir, case_id) if os.path.exists(case_dir): shutil.rmtree(case_dir, ignore_errors=True) return {"status": "deleted", "case_id": case_id} @app.post("/api/tools/application") async def tool_application(files: List[UploadFile] = File(default_factory=list)): tool_id = uuid4().hex tool_dir = os.path.join(uploads_dir, "tools", "application", tool_id) file_paths = await save_uploads(files, tool_dir) reader = DocumentReader() contents = [reader.process_input(path) for path in file_paths] content = "\n\n".join(contents) client = RectifyClient_application() result = client.extract_legal_document(content) return {"result": result, "files": build_file_items(file_paths, uploads_dir)} @app.post("/api/tools/transcript") async def tool_transcript(files: List[UploadFile] = File(default_factory=list)): tool_id = uuid4().hex tool_dir = os.path.join(uploads_dir, "tools", "transcript", tool_id) file_paths = await save_uploads(files, tool_dir) reader = DocumentReader() contents = [reader.process_input(path) for path in file_paths] content = "\n\n".join(contents) client = RectifyClient_transcript() result = client.clean_text(content) return {"result": result or "", "files": build_file_items(file_paths, uploads_dir)} @app.post("/api/tools/law") def tool_law(payload: Dict[str, Any]): query = (payload.get("query") or "").strip() result = law_rag_run(query, with_score=True) if query else [] return {"result": result} return app app = create_app()