# main_db.py
# Updated main pipeline with skip-on-existing-success and explicit DUPLICATE status handling
import os
import csv
import json
import shutil
import hashlib
import tempfile
from pathlib import Path
from datetime import datetime, timedelta
from typing import List, Dict, Tuple, Optional
import inspect
import time
import re

import fitz  # PyMuPDF
from pdf2image import convert_from_path
from dotenv import load_dotenv
from apscheduler.schedulers.blocking import BlockingScheduler
import pytz

# --- Your modules ---
# ARDEM LLM adapter; it preserves the existing inference function contract so
# the remaining S3/DB/dashboard workflow stays unchanged.
from ardem_inference_PA import infer_multiple_images, save_json_and_excel

# Import s3 helpers under private names; we add safe wrappers below
from s3_utils import list_pdfs_in_prefix as _list_pdfs_in_prefix
from s3_utils import download_pdf_from_s3 as _download_pdf_from_s3

# uses signature: send_notification(subject, records, note=None, totals=None)
from email_utils import send_notification

# NEW: DB write helpers (+ md5 lookup)
# Note: keep import path as in your project
from db_repo import (
    upsert_file,
    upsert_output,
    find_by_md5,
    list_files as db_list_files,
    file_exists_success,
    # ProcessData helpers (added)
    get_guid_by_date_filename,
    upsert_processdata,
    get_processdata_by_date_filename,
    # Rule configuration helpers (added)
    get_rule_configuration_by_state,
    get_state_rule_count,
)

# ---------------------------
# Environment & configuration
# ---------------------------
load_dotenv()

AWS_ACCESS_KEY = os.getenv("AWS_ACCESS_KEY")
AWS_SECRET_KEY = os.getenv("AWS_SECRET_KEY")
AWS_REGION = os.getenv("AWS_REGION")
S3_BUCKET = os.getenv("BUCKET_NAME")  # may be used by wrapper if needed


# Base backend directory (VITU_BACKEND)
BACKEND_DIR = Path(__file__).resolve().parents[0]

# ---------------------------
# Poppler path (for pdf2image)
# ---------------------------
POPLER_BIN_PATH = str(
    BACKEND_DIR
    / "Release-24.08.0-0"
    / "poppler-24.08.0"
    / "Library"
    / "bin"
)
# Example resolved path:
# C:\Users\Abhishek.Mishra\Desktop\VITU_BACKEND\Release-24.08.0-0\poppler-24.08.0\Library\bin

# ---------------------------
# Conversion (image temp) directory
# ---------------------------
CONVERTED_IMG_DIR = BACKEND_DIR / "conv"
CONVERTED_IMG_DIR.mkdir(parents=True, exist_ok=True)
# Resolves to: C:\Users\Abhishek.Mishra\Desktop\VITU_BACKEND\conv

# ---------------------------
# Dashboard data directory
# ---------------------------
DATA_DIR = BACKEND_DIR / "data"
DATA_DIR.mkdir(parents=True, exist_ok=True)
# Resolves to: C:\Users\Abhishek.Mishra\Desktop\VITU_BACKEND\data

# ---------------------------
# Dashboard conversion image directory (for serving via /static/conv)
# ---------------------------
DASH_CONV_DIR = BACKEND_DIR / "conv"  # same as CONVERTED_IMG_DIR
DASH_CONV_DIR.mkdir(parents=True, exist_ok=True)
# Resolves to: C:\Users\Abhishek.Mishra\Desktop\VITU_BACKEND\conv

# We're focusing PA now (can override)
ACTIVE_PREFIX = os.getenv("VITU_ACTIVE_PREFIX", "PA").upper()

# Timezone
TZ = pytz.timezone("Asia/Kolkata")

# Feature toggle: if you ever want to run without DB writes for suspects, set USE_DB = False
USE_DB = True

# ---------------- Numeric validation regexes & helpers for suspect checks -----------
NUMERIC_VALID_RE = re.compile(r'^(?:\d{1,3}(?:,\d{3})*|\d+)\.\d{2}$')
TXN_RE = re.compile(r'^\d{8,10}$')
PROCESS_DATE_RE = re.compile(r'^\d{4}-\d{2}-\d{2}$')
STATE_RE = re.compile(r'^[A-Z]{2}$')


def _is_multi_value(s: str):
    # heuristically detect multiple numeric tokens
    tokens = re.findall(r'[\d,.\-A-Za-z]+', s)
    num_tokens = sum(1 for t in tokens if re.search(r'\d', t))
    return num_tokens > 1


def fetch_state_rules(state_code: str) -> Dict[str, List[Dict]]:
    """
    Fetch validation rules from database for given state
    Returns: dict of {column_name: [rules]}
    """
    try:
        rules_by_column = get_rule_configuration_by_state(state_code)
        print(f"Loaded {len(rules_by_column)} columns with rules for state {state_code}")
        return rules_by_column
    except Exception as e:
        print(f"Error fetching rules for state {state_code}: {e}")
        return {}


def validate_field_with_rules(field_name: str, field_value: str, rules_list: List[Dict]) -> Tuple[bool, Optional[str]]:
    """
    Validate a field value against a list of rules
    Returns: (is_valid, reason_code) tuple
    """
    if not rules_list or not field_value:
        return True, None
    
    field_value = str(field_value).strip()
    
    for rule in rules_list:
        try:
            pattern = rule['regex']
            # Fix double-escaped patterns from database
            if '\\\\' in pattern:
                pattern = pattern.replace('\\\\', '\\')
            # Compile and test regex
            regex_obj = re.compile(pattern)
            
            if rule['rule_id'] == 'non_numeric':
                # Special case: non_numeric rule should flag if alphabetic chars found
                if regex_obj.search(field_value):
                    return False, "non_numeric"
            elif rule['rule_id'] == 'multi_value':
                # Special case: multi_value detection
                tokens = re.findall(pattern, field_value)
                if len([t for t in tokens if re.search(r'\d', t)]) > 1:
                    return False, "multi_value"
            else:
                # Standard regex validation
                if not regex_obj.match(field_value):
                    return False, rule['rule_id']
                    
        except re.error as e:
            print(f"Invalid regex in rule {rule['rule_id']}: {pattern} - {e}")
            continue
    
    return True, None


