backend/scripts/http_verify_catalog.py

#!/usr/bin/env python3
"""HTTP smoke test katalogu (HF test env lub lokalny). Bez maskowania closed=0."""
from __future__ import annotations

import json
import os
import sys
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path

BASE = os.environ.get("CATALOG_API_BASE", "https://bogdan555-grantforge-api.hf.space")
TIMEOUT = int(os.environ.get("CATALOG_HTTP_TIMEOUT", "45"))
SCRATCH = Path(
    os.environ.get(
        "CATALOG_SCRATCH",
        "/tmp/grok-goal-226f39df5eac/implementer",
    )
)
SCRATCH.mkdir(parents=True, exist_ok=True)
LOG_PATH = SCRATCH / "http_verify.log"


class _Tee:
    def __init__(self, *streams):
        self._streams = streams

    def write(self, data: str) -> None:
        for s in self._streams:
            s.write(data)
            s.flush()

    def flush(self) -> None:
        for s in self._streams:
            s.flush()


def _post(path: str, payload: dict) -> tuple[int, dict | str]:
    url = f"{BASE.rstrip('/')}{path}"
    data = json.dumps(payload).encode("utf-8")
    req = urllib.request.Request(
        url, data=data, headers={"Accept": "application/json", "Content-Type": "application/json"}
    )
    try:
        with urllib.request.urlopen(req, timeout=TIMEOUT) as resp:
            body = resp.read().decode("utf-8", errors="replace")
            try:
                return resp.status, json.loads(body)
            except json.JSONDecodeError:
                return resp.status, body[:2000]
    except urllib.error.HTTPError as e:
        body = e.read().decode("utf-8", errors="replace")[:2000]
        return e.code, body


def _get(path: str) -> tuple[int, dict | str]:
    url = f"{BASE.rstrip('/')}{path}"
    req = urllib.request.Request(url, headers={"Accept": "application/json"})
    try:
        with urllib.request.urlopen(req, timeout=TIMEOUT) as resp:
            body = resp.read().decode("utf-8", errors="replace")
            try:
                return resp.status, json.loads(body)
            except json.JSONDecodeError:
                return resp.status, body[:2000]
    except urllib.error.HTTPError as e:
        body = e.read().decode("utf-8", errors="replace")[:2000]
        return e.code, body


def main() -> int:
    log_file = LOG_PATH.open("w", encoding="utf-8")
    orig_stdout = sys.stdout
    sys.stdout = _Tee(orig_stdout, log_file)
    try:
        return _run_checks()
    finally:
        sys.stdout = orig_stdout
        log_file.close()


