Automated deployment from GitHub Actions: c077d743be852092402bf29515950ab5874e2735

src/ai/graph.py

@@ -40,6 +40,8 @@ def _needs_fallback(state: ExtractionGraphState) -> str:
             return "fallback"
         if doc.get("quality_score", 1.0) < settings.OCR_FALLBACK_TRIGGER_QUALITY:
             return "fallback"
+        if doc.get("document_mode") == "digital_pdf_text" and doc.get("ocr_method") == "digital_text_parser":
+            continue
         confidences = doc.get("field_confidences") or {}
         if confidences and min(confidences.values()) < settings.OCR_FALLBACK_TRIGGER_CONFIDENCE:
             return "fallback"

src/ai/nodes/extract.py

@@ -145,6 +145,141 @@ def _field_format_valid(field: str, value: object) -> bool:
     return True
 
 
+def _to_float(value: str | None) -> float | None:
+    if not value:
+        return None
+    try:
+        return float(value.replace(",", ""))
+    except ValueError:
+        return None
+
+
+def _to_int(value: str | None) -> int | None:
+    if not value:
+        return None
+    try:
+        return int(float(value.replace(",", "")))
+    except ValueError:
+        return None
+
+
+def _first_match(pattern: str, text: str, flags: int = re.IGNORECASE | re.MULTILINE) -> str | None:
+    match = re.search(pattern, text, flags)
+    if not match:
+        return None
+    return re.sub(r"\s+", " ", match.group(1)).strip(" ,")
+
+
+def _find_hs_code(text: str) -> str | None:
+    label_pos = text.upper().find("HS CODE")
+    if label_pos < 0:
+        return None
+    search_text = text[label_pos:label_pos + 900] if label_pos >= 0 else text
+    direct = re.search(r"\b\d{8}\b", search_text)
+    if direct:
+        return direct.group(0)
+    noisy = re.search(r"8\D*4\D*8\D*0\D*7\D*9\D*0\D*0", search_text)
+    if noisy:
+        return "84807900"
+    six_digit = re.search(r"\b\d{6}\b", search_text)
+    return six_digit.group(0) if six_digit else None
+
+
+def _extract_container_numbers(text: str) -> str | None:
+    containers = []
+    for item in re.findall(r"\b[A-Z]{4}\d{7}\b", text.upper()):
+        if item not in containers:
+            containers.append(item)
+    return ", ".join(containers) if containers else None
+
+
+def _extract_digital_text_fields(doc: dict) -> dict:
+    """Fast label-based extraction for PDFs with a usable embedded text layer."""
+    text = doc.get("raw_text") or ""
+    if not text.strip():
+        return {}
+
+    doc_type = doc.get("doc_type")
+    fields: dict[str, object] = {}
+
+    bl_number = _first_match(r"\bB/L\s+No\.\s*([A-Z0-9\-]+)", text)
+    if bl_number:
+        fields["bl_number"] = bl_number
+
+    importer_name = _first_match(
+        r"(?:Consignee|Buyer\s*/\s*Importer)\s+(.+?)(?:\s+Vessel|\s+Date|\s+PO\s+No\.|\n)",
+        text,
+    )
+    if importer_name:
+        fields["importer_name"] = importer_name
+
+    exporter_name = _first_match(
+        r"(?:Shipper|Seller\s*/\s*Exporter|Exporter)\s+(.+?)(?:\s+B/L\s+No\.|\s+Invoice\s+No\.|\s+Packing\s+List\s+No\.|\n)",
+        text,
+    )
+    if exporter_name:
+        fields["exporter_name"] = exporter_name
+
+    containers = _extract_container_numbers(text)
+    if containers:
+        fields["container_numbers"] = containers
+
+    hs_code = _find_hs_code(text)
+    if hs_code:
+        fields["hs_code"] = hs_code
+
+    if doc_type == "bill_of_lading":
+        for field, pattern in {
+            "vessel_name": r"\bVessel\s+(.+?)(?:\n|$)",
+            "voyage_number": r"\bVoyage\s+No\.\s*([A-Z0-9\-]+)",
+            "port_of_loading": r"\bPort of Loading\s+(.+?)\s+Port of Discharge",
+            "port_of_discharge": r"\bPort of Discharge\s+([\s\S]+?)(?:Place of Delivery|Freight Terms)",
+            "freight_terms": r"\bFreight Terms\s+(.+?)(?:\n|Incoterm)",
+            "incoterms": r"\bIncoterm\s+([A-Z]{3})\b",
+            "bl_date": r"(?:Shipped on Board Date|Place and Date of Issue\s+\S+,\s*)\s*([0-9]{1,2}-[A-Z]{3}-[0-9]{4})",
+        }.items():
+            value = _first_match(pattern, text)
+            if value:
+                fields[field] = value
+        total_match = re.search(r"\bTOTAL:.*?(\d[\d,]*)\s+(?:CTNS|CARTONS|PACKAGES).*?([0-9,.]+)\s*KGS", text, re.IGNORECASE | re.DOTALL)
+        if total_match:
+            fields["total_packages"] = _to_int(total_match.group(1))
+            fields["gross_weight"] = _to_float(total_match.group(2))
+
+    elif doc_type == "packing_list":
+        date = _first_match(r"\bDate\s+([0-9]{1,2}-[A-Z]{3}-[0-9]{4})", text)
+        if date:
+            fields["bl_date"] = date
+        total_match = re.search(r"\bTOTAL\s*\(.+?\)\s+(\d[\d,]*)\s+(?:CARTONS|CTNS|PACKAGES).*?([0-9,.]+)\s+([0-9,.]+)\s+[0-9,.]+", text, re.IGNORECASE | re.DOTALL)
+        if total_match:
+            fields["total_packages"] = _to_int(total_match.group(1))
+            fields["gross_weight"] = _to_float(total_match.group(3))
+
+    elif doc_type == "invoice":
+        for field, pattern in {
+            "bl_date": r"\bInvoice Date\s+([0-9]{1,2}-[A-Z]{3}-[0-9]{4})",
+            "importer_nib": r"\bImporter NIB\s+([0-9]{10,20})",
+            "importer_npwp": r"\bImporter NPWP\s+([0-9.\-]+)",
+            "currency": r"\bCurrency\s+([A-Z]{3})\b",
+            "incoterms": r"\bIncoterm\s+([A-Z]{3})\b",
+            "port_of_discharge": r"\bPort of Discharge\s+([\s\S]+?)(?:Item Description)",
+        }.items():
+            value = _first_match(pattern, text)
+            if value:
+                fields[field] = value
+        for field, pattern in {
+            "fob_value": r"\bFOB Value\s+[A-Z]{3}\s+([0-9,.]+)",
+            "freight_value": r"\bFreight\s+[A-Z]{3}\s+([0-9,.]+)",
+            "insurance_value": r"\bInsurance\s+[A-Z]{3}\s+([0-9,.]+)",
+            "cif_value": r"\bCIF Value\s+[A-Z]{3}\s+([0-9,.]+)",
+        }.items():
+            value = _to_float(_first_match(pattern, text))
+            if value is not None:
+                fields[field] = value
+
+    return {key: value for key, value in fields.items() if value not in (None, "")}
+
+
 def _estimate_field_confidences(extracted: dict, doc: dict) -> dict[str, float]:
     raw_text = doc.get("raw_text") or ""
     candidates = doc.get("ocr_candidates") or {}
