web/parse_worker.py - main - Paff/declarant

web/parse_worker.py
273 строки10 KB
1pash1985-gif
feat: populate certificate invoice number field from master invoice mapping
10 авг 2026, 09:14
10 авг 2026, 09:14d8339de
"""Background worker for AI parsing of a single report file.

Runs outside the gunicorn worker so long parses do not block the web API
or exceed proxy/browser timeouts. Started by reports_api.reparse_file().
"""
from __future__ import annotations

import json
import os
import re
import sys
import traceback
from dataclasses import asdict
from datetime import datetime
from pathlib import Path
from typing import Any

# Make project root and web/ importable when running as `python -m web.parse_worker`.
_PROJECT_ROOT = str(Path(__file__).resolve().parent.parent)
_WEB_DIR = str(Path(__file__).resolve().parent)
for _p in (_PROJECT_ROOT, _WEB_DIR):
    if _p not in sys.path:
        sys.path.insert(0, _p)

from reports_api import _load_meta, _parse_file_with_ai, _safe_report_paths  # noqa: E402


def _certificate_data_to_dict(cd: Any, hints: dict[str, str] | None) -> dict[str, Any]:
    """Convert cert_parser.CertificateData to the unified JSON schema.

    The unified schema contains all CertificateData fields (so /cert_parse
    can serve it directly) plus the AI-parser-compatible `cert_items` and
    `awb_groups` shapes the main report UI expects.

    For Ecuador/EUR.1 certificates the raw `awb_groups[*].certs` only hold
    the certificate number; we enrich each certificate from the
    `eur1_by_master` mapping and the per-item `master_invoice` so the
    comparison panel shows country, invoice number, boxes and stems.
    """
    result: dict[str, Any] = asdict(cd)

    # Group parsed items by master invoice for quick lookup.
    items_by_master: dict[str, list[dict[str, Any]]] = {}
    for it in result.get('items', []):
        mi = str(it.get('master_invoice') or it.get('invoice_nr') or '').strip()
        if mi:
            items_by_master.setdefault(mi, []).append(it)

    eur1_by_master = result.get('eur1_by_master') or {}

    # Build cert_items from the EUR1 <-> master invoice mapping so every
    # certificate gets its country, invoice number, and aggregated boxes/stems.
    cert_items: list[dict[str, Any]] = []
    cert_by_number: dict[str, dict[str, Any]] = {}
    for idx, (master_invoice, cert_number) in enumerate(eur1_by_master.items()):
        items = items_by_master.get(master_invoice, [])
        total_boxes = sum(
            _to_number(it.get('boxes')) for it in items
        )
        total_stems = sum(
            _to_number(it.get('stems')) for it in items
        )
        cert_item = {
            'cert_index': idx,
            'cert_type': 'EUR1',
            'country': result.get('country', ''),
            'cert_numbers': [cert_number] if cert_number else [],
            'cert_number': cert_number,
            'awb': '',
            'exporter': result.get('exporter', ''),
            'consignee': result.get('consignee', ''),
            'date': result.get('date', ''),
            'invoice_nr': master_invoice,
            'master_invoice': master_invoice,
            'master_invoices': [master_invoice] if master_invoice else [],
            'items': items,
            'total_stems': total_stems if total_stems else None,
            'total_boxes': total_boxes if total_boxes else None,
        }
        cert_items.append(cert_item)
        if cert_number:
            cert_by_number[cert_number] = cert_item

    # Assign AWB to each cert_item from the original awb_groups structure.
    awb_groups = result.get('awb_groups', [])
    for group in awb_groups:
        awb = group.get('awb', '')
        for cert in group.get('certs', []):
            cert_number = cert.get('cert_number', '')
            if cert_number in cert_by_number:
                cert_by_number[cert_number]['awb'] = awb

    # Replace the sparse certs inside awb_groups with the enriched cert_items
    # so the report UI's regex panel shows the same data.
    enriched_groups: list[dict[str, Any]] = []
    for group in awb_groups:
        awb = group.get('awb', '')
        group_certs = [
            cert_by_number[c.get('cert_number', '')]
            for c in group.get('certs', [])
            if c.get('cert_number', '') in cert_by_number
        ]
        if group_certs or awb:
            enriched_groups.append({
                **group,
                'certs': group_certs,
            })
    # If some certificates have no AWB group, keep them as a fallback group.
    grouped_certs = {
        ci['cert_number']
        for g in enriched_groups
        for ci in g.get('certs', [])
    }
    ungrouped = [ci for ci in cert_items if ci['cert_number'] not in grouped_certs]
    if ungrouped:
        enriched_groups.append({'awb': '', 'exporter': '', 'certs': ungrouped})
    result['awb_groups'] = enriched_groups

    result['cert_items'] = cert_items

    # Top-level list of all certificate numbers (AI-parser convention).
    result['cert_numbers'] = [
        ci['cert_number'] for ci in cert_items if ci.get('cert_number')
    ]

    # Ensure master_invoices is populated from items if empty.
    if not result.get('master_invoices'):
        masters: list[str] = []
        seen: set[str] = set()
        for ci in cert_items:
            for it in ci.get('items', []):
                mi = it.get('master_invoice') or it.get('invoice_nr')
                if mi:
                    key = str(mi).strip().lower()
                    if key and key not in seen:
                        seen.add(key)
                        masters.append(str(mi))
        result['master_invoices'] = masters

    # Provenance markers used by the report UI and ai_bridge.
    result['_doc_type'] = 'certificate'
    result['_parser'] = 'cert_parser'
    if hints:
        for k in ('_source_filename', '_folder_hint', '_user_hint'):
            v = hints.get(k)
            if v:
                result[k] = str(v)
    return result