def get_fallback_rules() -> Dict[str, List[Dict]]:
    """
    Return hardcoded rules as fallback if database is unavailable
    """
    return {
        "Taxable Selling Price / Purchase Price": [
            {'rule_id': 'money_2dec', 'regex': r'^(?:\d{1,3}(?:,\d{3})*|\d+)\.\d{2}$', 'label': 'Money (2 decimals)'}
        ],
        "Sales Tax Credit Reciprocity": [
            {'rule_id': 'money_2dec', 'regex': r'^(?:\d{1,3}(?:,\d{3})*|\d+)\.\d{2}$', 'label': 'Money (2 decimals)'}
        ],
        "Sales Tax": [
            {'rule_id': 'money_2dec', 'regex': r'^(?:\d{1,3}(?:,\d{3})*|\d+)\.\d{2}$', 'label': 'Money (2 decimals)'}
        ],
        "Sales Tax Rate": [
            {'rule_id': 'money_2dec', 'regex': r'^(?:\d{1,3}(?:,\d{3})*|\d+)\.\d{2}$', 'label': 'Money (2 decimals)'}
        ],
        "Total State Taxes": [
            {'rule_id': 'money_2dec', 'regex': r'^(?:\d{1,3}(?:,\d{3})*|\d+)\.\d{2}$', 'label': 'Money (2 decimals)'}
        ],
        "Registration Fee": [
            {'rule_id': 'money_2dec', 'regex': r'^(?:\d{1,3}(?:,\d{3})*|\d+)\.\d{2}$', 'label': 'Money (2 decimals)'}
        ],
        "Title Fee": [
            {'rule_id': 'money_2dec', 'regex': r'^(?:\d{1,3}(?:,\d{3})*|\d+)\.\d{2}$', 'label': 'Money (2 decimals)'}
        ],
        "Plate Fee": [
            {'rule_id': 'money_2dec', 'regex': r'^(?:\d{1,3}(?:,\d{3})*|\d+)\.\d{2}$', 'label': 'Money (2 decimals)'}
        ],
        "Lien Fee": [
            {'rule_id': 'money_2dec', 'regex': r'^(?:\d{1,3}(?:,\d{3})*|\d+)\.\d{2}$', 'label': 'Money (2 decimals)'}
        ],
        "Transfer Fee": [
            {'rule_id': 'money_2dec', 'regex': r'^(?:\d{1,3}(?:,\d{3})*|\d+)\.\d{2}$', 'label': 'Money (2 decimals)'}
        ],
        "Motor Vehicle Fees": [
            {'rule_id': 'money_2dec', 'regex': r'^(?:\d{1,3}(?:,\d{3})*|\d+)\.\d{2}$', 'label': 'Money (2 decimals)'}
        ],
        "Local Use / County Fee": [
            {'rule_id': 'money_2dec', 'regex': r'^(?:\d{1,3}(?:,\d{3})*|\d+)\.\d{2}$', 'label': 'Money (2 decimals)'}
        ],
        "Increase Fee": [
            {'rule_id': 'money_2dec', 'regex': r'^(?:\d{1,3}(?:,\d{3})*|\d+)\.\d{2}$', 'label': 'Money (2 decimals)'}
        ],
        "Exchange Fee": [
            {'rule_id': 'money_2dec', 'regex': r'^(?:\d{1,3}(?:,\d{3})*|\d+)\.\d{2}$', 'label': 'Money (2 decimals)'}
        ],
        "Specialty Plate Fee": [
            {'rule_id': 'money_2dec', 'regex': r'^(?:\d{1,3}(?:,\d{3})*|\d+)\.\d{2}$', 'label': 'Money (2 decimals)'}
        ],
        "Additional Charge": [
            {'rule_id': 'money_2dec', 'regex': r'^(?:\d{1,3}(?:,\d{3})*|\d+)\.\d{2}$', 'label': 'Money (2 decimals)'}
        ],
        "Total Title & Registration Fee": [
            {'rule_id': 'money_2dec', 'regex': r'^(?:\d{1,3}(?:,\d{3})*|\d+)\.\d{2}$', 'label': 'Money (2 decimals)'}
        ],
        "Total Penalties": [
            {'rule_id': 'money_2dec', 'regex': r'^(?:\d{1,3}(?:,\d{3})*|\d+)\.\d{2}$', 'label': 'Money (2 decimals)'}
        ],
        "Total Other Fees": [
            {'rule_id': 'money_2dec', 'regex': r'^(?:\d{1,3}(?:,\d{3})*|\d+)\.\d{2}$', 'label': 'Money (2 decimals)'}
        ],
        "Total Sales Taxes + Reg Fee + Penalties": [
            {'rule_id': 'money_2dec', 'regex': r'^(?:\d{1,3}(?:,\d{3})*|\d+)\.\d{2}$', 'label': 'Money (2 decimals)'}
        ],
        "Transaction ID": [
            {'rule_id': 'txn_8_10_digits', 'regex': r'^\d{8,10}$', 'label': 'Transaction ID (8-10 digits)'}
        ],
        "Process Date": [
            {'rule_id': 'yyyy_mm_dd_date', 'regex': r'^\d{4}-\d{2}-\d{2}$', 'label': 'Date (YYYY-MM-DD)'}
        ],
        "Titling State": [
            {'rule_id': 'state_2letters', 'regex': r'^[A-Z]{2}$', 'label': 'State (2 letters)'}
        ]
    }


def perform_consistency_checks(normalized: Dict[str, str], mark_function):
    """
    Perform business logic consistency checks (kept separate from field validation)
    """
    def _safe_float(s):
        try:
            return float(str(s).replace(",", "").strip())
        except Exception:
            return None
    
    # Total Title & Registration Fee consistency check
    fee_components = [
        "Registration Fee", "Title Fee", "Plate Fee", "Lien Fee", "Transfer Fee",
        "Motor Vehicle Fees", "Local Use / County Fee", "Increase Fee",
        "Exchange Fee", "Specialty Plate Fee", "Additional Charge"
    ]
    
    total_title_reg = _safe_float(normalized.get("Total Title & Registration Fee", "0") or "0")
    if total_title_reg is not None:
        component_sum = 0.0
        valid_sum = True
        
        for component in fee_components:
            component_val = _safe_float(normalized.get(component, "0") or "0")
            if component_val is None:
                valid_sum = False
                break
            component_sum += component_val
        
        if valid_sum and abs(component_sum - total_title_reg) > 0.01:
            mark_function("Total Title & Registration Fee", "consistency_mismatch")
    
    # Combined total consistency check
    total_combined = _safe_float(normalized.get("Total Sales Taxes + Reg Fee + Penalties", "0") or "0")
    sales_tax = _safe_float(normalized.get("Sales Tax", "0") or "0")
    penalties = _safe_float(normalized.get("Total Penalties", "0") or "0")
    
    if all(v is not None for v in [total_combined, sales_tax, total_title_reg, penalties]):
        expected_total = sales_tax + total_title_reg + penalties
        if abs(expected_total - total_combined) > 0.01:
            mark_function("Total Sales Taxes + Reg Fee + Penalties", "consistency_mismatch")


def run_suspect_checks(normalized: Dict[str, str], state: str = None) -> Tuple[Dict[str, str], Dict[str, str]]:
    """
    Enhanced suspect validation using database rules with hardcoded fallback
    
    Args:
        normalized: Normalized field data
        state: State code (e.g., 'PA', 'CT') - determines which rules to load
    
    Returns:
        - suspect_map: {field: reason_code}  
        - suspect_values: {field: original_value}
    """
    suspect_map = {}
    suspect_values = {}
    
    def mark(field, reason):
        suspect_map[field] = reason
        suspect_values[field] = str(normalized.get(field, "") or "")
    
    # Determine state from context if not provided
    if not state:
        state = "PA"  # Default for main_db.py
    
    print(f"Running suspect checks for state: {state}")
    
    # Try to load rules from database
    db_rules = fetch_state_rules(state)
    
    if not db_rules:
        print(f"No database rules found for {state}, using fallback rules")
        db_rules = get_fallback_rules()
    
    # Track validation statistics
    total_fields = len(normalized)
    validated_fields = 0
    
    # Validate each field that has rules configured
    for field_name, field_value in normalized.items():
        if field_name in db_rules:
            validated_fields += 1
            is_valid, reason = validate_field_with_rules(field_name, field_value, db_rules[field_name])
            if not is_valid:
                mark(field_name, reason)
    
    # Additional consistency checks (these remain hardcoded as business logic)
    perform_consistency_checks(normalized, mark)
    
    # Log validation summary
    suspect_count = len(suspect_map)
    print(f"Suspect validation completed for {state}:")
    print(f"  - Total fields processed: {total_fields}")
    print(f"  - Fields with database rules: {validated_fields}")
    print(f"  - Suspect fields identified: {suspect_count}")
    print(f"  - Database rule coverage: {(validated_fields/total_fields)*100:.1f}%")
    
    return suspect_map, suspect_values


