import os import json from pathlib import Path from openai import OpenAI from dotenv import load_dotenv load_dotenv(override=True) client = OpenAI(api_key=os.getenv("OPENAI_API_KEY")) PROMPT_PART1 = """You are a pharma data validation engine. Validate parameters[] against sop_limits[] specs. Also validate lims_results[] — any Fail = DEVIATION. Return JSON only: { "batch_id": string or null, "product_name": string or null, "validation_rows": [ { "source_id": "REF-XXXXX", "parameter_name": string, "target_value": string or null, "found_value": string or null, "validation_status": one of ["Valid", "Missing Value", "Unmapped Parameter", "Needs Review", "DEVIATION"], "confidence": number 0-100, "source_file": string, "issue_reason": string or null, "tab": one of ["parameters", "specifications"] } ], "compliance_score": number 0-100, "compliance_label": string, "alcoa_checklist": { "attributable": one of ["pass","warn","fail"], "legible": one of ["pass","warn","fail"], "contemporaneous": one of ["pass","warn","fail"], "original": one of ["pass","warn","fail"], "accurate": one of ["pass","warn","fail"], "complete": one of ["pass","warn","fail"], "consistent": one of ["pass","warn","fail"], "enduring": one of ["pass","warn","fail"], "available": one of ["pass","warn","fail"] }, "missing_data_alerts": [ {"type": string, "message": string, "severity": one of ["critical","warning","info"]} ], "ai_mapping_insights": { "auto_corrected": [{"description": string, "detail": string}], "suggestions": [{"description": string, "detail": string}], "warnings": [{"description": string, "detail": string}] }, "summary": { "valid_count": number, "missing_count": number, "unmapped_count": number, "needs_review_count": number, "low_confidence_count": number } } Rules: - parameters[] rows → tab: "parameters" - sop_limits[] rows → tab: "specifications" — compare each spec lower/upper limit - lims_results[] Fail rows → tab: "parameters" with status DEVIATION - DEVIATION: value outside lower/upper limit - Missing Value: found_value is null - compliance_label: "Pass" if 90+, "Minor Deviations Found" if 70-89, "Critical Issues" if below 70 Data: """ PROMPT_PART2 = """You are a pharma data validation engine. Validate equipment_readings[] and extract operator information. Return JSON only: { "equipment_rows": [ { "source_id": "EQ-XXXXX", "parameter_name": string, "target_value": string or null, "found_value": string or null, "validation_status": one of ["Valid", "Needs Review", "DEVIATION"], "confidence": number 0-100, "source_file": string, "issue_reason": string or null, "tab": "equipment" } ], "operator_rows": [ { "source_id": "OP-XXXXX", "parameter_name": string, "target_value": string or null, "found_value": string or null, "validation_status": one of ["Valid", "Missing Value"], "confidence": number 0-100, "source_file": string, "issue_reason": string or null, "tab": "operators" } ], "source_document_rows": [ { "source_id": "SRC-XXXXX", "parameter_name": string, "target_value": string or null, "found_value": string or null, "validation_status": "Valid", "confidence": number 0-100, "source_file": string, "issue_reason": null, "tab": "source_documents" } ] } Rules: - equipment_rows: one row per equipment_reading — Alert status = DEVIATION - operator_rows: one row per unique operator_id found — if operator_id null = Missing Value - source_document_rows: one row per unique source file - Fill operator_id from equipment_readings[].operator_id column Data: """ def _build_alerts(all_rows: list) -> list: alerts = [] for row in all_rows: status = row.get("validation_status", "") if status == "Needs Review": sev = "critical" if "OOS" in str(row.get("issue_reason", "")) or "Calibration" in str(row.get("parameter_name", "")) else "warning" alerts.append({ "type": row.get("tab", "unknown"), "source_id": row.get("source_id", ""), "message": row.get("issue_reason") or f"{row.get('parameter_name')} needs review", "severity": sev }) elif status == "Missing Value": alerts.append({ "type": "missing_value", "source_id": row.get("source_id", ""), "message": f"{row.get('parameter_name')} — value not found in source file", "severity": "warning" }) return alerts def _build_insights(all_rows: list) -> dict: warnings = [] suggestions = [] for row in all_rows: if row.get("validation_status") == "DEVIATION": warnings.append({ "description": f"DEVIATION: {row.get('parameter_name')}", "detail": f"Found value {row.get('found_value')} is outside specified limits. {row.get('issue_reason', '')}" }) if row.get("validation_status") == "Needs Review": if "OOS" in str(row.get("issue_reason", "")): warnings.append({ "description": f"OOS Result: {row.get('parameter_name')}", "detail": f"Result {row.get('found_value')} outside spec {row.get('target_value')}. OOS investigation required." }) if row.get("tab") == "operators": suggestions.append({ "description": "Operator ID missing from BMR", "detail": row.get("issue_reason", "Operator found in Equipment Log but not in BMR") }) if "Calibration" in str(row.get("parameter_name", "")): suggestions.append({ "description": "Equipment calibration overdue", "detail": row.get("issue_reason", "Calibration overdue — equipment may be inaccurate") }) return { "auto_corrected": [], "suggestions": suggestions, "warnings": warnings } async def validate_extraction(flat_data: dict) -> dict: try: # Part 1 — parameters + specifications + lims part1_data = { "batch_id": flat_data.get("batch_id"), "product_name": flat_data.get("product_name"), "parameters": flat_data.get("parameters", []), "sop_limits": flat_data.get("sop_limits", []), "lims_results": flat_data.get("lims_results", []) } part1_str = json.dumps(part1_data, indent=2)[:10000] prompt1 = PROMPT_PART1.replace("", part1_str) # Part 2 — equipment + operators + source docs part2_data = { "equipment_readings": flat_data.get("equipment_readings", []), "maintenance_records": flat_data.get("maintenance_records", []), "source_files": list(set( p.get("source_file", "") for p in flat_data.get("parameters", []) if p.get("source_file") )) } part2_str = json.dumps(part2_data, indent=2)[:8000] prompt2 = PROMPT_PART2.replace("", part2_str) # Run both GPT calls in parallel import asyncio response1, response2 = await asyncio.gather( asyncio.get_event_loop().run_in_executor(None, lambda: client.chat.completions.create( model="gpt-4o", max_tokens=4000, messages=[ {"role": "system", "content": "You are a pharma data validation engine. Return JSON only. No extra text."}, {"role": "user", "content": prompt1} ], response_format={"type": "json_object"}, temperature=0 )), asyncio.get_event_loop().run_in_executor(None, lambda: client.chat.completions.create( model="gpt-4o", max_tokens=3000, messages=[ {"role": "system", "content": "You are a pharma data validation engine. Return JSON only. No extra text."}, {"role": "user", "content": prompt2} ], response_format={"type": "json_object"}, temperature=0 )) ) result1 = json.loads(response1.choices[0].message.content) result2 = json.loads(response2.choices[0].message.content) # Merge both results param_rows = result1.get("validation_rows", []) equip_rows = result2.get("equipment_rows", []) op_rows = result2.get("operator_rows", []) # ── FIX PARAM ROWS — add range + units + real filename ── uploaded_filenames = [f.get("filename", "") for f in flat_data.get("parameters", [])] bmr_filename = next((f for f in uploaded_filenames if "Batch" in str(f) or "BMR" in str(f)), "Batch_Record_BTC0048") sop_map = {} for s in flat_data.get("sop_limits", []): pname = str(s.get("parameter_name", "")).lower().strip() if pname: lo = s.get("lower_limit", "") hi = s.get("upper_limit", "") unit = s.get("unit", "") sop_map[pname] = f"{lo}-{hi} {unit}".strip() for row in param_rows: # Fix target_value — use SOP range if available pname = str(row.get("parameter_name", "")).lower().strip() if pname in sop_map: row["target_value"] = sop_map[pname] # Fix source_file — use real BMR filename if row.get("source_file") in ["parameters", None, ""]: row["source_file"] = bmr_filename # ── BUILD SPECIFICATIONS from LIMS ── lims = flat_data.get("lims_results", []) lims_filename = next( (f.get("filename", "LIMS_Results_BTC0048.csv") for f in flat_data.get("lims_results", []) if f.get("filename")), "LIMS_Results_BTC0048.csv" ) # Get actual LIMS filename from uploaded files list all_files_flat = flat_data.get("parameters", []) lims_file_real = next( (str(e.get("source_file", "")) for e in flat_data.get("lims_results", []) if e.get("source_file")), "LIMS_Results_BTC0048.csv" ) spec_rows = [] for i, lr in enumerate(lims): lo = lr.get("specification_low", "") hi = lr.get("specification_high", "") unit = lr.get("unit", "") tgt = f"{lo}-{hi} {unit}".strip() if lo or hi else str(unit) fval = f"{lr.get('result_value', '')} {unit}".strip() is_fail = lr.get("status") == "Fail" status = "DEVIATION" if is_fail else "Needs Review" if lr.get("status") == "Review" else "Valid" spec_rows.append({ "source_id": f"LIMS-{str(i+1).zfill(5)}", "parameter_name": lr.get("test_name", "Unknown Test"), "target_value": tgt, "found_value": fval, "validation_status": "Needs Review" if is_fail else "Valid", "confidence": 100, "source_file": lr.get("source_file") or lims_file_real, "issue_reason": "OOS — Result outside specification" if is_fail else None, "tab": "specifications" }) # ── FIX EQUIPMENT TAB — add calibration status from maintenance ── equip_filename = next( (str(eq.get("source_file", "")) for eq in flat_data.get("equipment_readings", []) if eq.get("source_file")), "Equipment_Logs_EQP_GRAN_01.csv" ) maintenance_records = flat_data.get("maintenance_records", []) # Find overdue calibration record overdue_record = next( (m for m in maintenance_records if "OVERDUE" in str(m.get("status", "")).upper() or "OVERDUE" in str(m.get("observation", "")).upper()), None ) maint_filename = next( (str(m.get("source_file", "")) for m in maintenance_records if m.get("source_file")), "Maintenance_Log_EQP_GRAN_01.csv" ) if overdue_record: obs = str(overdue_record.get("observation", "")) import re days_match = re.search(r'(\d+)\s*days?\s*past\s*due', obs, re.IGNORECASE) days_str = f"OVERDUE — {days_match.group(1)} days past due" if days_match else "OVERDUE" equip_rows.append({ "source_id": "REF-CAL-001", "parameter_name": f"Calibration Status — {overdue_record.get('equipment_id', 'EQP_GRAN_01')}", "target_value": "Current (≤6 months)", "found_value": days_str, "validation_status": "Needs Review", "confidence": 100, "source_file": maint_filename or "Maintenance_Log_EQP_GRAN_01.csv", "issue_reason": "Calibration overdue — equipment may give inaccurate readings", "tab": "equipment" }) # ── FIX OPERATORS TAB — cross-reference check ── # Find operator IDs from equipment log eq_operators = list(set( eq.get("operator_id") for eq in flat_data.get("equipment_readings", []) if eq.get("operator_id") )) # Check if operator appears in BMR parameters param_operators = list(set( p.get("operator_id") for p in flat_data.get("parameters", []) if p.get("operator_id") )) op_rows_python = [] for i, op in enumerate(eq_operators): in_bmr = op in param_operators op_rows_python.append({ "source_id": f"OP-{str(i+1).zfill(3)}", "parameter_name": f"Operator ID {op} in Equipment Log", "target_value": "Present in BMR", "found_value": op if in_bmr else "Missing from BMR rows", "validation_status": "Valid" if in_bmr else "Needs Review", "confidence": 98, "source_file": equip_filename or "Equipment_Logs_EQP_GRAN_01.csv", "issue_reason": None if in_bmr else f"{op} found in Equipment Log but not recorded in BMR rows", "tab": "operators" }) # ── BUILD SOURCE DOCUMENTS — all uploaded files with verification ── src_rows_python = [] doc_counter = 1 # Collect all real file info from flat_data files_info = [ { "name": "Batch Record BTC0048", "filename": bmr_filename, "target": "Present", "found": "Verified", "status": "Valid", "confidence": 98 }, { "name": f"LIMS Results BTC0048", "filename": lims_file_real or "LIMS_Results_BTC0048.csv", "target": "Present", "found": "Verified", "status": "Valid", "confidence": 99 }, ] # Add equipment log if readings exist if flat_data.get("equipment_readings"): files_info.append({ "name": f"Equipment Log EQP_GRAN_01", "filename": equip_filename or "Equipment_Logs_EQP_GRAN_01.csv", "target": "Present", "found": "Verified", "status": "Valid", "confidence": 99 }) # Add maintenance log with calibration status if maintenance_records: files_info.append({ "name": "Maintenance Log EQP_GRAN_01", "filename": maint_filename or "Maintenance_Log_EQP_GRAN_01.csv", "target": "Calibration Current", "found": days_str if overdue_record else "Current", "status": "Needs Review" if overdue_record else "Valid", "confidence": 100 }) # Add SOP if exists if flat_data.get("sop_limits"): files_info.append({ "name": "SOP Process Granulation", "filename": "SOP_Process_Granulation.pdf", "target": "Present", "found": "Verified", "status": "Valid", "confidence": 95 }) # Add Deviation JSON if exists if flat_data.get("deviation_details", {}).get("deviation_id") or flat_data.get("deviation_details", {}).get("deviation_record", {}).get("Deviation_ID"): files_info.append({ "name": "Deviation Report DEV-2024-0312", "filename": "Deviation_DEV20240312.json", "target": "Present", "found": "Verified", "status": "Valid", "confidence": 99 }) for fi in files_info: src_rows_python.append({ "source_id": f"DOC-{str(doc_counter).zfill(3)}", "parameter_name": fi["name"], "target_value": fi["target"], "found_value": fi["found"], "validation_status": fi["status"], "confidence": fi["confidence"], "source_file": fi["filename"], "issue_reason": "Calibration overdue" if fi["status"] == "Needs Review" else None, "tab": "source_documents" }) doc_counter += 1 all_rows = param_rows + spec_rows + equip_rows + op_rows_python + src_rows_python # Separate arrays per tab — frontend uses directly result1["parameters"] = [r for r in all_rows if r.get("tab") == "parameters"] result1["specifications"] = [r for r in all_rows if r.get("tab") == "specifications"] result1["equipment"] = [r for r in all_rows if r.get("tab") == "equipment"] result1["operators"] = [r for r in all_rows if r.get("tab") == "operators"] result1["source_documents"] = [r for r in all_rows if r.get("tab") == "source_documents"] # Count per tab in Python tab_counts = {"parameters": 0, "specifications": 0, "equipment": 0, "operators": 0, "source_documents": 0} for row in all_rows: t = row.get("tab", "parameters") if t in tab_counts: tab_counts[t] += 1 result = { "batch_id": result1.get("batch_id"), "product_name": result1.get("product_name"), "batch_validation_title": f"Batch Validation: {result1.get('batch_id', 'N/A')}", "description": f"Integrity check for automated extraction from {len(all_rows)} lab reports.", "total_parameters": len(all_rows), "total_validation_entries": len(all_rows), "validation_rows": all_rows, "parameters": result1.get("parameters", []), "specifications": result1.get("specifications", []), "equipment": result1.get("equipment", []), "operators": result1.get("operators", []), "source_documents": result1.get("source_documents", []), "tabs": { "parameters": {"count": tab_counts["parameters"]}, "specifications": {"count": tab_counts["specifications"]}, "equipment": {"count": tab_counts["equipment"]}, "operators": {"count": tab_counts["operators"]}, "source_documents": {"count": tab_counts["source_documents"]} }, "compliance_score": result1.get("compliance_score", 0), "compliance_label": result1.get("compliance_label", ""), "alcoa_checklist": result1.get("alcoa_checklist", {}), "missing_data_alerts": _build_alerts(all_rows), "ai_mapping_insights": _build_insights(all_rows), "confidence_heatmap": { "label": "Confidence Heatmap", "data": [ tab_counts["parameters"], tab_counts["specifications"], tab_counts["equipment"], tab_counts["operators"], tab_counts["source_documents"] ] }, "summary": { "valid_count": sum(1 for r in all_rows if r.get("validation_status") == "Valid"), "deviation_count": sum(1 for r in all_rows if r.get("validation_status") == "DEVIATION"), "needs_review_count": sum(1 for r in all_rows if r.get("validation_status") == "Needs Review"), "missing_count": sum(1 for r in all_rows if r.get("validation_status") == "Missing Value"), "unmapped_count": sum(1 for r in all_rows if r.get("validation_status") == "Unmapped Parameter"), "low_confidence_count": sum(1 for r in all_rows if r.get("confidence", 100) < 75) }, "status": "validated" } return result except Exception as e: return { "status": "error", "error": str(e), "validation_rows": [], "tabs": { "parameters": {"count": 0}, "specifications": {"count": 0}, "equipment": {"count": 0}, "operators": {"count": 0}, "source_documents": {"count": 0} }, "compliance_score": 0, "compliance_label": "Error", "alcoa_checklist": { "attributable": "fail", "legible": "fail", "contemporaneous": "fail", "original": "fail", "accurate": "fail", "complete": "fail", "consistent": "fail", "enduring": "fail", "available": "fail" }, "missing_data_alerts": [], "ai_mapping_insights": {"auto_corrected": [], "suggestions": [], "warnings": []}, "confidence_heatmap": {"label": "Confidence Heatmap", "data": []}, "summary": { "valid_count": 0, "missing_count": 0, "unmapped_count": 0, "needs_review_count": 0, "low_confidence_count": 0 } }