from __future__ import annotations import logging import os import uuid logger = logging.getLogger("dscons.document_processing.pipeline") class DocumentPipelineMixin: """End-to-end background document processing pipeline orchestration.""" async def run_pipeline( self, doc_id: str, file_path_str: str, filename: str, file_format: str ) -> None: """Run text extraction, AI analysis, Deterministic rules, Cross-Agent Audit and Staging Quarantine.""" logger.info("Starting AI pipeline for doc_id=%s, file=%s", doc_id, filename) text = self.extract_text(file_path_str, file_format) if not text: text = f"Không có nội dung văn bản nào được trích xuất từ file {filename}." # 1. Agent Thảo trích xuất dữ liệu thô ai_result = await self.analyze_with_ai(text, filename) # 1.1. Tự động kiểm tra phát hiện UNC từ nội dung văn bản và định tuyến vào Hot Watcher from app.modules.banking.application.unc_classifier_and_router import ( UncClassifierAndRouter, ) f_bytes = None if os.path.exists(file_path_str): try: with open(file_path_str, "rb") as f: f_bytes = f.read() except Exception: pass is_unc_text, bank_from_text, _ = UncClassifierAndRouter.detect_unc_and_bank( filename=filename, text=text, content=f_bytes, ) if ( is_unc_text or ai_result.get("document_type") == "UNC" or ai_result.get("document_group") == "financial" ): bank_code = bank_from_text or "OTHER" ai_result["document_type"] = "UNC" ai_result["document_group"] = "financial" try: if f_bytes: target_folder, target_file = UncClassifierAndRouter.route_to_hot_folder( file_content=f_bytes, filename=filename, bank_code=bank_code, ) router_svc = UncClassifierAndRouter(self._erp_client) ingest_res = router_svc.ingest_statement_file( file_path=target_file, bank_code=bank_code, file_content=f_bytes, ) logger.info( "Pipeline auto-routed UNC doc_id=%s to %s (inserted %d txs)", doc_id, target_file, ingest_res.get("inserted", 0), ) except Exception as unc_err: logger.warning("Pipeline fallback UNC routing error: %s", unc_err) # 2. Kiểm tra quy tắc logic cứng (Deterministic Rules) deterministic_flags = self.validate_deterministic_rules(ai_result, text) # 3. Agent Thủy kiểm tra chéo độc lập (Cross-Agent Audit) cross_audit_res = await self.run_cross_agent_audit(text, ai_result, filename) # 4. Hợp nhất cảnh báo và tính toán điểm tin cậy tổng thể all_flags = deterministic_flags + (cross_audit_res.get("flags") or []) raw_score = int(cross_audit_res.get("confidence_score") or 85) error_count = len([f for f in all_flags if f.get("severity") == "error"]) final_confidence_score = max(0, min(100, raw_score - (error_count * 20))) # 5. Phân luồng cách ly (Quarantine Routing) # Nếu điểm < 80 hoặc có lỗi nghiêm trọng -> Tự động gắn cờ cách ly (Quarantined) if final_confidence_score < 80 or error_count > 0: quarantine_status = "quarantined" status = "quarantined" verification_status = "quarantined" else: quarantine_status = "clear" status = "pending_review" verification_status = "pending_review" # Calculate audit readiness score based on signature status sig_status = ai_result.get("signature_status") or "draft" audit_score = ( 100 if sig_status == "fully_executed" else (75 if sig_status == "signed_unsigned_seal" else 50) ) # Fetch existing document metadata to merge gracefully existing_doc = self._erp_client.get_document(doc_id) or {} existing_code = existing_doc.get("document_code") final_doc_code = ai_result.get("document_code") or ( existing_code if existing_code and not existing_code.startswith("TEMP-") else f"TEMP-{uuid.uuid4().hex[:6].upper()}" ) # Sanitize document_title and summary raw_title = ai_result.get("document_title") clean_title = ( self.sanitize_document_title(raw_title, filename) if raw_title else ( existing_doc.get("document_title") or self.sanitize_document_title(filename, filename) ) ) raw_summary = ( ai_result.get("summary_content") or existing_doc.get("summary_content") or "" ) clean_summary = ( self.sanitize_document_title(raw_summary, "") if raw_summary else raw_summary ) issuer = ( ai_result.get("issuer_name") or existing_doc.get("issuer_name") or "Trung tâm Dịch vụ sự nghiệp công xã Kiến Minh" ) partner = ( ai_result.get("partner_name") or existing_doc.get("partner_name") or issuer ) issue_date = ai_result.get("issue_date") or existing_doc.get("issue_date") # Map fields to match database columns update_payload = { "document_code": final_doc_code, "document_title": clean_title, "document_group": ai_result.get("document_group") or existing_doc.get("document_group") or "dispatch", "document_type": ai_result.get("document_type") or existing_doc.get("document_type") or "CV", "category": "incoming", "signature_status": sig_status, "document_direction": ai_result.get("document_direction") or "incoming", "project_stage": ai_result.get("project_stage") or "stage_01_initiation", "partner_name": partner, "partner_tax_code": ai_result.get("partner_tax_code") or existing_doc.get("partner_tax_code"), "issuer_name": issuer, "issuing_authority": issuer or "Không rõ", "signer_name": ai_result.get("signer_name") or existing_doc.get("signer_name"), "issue_date": issue_date, "sign_date": issue_date, "summary_content": clean_summary, "ai_analysis_summary": clean_summary, "ai_risk_assessment": ai_result.get("ai_risk_assessment") or existing_doc.get("ai_risk_assessment"), "ai_legal_references": ai_result.get("ai_legal_references") or existing_doc.get("ai_legal_references") or [], "audit_readiness_score": audit_score, "ocr_confidence_score": final_confidence_score, "ocr_validation_flags": all_flags, "cross_audit_summary": cross_audit_res.get("cross_audit_summary") or "Đã đối soát chéo hoàn tất.", "quarantine_status": quarantine_status, "status": status, "verification_status": verification_status, } try: self._erp_client.update_document(doc_id, update_payload) logger.info( "Successfully updated document doc_id=%s with Cross-Agent Audit", doc_id ) # 6. Tự động khởi tạo hoặc liên kết Dự Án nếu đây là Thông Báo Mời Thầu / Quyết Định Dự Án mới doc_type = update_payload.get("document_type") or "CV" if not existing_doc.get("project_id") and ( doc_type in ["TB", "QD", "HD"] or "mời thầu" in clean_title.lower() or "dự án" in clean_title.lower() ): logger.info( "Auto-initiating tender project from document doc_id=%s (%s)", doc_id, clean_title, ) proj_res = self._erp_client.create_project_from_document(doc_id) logger.info("Project initiation result: %s", proj_res.get("message")) except Exception as e: logger.error( "Failed to save AI extraction or initiate project for doc_id=%s: %s", doc_id, e, )