# ---------- S3 helper wrappers (signature-agnostic) ----------
def list_pdfs_safe(prefix: str) -> List[str]:
    """Call list_pdfs_in_prefix no matter if it expects (prefix) or (bucket, prefix)."""
    sig = inspect.signature(_list_pdfs_in_prefix)
    params = list(sig.parameters)
    if len(params) == 1:
        return _list_pdfs_in_prefix(prefix)
    elif len(params) == 2:
        return _list_pdfs_in_prefix(S3_BUCKET, prefix)
    else:
        raise TypeError("Unsupported list_pdfs_in_prefix signature")


def download_pdf_safe(key: str, local_path: str) -> str:
    """Call download_pdf_from_s3 whether it is (key, local_path) or (bucket, key, local_path).

    Returns:
      md5 hex string (lowercase) when successful.

    Raises:
      re-raises underlying exceptions so caller can handle failures.
    """
    sig = inspect.signature(_download_pdf_from_s3)
    params = list(sig.parameters)
    if len(params) == 2:
        return _download_pdf_from_s3(key, local_path)
    elif len(params) == 3:
        return _download_pdf_from_s3(S3_BUCKET, key, local_path)
    else:
        raise TypeError("Unsupported download_pdf_from_s3 signature")


# ------------- Helpers -------------
def _today_iso() -> str:
    return datetime.now(TZ).strftime("%Y-%m-%d")


def _date_dir(day_iso: str) -> Path:
    p = DATA_DIR / day_iso
    p.mkdir(parents=True, exist_ok=True)
    return p


def get_today_s3_prefix() -> str:
    """Build the current PA ARDEM input-folder key.

    The source folder is always the previous calendar day.  For example, a
    run at 09:00 IST on July 29, 2026 reads:
    ``VITU/2026/July-2026/07282026/09:00/Input_Files/PA_07.28/``.

    The bucket is deliberately not included because the existing S3 helpers
    receive it separately from ``BUCKET_NAME``.
    """
    source_day = datetime.now(TZ) - timedelta(days=1)
    year = source_day.strftime("%Y")
    month_folder = source_day.strftime("%B-%Y")
    compact_date = source_day.strftime("%m%d%Y")
    time_folder = source_day.strftime("%H:%M")
    state_day = source_day.strftime("%m.%d")
    return (
        f"VITU/{year}/{month_folder}/{compact_date}/{time_folder}/"
        f"Input_Files/{ACTIVE_PREFIX}_{state_day}/"
    )


def convert_all_pages_to_images(pdf_path: str, output_dir: Path) -> List[Path]:
    """
    Convert all pages of a PDF to PNG images in output_dir, returns list of paths.
    """
    image_paths: List[Path] = []
    base_name = Path(pdf_path).stem

    try:
        with fitz.open(pdf_path) as doc:
            page_numbers = list(range(1, doc.page_count + 1))
    except Exception as e:
        print(f"Could not open PDF {pdf_path}: {e}")
        return []

    for page_num in page_numbers:
        try:
            imgs = convert_from_path(
                pdf_path,
                dpi=300,
                first_page=page_num,
                last_page=page_num,
                poppler_path=POPLER_BIN_PATH
            )
            if imgs:
                image_path = output_dir / f"{base_name}_page_{page_num}.png"
                imgs[0].save(str(image_path), "PNG")
                image_paths.append(image_path)
        except Exception as e:
            print(f"Skipping page {page_num}: {e}")

    return image_paths


def _append_run_log(row: Dict):
    """
    Append one row to data/run_log.csv (create file with header if needed).
    """
    run_log = DATA_DIR / "run_log.csv"
    run_log.parent.mkdir(parents=True, exist_ok=True)
    header = ["date", "prefix", "filename", "pages", "started", "ended", "durationSec", "status"]
    write_header = not run_log.exists()
    with run_log.open("a", newline="", encoding="utf-8") as f:
        w = csv.DictWriter(f, fieldnames=header)
        if write_header:
            w.writeheader()
        w.writerow({k: row.get(k, "") for k in header})


def _append_audit_log(entry: Dict):
    """
    Append a JSON line to data/audit_log.jsonl
    """
    audit = DATA_DIR / "audit_log.jsonl"
    audit.parent.mkdir(parents=True, exist_ok=True)
    with audit.open("a", encoding="utf-8") as f:
        f.write(json.dumps(entry, ensure_ascii=False) + "\n")


def _write_daily_files(day_iso: str, file_rows: List[Dict]):
    """
    Write daily snapshots needed by the dashboard (still kept for compatibility):
      - files_<date>.json
      - daily_metrics_<date>.json
      - charts_<date>.json
    """
    day_dir = _date_dir(day_iso)

    # files_<date>.json
    (day_dir / f"files_{day_iso}.json").write_text(json.dumps(file_rows, indent=2), encoding="utf-8")

    # KPIs
    received = len({r["filename"] for r in file_rows})
    processed = sum(1 for r in file_rows if str(r.get("status", "")).upper() == "SUCCESS")
    failed = sum(1 for r in file_rows if str(r.get("status", "")).upper() == "FAILED")
    duplicate_count = sum(1 for r in file_rows if str(r.get("status", "")).upper() == "DUPLICATE")
    pages_total = sum(int(r.get("pages") or 0) for r in file_rows)
    durations = []
    for r in file_rows:
        try:
            durations.append(float(r.get("durationSec", 0)))
        except Exception:
            pass

    avg_pages = round(pages_total / received, 2) if received else 0.0
    avg_time = round(sum(durations) / len(durations), 2) if durations else 0.0

    daily_metrics = {
        "received": received,
        "processed": processed,
        "failed": failed,
        "duplicates": duplicate_count,
        "pages": pages_total,
        "avgPages": avg_pages,
        "avgTimeSec": avg_time,
    }
    (day_dir / f"daily_metrics_{day_iso}.json").write_text(json.dumps(daily_metrics, indent=2), encoding="utf-8")

    # Charts (status pie + duration histogram)
    edges = [0, 5, 10, 20, 30]
    counts = [0, 0, 0, 0]
    for d in durations:
        if 0 <= d < 5: counts[0] += 1
        elif d < 10: counts[1] += 1
        elif d < 20: counts[2] += 1
        else: counts[3] += 1

    charts = {
        "status_pie": {"success": processed, "failed": failed},
        "time_hist": {"bin_edges": edges, "counts": counts},
    }
    (day_dir / f"charts_{day_iso}.json").write_text(json.dumps(charts, indent=2), encoding="utf-8")


def _infer_status_from_result(result: Dict) -> str:
    """
    Consider SUCCESS if at least one key field is present.
    """
    keys_of_interest = {
        "State", "Titling State", "Transaction ID", "Process Date",
        "Taxable Selling Price/ Purchase Price", "Sales Tax",
        "Motor Vehicle Fees", "Local Use/ County Fee", "Plate Fee"
    }
    for k in keys_of_interest:
        v = (result or {}).get(k)
        if isinstance(v, str) and v.strip():
            return "SUCCESS"
    return "FAILED"


