from __future__ import annotations import json import logging from datetime import date from typing import Any logger = logging.getLogger(__name__) class ErpDocumentVersionMixin: def create_document_version( self, parent_doc_id: str, payload: dict[str, Any] ) -> dict[str, Any]: """Tạo phiên bản mới (sửa đổi / bản đã ký đóng dấu) cho một văn bản đã tồn tại.""" parent_doc = self.get_document(parent_doc_id) if not parent_doc: raise ValueError(f"Không tìm thấy văn bản gốc có ID {parent_doc_id}") root_id = parent_doc.get("root_document_id") or parent_doc["id"] next_index = (parent_doc.get("version_index") or 1) + 1 # Determine version string (e.g. v1.1, v1.2, or v2.0 if fully_executed) signature_status = payload.get("signature_status") or "draft" if signature_status == "fully_executed": version_number = f"v{next_index}.0_Final" else: version_number = f"v1.{next_index - 1}" with self.get_connection() as conn: with conn.cursor() as cur: # Đánh dấu các bản cũ trong cùng chùm thành is_active_version = false cur.execute( """ UPDATE erp_documents SET is_active_version = FALSE, updated_at = NOW() WHERE root_document_id = %s OR id = %s; """, (root_id, root_id), ) # Insert phiên bản mới new_payload = { "company_id": parent_doc.get("company_id"), "project_id": payload.get("project_id") or parent_doc.get("project_id"), "parent_document_id": parent_doc_id, "root_document_id": root_id, "document_code": payload.get("document_code") or parent_doc.get("document_code"), "document_title": payload.get("document_title") or parent_doc.get("document_title"), "document_group": payload.get("document_group") or parent_doc.get("document_group"), "document_type": payload.get("document_type") or parent_doc.get("document_type"), "category": payload.get("category") or parent_doc.get("category", "incoming"), "issue_date": payload.get("issue_date") or parent_doc.get("issue_date") or date.today(), "issuer_name": payload.get("issuer_name") or parent_doc.get("issuer_name"), "signer_name": payload.get("signer_name") or parent_doc.get("signer_name"), "file_path": payload.get("file_path") or payload.get("file_url"), "file_url": payload.get("file_url"), "file_format": payload.get("file_format", "pdf"), "file_size_bytes": payload.get("file_size_bytes", 0), "file_hash": payload.get("file_hash") or parent_doc.get("file_hash"), "verification_status": payload.get( "verification_status", "verified" ), "status": payload.get("status", "verified"), "ai_analysis_summary": payload.get("ai_analysis_summary") or payload.get("summary_content") or parent_doc.get("ai_analysis_summary"), "summary_content": payload.get("summary_content") or parent_doc.get("summary_content"), "ai_risk_assessment": payload.get("ai_risk_assessment") or parent_doc.get("ai_risk_assessment"), "ai_legal_references": json.dumps( payload.get("ai_legal_references") or [] ) if isinstance(payload.get("ai_legal_references"), (list, dict)) else (payload.get("ai_legal_references") or "[]"), "is_gold_reference": payload.get("is_gold_reference", False), "version_number": version_number, "version_index": next_index, "is_active_version": True, "signature_status": signature_status, "document_direction": payload.get("document_direction") or parent_doc.get("document_direction", "incoming"), "partner_tax_code": payload.get("partner_tax_code") or parent_doc.get("partner_tax_code"), "partner_name": payload.get("partner_name") or parent_doc.get("partner_name"), "project_stage": payload.get("project_stage") or parent_doc.get("project_stage", "stage_01_initiation"), "change_summary": payload.get("change_summary") or f"Bản cập nhật {version_number} thay thế cho {parent_doc.get('version_number', 'v1.0')}", "diff_analysis": json.dumps(payload.get("diff_analysis") or {}), "audit_readiness_score": payload.get("audit_readiness_score", 100), } sql = """ INSERT INTO erp_documents ( company_id, project_id, parent_document_id, root_document_id, document_code, document_title, document_group, document_type, category, issue_date, issuer_name, signer_name, file_path, file_url, file_format, file_size_bytes, file_hash, verification_status, status, ai_analysis_summary, summary_content, ai_risk_assessment, ai_legal_references, is_gold_reference, version_number, version_index, is_active_version, signature_status, document_direction, partner_tax_code, partner_name, project_stage, change_summary, diff_analysis, audit_readiness_score ) VALUES ( %(company_id)s, %(project_id)s, %(parent_document_id)s, %(root_document_id)s, %(document_code)s, %(document_title)s, %(document_group)s, %(document_type)s, %(category)s, %(issue_date)s, %(issuer_name)s, %(signer_name)s, %(file_path)s, %(file_url)s, %(file_format)s, %(file_size_bytes)s, %(file_hash)s, %(verification_status)s, %(status)s, %(ai_analysis_summary)s, %(summary_content)s, %(ai_risk_assessment)s, %(ai_legal_references)s::jsonb, %(is_gold_reference)s, %(version_number)s, %(version_index)s, %(is_active_version)s, %(signature_status)s, %(document_direction)s, %(partner_tax_code)s, %(partner_name)s, %(project_stage)s, %(change_summary)s, %(diff_analysis)s::jsonb, %(audit_readiness_score)s ) RETURNING *; """ cur.execute(sql, new_payload) row = cur.fetchone() conn.commit() return row def list_document_versions(self, doc_id: str) -> list[dict[str, Any]]: """Lấy toàn bộ lịch sử các phiên bản trong chùm văn bản.""" doc = self.get_document(doc_id) if not doc: return [] root_id = doc.get("root_document_id") or doc["id"] sql = """ SELECT d.*, p.project_name FROM erp_documents d LEFT JOIN projects p ON d.project_id = p.id WHERE d.root_document_id = %s OR d.id = %s ORDER BY d.version_index ASC, d.created_at ASC; """ with self.get_connection() as conn, conn.cursor() as cur: cur.execute(sql, (root_id, root_id)) return cur.fetchall() def update_document(self, doc_id: str, updates: dict[str, Any]) -> dict[str, Any]: """Cập nhật thông tin / trạng thái văn bản.""" if not updates: doc = self.get_document(doc_id) if not doc: raise ValueError(f"Document {doc_id} not found") return doc set_clauses = [] params: dict[str, Any] = {} for key, val in updates.items(): if key in ( "ai_legal_references", "diff_analysis", "audit_check_notes", "ocr_validation_flags", ) and isinstance(val, (list, dict)): set_clauses.append(f"{key} = %({key})s::jsonb") params[key] = json.dumps(val) else: set_clauses.append(f"{key} = %({key})s") params[key] = val params["doc_id"] = doc_id sql = f""" UPDATE erp_documents SET {", ".join(set_clauses)}, updated_at = NOW() WHERE id = %(doc_id)s RETURNING *; """ with self.get_connection() as conn, conn.cursor() as cur: cur.execute(sql, params) conn.commit() row = cur.fetchone() if not row: raise ValueError(f"Document {doc_id} not found") return row