backend/scripts/verify_catalog.py

#!/usr/bin/env python3
"""Weryfikacja katalogu na realnym JSON + SQLite (verbose → scratch logs)."""
from __future__ import annotations

import json
import os
import sys
from pathlib import Path

BACKEND = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(BACKEND))
os.environ.setdefault("DATABASE_URL", "sqlite:///:memory:")

SCRATCH = Path(
    os.environ.get(
        "CATALOG_SCRATCH",
        "/tmp/grok-goal-226f39df5eac/implementer",
    )
)
SCRATCH.mkdir(parents=True, exist_ok=True)

from core.subscription.db import Base, SessionLocal, engine, init_models  # noqa: E402
from core.grants.wyszukiwarka_import import (  # noqa: E402
    import_wyszukiwarka_file,
    resolve_wyszukiwarka_path,
)
from core.grants.catalog_service import (  # noqa: E402
    get_catalog_stats,
    list_catalog_sources,
    search_catalog,
    nabory_search,
    _matches_filter,
    _effective_company_sizes,
)
from core.grants.catalog_filters import is_catalog_visible  # noqa: E402
from core.grants.wyszukiwarka_mapper import load_wyszukiwarka_json  # noqa: E402


def _assert(condition: bool, msg: str) -> None:
    if not condition:
        raise AssertionError(msg)


def _sample(items: list, n: int = 5) -> list:
    return [
        {
            "name": (x.get("name") or "")[:60],
            "status": x.get("status"),
            "region": x.get("eligible_regions"),
            "beneficiary_tags": x.get("beneficiary_tags"),
            "company_sizes": x.get("eligible_company_sizes"),
            "effective_sizes": _effective_company_sizes(x),
            "deadline": x.get("deadline"),
            "regulation": bool(x.get("precise_regulation_url")),
            "confidence_score": x.get("confidence_score"),
        }
        for x in items[:n]
    ]


def _log_section(path: Path, title: str, payload: dict | list | str) -> None:
    with path.open("a", encoding="utf-8") as f:
        f.write(f"\n=== {title} ===\n")
        if isinstance(payload, str):
            f.write(payload + "\n")
        else:
            f.write(json.dumps(payload, ensure_ascii=False, indent=2) + "\n")


def field_coverage_report(json_path: Path) -> dict:
    raw = json.loads(json_path.read_text(encoding="utf-8"))
    grants = raw.get("grants") or []
    mapped = load_wyszukiwarka_json(raw)
    n = len(grants)
    m = len(mapped)

    def _raw_filled(key: str) -> int:
        return sum(1 for g in grants if str(g.get(key) or "").strip())

    def _map_filled(key: str) -> int:
        return sum(1 for g in mapped if g.get(key))

    return {
        "raw_total": n,
        "mapped_total": m,
        "raw_termin_do": _raw_filled("termin_do"),
        "raw_link_regulamin": _raw_filled("link_regulamin"),
        "raw_beneficjenci": _raw_filled("beneficjenci"),
        "mapped_deadline": _map_filled("deadline"),
        "mapped_precise_regulation_url": sum(
            1 for g in mapped if str(g.get("precise_regulation_url") or "").startswith("http")
        ),
        "mapped_beneficiary_tags": sum(1 for g in mapped if g.get("beneficiary_tags")),
        "mapped_eligible_regions": sum(1 for g in mapped if g.get("eligible_regions")),
        "mapped_eligible_company_sizes": sum(1 for g in mapped if g.get("eligible_company_sizes")),
        "mapped_description_nonempty": sum(1 for g in mapped if str(g.get("description") or "").strip()),
        "note": (
            "Pełne dane zależą od źródeł — wiele portali nie publikuje terminu/kwoty; "
            "mapper wypełnia z dostępnych pól + inferencja MŚP/regionów."
        ),
    }