def _copy_images_to_conv(image_paths: List[Path]):
    """
    Copy generated PNGs into the dashboard conv directory so FastAPI can serve /static/conv/<file>.
    """
    for p in image_paths:
        try:
            shutil.copy2(str(p), str(DASH_CONV_DIR / p.name))
        except Exception as e:
            print(f"Copy failed for {p.name}: {e}")


# ---------- Email records helpers ----------
def _to_hms_from_iso(iso_str: str) -> str:
    """
    Parse 'YYYY-MM-DDTHH:MM:SS+05:30' to 'HH:MM:SS'. Returns '' if invalid.
    """
    if not iso_str:
        return ""
    try:
        dt = datetime.fromisoformat(iso_str)
        return dt.strftime("%H:%M:%S")
    except Exception:
        return iso_str[-8:] if len(iso_str) >= 8 else ""


def convert_date_to_iso(date_str: str) -> str:
    """
    Convert date from MM/DD/YYYY or M/D/YYYY to YYYY-MM-DD format.
    Handles both single and double-digit months/days.
    Returns empty string if conversion fails.
    """
    if not date_str:
        return ""
    try:
        date_str = date_str.strip()
        # Try MM/DD/YYYY format first (handles both 12/5/2025 and 12/05/2025)
        dt = datetime.strptime(date_str, "%m/%d/%Y")
        return dt.strftime("%Y-%m-%d")
    except Exception:
        # If it fails, return as-is (might already be in correct format or invalid)
        return date_str


# ------------- Normalization helper -------------
def normalize_fields(result: Dict, day_iso: str, filename: str = "") -> Dict:
    """
    Expand a sparse inference JSON into the canonical schema.
    Now includes the 5 extra fields:
      - Total Title & Registration Fee            (computed sum of fee fields)
      - Total Penalties                           (always "0.00")
      - Total Other Fees                           (always "0.00")
      - State Receipt Total                        (always blank "")
      - Total Sales Taxes + Reg Fee + Penalties    (Sales Tax + Total Title & Registration Fee + Total Penalties)
    """
    if result is None:
        result = {}

    # Helper to fetch with small-key-variations
    def fetch(*keys, default=""):
        for k in keys:
            if k in result and result.get(k) not in (None, ""):
                return result.get(k)
        return default

    # Canonical keys and default types
    numeric_fields = {
        "Taxable Selling Price / Purchase Price",
        "Sales Tax Credit Reciprocity",
        "Sales Tax",
        "Sales Tax Rate",
        "Total State Taxes",
        "Registration Fee",
        "Title Fee",
        "Plate Fee",
        "Lien Fee",
        "Transfer Fee",
        "Motor Vehicle Fees",
        "Local Use / County Fee",
        "Increase Fee",
        "Exchange Fee",
        "Specialty Plate Fee",
        "Additional Charge",
        # new numeric outputs:
        "Total Title & Registration Fee",
        "Total Penalties",
        "Total Other Fees",
        "Total Sales Taxes + Reg Fee + Penalties"
    }
    text_fields = {
        "Delivery Date",
        "Titling State",
        "Transaction ID",
        "Process Date",
        "State Receipt Total"  # intentionally blank text field
    }

    # Build normalized dict in the exact requested order
    normalized: Dict[str, str] = {}

    # Text fields (Delivery Date uses day_iso)
    normalized["Delivery Date"] = day_iso or ""
    
    # Titling State: fetch from result, or extract from filename (first 2 letters: PA-xxx.pdf, CT-xxx.pdf, GA-xxx.pdf)
    titling_state = fetch("Titling State", "State", default="")
    if not titling_state and filename:
        # Extract state code from filename like "PA-12345.pdf" or "CT_99999.pdf"
        match = re.match(r'^([A-Z]{2})[-_]', filename)
        if match:
            titling_state = match.group(1)
    normalized["Titling State"] = titling_state
    
    # Transaction ID: fetch from result, or extract from filename (format: PA-62458199.pdf)
    txn_id_raw = fetch("Transaction ID", "TransactionID", default="")

    # STRICT check: must be exactly 8–10 digits (no spaces, no chars)
    if isinstance(txn_id_raw, str) and re.fullmatch(r'\d{8,10}', txn_id_raw):
        txn_id = txn_id_raw
    else:
        # fallback to filename
        txn_id = ""
        if filename:
            match = re.search(r'[-_](\d+)(?:\.\w+)?$', filename)
            if match:
                txn_id = match.group(1)

    normalized["Transaction ID"] = txn_id
    
    # Process Date: convert from MM/DD/YYYY to YYYY-MM-DD
    process_date_raw = fetch("Process Date", "ProcessDate", default="")
    normalized["Process Date"] = convert_date_to_iso(process_date_raw)

    # Numeric fields - use fetch to check variants where helpful
    normalized["Taxable Selling Price / Purchase Price"] = fetch(
        "Taxable Selling Price / Purchase Price",
        "Taxable Selling Price/ Purchase Price",
        "Taxable Selling Price /Purchase Price",
        default="0.00"
    )
    normalized["Sales Tax Credit Reciprocity"] = fetch("Sales Tax Credit Reciprocity", default="0.00")
    normalized["Sales Tax"] = fetch("Sales Tax", default="0.00")
    normalized["Sales Tax Rate"] = fetch("Sales Tax Rate", default="0.00")

    # Total State Taxes = copy of Sales Tax (if Sales Tax present) else check Total State Taxes key
    total_state = fetch("Total State Taxes", default=None)
    if total_state not in (None, ""):
        normalized["Total State Taxes"] = total_state
    else:
        normalized["Total State Taxes"] = normalized["Sales Tax"] or "0.00"

    normalized["Registration Fee"] = fetch("Registration Fee", default="0.00")
    normalized["Title Fee"] = fetch("Title Fee", default="0.00")
    normalized["Plate Fee"] = fetch("Plate Fee", default="0.00")
    normalized["Lien Fee"] = fetch("Lien Fee", default="0.00")
    normalized["Transfer Fee"] = fetch("Transfer Fee", default="0.00")
    normalized["Motor Vehicle Fees"] = fetch("Motor Vehicle Fees", default="0.00")
    # Local Use variations
    normalized["Local Use / County Fee"] = fetch(
        "Local Use / County Fee",
        "Local Use/ County Fee",
        "Local Use /County Fee",
        "Local Use/County Fee",
        default="0.00"
    )
    normalized["Increase Fee"] = fetch("Increase Fee", default="0.00")
    normalized["Exchange Fee"] = fetch("Exchange Fee", default="0.00")
    normalized["Specialty Plate Fee"] = fetch("Specialty Plate Fee", default="0.00")
    normalized["Additional Charge"] = fetch("Additional Charge", default="0.00")

    # ---------- Compute the 5 extra fields ----------
    # Convert numeric-like strings to floats safely
    def to_float_safe(x):
        try:
            if x is None or x == "":
                return 0.0
            # Allow comma thousand separators
            s = str(x).replace(",", "").strip()
            return float(s)
        except Exception:
            return 0.0

    # Sum of fee-related fields for "Total Title & Registration Fee".
    fee_fields_for_total = [
        "Registration Fee", "Title Fee", "Plate Fee", "Lien Fee", "Transfer Fee",
        "Motor Vehicle Fees", "Local Use / County Fee", "Increase Fee",
        "Exchange Fee", "Specialty Plate Fee", "Additional Charge"
    ]
    total_title_reg = 0.0
    for ff in fee_fields_for_total:
        total_title_reg += to_float_safe(normalized.get(ff, "0.00"))

    # Format back to string with two decimal places
    normalized["Total Title & Registration Fee"] = f"{total_title_reg:.2f}"

    # Total Penalties and Total Other Fees are required to be always 0.00
    normalized["Total Penalties"] = "0.00"
    normalized["Total Other Fees"] = "0.00"

    # State Receipt Total must remain blank
    normalized["State Receipt Total"] = ""

    # Total Sales Taxes + Reg Fee + Penalties = Sales Tax + Total Title & Registration Fee + Total Penalties
    sales_tax_val = to_float_safe(normalized.get("Sales Tax", "0.00"))
    penalties_val = to_float_safe(normalized.get("Total Penalties", "0.00"))
    total_sales_plus_reg_plus_penalties = sales_tax_val + total_title_reg + penalties_val
    normalized["Total Sales Taxes + Reg Fee + Penalties"] = f"{total_sales_plus_reg_plus_penalties:.2f}"

    # Ensure everything is a string (DB expects JSON)
    for k in list(normalized.keys()):
        v = normalized[k]
        if v is None:
            normalized[k] = "0.00" if k in numeric_fields else ""
        else:
            normalized[k] = str(v)

    return normalized