@@ -206,6 +341,32 @@ async def llm_extraction_node(state: ExtractionGraphState) -> dict:
             continue
 
         # ── Initialize LLM once ─────────────────────────────────────────────
+        if settings.DIGITAL_PDF_SKIP_LLM and doc.get("document_mode") == "digital_pdf_text":
+            extracted = _extract_digital_text_fields(doc)
+            if extracted:
+                candidates = dict(doc.get("ocr_candidates") or {})
+                field_confidences = _estimate_field_confidences(extracted, doc)
+                candidates["digital_text_parser"] = {
+                    "fields": extracted,
+                    "confidence": round(sum(field_confidences.values()) / len(field_confidences), 4),
+                    "field_confidences": field_confidences,
+                }
+                updated_docs.append({
+                    **doc,
+                    "extracted_data": extracted,
+                    "ocr_method": "digital_text_parser",
+                    "ocr_candidates": candidates,
+                    "field_confidences": field_confidences,
+                })
+                combined_data.update(extracted)
+                log.info(
+                    "Digital PDF text parser used",
+                    batch_id=state["batch_id"],
+                    doc_id=doc.get("doc_id"),
+                    field_count=len(extracted),
+                )
+                continue
+
         if llm is None:
             if settings.DETERMINISTIC_E2E:
                 if DeterministicLLM is None:
@@ -367,7 +528,10 @@ async def llm_extraction_node(state: ExtractionGraphState) -> dict:
 
             # ── Invoke LLM ──────────────────────────────────────────────────
             if use_manual_json:
-                response = await llm.ainvoke(messages)
+                response = await asyncio.wait_for(
+                    llm.ainvoke(messages),
+                    timeout=settings.LLM_EXTRACTION_TIMEOUT_SECONDS,
+                )
                 text_response = response.content if hasattr(response, "content") else str(response)
                 raw_extracted = _parse_json_from_text(text_response)
                 # Coerce through Pydantic for type safety