def main() -> int:
    filters_log = SCRATCH / "catalog_filters.log"
    search_log = SCRATCH / "search_catalog.log"
    stats_path = SCRATCH / "catalog_stats.json"
    for p in (filters_log, search_log):
        p.write_text("", encoding="utf-8")

    init_models()
    Base.metadata.create_all(bind=engine)
    db = SessionLocal()

    path = resolve_wyszukiwarka_path()
    _assert(path is not None, "Brak pliku dotacje-*.json")
    coverage = field_coverage_report(path)
    print("FIELD_COVERAGE:", json.dumps(coverage, ensure_ascii=False, indent=2))
    _log_section(filters_log, "FIELD_COVERAGE", coverage)

    imp = import_wyszukiwarka_file(path, db=db)
    print("IMPORT:", json.dumps(imp, ensure_ascii=False))
    _log_section(filters_log, "IMPORT", imp)

    stats = get_catalog_stats(db)
    print("STATS:", json.dumps(stats, ensure_ascii=False, indent=2))
    stats_path.write_text(json.dumps(stats, ensure_ascii=False, indent=2), encoding="utf-8")

    _assert(stats["has_data"], "has_data false")
    _assert(stats["wyszukiwarka_imported"] > 0, "wyszukiwarka_imported=0")
    _assert(stats["visible_in_catalog"] > 100, f"visible_in_catalog={stats['visible_in_catalog']}")
    _assert(stats["with_regulation_url"] > 50, f"with_regulation_url={stats['with_regulation_url']}")
    _assert(len(stats["top_sources"]) > 0, "top_sources empty")
    _assert(stats["closed"] > 0, "closed count=0 in DB after import")

    sources = list_catalog_sources(db)
    _assert(len(sources) > 5, "list_catalog_sources too few")
    _log_section(filters_log, "TOP_SOURCES", sources[:8])

    # FILTER: OZE + mazowieckie + active/planned
    r1 = search_catalog(
        db,
        query="OZE fotowoltaika",
        filters={"statuses": ["active", "planned"], "region": "mazowieckie"},
        limit=30,
    )
    section1 = {
        "filter": "OZE+mazowieckie+active/planned",
        "count": len(r1),
        "sample": _sample(r1),
    }
    print(f"FILTER OZE+mazowieckie: count={len(r1)}")
    _log_section(filters_log, "FILTER: OZE+mazowieckie", section1)
    _assert(len(r1) >= 1, "OZE+mazowieckie: 0 results")
    for item in r1:
        _assert(item["status"] in ("active", "planned"), f"bad status {item.get('status')}")
        _assert(is_catalog_visible(item), f"hidden: {item.get('name')}")
        regions = [x.lower() for x in (item.get("eligible_regions") or [])]
        hay = " ".join(regions)
        _assert(
            "mazowieckie" in hay or "cała polska" in hay or "cala polska" in hay,
            f"region mismatch: {item.get('name')} -> {regions}",
        )
        _assert(_matches_filter(item, {"region": "mazowieckie"}), item.get("name"))

    # FILTER: fundusze europejskie + mśp + mikro
    r2 = search_catalog(
        db,
        query="fundusze europejskie",
        filters={"beneficiary": "mśp", "company_size": "mikro"},
        limit=30,
    )
    section2 = {
        "filter": "FE+mśp+mikro",
        "count": len(r2),
        "sample": _sample(r2),
    }
    print(f"FILTER FE+mśp+mikro: count={len(r2)}")
    _log_section(search_log, "FILTER: FE+mśp+mikro", section2)
    _assert(len(r2) >= 1, "FE+mśp+mikro: 0 results")
    for item in r2:
        hay = f"{item.get('beneficjenci', '')} {' '.join(item.get('beneficiary_tags') or [])}".lower()
        _assert("mśp" in hay or "msp" in hay, f"beneficiary miss: {item.get('name')}")
        eff = _effective_company_sizes(item)
        _assert(eff and any("mikro" in s for s in eff), f"size miss: {item.get('name')} {eff}")
        _assert(
            _matches_filter(item, {"beneficiary": "mśp", "company_size": "mikro"}),
            item.get("name"),
        )
        _assert("confidence_score" in item, "missing confidence_score")
    scores = [x["confidence_score"] for x in r2]
    _assert(scores == sorted(scores, reverse=True), "not sorted by confidence_score")

    # FILTER: closed + mazowieckie (nabory_search — ta sama ścieżka co GET /nabory)
    closed_resp = nabory_search(db, status="closed", region="mazowieckie", limit=15)
    r_closed = closed_resp["nabory"]
    section3 = {
        "filter": "closed+mazowieckie via nabory_search",
        "count": len(r_closed),
        "has_catalog_stats": "catalog_stats" in closed_resp,
        "sample": _sample(r_closed),
    }
    print(f"FILTER closed+mazowieckie (nabory_search): count={len(r_closed)}")
    _log_section(filters_log, "FILTER: closed+region", section3)
    _assert(len(r_closed) >= 1, "closed+region: 0 results")
    _assert(closed_resp["catalog_stats"]["closed"] > 0, "stats.closed=0")
    for item in r_closed:
        _assert(item["status"] == "closed", f"not closed: {item.get('name')}")
        _assert(
            _matches_filter(item, {"status": "closed", "region": "mazowieckie"}),
            item.get("name"),
        )

    # FILTER: company_size=mikro — zero leakerów
    r_size = search_catalog(
        db,
        query="",
        filters={"company_size": "mikro", "statuses": ["active", "planned", "closed"]},
        limit=500,
    )
    section4 = {"filter": "company_size=mikro", "count": len(r_size), "sample": _sample(r_size, 5)}
    _log_section(search_log, "FILTER: company_size=mikro", section4)
    for item in r_size:
        eff = _effective_company_sizes(item)
        _assert(eff and any("mikro" in s for s in eff), f"leaker: {item.get('name')} {eff}")

    # Stale active hidden
    r3 = search_catalog(db, query="2019", filters={"statuses": ["active"]}, limit=20)
    stale = [x for x in r3 if (x.get("program_year") or 2024) < 2023]
    _log_section(
        search_log,
        "FILTER: active+query 2019 (stale check)",
        {"count": len(r3), "stale_visible": len(stale), "stale_names": [s.get("name") for s in stale]},
    )
    _assert(len(stale) == 0, f"stale active visible: {[s.get('name') for s in stale]}")

    db.close()
    print(f"VERIFY_CATALOG: OK (logs → {SCRATCH})")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())