# ---------------- Main worker ----------------
def run_inference_on_all_pdfs():
    """
    Download, process, and save output for today's PDFs + write dashboard logs/snapshots.
    Also writes summary + fields JSON into the SQL database.
    """
    s3_prefix = get_today_s3_prefix()
    day_iso = _today_iso()
    run_id = datetime.now(TZ).strftime("%m%d%Y-%H%M%S")
    run_started_ts = datetime.now(TZ)

    # --- Decide Excel save locations up front ---
    day_dir = _date_dir(day_iso)  # ...\Backend_db\data\<YYYY-MM-DD>\  (ensured to exist)
    today_str = datetime.now(TZ).strftime("%m%d%Y")
    excel_filename = f"ARDEM_VITU_{ACTIVE_PREFIX}_{today_str}_Output.xlsx"
    excel_output_path = day_dir / excel_filename         # primary (FastAPI searches here first)
    excel_secondary_path = DATA_DIR / excel_filename     # optional convenience copy (2nd search location)

    print(f"\nLooking for PDFs in S3 path: {s3_prefix}")
    file_rows: List[Dict] = []
    filename_to_s3key: Dict[str, str] = {}

    try:
        # Use signature-agnostic wrapper
        pdf_keys = list_pdfs_safe(s3_prefix)
        if not pdf_keys:
            msg_subject = "No PDFs Found"
            print(f"{msg_subject} at {s3_prefix}")
            send_notification(
                msg_subject,
                records=[],
                note=f"No PDFs found at this S3 location: {s3_prefix}",
                totals={"received": 0, "success": 0}
            )
            _append_audit_log({
                "date": day_iso,
                "run_id": run_id,
                "started": run_started_ts.isoformat(timespec="seconds"),
                "ended": datetime.now(TZ).isoformat(timespec="seconds"),
                "received": 0,
                "processed": 0,
                "failed": 0,
                "duplicates": 0
            })
            return

        print(f"Found {len(pdf_keys)} PDFs.")

        # -----------------------
        # Prepare historical seen maps (strictly before current day)
        # -----------------------
        seen_filenames = set()
        seen_pdf_md5 = set()
        seen_txn_md5 = set()
        # maps to recover original filename for a given md5 or filename key
        pdf_md5_to_filename = {}
        filename_norm_to_original = {}

        try:
            day_before = (datetime.fromisoformat(day_iso).date() - timedelta(days=1)).isoformat()
            historical_rows = db_list_files("1900-01-01", day_before, ACTIVE_PREFIX) if db_list_files else []
        except Exception:
            historical_rows = []

        for h in (historical_rows or []):
            fn = str(h.get("filename") or "").strip()
            fn_norm = fn.lower()
            if fn:
                seen_filenames.add(fn_norm)
                filename_norm_to_original[fn_norm] = fn
            p = (h.get("pdf_md5") or h.get("pdfMd5") or "") or ""
            p_norm = str(p).strip().lower()
            if p_norm:
                seen_pdf_md5.add(p_norm)
                # record a representative filename for this md5 (may be overwritten but that's fine)
                pdf_md5_to_filename[p_norm] = h.get("filename") or pdf_md5_to_filename.get(p_norm)
            t = (h.get("txn_md5") or h.get("txnMd5") or "") or ""
            t_norm = str(t).strip().lower()
            if t_norm:
                seen_txn_md5.add(t_norm)

        # -----------------------
        # Process files in deterministic order (the order returned by list_pdfs_safe)
        # -----------------------
        for pdf_key in pdf_keys:
            filename = os.path.basename(pdf_key)
            filename_to_s3key[filename] = pdf_key
            base = Path(filename).stem
            started_ts = datetime.now(TZ)
            pages_count = 0

            # -------------------------
            # SKIP if this exact filename already exists in DB as SUCCESS for the same date.
            # This prevents previously-successful rows from being reprocessed and turned into DUPLICATE.
            # -------------------------
            try:
                if file_exists_success(day_iso, filename):
                    print(f"[SKIP] {filename} already SUCCESS in DB for {day_iso}; skipping reprocessing.")
                    # Do not overwrite existing row; do not write a new row for this run.
                    continue
            except Exception as e:
                # If we cannot check the DB for any reason, continue processing (defensive).
                print(f"[WARN] Could not check DB for existing SUCCESS file {filename}: {e}")

            print(f"\n[DOWNLOAD] Downloading {filename}...")
            # Create a unique temporary file with process ID and timestamp to avoid conflicts
            # when running multiple schedulers in parallel
            process_id = os.getpid()
            timestamp = int(time.time() * 1000000)  # microsecond precision
            temp_dir = Path(tempfile.gettempdir()) / f"vitu_pa_{process_id}"
            temp_dir.mkdir(parents=True, exist_ok=True)
            
            temp_filename = f"{timestamp}_{filename}"
            tmp_pdf_path = temp_dir / temp_filename
            
            try:
                # Download and compute pdf_md5 (download_pdf_safe now returns md5)
                md5 = download_pdf_safe(pdf_key, str(tmp_pdf_path))
            except Exception as e:
                print(f"Download failed for {filename}: {e}")
                # Clean up temp file if it exists
                try:
                    if tmp_pdf_path.exists():
                        tmp_pdf_path.unlink()
                except:
                    pass
                ended_ts = datetime.now(TZ)
                duration = (ended_ts - started_ts).total_seconds()
                row = {
                    "date": day_iso,
                    "prefix": ACTIVE_PREFIX,
                    "filename": filename,
                    "pages": 0,
                    "started": started_ts.isoformat(timespec="seconds"),
                    "ended": ended_ts.isoformat(timespec="seconds"),
                    "durationSec": round(duration, 2),
                    "status": "FAILED",
                    "pdf_md5": None,
                    "txn_md5": None,
                    "rejection_reason": f"Download failed: {e}",
                    "rejected_for": None,
                    "run_id": run_id
                }
                try:
                    upsert_file(row, s3_key=pdf_key, run_id=run_id)
                    upsert_output(day_iso, filename, normalize_fields({}, day_iso, filename))
                except Exception as db_err:
                    print(f"[WARN] DB upsert failed for {filename}: {db_err}")
                _append_run_log(row)
                file_rows.append(row)
                continue

            # Persist a copy for the viewer at data/<date>/<filename>.pdf
            persist_path = day_dir / filename
            try:
                shutil.copy2(str(tmp_pdf_path), persist_path)
                print(f"Saved PDF for viewer: {persist_path}")
            except Exception as e:
                print(f"Could not save PDF to {persist_path}: {e}")

            # -------------------------
            # STAGE 1/2 duplicate check: filename and pdf_md5 (before conversion/inference)
            # -------------------------
            is_duplicate = False
            dup_reason = None
            dup_for = None
            dup_of_filename = None

            fn_norm = filename.strip().lower()
            md5_norm = (md5 or "").strip().lower()

            # check filename historical/seen
            if fn_norm and fn_norm in seen_filenames:
                is_duplicate = True
                dup_for = "filename"
                dup_of_filename = filename_norm_to_original.get(fn_norm) or filename
                dup_reason = f"Duplicate filename: {dup_of_filename}"
            # check pdf_md5 historical/seen
            elif md5_norm and md5_norm in seen_pdf_md5:
                is_duplicate = True
                dup_for = "pdf_md5"
                dup_of_filename = pdf_md5_to_filename.get(md5_norm)
                if dup_of_filename:
                    dup_reason = f"Duplicate of {dup_of_filename} (same pdf_md5)"
                else:
                    # try DB lookup to recover filename
                    try:
                        found = find_by_md5(md5 or None, None)
                        if found and len(found) > 0:
                            dup_of_filename = found[0].get("filename")
                            dup_reason = f"Duplicate of {dup_of_filename} (same pdf_md5)"
                        else:
                            dup_reason = f"Duplicate of existing file (same pdf_md5)"
                    except Exception:
                        dup_reason = f"Duplicate of existing file (same pdf_md5)"
            else:
                # safety: call find_by_md5 to detect pdf_md5 matches in DB if available
                try:
                    # find_by_md5 may return rows that match pdf_md5 or txn_md5
                    found = find_by_md5(md5 or None, None)
                    if found:
                        # check for same pdf_md5 match
                        for d in found:
                            if d.get("pdf_md5") and md5 and d.get("pdf_md5") == md5:
                                is_duplicate = True
                                dup_for = "pdf_md5"
                                dup_of_filename = d.get("filename")
                                dup_reason = f"Duplicate of {dup_of_filename} (same pdf_md5)" if dup_of_filename else "Duplicate (same pdf_md5)"
                                break
                except Exception:
                    # ignore errors here; continue to conversion/inference if not duplicate
                    pass

            if is_duplicate:
                # mark as DUPLICATE and skip conversion/inference
                ended_ts = datetime.now(TZ)
                duration = (ended_ts - started_ts).total_seconds()
                row = {
                    "date": day_iso,
                        "prefix": ACTIVE_PREFIX,
                        "filename": filename,
                        "pages": pages_count,
                        "started": started_ts.isoformat(timespec="seconds"),
                        "ended": ended_ts.isoformat(timespec="seconds"),
                        "durationSec": round(duration, 2),
                        "status": "DUPLICATE",
                        "pdf_md5": md5 or "",
                        "txn_md5": "",
                    "rejection_reason": dup_reason,
                    "rejected_for": dup_for,
                    "run_id": run_id
                }

                # Persist duplicate marker to DB
                try:
                    upsert_file(row, s3_key=pdf_key, run_id=run_id)
                    # Still write a minimal normalized output for consistency
                    norm_min = normalize_fields({}, day_iso, filename)
                    norm_min["_pdf_md5"] = md5 or ""
                    if dup_of_filename:
                        norm_min["_duplicate_of"] = dup_of_filename
                    upsert_output(day_iso, filename, norm_min)
                    
                    # Also upsert ProcessData row indicating DUPLICATE (no suspect info)
                    if USE_DB:
                        # try to get GUID (retry a few times)
                        guid = None
                        tries = 0
                        while tries < 3:
                            guid = get_guid_by_date_filename(day_iso, filename)
                            if guid:
                                break
                            tries += 1
                            time.sleep(0.25)
                        if guid:
                            # For duplicates, treat as "no suspect fields" for QCCorrectedData purposes:
                            # store ExtractedData in QCCorrectedData so downloads can use it directly.
                            now_iso = datetime.now().isoformat(timespec="seconds")
                            proc_row = {
                                "GUID": guid,
                                "FileName": filename,
                                "State_Name": ACTIVE_PREFIX,
                                "DataExtractedDateTime": now_iso,
                                "SuspectCheckingDateTime": now_iso,
                                "IsDataExtracted": "Yes",
                                # No suspect fields for DUPLICATE marker itself
                                "IsSuspectValidated": "Yes",
                                "IsQCCompleted": "Yes",
                                "IsDelivered": False,
                                "AWSS3Path": pdf_key,
                                "ExtractedData": json.dumps(norm_min, ensure_ascii=False),
                                "IsSuspectData": None,
                                "SuspectExtractData": None,
                                # Store the full JSON into QCCorrectedData (so audit/download can prefer QCCorrectedData)
                                "QCCorrectedData": json.dumps(norm_min, ensure_ascii=False),
                                "QCCheckingDateTime": now_iso
                            }
                            upsert_processdata(proc_row)
                        else:
                            # file-mode fallback
                            fallback = {
                                "ExtractedData": norm_min,
                                "IsSuspectData": None,
                                "SuspectExtractData": None,
                                "SuspectCheckingDateTime": datetime.now().isoformat(timespec="seconds")
                            }
                            (day_dir / f"{Path(filename).stem}_suspects.json").write_text(json.dumps(fallback, ensure_ascii=False), encoding="utf-8")
                except Exception as db_err:
                    print(f"[WARN] DB upsert failed for {filename}: {db_err}")

                # update seen maps so later files in same run see this filename/md5
                if fn_norm:
                    seen_filenames.add(fn_norm)
                    filename_norm_to_original[fn_norm] = filename
                if md5_norm:
                    seen_pdf_md5.add(md5_norm)
                    if dup_of_filename:
                        pdf_md5_to_filename[md5_norm] = dup_of_filename
                    else:
                        pdf_md5_to_filename[md5_norm] = filename

                _append_run_log(row)
                file_rows.append(row)
                # skip to next file
                continue

            # Not duplicate by filename/pdf_md5 so proceed to convert + infer
            print("Converting all pages to images...")
            image_paths = convert_all_pages_to_images(str(tmp_pdf_path), CONVERTED_IMG_DIR)
            pages_count = len(image_paths)

            if not image_paths:
                print(f"Skipping {filename}, no images generated.")
                ended_ts = datetime.now(TZ)
                duration = (ended_ts - started_ts).total_seconds()
                row = {
                    "date": day_iso,
                    "prefix": ACTIVE_PREFIX,
                    "filename": filename,
                    "pages": pages_count,
                    "started": started_ts.isoformat(timespec="seconds"),
                    "ended": ended_ts.isoformat(timespec="seconds"),
                    "durationSec": round(duration, 2),
                    "status": "FAILED",
                    "pdf_md5": md5 or "",
                    "txn_md5": "",
                    "run_id": run_id
                }

                try:
                    upsert_file(row, s3_key=pdf_key, run_id=run_id)
                    norm = normalize_fields({}, day_iso, filename)  # backfill canonical defaults for failed rows
                    norm["_pdf_md5"] = md5 or ""
                    norm["_txn_md5"] = ""
                    upsert_output(day_iso, filename, norm)

                    # Persist ProcessData row (failed) as well
                    if USE_DB:
                        guid = None
                        tries = 0
                        while tries < 3:
                            guid = get_guid_by_date_filename(day_iso, filename)
                            if guid:
                                break
                            tries += 1
                            time.sleep(0.25)
                        if guid:
                            now_iso = datetime.now().isoformat(timespec="seconds")
                            # For failed rows with no suspect checks (no extracted values), we store ExtractedData into QCCorrectedData
                            # so audit/download can still present the same fields.
                            proc_row = {
                                "GUID": guid,
                                "FileName": filename,
                                "State_Name": ACTIVE_PREFIX,
                                "DataExtractedDateTime": now_iso,
                                "SuspectCheckingDateTime": now_iso,
                                "IsDataExtracted": "Yes",
                                "IsSuspectValidated": "Yes",
                                "IsQCCompleted": "Yes",
                                    "IsDelivered": False,
                                    "AWSS3Path": pdf_key,
                                    "ExtractedData": json.dumps(norm, ensure_ascii=False),
                                    "IsSuspectData": None,
                                "SuspectExtractData": None,
                                "QCCorrectedData": json.dumps(norm, ensure_ascii=False),
                                "QCCheckingDateTime": now_iso
                            }
                            upsert_processdata(proc_row)
                        else:
                            fallback = {
                                "ExtractedData": norm,
                                "IsSuspectData": None,
                                "SuspectExtractData": None,
                                "SuspectCheckingDateTime": datetime.now().isoformat(timespec="seconds")
                            }
                            (day_dir / f"{Path(filename).stem}_suspects.json").write_text(json.dumps(fallback, ensure_ascii=False), encoding="utf-8")

                except Exception as db_err:
                    print(f"[WARN] DB upsert failed for {filename}: {db_err}")

                _append_run_log(row)
                file_rows.append(row)
                continue

            print(f"Running inference on {len(image_paths)} pages...")
            result = infer_multiple_images([str(p) for p in image_paths])

            # Normalize result into canonical schema (now includes the 5 new fields)
            norm = normalize_fields(result, day_iso, filename)
            
            # Attach md5s into the normalized JSON artifact for traceability
            norm["_pdf_md5"] = md5 or ""
            norm["_txn_md5"] = result.get("_txn_md5", "") or ""
            txn_md5_val = norm["_txn_md5"]
            txn_md5_norm = (txn_md5_val or "").strip().lower()

            # Run suspect checks on normalized output with PA state
            suspect_map, suspect_values = run_suspect_checks(norm, "PA")

            # STAGE 3 duplicate check: txn_md5 (after inference)
            dup_after_txn = False
            dup_after_reason = None
            dup_after_for = None
            dup_after_of_filename = None

            if txn_md5_norm and txn_md5_norm in seen_txn_md5:
                dup_after_txn = True
                dup_after_for = "txn_md5"
                dup_after_reason = f"Duplicate Transaction ID found in this run"
            else:
                # safety: check DB via find_by_md5 for txn_md5
                try:
                    found2 = find_by_md5(None, txn_md5_val or None)
                    if found2:
                        for d in found2:
                            if d.get("txn_md5") and txn_md5_val and d.get("txn_md5") == txn_md5_val:
                                dup_after_txn = True
                                dup_after_for = "txn_md5"
                                dup_after_of_filename = d.get("filename")
                                dup_after_reason = f"Duplicate Transaction ID with {dup_after_of_filename}" if dup_after_of_filename else "Duplicate Transaction ID"
                                break
                except Exception:
                    pass

                # Determine final status: DUPLICATE if txn-md5 duplicate, else inference-derived status
                if dup_after_txn:
                    status = "DUPLICATE"
                    rejection_reason = dup_after_reason
                    rejected_for = dup_after_for
                else:
                    status = _infer_status_from_result(result)
                    rejection_reason = None
                    rejected_for = None

                ended_ts = datetime.now(TZ)
                duration = (ended_ts - started_ts).total_seconds()
                row = {
                    "date": day_iso,
                    "prefix": ACTIVE_PREFIX,
                    "filename": filename,
                    "pages": pages_count,
                    "started": started_ts.isoformat(timespec="seconds"),
                    "ended": ended_ts.isoformat(timespec="seconds"),
                    "durationSec": round(duration, 2),
                    "status": status,
                    "pdf_md5": md5 or "",
                    "txn_md5": txn_md5_val or "",
                    "rejection_reason": rejection_reason,
                    "rejected_for": rejected_for,
                    "run_id": run_id
                }

                # --- write to DB (summary + normalized fields JSON) ---
                try:
                    upsert_file(row, s3_key=pdf_key, run_id=run_id)
                    # attach duplicate-of hint into norm if available
                    if dup_after_of_filename:
                        norm["_duplicate_of"] = dup_after_of_filename
                    upsert_output(day_iso, filename, norm)

                    # Persist ProcessData row including suspect info (if any)
                    if USE_DB:
                        guid = None
                        tries = 0
                        while tries < 3:
                            guid = get_guid_by_date_filename(day_iso, filename)
                            if guid:
                                break
                            tries += 1
                            time.sleep(0.25)
                        if guid:
                            now_iso = datetime.now().isoformat(timespec="seconds")

                            # If there are suspect fields, store IsSuspectData/SuspectExtractData and leave QCCorrectedData empty.
                            # If there are no suspect fields, store the full ExtractedData in QCCorrectedData and mark QC completed.
                            if suspect_map:
                                proc_row = {
                                    "GUID": guid,
                                    "FileName": filename,
                                    "State_Name": ACTIVE_PREFIX,
                                    "DataExtractedDateTime": now_iso,
                                    "SuspectCheckingDateTime": now_iso,
                                    "IsDataExtracted": "Yes",
                                    "IsSuspectValidated": "No",
                                    "IsQCCompleted": "No",
                                    "IsDelivered": False,
                                    "AWSS3Path": pdf_key,
                                    "ExtractedData": json.dumps(norm, ensure_ascii=False),
                                    "IsSuspectData": json.dumps(suspect_map, ensure_ascii=False),
                                    "SuspectExtractData": json.dumps(suspect_values, ensure_ascii=False),
                                    "QCCorrectedData": None,
                                    "QCCheckingDateTime": None
                                }
                            else:
                                # No suspects -> store ExtractedData into QCCorrectedData and mark QC as completed.
                                proc_row = {
                                    "GUID": guid,
                                    "FileName": filename,
                                    "State_Name": ACTIVE_PREFIX,
                                    "DataExtractedDateTime": now_iso,
                                    "SuspectCheckingDateTime": now_iso,
                                    "IsDataExtracted": "Yes",
                                    "IsSuspectValidated": "Yes",
                                    "IsQCCompleted": "Yes",
                                    "IsDelivered": False,
                                    "AWSS3Path": pdf_key,
                                    "ExtractedData": json.dumps(norm, ensure_ascii=False),
                                    "IsSuspectData": None,
                                    "SuspectExtractData": None,
                                    "QCCorrectedData": json.dumps(norm, ensure_ascii=False),
                                    "QCCheckingDateTime": now_iso
                                }
                            upsert_processdata(proc_row)
                        else:
                            # fallback: write suspects JSON to file so UI can still read it in file-mode
                            fallback = {
                                "ExtractedData": norm,
                                "IsSuspectData": suspect_map if suspect_map else None,
                                "SuspectExtractData": suspect_values if suspect_values else None,
                                "SuspectCheckingDateTime": datetime.now().isoformat(timespec="seconds")
                            }
                            (day_dir / f"{Path(filename).stem}_suspects.json").write_text(json.dumps(fallback, ensure_ascii=False), encoding="utf-8")
                    else:
                        # If not using DB at all, write suspects JSON to data dir for UI
                        fallback = {
                            "ExtractedData": norm,
                            "IsSuspectData": suspect_map if suspect_map else None,
                            "SuspectExtractData": suspect_values if suspect_values else None,
                            "SuspectCheckingDateTime": datetime.now().isoformat(timespec="seconds")
                        }
                        (day_dir / f"{Path(filename).stem}_suspects.json").write_text(json.dumps(fallback, ensure_ascii=False), encoding="utf-8")

                except Exception as db_err:
                    print(f"[WARN] DB upsert failed for {filename}: {db_err}")

                # update seen maps (so later files in same run are checked against these)
                if fn_norm:
                    seen_filenames.add(fn_norm)
                    filename_norm_to_original[fn_norm] = filename
                if md5_norm:
                    seen_pdf_md5.add(md5_norm)
                    # map md5 to filename (if not already present)
                    if md5_norm not in pdf_md5_to_filename:
                        pdf_md5_to_filename[md5_norm] = filename
                if txn_md5_norm:
                    seen_txn_md5.add(txn_md5_norm)

                _append_run_log(row)
                file_rows.append(row)
                
                # Clean up temporary file
                try:
                    if tmp_pdf_path.exists():
                        tmp_pdf_path.unlink()
                        print(f"Cleaned up temporary file: {tmp_pdf_path}")
                except Exception as cleanup_err:
                    print(f"[WARN] Could not clean up temp file {tmp_pdf_path}: {cleanup_err}")

        # Clean up any remaining temporary directories
        try:
            process_id = os.getpid()
            temp_dir = Path(tempfile.gettempdir()) / f"vitu_pa_{process_id}"
            if temp_dir.exists():
                shutil.rmtree(temp_dir, ignore_errors=True)
                print(f"Cleaned up temp directory: {temp_dir}")
        except Exception:
            pass

        # After all files: write daily snapshot JSONs (for dashboard compatibility)
        _write_daily_files(day_iso, file_rows)

        # (Optional) also place a convenience copy in data\ so /api/audit/download can find it there too
        try:
            if excel_output_path.exists():
                shutil.copy2(str(excel_output_path), str(excel_secondary_path))
                print(f"Copied workbook to {excel_secondary_path}")
        except Exception as e:
            print(f"[WARN] Could not copy Excel to {excel_secondary_path}: {e}")

        # Audit line
        run_ended_ts = datetime.now(TZ)
        duplicates_count = sum(1 for r in file_rows if str(r.get("status", "")).upper() == "DUPLICATE")
        _append_audit_log({
            "date": day_iso,
            "run_id": run_id,
            "started": run_started_ts.isoformat(timespec="seconds"),
            "ended": run_ended_ts.isoformat(timespec="seconds"),
            "received": len(pdf_keys),
            "processed": sum(1 for r in file_rows if r["status"] == "SUCCESS"),
            "failed": sum(1 for r in file_rows if r["status"] == "FAILED"),
            "duplicates": duplicates_count
        })

        # -------- Build email records (SUCCESS items only) --------
        email_records: List[Dict] = []
        for r in file_rows:
            if str(r.get("status", "")).upper() != "SUCCESS":
                continue
            fname = r.get("filename", "")
            s3_key = filename_to_s3key.get(fname, "")
            email_records.append({
                "filename": fname,
                "processed_time": r.get("ended", "")[-8:],  # HH:MM:SS
                "s3_location": s3_key,  # VITU/YYYY/MONTH/MM.DD_Files/PA_MM.DD/...
            })

        subject = "Task Complete" if email_records else "Task Complete (0 Success)"
        send_notification(
            subject,
            records=email_records,
            totals={"received": len(pdf_keys), "success": len(email_records)}
        )

        print(f"\nAll data saved to: {excel_output_path}")

    except Exception as e:
        error_msg = f"Unexpected error during processing: {str(e)}"
        print(error_msg)
        # Safe string conversion for email - avoid unicode issues in NSSM service
        error_detail = str(e).encode('ascii', 'ignore').decode('ascii') if e else "Unknown error"
        send_notification(
            "Task Failed",
            records=[],
            note=f"Error while processing.\nS3 path: {get_today_s3_prefix()}\nDetails: {error_detail}",
            totals={"received": 0, "success": 0}
        )