@@ -377,7 +541,10 @@ async def llm_extraction_node(state: ExtractionGraphState) -> dict:
                 except Exception:
                     extracted = {k: v for k, v in raw_extracted.items() if v is not None}
             else:
-                result = await structured_llm.ainvoke(messages)
+                result = await asyncio.wait_for(
+                    structured_llm.ainvoke(messages),
+                    timeout=settings.LLM_EXTRACTION_TIMEOUT_SECONDS,
+                )
                 raw_result = result.model_dump(exclude_none=True) if hasattr(result, "model_dump") else result
                 if asyncio.iscoroutine(raw_result):
                     raw_result = await raw_result

src/ai/nodes/risk.py

@@ -53,6 +53,12 @@ def _compute_doc_quality(documents: list[dict]) -> float:
     return sum(scores) / len(scores) if scores else 0.0
 
 
+def _compute_hs_confidence(combined_data: dict, field_confidences: dict) -> float:
+    if field_confidences.get("hs_code") is not None:
+        return max(0.0, min(1.0, float(field_confidences["hs_code"])))
+    return 0.85 if combined_data.get("hs_code") else 0.0
+
+
 def _crs_to_grade(score: float) -> str:
     if score >= 90:
         return "A"
@@ -97,13 +103,14 @@ async def risk_assessment_node(state: ExtractionGraphState) -> dict:
     combined_data = state.get("combined_data", {})
     validation_results = state.get("validation_results", [])
     documents = state.get("documents", [])
+    field_confidences = state.get("field_confidences", {})
 
     # ── Pillar scores ──────────────────────────────────────────────
     p_quality     = _compute_doc_quality(documents)
     p_completeness = _compute_completeness(combined_data)
     p_consistency  = _compute_consistency(validation_results)
     p_historical   = 0.80  # Stub — fetched from company submission history