def _run_checks() -> int:
    print(f"HTTP_VERIFY base={BASE}")
    checks: list[str] = []

    code, stats = _get("/api/grants/catalog/stats")
    print(f"GET /api/grants/catalog/stats -> HTTP {code}")
    print(json.dumps(stats, ensure_ascii=False, indent=2) if isinstance(stats, dict) else stats)
    closed_in_db = 0
    if code != 200:
        checks.append(f"stats HTTP {code}")
    elif isinstance(stats, dict):
        if not stats.get("has_data"):
            checks.append("has_data false")
        if stats.get("visible_in_catalog", 0) < 100:
            checks.append(f"visible_in_catalog={stats.get('visible_in_catalog')}")
        closed_in_db = int(stats.get("closed") or 0)

    search_body = {
        "query": "PARP",
        "filters": {"status": "active", "region": "mazowieckie"},
        "limit": 5,
    }
    code2, search = _post("/api/grants/catalog/search", search_body)
    print(f"\nPOST /api/grants/catalog/search -> HTTP {code2}")
    print(json.dumps(search_body, ensure_ascii=False))
    if isinstance(search, dict):
        grants = search.get("grants") or []
        print(
            json.dumps(
                {
                    "count": search.get("count"),
                    "sample": [
                        {
                            "name": (g.get("name") or "")[:50],
                            "status": g.get("status"),
                            "regions": g.get("eligible_regions"),
                        }
                        for g in grants[:3]
                    ],
                },
                ensure_ascii=False,
                indent=2,
            )
        )
        if code2 != 200:
            checks.append(f"catalog/search HTTP {code2}")
        elif search.get("count", 0) < 1:
            checks.append("catalog/search count=0")
        else:
            for g in grants:
                if g.get("status") != "active":
                    checks.append(f"search returned non-active: {g.get('name')}")
                    break
    else:
        print(search)
        checks.append("catalog/search non-json")

    closed_body = {
        "query": "",
        "filters": {"status": "closed", "region": "mazowieckie"},
        "limit": 5,
    }
    code3, closed = _post("/api/grants/catalog/search", closed_body)
    print(f"\nPOST /api/grants/catalog/search (closed) -> HTTP {code3}")
    if isinstance(closed, dict):
        grants_c = closed.get("grants") or []
        closed_count = closed.get("count", 0)
        print(
            json.dumps(
                {
                    "count": closed_count,
                    "sample_status": [g.get("status") for g in grants_c[:3]],
                    "sample_names": [(g.get("name") or "")[:50] for g in grants_c[:3]],
                },
                ensure_ascii=False,
                indent=2,
            )
        )
        if code3 != 200:
            checks.append(f"closed search HTTP {code3}")
        elif closed_count < 1:
            if closed_in_db > 0:
                checks.append(
                    f"closed+mazowieckie count=0 but stats.closed={closed_in_db} "
                    "(wymaga FORCE_WYSZUKIWARKA_REIMPORT=true + redeploy)"
                )
            else:
                checks.append("closed+mazowieckie count=0 and stats.closed=0")
        else:
            bad = [g for g in grants_c if g.get("status") != "closed"]
            if bad:
                checks.append(f"closed search has non-closed: {len(bad)}")
    else:
        checks.append("closed search failed")

    nabory_qs = urllib.parse.urlencode(
        {"status": "active", "region": "mazowieckie", "q": "PARP", "limit": "5"}
    )
    nabory_path = f"/api/grants/nabory?{nabory_qs}"
    code4, nabory = _get(nabory_path)
    print(f"\nGET {nabory_path} -> HTTP {code4}")
    if isinstance(nabory, dict):
        nabory_list = nabory.get("nabory") or []
        has_catalog_stats = "catalog_stats" in nabory
        print(
            json.dumps(
                {
                    "count": nabory.get("count"),
                    "has_catalog_stats": has_catalog_stats,
                    "catalog_visible": (nabory.get("catalog_stats") or {}).get(
                        "visible_in_catalog"
                    ),
                    "sample": [
                        {
                            "name": (n.get("name") or "")[:50],
                            "status": n.get("status"),
                            "program_year": n.get("program_year"),
                        }
                        for n in nabory_list[:5]
                    ],
                },
                ensure_ascii=False,
                indent=2,
            )
        )
        if code4 != 200:
            checks.append(f"nabory HTTP {code4}")
        elif not has_catalog_stats:
            checks.append("nabory missing catalog_stats (stary endpoint — wymaga redeploy)")
        elif nabory.get("count", 0) < 1:
            checks.append("nabory count=0 for active+PARP+mazowieckie")
        else:
            stale = [
                n
                for n in nabory_list
                if (n.get("program_year") or 2024) < 2023 and n.get("status") == "active"
            ]
            if stale:
                checks.append(f"nabory returned stale active: {len(stale)}")
    else:
        print(nabory)
        checks.append("nabory non-json")

    nabory_closed_qs = urllib.parse.urlencode(
        {"status": "closed", "region": "mazowieckie", "limit": "5"}
    )
    nabory_closed_path = f"/api/grants/nabory?{nabory_closed_qs}"
    code5, nabory_closed = _get(nabory_closed_path)
    print(f"\nGET {nabory_closed_path} -> HTTP {code5}")
    if isinstance(nabory_closed, dict):
        print(
            json.dumps(
                {
                    "count": nabory_closed.get("count"),
                    "has_catalog_stats": "catalog_stats" in nabory_closed,
                    "sample_status": [
                        n.get("status") for n in (nabory_closed.get("nabory") or [])[:3]
                    ],
                },
                ensure_ascii=False,
                indent=2,
            )
        )
        if code5 == 200 and "catalog_stats" in nabory_closed:
            if nabory_closed.get("count", 0) < 1 and closed_in_db > 0:
                checks.append("nabory closed+region count=0 despite stats.closed>0")
    else:
        print(nabory_closed)

    if checks:
        print("\nHTTP_VERIFY FAIL:", checks)
        return 1
    print("\nHTTP_VERIFY: OK")
    print(f"Log saved: {LOG_PATH}")
    return 0


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