def _to_number(value: Any) -> float:
    """Safely convert a value to a float; non-numeric values become 0."""
    if value is None:
        return 0.0
    if isinstance(value, (int, float)):
        return float(value)
    try:
        return float(str(value).replace(',', '.').strip())
    except Exception:
        return 0.0


def _parse_certificate_with_regex(
    fpath: Path,
    hints: dict[str, str] | None,
) -> dict[str, Any]:
    """Run the local regex/OCR certificate parser in the background worker."""
    try:
        import cert_parser
    except Exception as e:
        return {'error': f'cert_parser недоступен: {e}', '_doc_type': 'certificate'}
    try:
        cd = cert_parser.parse_certificate(str(fpath))
    except Exception as e:
        tb = traceback.format_exc()
        return {'error': f'cert_parser: {e}\n{tb}', '_doc_type': 'certificate'}
    return _certificate_data_to_dict(cd, hints)


def _extract_source_text_for_corrections(fpath: Path, parsed_dir: Path) -> None:
    """Save extracted text next to the JSON for corrections context."""
    try:
        from ai_parser import extract_text_from_excel, extract_text_from_pdf
        ext = fpath.suffix.lower()
        source_text = ''
        if ext == '.pdf':
            source_text = extract_text_from_pdf(str(fpath))
        elif ext in ('.xls', '.xlsx', '.xlsm'):
            source_text = extract_text_from_excel(str(fpath))
        if source_text:
            (parsed_dir / (fpath.stem + '.txt')).write_text(
                source_text, encoding='utf-8',
            )
    except Exception:
        pass


def main() -> int:
    if len(sys.argv) < 3:
        print('Usage: python -m web.parse_worker <report_id> <file_name> [marking_hint]', file=sys.stderr)
        return 1

    report_id = sys.argv[1]
    name = sys.argv[2]
    override_mark = (sys.argv[3] if len(sys.argv) > 3 else '').strip()

    got = _safe_report_paths(report_id)
    if not got:
        print(f'Report not found: {report_id}', file=sys.stderr)
        return 1
    paths, _date, _car = got

    fpath = paths['sources'] / name
    if not fpath.is_file():
        print(f'File not found: {fpath}', file=sys.stderr)
        return 1

    m = re.match(r'([a-z]+)_', name)
    doc_type = m.group(1) if m else 'other'

    meta = _load_meta(paths)
    hints = (meta.get('hints') or {}).get(name) or {}
    hint_for_ai: dict[str, str] = {}
    if hints.get('orig_name'):
        hint_for_ai['_source_filename'] = hints['orig_name']
    if hints.get('folder_hint'):
        hint_for_ai['_folder_hint'] = hints['folder_hint']
    if hints.get('user_hint'):
        hint_for_ai['_user_hint'] = hints['user_hint']
    if override_mark:
        hint_for_ai['_user_hint'] = override_mark

    parsed_dir = paths['parsed']
    parsed_dir.mkdir(parents=True, exist_ok=True)
    lock_path = parsed_dir / (fpath.stem + '.parsing')
    out_path = parsed_dir / (fpath.stem + '.json')

    _extract_source_text_for_corrections(fpath, parsed_dir)

    lock_path.write_text(
        json.dumps({'started': datetime.now().isoformat()}, ensure_ascii=False),
        encoding='utf-8',
    )
    parsed: dict = {}
    try:
        if doc_type == 'certificate':
            parsed = _parse_certificate_with_regex(fpath, hint_for_ai)
        else:
            parsed = _parse_file_with_ai(fpath, doc_type, hints=hint_for_ai)
    except Exception as e:
        tb = traceback.format_exc()
        print(f'Parse worker error for {name}: {e}\n{tb}', file=sys.stderr)
        parsed = {'error': f'AI-парсинг: {e}'}
    finally:
        try:
            lock_path.unlink(missing_ok=True)
        except Exception:
            pass

    try:
        out_path.write_text(
            json.dumps(parsed, ensure_ascii=False, indent=2),
            encoding='utf-8',
        )
    except Exception as e:
        print(f'Failed to write parsed JSON for {name}: {e}', file=sys.stderr)
        return 1

    return 0


if __name__ == '__main__':
    sys.exit(main())