Spaces:
Runtime error
Runtime error
| from fastapi import APIRouter, UploadFile, File, HTTPException, Depends | |
| from fastapi.responses import StreamingResponse | |
| from fastapi.concurrency import run_in_threadpool | |
| from core.config import settings | |
| from ml_pipeline.engine import IntelligentDocumentProcessor | |
| from api.dependencies import get_db | |
| from database.repository import DocumentRepository | |
| import aiofiles | |
| import os | |
| import uuid | |
| import io | |
| import pandas as pd | |
| router = APIRouter() | |
| # Load the ML engine directly into the API memory (Bypassing Celery/Redis) | |
| print("Loading ML Models directly into FastAPI...") | |
| ocr_engine = IntelligentDocumentProcessor() | |
| def merge_page_data(existing_data: dict, new_page_data: dict) -> dict: | |
| merged = existing_data.copy() | |
| for key, val in new_page_data.items(): | |
| if isinstance(val, dict): | |
| if key not in merged or not isinstance(merged[key], dict): | |
| merged[key] = val.copy() | |
| else: | |
| for sub_key, sub_val in val.items(): | |
| if isinstance(sub_val, list): | |
| if sub_key not in merged[key] or not isinstance(merged[key][sub_key], list): | |
| merged[key][sub_key] = sub_val.copy() | |
| else: | |
| merged[key][sub_key].extend(sub_val) | |
| else: | |
| merged[key][sub_key] = sub_val | |
| elif isinstance(val, list): | |
| if key not in merged or not isinstance(merged[key], list): | |
| merged[key] = val.copy() | |
| else: | |
| merged[key].extend(val) | |
| else: | |
| merged[key] = val | |
| return merged | |
| def get_val(d: dict, keys: list, default="") -> str: | |
| if not isinstance(d, dict): | |
| return default | |
| for k in keys: | |
| if k in d: | |
| return str(d[k]) if d[k] is not None else default | |
| if k.lower() in d: | |
| return str(d[k.lower()]) if d[k.lower()] is not None else default | |
| if k.title() in d: | |
| return str(d[k.title()]) if d[k.title()] is not None else default | |
| if k.upper() in d: | |
| return str(d[k.upper()]) if d[k.upper()] is not None else default | |
| spaced_k = k.replace("_", " ") | |
| if spaced_k in d: | |
| return str(d[spaced_k]) if d[spaced_k] is not None else default | |
| if spaced_k.lower() in d: | |
| return str(d[spaced_k.lower()]) if d[spaced_k.lower()] is not None else default | |
| if spaced_k.title() in d: | |
| return str(d[spaced_k.title()]) if d[spaced_k.title()] is not None else default | |
| if spaced_k.upper() in d: | |
| return str(d[spaced_k.upper()]) if d[spaced_k.upper()] is not None else default | |
| return default | |
| async def upload_and_process_document( | |
| file: UploadFile = File(None), | |
| filename: str = None, | |
| page: int = 0, | |
| task_id: str = None, | |
| db = Depends(get_db) | |
| ): | |
| """ | |
| Accepts an industrial scan page-by-page. For initial page (page=0), accepts a file upload. | |
| For subsequent pages, accepts filename to reuse the saved document path. | |
| Integrates results incrementally into MongoDB. | |
| """ | |
| if not file and not filename: | |
| raise HTTPException(status_code=400, detail="Either file or filename must be provided.") | |
| if file: | |
| allowed_types = ["image/jpeg", "image/png", "application/pdf"] | |
| file_extension = file.filename.split(".")[-1].lower() | |
| # Verify content type or extension matches allowed files (more robust for browser variations) | |
| is_allowed = ( | |
| file.content_type in allowed_types or | |
| file_extension in ["pdf", "jpg", "jpeg", "png"] | |
| ) | |
| if not is_allowed: | |
| raise HTTPException(status_code=400, detail="Unsupported file type. Use JPG, PNG, or PDF.") | |
| unique_filename = f"{uuid.uuid4().hex}.{file_extension}" | |
| file_path = os.path.join(settings.UPLOAD_DIR, unique_filename) | |
| # Save file | |
| async with aiofiles.open(file_path, 'wb') as out_file: | |
| content = await file.read() | |
| await out_file.write(content) | |
| filename_used = unique_filename | |
| else: | |
| file_path = os.path.join(settings.UPLOAD_DIR, filename) | |
| if not os.path.exists(file_path): | |
| raise HTTPException(status_code=404, detail=f"Saved file {filename} not found on server.") | |
| filename_used = filename | |
| try: | |
| # Process the specific page | |
| result_payload = await run_in_threadpool(ocr_engine.process_document, file_path, page_num=page) | |
| # Enhanced debugging log | |
| print(f"DEBUG - Extracted page results payload: {result_payload}") | |
| if isinstance(result_payload, dict) and "error" in result_payload: | |
| # Dynamically set to 503 if Google is busy, else 500 | |
| error_msg = result_payload['error'] | |
| status = 503 if "503" in error_msg or "UNAVAILABLE" in error_msg else 500 | |
| raise HTTPException( | |
| status_code=status, | |
| detail=f"AI Extraction Pipeline Error: {error_msg}" | |
| ) | |
| extracted_data = result_payload["extracted_data"] | |
| total_pages = result_payload["total_pages"] | |
| # Save to database (MongoDB) and manage page-by-page list if task_id is provided | |
| repo = DocumentRepository(db) | |
| if task_id: | |
| existing_doc = await repo.get_document(task_id) | |
| if existing_doc: | |
| existing_extracted = existing_doc.get("extracted_data", {}) or {} | |
| pages = existing_extracted.get("pages", []) or [] | |
| while len(pages) <= page: | |
| pages.append(None) | |
| pages[page] = extracted_data | |
| accumulated_data = { | |
| "pages": pages, | |
| "total_pages": total_pages, | |
| "filename": filename_used | |
| } | |
| await repo.save_document(task_id, accumulated_data) | |
| else: | |
| accumulated_data = { | |
| "pages": [extracted_data], | |
| "total_pages": total_pages, | |
| "filename": filename_used | |
| } | |
| await repo.save_document(task_id, accumulated_data) | |
| else: | |
| task_id = uuid.uuid4().hex | |
| accumulated_data = { | |
| "pages": [extracted_data], | |
| "total_pages": total_pages, | |
| "filename": filename_used | |
| } | |
| await repo.save_document(task_id, accumulated_data) | |
| return { | |
| "message": "Page processed successfully", | |
| "filename": filename_used, | |
| "task_id": task_id, | |
| "data": accumulated_data, | |
| "current_page": page, | |
| "total_pages": total_pages, | |
| "has_next_page": page < total_pages - 1 | |
| } | |
| except HTTPException as he: | |
| raise he | |
| except Exception as e: | |
| raise HTTPException(status_code=500, detail=f"Processing failed inside route: {str(e)}") | |
| async def get_all_processed_documents(db = Depends(get_db)): | |
| """ | |
| Retrieves all processed document records from the database. | |
| """ | |
| try: | |
| repo = DocumentRepository(db) | |
| records = await repo.get_all_documents() | |
| return records | |
| except Exception as e: | |
| raise HTTPException(status_code=500, detail=f"Failed to retrieve records: {str(e)}") | |
| async def export_all_data_to_excel(db = Depends(get_db)): | |
| """ | |
| Aggregates all processed document records and converts them to a multi-sheet Excel file. | |
| Worksheets: | |
| 1. Product Specifications | |
| 2. Moulding & Consumption | |
| 3. QA & Inspection Checks | |
| 4. Consumable Materials | |
| 5. Verification Signatures | |
| """ | |
| try: | |
| repo = DocumentRepository(db) | |
| documents = await repo.get_all_documents() | |
| sheet1_rows = [] | |
| sheet2_rows = [] | |
| sheet3_rows = [] | |
| sheet4_rows = [] | |
| sheet5_rows = [] | |
| def normalize_document(doc) -> list: | |
| task_id = doc.get("task_id", "N/A") | |
| data = doc.get("extracted_data", {}) or {} | |
| if not data: | |
| return [] | |
| pages = [] | |
| # 1. Structured/New multi-page format | |
| if "pages" in data: | |
| raw_pages = data.get("pages", []) or [] | |
| for idx, page_data in enumerate(raw_pages): | |
| if page_data is None: | |
| continue | |
| pages.append({ | |
| "task_id": task_id, | |
| "page_no": f"Page {idx+1}", | |
| "metadata": page_data.get("document_metadata", {}) or {}, | |
| "product_details": page_data.get("product_details", {}) or {}, | |
| "moulding_details": page_data.get("moulding_details", {}) or {}, | |
| "inspection_parameters": page_data.get("inspection_parameters", {}) or {}, | |
| "refractory_and_sand": page_data.get("refractory_sleeve_and_sand_consumption", {}) or {}, | |
| "materials_table": page_data.get("materials_table", []) or [], | |
| "signatures": page_data.get("signatures", {}) or {}, | |
| "qa_parameters": page_data.get("qa_parameters", []) or [] | |
| }) | |
| # 2. Legacy Queue Pages format (Original / 6-Page schemas) | |
| elif "queue_pages" in data: | |
| raw_pages = data.get("queue_pages", []) or [] | |
| for idx, page in enumerate(raw_pages): | |
| prod = page.get("production_plan", {}) or {} | |
| qa = page.get("qa_parameters", {}) or {} | |
| pour = page.get("pouring_details", {}) or {} | |
| pages.append({ | |
| "task_id": task_id, | |
| "page_no": page.get("page_number") or f"Page {idx+1}", | |
| "metadata": { | |
| "heat_no": prod.get("heat_no", ""), | |
| "planning_date": prod.get("planning_date", ""), | |
| "pouring_date": prod.get("pouring_date", ""), | |
| }, | |
| "product_details": { | |
| "customer": prod.get("customer", ""), | |
| "grade": prod.get("grade", ""), | |
| "casting_weight": prod.get("casting_weight", ""), | |
| "liquid_weight": pour.get("pouring_weight", ""), | |
| }, | |
| "moulding_details": { | |
| "top": {"moulding_time": pour.get("pouring_time", "")} | |
| }, | |
| "inspection_parameters": { | |
| "mould_checking": qa.get("hardness_mould", ""), | |
| "sleeve_size_oven": qa.get("hardness_core", "") | |
| }, | |
| "refractory_and_sand": {}, | |
| "materials_table": [], | |
| "signatures": {}, | |
| "qa_parameters": [f"Mould Hardness: {qa.get('hardness_mould', '')}", f"Core Hardness: {qa.get('hardness_core', '')}"] | |
| }) | |
| # 3. Dynamic schema / Single page formats | |
| else: | |
| metadata = data.get("document_metadata", {}) or {} | |
| prod = data.get("product_details", {}) or {} | |
| pour = data.get("pouring_details", {}) or {} | |
| inspect = data.get("inspection_parameters", {}) or {} | |
| signatures = data.get("signatures", {}) or {} | |
| # Consumables / Sleeves / Batch Summary | |
| materials = [] | |
| if "materials_table" in data: | |
| raw_mats = data.get("materials_table", []) or [] | |
| for row in raw_mats: | |
| materials.append({ | |
| "sle_code": row.get("sle_code", ""), | |
| "sle_name": row.get("sle_name", ""), | |
| "slv_qty": row.get("slv_qty", ""), | |
| "actual_qty": row.get("actual_qty", "") | |
| }) | |
| elif "tables" in data and "batch_summary" in data.get("tables", {}): | |
| for row in data.get("tables", {}).get("batch_summary", []): | |
| materials.append({ | |
| "sle_code": row.get("material_code", ""), | |
| "sle_name": row.get("material_description", ""), | |
| "slv_qty": row.get("t_qty", ""), | |
| "actual_qty": row.get("batch_no", "") | |
| }) | |
| elif "batch_summary" in data: | |
| for row in data.get("batch_summary", []): | |
| materials.append({ | |
| "sle_code": row.get("material_code", ""), | |
| "sle_name": row.get("material_description", ""), | |
| "slv_qty": row.get("t_qty", ""), | |
| "actual_qty": row.get("batch_no", "") | |
| }) | |
| pages.append({ | |
| "task_id": task_id, | |
| "page_no": "Page 1", | |
| "metadata": metadata, | |
| "product_details": prod, | |
| "moulding_details": data.get("moulding_details", {}) or {}, | |
| "inspection_parameters": inspect, | |
| "refractory_and_sand": data.get("refractory_sleeve_and_sand_consumption", {}) or {}, | |
| "materials_table": materials, | |
| "signatures": signatures, | |
| "qa_parameters": data.get("qa_parameters", []) or [] | |
| }) | |
| return pages | |
| for doc in documents: | |
| pages = normalize_document(doc) | |
| for page in pages: | |
| # --- SHEET 1: Product Specifications --- | |
| m = page["metadata"] | |
| p = page["product_details"] | |
| sheet1_rows.append({ | |
| "Task ID": page["task_id"], | |
| "Page No": page["page_no"], | |
| "Form ID": get_val(m, ["form_id", "form id"]), | |
| "Planning Date": get_val(m, ["planning_date", "planning date", "date"]), | |
| "Pouring Date": get_val(m, ["pouring_date", "pouring date"]), | |
| "Heat No": get_val(m, ["heat_no", "heat no"]), | |
| "Customer": get_val(p, ["customer"]), | |
| "Description": get_val(p, ["description"]), | |
| "Grade": get_val(p, ["grade"]), | |
| "Casting Weight (kg)": get_val(p, ["casting_weight", "casting weight"]), | |
| "Liquid Weight (kg)": get_val(p, ["liquid_weight", "liquid weight"]), | |
| "Qty": get_val(p, ["qty", "quantity"]), | |
| "Sample / Bulk": get_val(p, ["sample_bulk", "sample/bulk"]), | |
| "Finish Type": get_val(p, ["finish_type", "finish type"]), | |
| "Pattern Code": get_val(p, ["pattern_code", "pattern code"]), | |
| "Pattern Serial No": get_val(p, ["pattern_serial_no", "pattern serial no"]), | |
| "Pattern Type": get_val(p, ["pattern_type", "pattern type"]), | |
| "Drawing Number": get_val(p, ["drawing_number", "drawing number"]), | |
| "Part No": get_val(p, ["part_no", "part no"]), | |
| "Pcs in Box": get_val(p, ["pcs_in_box", "pcs in box"]), | |
| "No of Core Boxes": get_val(p, ["no_of_core_boxes", "no of core boxes"]), | |
| "No of Cores": get_val(p, ["no_of_cores", "no of cores"]), | |
| "Method Remarks": get_val(p, ["method_remarks", "method remarks"]) | |
| }) | |
| # --- SHEET 2: Moulding & Consumption --- | |
| md = page["moulding_details"] or {} | |
| rc = page["refractory_and_sand"] or {} | |
| notes = get_val(rc, ["notes"]) | |
| has_top_bottom = ("top" in md or "bottom" in md or "top" in rc or "bottom" in rc) | |
| if has_top_bottom: | |
| for part in ["top", "bottom"]: | |
| part_md = md.get(part, {}) or {} | |
| part_rc = rc.get(part, {}) or {} | |
| sheet2_rows.append({ | |
| "Task ID": page["task_id"], | |
| "Page No": page["page_no"], | |
| "Heat No": get_val(m, ["heat_no", "heat no"]), | |
| "Part (Top/Bottom)": part.upper(), | |
| "Contractor": get_val(part_md, ["contractor"]), | |
| "Moulder": get_val(part_md, ["moulder"]), | |
| "Moulding Date": get_val(part_md, ["moulding_date", "moulding date"]), | |
| "Moulding Time": get_val(part_md, ["moulding_time", "moulding time"]), | |
| "Coating Details": get_val(part_md, ["coating_details", "coating details"]), | |
| "Coating Date": get_val(part_md, ["coating_date", "coating date"]), | |
| "Coating Time": get_val(part_md, ["coating_time", "coating time"]), | |
| "Chromite Sand (kg)": get_val(part_rc, ["chromite_sand", "chromite sand"]), | |
| "Silica Sand (kg)": get_val(part_rc, ["silica_sand", "silica sand"]), | |
| "Sinotherm Resin (kg)": get_val(part_rc, ["sinotherm", "resin"]), | |
| "Activator (kg)": get_val(part_rc, ["activator"]), | |
| "Sparklex 100A Isomol (kg)": get_val(part_rc, ["sparklex_100a_isomol", "sparklex_100a_isomol (kg)", "sparklex"]), | |
| "Process Notes": notes | |
| }) | |
| else: | |
| sheet2_rows.append({ | |
| "Task ID": page["task_id"], | |
| "Page No": page["page_no"], | |
| "Heat No": get_val(m, ["heat_no", "heat no"]), | |
| "Part (Top/Bottom)": "GENERAL/FLAT", | |
| "Contractor": get_val(md, ["contractor"]), | |
| "Moulder": get_val(md, ["moulder"]), | |
| "Moulding Date": get_val(md, ["moulding_date", "moulding date"]), | |
| "Moulding Time": get_val(md, ["moulding_time", "moulding time"]), | |
| "Coating Details": get_val(md, ["coating_details", "coating details"]), | |
| "Coating Date": get_val(md, ["coating_date", "coating date"]), | |
| "Coating Time": get_val(md, ["coating_time", "coating time"]), | |
| "Chromite Sand (kg)": get_val(rc, ["chromite_sand", "chromite sand"]), | |
| "Silica Sand (kg)": get_val(rc, ["silica_sand", "silica sand"]), | |
| "Sinotherm Resin (kg)": get_val(rc, ["sinotherm", "resin"]), | |
| "Activator (kg)": get_val(rc, ["activator"]), | |
| "Sparklex 100A Isomol (kg)": get_val(rc, ["sparklex_100a_isomol", "sparklex_100a_isomol (kg)", "sparklex"]), | |
| "Process Notes": notes | |
| }) | |
| # --- SHEET 3: QA & Inspection Checks --- | |
| ip = page["inspection_parameters"] or {} | |
| qa_list = page["qa_parameters"] or [] | |
| if isinstance(qa_list, list): | |
| qa_summary = "; ".join([str(item) for item in qa_list if item]) | |
| else: | |
| qa_summary = str(qa_list) | |
| sheet3_rows.append({ | |
| "Task ID": page["task_id"], | |
| "Page No": page["page_no"], | |
| "Heat No": get_val(m, ["heat_no", "heat no"]), | |
| "Process Type": get_val(ip, ["process", "process type", "Pattern Finishing Process"]), | |
| "Pattern Finishing": get_val(ip, ["pattern_finishing", "pattern finishing"]), | |
| "Chill Size / Thickness": get_val(ip, ["chill_size_thickness", "chill size thickness"]), | |
| "Chill Slot Blasted": get_val(ip, ["chill_slot_blasted", "chill slot blasted"]), | |
| "Chill Finishing": get_val(ip, ["chill_finishing", "chill finishing"]), | |
| "Sleeve Size Oven": get_val(ip, ["sleeve_size_oven", "sleeve size oven"]), | |
| "Refractory Sleeve Check": get_val(ip, ["refactory_sleeve", "refractory sleeve", "refactory sleeve check"]), | |
| "Lettering Checking": get_val(ip, ["lettering_checking", "lettering checking"]), | |
| "Mould Checking": get_val(ip, ["mould_checking", "mould checking"]), | |
| "QA Requirements Summary": qa_summary | |
| }) | |
| # --- SHEET 4: Consumable Materials --- | |
| raw_mats = page["materials_table"] or [] | |
| materials = [] | |
| if isinstance(raw_mats, list): | |
| for row in raw_mats: | |
| code = get_val(row, ["sle_code", "code"]) | |
| desc = get_val(row, ["sle_name", "description", "item"]) | |
| qty = get_val(row, ["slv_qty", "qty", "quantity"]) | |
| act = get_val(row, ["actual_qty", "batch_no"]) | |
| if not code and not desc and not qty: | |
| continue | |
| materials.append({ | |
| "code": code, | |
| "desc": desc, | |
| "qty": qty, | |
| "act": act | |
| }) | |
| for mat in materials: | |
| sheet4_rows.append({ | |
| "Task ID": page["task_id"], | |
| "Page No": page["page_no"], | |
| "Heat No": get_val(m, ["heat_no", "heat no"]), | |
| "Material/Sleeve Code": mat["code"], | |
| "Description": mat["desc"], | |
| "Quantity": mat["qty"], | |
| "Actual Quantity / Batch No": mat["act"] | |
| }) | |
| # --- SHEET 5: Verification Signatures --- | |
| sig = page["signatures"] or {} | |
| sheet5_rows.append({ | |
| "Task ID": page["task_id"], | |
| "Page No": page["page_no"], | |
| "Heat No": get_val(m, ["heat_no", "heat no"]), | |
| "Planned By": get_val(sig, ["planned_by", "Planned By"]), | |
| "Pattern Inspected By": get_val(sig, ["pattern_inspected_by", "Pattern Inspected By"]), | |
| "QA Checked By": get_val(sig, ["qa_checked_by", "QA Checked By"]), | |
| "Core Inspected By": get_val(sig, ["core_inspected_by", "Core Inspected By"]), | |
| "Mould Inspected By": get_val(sig, ["mould_inspected_by", "Mould Inspected By"]), | |
| "Closing Inspected By": get_val(sig, ["closing_inspected_by", "Closing Inspected By"]), | |
| "Pouring Inspected By": get_val(sig, ["pouring_inspected_by", "Pouring Inspected By"]), | |
| "Pre Production Inspected By": get_val(sig, ["pre_production_inspected_by", "Pre Production Inspected By"]) | |
| }) | |
| # Convert to Pandas DataFrames | |
| df_sheet1 = pd.DataFrame(sheet1_rows) if sheet1_rows else pd.DataFrame(columns=["Task ID", "Page No", "Heat No", "Customer", "Grade"]) | |
| df_sheet2 = pd.DataFrame(sheet2_rows) if sheet2_rows else pd.DataFrame(columns=["Task ID", "Page No", "Heat No", "Part (Top/Bottom)", "Moulder"]) | |
| df_sheet3 = pd.DataFrame(sheet3_rows) if sheet3_rows else pd.DataFrame(columns=["Task ID", "Page No", "Heat No", "Process Type", "QA Requirements Summary"]) | |
| df_sheet4 = pd.DataFrame(sheet4_rows) if sheet4_rows else pd.DataFrame(columns=["Task ID", "Page No", "Heat No", "Material/Sleeve Code", "Description", "Quantity"]) | |
| df_sheet5 = pd.DataFrame(sheet5_rows) if sheet5_rows else pd.DataFrame(columns=["Task ID", "Page No", "Heat No", "Planned By", "Mould Inspected By"]) | |
| # Write to memory buffer | |
| buffer = io.BytesIO() | |
| with pd.ExcelWriter(buffer, engine='openpyxl') as writer: | |
| df_sheet1.to_excel(writer, index=False, sheet_name='Product Specifications') | |
| df_sheet2.to_excel(writer, index=False, sheet_name='Moulding & Consumption') | |
| df_sheet3.to_excel(writer, index=False, sheet_name='QA & Inspection Checks') | |
| df_sheet4.to_excel(writer, index=False, sheet_name='Consumable Materials') | |
| df_sheet5.to_excel(writer, index=False, sheet_name='Verification Signatures') | |
| buffer.seek(0) | |
| return StreamingResponse( | |
| buffer, | |
| media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", | |
| headers={"Content-Disposition": "attachment; filename=manufacturing_records.xlsx"} | |
| ) | |
| except Exception as e: | |
| raise HTTPException(status_code=500, detail=f"Failed to export data: {str(e)}") | |
| async def get_processing_status(task_id: str): | |
| return {"task_id": task_id, "status": "SYNC_MODE_ACTIVE", "message": "Redis is disabled. Check the main /process route for output."} |