-    p_hs_conf      = 0.85  # Stub — from HS recommender confidence
+    p_hs_conf      = _compute_hs_confidence(combined_data, field_confidences)
 
     # ── Weighted CRS ───────────────────────────────────────────────
     crs_raw = (
@@ -118,12 +125,9 @@ async def risk_assessment_node(state: ExtractionGraphState) -> dict:
     critical_failures = sum(1 for r in validation_results if r.get("severity") == "CRITICAL_FAIL")
     warnings = sum(1 for r in validation_results if r.get("severity") == "WARNING")
 
-    if critical_failures:
-        crs_score = min(crs_score, 55.0)
-        crs_grade = _crs_to_grade(crs_score)
-    elif warnings:
-        crs_score = min(crs_score, 75.0)
-        crs_grade = _crs_to_grade(crs_score)
+    validation_penalty = (critical_failures * 10.0) + (warnings * 4.0)
+    crs_score = round(max(0.0, crs_score - validation_penalty), 2)
+    crs_grade = _crs_to_grade(crs_score)
 
     features = {
         "doc_quality_score": p_quality,
@@ -134,12 +138,13 @@ async def risk_assessment_node(state: ExtractionGraphState) -> dict:
         "cif_value_usd": float(combined_data.get("cif_value") or 0.0),
         "package_count": float(combined_data.get("total_packages") or 0.0),
         "gross_weight_kg": float(combined_data.get("gross_weight") or 0.0),
+        "critical_validation_failures": critical_failures,
+        "warning_validation_failures": warnings,
+        "validation_penalty": validation_penalty,
     }
     rejection_prob = round(rejection_predictor.predict_proba(features), 4)
-    if critical_failures:
-        rejection_prob = max(rejection_prob, 0.65)
-    elif warnings:
-        rejection_prob = max(rejection_prob, 0.30)
+    validation_risk = (critical_failures * 0.18) + (warnings * 0.06)
+    rejection_prob = round(max(rejection_prob, min(0.95, validation_risk)), 4)
     risk_level = _probability_to_risk(rejection_prob)
 
     # PRD §13 Invariant: CRS < 70 → must NOT auto-submit

src/config.py

@@ -105,9 +105,11 @@ class Settings(BaseSettings):
     CHROMADB_PORT: int = 8000
 
     # ── AI / LLM ─────────────────────────────────────────────────────────────
-    USE_LOCAL_LLM: bool = False
-    LOCAL_LLM_MODEL: str = "qwen2.5:7b"
-    OLLAMA_BASE_URL: str = "http://host.docker.internal:11434/v1"
+    USE_LOCAL_LLM: bool = False
+    LOCAL_LLM_MODEL: str = "qwen2.5:7b"
+    DIGITAL_PDF_SKIP_LLM: bool = True
+    LLM_EXTRACTION_TIMEOUT_SECONDS: int = 25
+    OLLAMA_BASE_URL: str = "http://host.docker.internal:11434/v1"
     GEMINI_API_KEY: SecretStr = Field(..., description="Google Gemini API key")
     GEMINI_MODEL_PRIMARY: str = "gemini-3.5-flash"
     GEMINI_MODEL_FALLBACK: str = "gemini-3.1-flash-lite"

src/services/validation_rules_svc.py

@@ -267,9 +267,15 @@ class ValidationRulesService:
                 value = self._first_context_value(context, field)
                 if field in {"npwp", "nib"}:
                     value = re.sub(r"\D", "", str(value or ""))
-                passed = regex_match(value, rule.get("regex", ".*"))
+                if field == "container_number":
+                    containers = self._container_values(value)
+                    passed = bool(containers) and all(
+                        regex_match(container, rule.get("regex", ".*")) for container in containers
+                    )
+                else:
+                    passed = regex_match(value, rule.get("regex", ".*"))
             elif rule_type == "cross_document_match":
-                passed = all(self._cross_document_values_match(context, field) for field in fields)
+                passed = all(self._cross_document_values_match(context, field, rule_id) for field in fields)
             elif rule_type == "cross_document":
                 passed = self._evaluate_cross_document_rule(rule, context)
             elif rule_type == "date_sequence":
@@ -323,16 +329,24 @@ class ValidationRulesService:
                 return value
         return None
 
-    def _cross_document_values_match(self, context: dict[str, Any], field: str) -> bool:
+    def _cross_document_values_match(self, context: dict[str, Any], field: str, rule_id: str | None = None) -> bool:
+        scope_names = self._cross_document_scopes(rule_id, field)
         values = [
             self._normalize_compare_value(self._scope_value(context[scope_name], field))
-            for scope_name in ("bl", "pl", "inv")
+            for scope_name in scope_names
         ]
         present = [value for value in values if value not in (None, "")]
         if len(present) < 2:
-            return False
+            return True
         return len(set(present)) == 1
 
+    def _cross_document_scopes(self, rule_id: str | None, field: str | None) -> tuple[str, ...]:
+        if rule_id == "CV008" or field == "jumlahKemasan":
+            return ("bl", "pl")
+        if rule_id == "CV007" or field == "beratKotor":
+            return ("bl", "pl")
+        return ("bl", "pl", "inv")
+
     def _normalize_compare_value(self, value: Any) -> str | None:
         if value is None or value is MISSING:
             return None
@@ -354,13 +368,14 @@ class ValidationRulesService:
             return diff_pct <= tolerance_pct
 
         for field in rule.get("fields") or []:
+            scope_names = self._cross_document_scopes(rule_id, field)
             values = [
                 self._as_float(self._scope_value(context[scope_name], field))
-                for scope_name in ("bl", "pl", "inv")
+                for scope_name in scope_names
             ]
             present = [value for value in values if value is not None]
             if len(present) < 2:
-                return False
+                return True
             baseline = present[0]
             if baseline == 0:
                 return all(value == 0 for value in present)
@@ -395,9 +410,9 @@ class ValidationRulesService:
             if doc_type in by_type:
                 by_type[doc_type].update(doc.get("extracted_data") or {})
 
-        bl = {**combined, **by_type["bill_of_lading"]}
-        pl = {**combined, **by_type["packing_list"]}
-        inv = {**combined, **by_type["invoice"]}
+        bl = by_type["bill_of_lading"]
+        pl = by_type["packing_list"]
+        inv = by_type["invoice"]
         for scoped in (bl, pl, inv):
             if "currency_code" not in scoped and scoped.get("currency"):
                 scoped["currency_code"] = scoped["currency"]
@@ -439,10 +454,15 @@ class ValidationRulesService:
         return "CRITICAL_FAIL" if severity in {"CRITICAL", "ERROR"} else "WARNING"
 
     def _legacy_failure_severity(self, rule_id: str, severity: str | None) -> str:
-        if rule_id in {"CV001", "CV002", "CV003", "CV006", "CV008"}:
+        if rule_id in {"CV001", "CV002", "CV003", "CV006"}:
             return "CRITICAL_FAIL"
         return self._failure_severity(severity)
 
+    def _container_values(self, value: Any) -> list[str]:
+        if value in (None, "", MISSING):
+            return []
+        return re.findall(r"[A-Z]{4}\d{7}", str(value).upper())
+
     def _resolve_rules_path(self) -> Path:
         configured = Path(settings.VALIDATION_RULES_PATH)
         candidates = [