# ------------- Day check wrapper -------------
def run_pipeline_with_day_check():
    """
    Wrapper function that checks the day of the week before running the pipeline.
    Skips execution on Sundays (6) and Mondays (0) since there are no files from weekend.
    """
    today = datetime.now(TZ)
    day_of_week = today.weekday()  # Monday=0, Sunday=6
    day_name = today.strftime("%A")
    
    if day_of_week == 6:  # Sunday
        print(f"[SKIP] Skipping execution on {day_name} - No files expected from Saturday")
        return
    elif day_of_week == 0:  # Monday
        print(f"[SKIP] Skipping execution on {day_name} - No files expected from Sunday")
        return
    else:
        print(f"[RUN] Running PA pipeline on {day_name}")
        run_inference_on_all_pdfs()

def run_pipeline():
    print(f"\n[START] {ACTIVE_PREFIX} pipeline started")
    try:
        run_pipeline_with_day_check()
    except Exception as e:
        print(f"[ERROR] {ACTIVE_PREFIX} pipeline failed:", e)
    print(f"[END] {ACTIVE_PREFIX} pipeline finished\n")


# ------------- Scheduler (every 10 minutes from 9:00 AM IST) -------------
def schedule_next_minute_job():
    """Run PA checks every 10 minutes from 09:00 through 23:50 IST.

    ``run_pipeline`` catches an individual-run failure, therefore an ARDEM/GPU,
    network, S3, or database outage does not stop APScheduler.  The next
    ten-minute slot automatically retries the full discovery-and-processing
    flow.  Overlapping runs are prevented to protect duplicate checks and DB
    writes when an external service is slow.
    """
    scheduler = BlockingScheduler(timezone=TZ)
    scheduler.add_job(
        func=run_pipeline,
        trigger="cron",
        hour="9-23",
        minute="*/10",
        timezone=TZ,
        id="pa_pipeline_every_10_minutes",
        replace_existing=True,
        max_instances=1,
        coalesce=True,
        misfire_grace_time=600,
    )
    print("[SCHEDULE] PA pipeline: every 10 minutes from 09:00 to 23:50 IST")
    print("[INFO] Scheduler starting... Press Ctrl+C to stop")

    try:
        scheduler.start()

    except KeyboardInterrupt:
        print("\n[STOP] Scheduler stopped by user")
        scheduler.shutdown()

    except Exception as e:
        print(f"[ERROR] Scheduler error: {e}")
        scheduler.shutdown()



# ------------- Entrypoint -------------
if __name__ == "__main__":
    import sys

    if len(sys.argv) > 1 and sys.argv[1].lower() == "run":
        print("[MANUAL] Running PA pipeline immediately...")
        run_pipeline()
    else:
        schedule_next_minute_job()
