Spaces:
Running on Zero
Running on Zero
| import pdfplumber | |
| import re | |
| from pdf_engine import extract_header_value, apply_value_replacement | |
| from parser_welspun import extract_welspun_items | |
| from parser_bkt import extract_bkt_items | |
| # 🚀 LayoutLMv3 & Document AI Integration Imports | |
| from transformers import LayoutLMv3Processor, LayoutLMv3ForTokenClassification | |
| import torch | |
| from PIL import Image | |
| import io | |
| def process_document_with_layoutlmv3(image_or_pdf_bytes): | |
| """ | |
| ZeroGPU / GPU environment ke liye LayoutLMv3 processing function. | |
| Yeh document ki visual bounding boxes aur text dono ko process karta hai. | |
| """ | |
| try: | |
| # Note: Model loading aur inference yahan ZeroGPU decorator ke antargat run hoga | |
| processor = LayoutLMv3Processor.from_pretrained("microsoft/layoutlmv3-base") | |
| # Model initialization aur inference logic yahan aage configure kiya ja sakta hai | |
| return True, "LayoutLMv3 processor initialized successfully." | |
| except Exception as e: | |
| return False, str(e) | |
| def run_universal_test_suite_backend(selected_shipper, shipper_database, pdf_lines, pdf_text, test_category, target_field): | |
| """ | |
| Streamlit-free backend test suite jo Gradio ke through single field ya row-by-row inspection run karta hai. | |
| """ | |
| if not pdf_lines: | |
| return "💡 कृपया पहले इनवॉइस PDF अपलोड करें, फिर टेस्ट रन करें।" | |
| shipper_info = shipper_database.get(selected_shipper, {}) | |
| header_rules = shipper_info.get("mapping_rules", {}) | |
| item_rules = shipper_info.get("item_table_rules", {}) | |
| assigned_parser = shipper_info.get("item_table_rule_name", "parser_welspun").strip().lower() | |
| if test_category == "Header Fields Rules": | |
| if target_field not in header_rules: | |
| return f"⚠️ Header Field '{target_field}' nahi mila." | |
| rule_data = header_rules[target_field] | |
| ky = rule_data.get("keyword", "") | |
| pos = rule_data.get("position", "Right (आगे)") | |
| cl = rule_data.get("cell", "").strip() | |
| m_mode = rule_data.get("match_mode", "Exact Word") | |
| stop_kw = rule_data.get("stop_kw", "") | |
| final_flt = rule_data.get("filter", "None") | |
| final_val = extract_header_value(pdf_lines, pdf_text, ky, pos, m_mode, stop_kw, final_flt) | |
| display_cell = cl if cl else 'Not Set' | |
| if cl and cl.isalpha(): | |
| display_cell = f"{cl.upper()}2 (Dynamic Auto-Increment Row)" | |
| result_summary = ( | |
| f"🔍 Inspection Result: {target_field}\n" | |
| f"----------------------------------------\n" | |
| f"📋 Parameters:\n" | |
| f"- Target Excel Cell: {display_cell}\n" | |
| f"- Keyword: {ky if ky else 'N/A'}\n" | |
| f"- Match Mode: {m_mode}\n" | |
| f"- Filter: {final_flt}\n\n" | |
| f"🎯 Extracted Value: {final_val if final_val else 'BLANK / NOT FOUND'}" | |
| ) | |
| return result_summary | |
| else: | |
| if target_field not in item_rules: | |
| return f"⚠️ Item Field '{target_field}' nahi mila." | |
| rule_info = item_rules[target_field] | |
| col_letter = rule_info.get("col", "").upper() | |
| rule_type = rule_info.get("type", "PDF Row Item") | |
| rule_val = rule_info.get("rule", "") | |
| if "bkt" in assigned_parser: | |
| parsed_items = extract_bkt_items(pdf_lines) | |
| else: | |
| parsed_items = extract_welspun_items(pdf_lines, pdf_text=pdf_text) | |
| if not parsed_items: | |
| return "⚠️ इस PDF में कोई आइटम रो नहीं मिली या पार्सर से डेटा एक्सट्रेक्ट नहीं हुआ।" | |
| preview_lines = [f"📊 Row-by-Row Preview for Column {col_letter}:"] | |
| for idx, item in enumerate(parsed_items): | |
| excel_row_num = 2 + idx | |
| cell_target = f"{col_letter}{excel_row_num}" | |
| nums = item.get("nums", []) | |
| r_val_lower = str(rule_val).lower().strip() | |
| f_name_lower = target_field.lower().strip() | |
| extracted_cell_val = "" | |
| if rule_type == "Constant Text": | |
| extracted_cell_val = apply_value_replacement(rule_val, rule_val) | |
| elif rule_type == "Excel Cell Reference": | |
| extracted_cell_val = f"={rule_val}" | |
| elif rule_type == "Smart Detection": | |
| desc = item.get("description_text", "").upper() | |
| extracted_cell_val = "PCS" if ("PCS" in desc or "PC" in desc) else (rule_val if rule_val else "SET") | |
| else: | |
| if "igst %" in r_val_lower or "igst rate" in f_name_lower: | |
| extracted_cell_val = nums[5] if len(nums) > 5 else "" | |
| elif "igst amt" in r_val_lower or "igst amount" in f_name_lower: | |
| extracted_cell_val = nums[6] if len(nums) > 6 else "" | |
| elif "hs" in r_val_lower or "ritc" in f_name_lower or "hs code" in r_val_lower: | |
| extracted_cell_val = item.get("hs_code", "") | |
| elif "description" in r_val_lower or "description" in f_name_lower: | |
| extracted_cell_val = item.get("description_text", "") | |
| elif "dbk" in r_val_lower or "drawback" in f_name_lower or col_letter == "S": | |
| extracted_cell_val = item.get("dbk_found", "") | |
| elif "weight" in r_val_lower or "net wt" in f_name_lower: | |
| extracted_cell_val = nums[0] if len(nums) > 0 else "" | |
| elif "qty" in r_val_lower or "quantity" in f_name_lower: | |
| extracted_cell_val = nums[1] if len(nums) > 1 else "" | |
| elif "rate" in r_val_lower: | |
| extracted_cell_val = nums[2] if len(nums) > 2 else "" | |
| elif "amount" in r_val_lower or "goods value" in f_name_lower: | |
| extracted_cell_val = nums[3] if len(nums) > 3 else "" | |
| elif "taxable" in r_val_lower: | |
| extracted_cell_val = nums[4] if len(nums) > 4 else "" | |
| else: | |
| extracted_cell_val = rule_val | |
| if "=" in str(rule_val): | |
| extracted_cell_val = apply_value_replacement(str(extracted_cell_val), str(rule_val)) | |
| preview_lines.append(f"Row {idx+1} | Cell: {cell_target} | Value: {extracted_cell_val}") | |
| return "\n".join(preview_lines) |