#!/usr/bin/env python3 """Pełna weryfikacja Hybrid Grant Intelligence System (plan gating steps 1-6).""" from __future__ import annotations import asyncio import json import os import subprocess import sys from pathlib import Path BACKEND = Path(__file__).resolve().parents[1] sys.path.insert(0, str(BACKEND)) SCRATCH = Path(os.environ.get("CATALOG_SCRATCH", "/tmp/grok-goal-935580f22161/implementer")) SCRATCH.mkdir(parents=True, exist_ok=True) DB_PATH = SCRATCH / "verify_hgis.db" if DB_PATH.exists(): DB_PATH.unlink() os.environ["DATABASE_URL"] = f"sqlite:///{DB_PATH}" from core.subscription.db import SessionLocal, init_models # noqa: E402 from scripts.migrate_hgis_schema import migrate # noqa: E402 from core.grants.wyszukiwarka_import import import_wyszukiwarka_file, resolve_wyszukiwarka_path # noqa: E402 from core.grants.catalog_service import ( # noqa: E402 get_catalog_stats, nabory_search, search_catalog, ) from core.grants.catalog_visibility import is_catalog_visible, is_program_too_old # noqa: E402 from core.grants.completeness import grant_dict_from_row, compute_completeness # noqa: E402 from core.grants.models import Grant, GrantVersion, HumanVerificationItem # noqa: E402 from core.grants.research_snapshot import SNAPSHOT_HASH_KEY # noqa: E402 from core.grants.versioning import detect_field_changes, record_changes, get_recent_versions # noqa: E402 from core.grants.live_research import ( # noqa: E402 run_live_research_cycle, research_source_state, detect_grant_changes, ) from core.grants.human_verification import list_pending_verifications, resolve_verification, hvq_stats # noqa: E402 from core.grants.recommendations import recommend_grants_for_company # noqa: E402 from core.grants.hybrid_search import hybrid_search_catalog, is_semantic_search_enabled # noqa: E402 from core.grants.hybrid_search_status import get_hybrid_search_status # noqa: E402 from core.grants.source_research import run_organic_aggregator_refresh, collect_representative_sources # noqa: E402 from core.grants.regulation_backfill import ( # noqa: E402 regulation_completeness_report, check_regulation_links_sample, ) from core.grants.catalog_vector_index import index_grants_to_pinecone, is_pinecone_indexing_enabled # noqa: E402 from agents.grant_research_agent import run_grant_research_agent, LANGGRAPH_AVAILABLE # noqa: E402 # Shared state between steps (real detection output, not synthetic) _PIPELINE_STATE: dict = {} def _write(name: str, data) -> None: path = SCRATCH / name if isinstance(data, (dict, list)): path.write_text(json.dumps(data, ensure_ascii=False, indent=2, default=str), encoding="utf-8") else: path.write_text(str(data), encoding="utf-8") print(f" wrote {path}") def step1_model_enrichment(db) -> dict: src = resolve_wyszukiwarka_path() assert src, "WYSZUKIWARKA JSON not found" r1 = import_wyszukiwarka_file(src, db=db) assert r1["total_mapped"] > 1000 assert r1["inserted"] == r1["total_mapped"], f"first import should insert all: {r1}" assert r1["catalog_hidden_flagged"] > 0, f"catalog_hidden must be set on first import: {r1}" seeded_on_import = sum( 1 for g in db.query(Grant).limit(50).all() if (g.raw_data or {}).get(SNAPSHOT_HASH_KEY) ) assert seeded_on_import == 0, "import must not seed research snapshots" row = ( db.query(Grant) .filter(Grant.eligible_pkd != None) # noqa: E711 .filter(Grant.regulation_url != None) # noqa: E711 .first() ) if not row: row = db.query(Grant).first() d = grant_dict_from_row(row) comp = compute_completeness(d) assert d.get("name") and d.get("status") assert d.get("official_page_url") or d.get("url"), "missing official page" with_pkd = db.query(Grant).filter(Grant.eligible_pkd != None).count() # noqa: E711 with_budget = db.query(Grant).filter(Grant.budget_max_pln != None).count() # noqa: E711 with_reg = db.query(Grant).filter(Grant.regulation_url != None).filter(Grant.regulation_url != "").count() # noqa: E711 assert with_pkd > 100, f"eligible_pkd underpopulated: {with_pkd}" assert with_reg > 200, f"regulation_url underpopulated: {with_reg}" old_items = [g for g in db.query(Grant).all() if is_program_too_old(grant_dict_from_row(g))] hidden_visible = sum(1 for g in old_items if is_catalog_visible(grant_dict_from_row(g))) assert hidden_visible == 0, "old programs must be hidden by default" ch = detect_field_changes( d["id"], {**d, "deadline": "2020-01-01"}, d, ) assert any(c["field_name"] == "deadline" for c in ch) rec = record_changes(db, ch[:1], source="verify_sample") db.commit() versions = get_recent_versions(db, limit=20) assert len(versions) >= 1 stats = get_catalog_stats(db) enrichment = r1.get("enrichment", {}) reg_report = regulation_completeness_report(db) reg_links = asyncio.run(check_regulation_links_sample(db, limit=30)) fixture_path = BACKEND / "tests" / "fixtures" / "real_grant_sample.json" assert fixture_path.exists(), f"missing real grant fixture: {fixture_path}" real_sample = json.loads(fixture_path.read_text(encoding="utf-8")) assert real_sample.get("name") and (real_sample.get("official_page_url") or real_sample.get("url")) out = { "import_latest": r1, "source_file": str(src), "versioning_sample_changes": ch[:2], "sample": {k: d.get(k) for k in ( "name", "program_year", "catalog_hidden", "eligible_pkd", "budget_max_pln", "regulation_url", "official_page_url", "application_url", "completeness_score", )}, "completeness": comp, "old_programs_total": len(old_items), "old_hidden_by_visibility": len(old_items) - hidden_visible, "versions_created": len(versions), "with_pkd": with_pkd, "with_budget": with_budget, "with_regulation_url": with_reg, "catalog_hidden_flagged_import": r1.get("catalog_hidden_flagged", 0), "enrichment": enrichment, "enrichment_pct_regulation": enrichment.get("pct_regulation"), "regulation_url_note": ( f"Source JSON provides ~{enrichment.get('pct_regulation')}% regulation links; " "full coverage requires regulation_ingest/crawl4ai pipeline" ), "regulation_completeness": reg_report, "regulation_backfill": r1.get("regulation_backfill"), "regulation_links_sample": reg_links, "real_grant_fixture": { "path": str(fixture_path), "name": real_sample.get("name"), "source": real_sample.get("source"), "has_regulation_url": str(real_sample.get("regulation_url") or "").startswith("http"), }, "stats": stats, } _write("model_enrichment.log", out) return out def step2_live_research(db) -> dict: organic = run_organic_aggregator_refresh(db) assert organic.get("ok"), f"organic aggregator refresh failed: {organic}" detection = organic["detection"] assert detection.get("changed", 0) >= 5, f"organic refresh must detect changes: {detection}" assert detection.get("change_count", 0) >= 5 _PIPELINE_STATE["detection"] = detection _PIPELINE_STATE["organic"] = organic cycle = run_live_research_cycle(db) assert cycle["jobs"].get("processed", 0) >= 1, "job queue must process jobs" rss = research_source_state(db) sources = collect_representative_sources(db, limit=10) out = { "organic_aggregator_refresh": { "older_file": organic.get("older_file"), "newer_file": organic.get("newer_file"), "import_older_total": organic["import_older"].get("total_mapped"), "import_newer_updated": organic["import_newer"].get("updated"), }, "detection": detection, "changed_fields": [c.get("field_name") for c in detection.get("changes", [])[:20]], "cycle": cycle, "source_state": rss, "representative_sources": sources[:5], "note": "Two real WYSZUKIWARKA snapshots (2026-06-20 → latest) — no scratch mutations", } _write("live_research.log", out) return out def step3_agent(db) -> dict: detection = _PIPELINE_STATE.get("detection", {}) changes = detection.get("changes") or [] assert changes, "agent must use real changes from detect_grant_changes" target_id = changes[0].get("grant_source_id") or _PIPELINE_STATE.get("target_id") row = ( db.query(Grant) .filter(Grant.source_id == target_id) .first() ) rec_row = ( db.query(Grant) .filter(Grant.status == "active") .filter(Grant.operator.like("%PARP%")) .first() ) if not rec_row: rec_row = row or db.query(Grant).filter(Grant.status == "active").first() sample = grant_dict_from_row(row or rec_row) rec_grant = grant_dict_from_row(rec_row) company = {"size": "mikro", "region": "mazowieckie", "pkd": ["62.01"]} a1 = run_grant_research_agent(rec_grant, changes=changes, company_profile=company) a2 = run_grant_research_agent(rec_grant, changes=changes, company_profile=company) assert a1.get("change_summary") assert a1.get("enriched_metadata") assert len(a1.get("recommendations", [])) >= 1 rec = a1["recommendations"][0] why = rec.get("why_it_fits", "") assert len(why) > 40 and "«" in why and rec.get("relevance_score", 0) >= 55 assert a2.get("change_summary") == a1.get("change_summary") if not LANGGRAPH_AVAILABLE: print(" WARN: langgraph not installed — sequential fallback used") out = { "langgraph_available": LANGGRAPH_AVAILABLE, "execution_mode": "langgraph" if LANGGRAPH_AVAILABLE else "sequential_fallback", "changes_from_detection": changes[:3], "change_target_grant": sample.get("name"), "recommendation_grant": rec_grant.get("name"), "run1": {k: a1.get(k) for k in ("change_analysis", "change_summary", "enriched_metadata", "recommendations", "langgraph")}, "run2_summary": a2.get("change_summary"), } _write("grant_research_agent.log", out) return out def step4_diagnostics(db) -> dict: pending_before = list_pending_verifications(db) hvq = hvq_stats(db) item = HumanVerificationItem( grant_source_id="hvq-test", change_type="field_update", field_name="deadline", old_value="2024-01-01", new_value="2026-12-31", summary="Test HVQ entry", priority=9, ) db.add(item) db.commit() pending = list_pending_verifications(db) assert any(p["grant_source_id"] == "hvq-test" for p in pending) item_id = next(p["id"] for p in pending if p["grant_source_id"] == "hvq-test") resolved = resolve_verification(db, item_id, resolution_note="verified OK", new_status="approved") assert resolved["ok"] filter_cases = { "beneficiary": {"beneficiary": "mśp"}, "company_size": {"company_size": "mikro"}, "region": {"region": "mazowieckie"}, "category": {"category": "innowac"}, "program_year": {"program_year": 2024}, "operator": {"operator": "PARP"}, "source": {"source": "parp"}, "status": {"status": "active"}, "pkd": {"pkd": "62"}, "q": {"q": "fundusze europejskie"}, "hybrid": {"q": "dotacja MŚP", "use_hybrid": True}, } results1 = {} results2 = {} for name, kw in filter_cases.items(): r1 = nabory_search(db, limit=10, **kw) r2 = nabory_search(db, limit=10, **kw) results1[name] = {"count": r1["count"], "has_stats": "catalog_stats" in r1} results2[name] = r2["count"] assert r1["count"] == r2["count"], f"inconsistent filter {name}" assert "catalog_stats" in r1 cat = search_catalog(db, query="dotacja", filters={"company_size": "mikro"}, limit=5) pinecone_index = index_grants_to_pinecone(db, limit=50) hybrid = hybrid_search_catalog(db, query="dotacja MŚP", filters={"status": "active"}, limit=5) hybrid_modes = {r.get("search_mode") for r in hybrid} hybrid_status = get_hybrid_search_status() company = {"size": "mikro", "region": "mazowieckie", "pkd": ["62.01"], "name": "IT"} rec1 = recommend_grants_for_company(db, company) rec2 = recommend_grants_for_company(db, company) assert rec1["count"] == rec2["count"] assert rec1["recommendations"][0]["why_it_fits"] dashboard = { "catalog_stats": get_catalog_stats(db), "research": research_source_state(db), "hvq": hvq_stats(db), "versions": get_recent_versions(db, limit=5), "pending_verifications": len(list_pending_verifications(db)), } out = { "dashboard": dashboard, "filters_run1": results1, "filters_run2": results2, "catalog_search_count": len(cat), "hybrid_search_count": len(hybrid), "hybrid_search_modes": sorted(hybrid_modes), "hybrid_search_status": hybrid_status, "pinecone_index": pinecone_index, "pinecone_indexing_enabled": is_pinecone_indexing_enabled(), "fts_active": hybrid_status.get("fts_active"), "pinecone_semantic_active": hybrid_status.get("pinecone_semantic_active"), "semantic_enabled": is_semantic_search_enabled(), "data_quality_limits": { "regulation_url_in_source_json_pct": "~23%", "with_regulation_url_in_db": dashboard["catalog_stats"].get("with_regulation_url"), "avg_completeness_visible": dashboard["catalog_stats"].get("completeness", {}).get("avg_completeness"), "improvement_path": "regulation_ingest.py + crawl4ai verify cron", }, "rec1_top": rec1["recommendations"][:2], "rec2_count": rec2["count"], "hvq_resolve": resolved, "links_in_rec": { "regulation_url": rec1["recommendations"][0].get("regulation_url"), "official_page_url": rec1["recommendations"][0].get("official_page_url"), }, } _write("diagnostics_filters_recs.log", out) return out def step5_docs() -> dict: doc = BACKEND / "docs" / "HYBRID_GRANT_INTELLIGENCE.md" assert doc.exists() and doc.stat().st_size > 500 text = doc.read_text(encoding="utf-8") for section in ("Warstwa", "ENABLE_", "recommendations", "migrate", "dashboard"): assert section.lower() in text.lower(), f"docs missing {section}" head = text[:1500] _write("tech_docs.log", head) return {"path": str(doc), "size": doc.stat().st_size, "sections_ok": True} def step6_tests() -> dict: use_verify_db = os.environ.get("PYTEST_USE_VERIFY_DB", "1").lower() in ("1", "true", "yes") pytest_db = DB_PATH if use_verify_db else SCRATCH / "pytest_step6_isolated.db" if not use_verify_db and pytest_db.exists(): pytest_db.unlink() env = { **os.environ, "CATALOG_SCRATCH": str(SCRATCH), "DATABASE_URL": f"sqlite:///{pytest_db}", "PYTEST_ISOLATED": "0" if use_verify_db else "1", "PYTEST_USE_VERIFY_DB": "1" if use_verify_db else "0", } tests = subprocess.run( [ sys.executable, "-m", "pytest", "tests/test_hgis.py", "tests/test_research_snapshot.py", "tests/test_wyszukiwarka_pipeline.py", "tests/test_catalog_filters.py", "tests/test_catalog_pipeline.py", "-q", ], cwd=BACKEND, capture_output=True, text=True, env=env, ) verify = subprocess.run( [sys.executable, "scripts/verify_catalog.py"], cwd=BACKEND, capture_output=True, text=True, env={**env, "DATABASE_URL": "sqlite:///:memory:"}, ) local_ep = subprocess.run( [sys.executable, "scripts/test_intelligence_local.py"], cwd=BACKEND, capture_output=True, text=True, env=env, ) out = { "pytest_db": str(pytest_db), "pytest_use_verify_db": use_verify_db, "pytest_exit": tests.returncode, "pytest_stdout": tests.stdout, "pytest_stderr": tests.stderr[-500:] if tests.stderr else "", "verify_catalog_exit": verify.returncode, "verify_catalog_stdout": verify.stdout[-2000:] if verify.stdout else "", "local_intelligence_exit": local_ep.returncode, "local_intelligence_stdout": local_ep.stdout[-1500:] if local_ep.stdout else "", } _write("pytest_and_verify.log", out) assert tests.returncode == 0, tests.stdout + tests.stderr return out def step0_shipped_code_manifest() -> dict: """Manifest absolutnych ścieżek kodu wdrożonego (grantforge-patch/backend).""" import hashlib rel_paths = [ "core/grants/models.py", "core/grants/research_snapshot.py", "core/grants/catalog_visibility.py", "core/grants/catalog_filters.py", "core/grants/live_research.py", "core/grants/source_research.py", "core/grants/hybrid_search_status.py", "core/grants/wyszukiwarka_import.py", "core/grants/wyszukiwarka_mapper.py", "core/grants/hybrid_search.py", "core/grants/recommendations.py", "core/grants/versioning.py", "core/grants/completeness.py", "core/grants/enrichment.py", "core/grants/regulation_backfill.py", "core/grants/catalog_vector_index.py", "core/grants/credibility.py", "core/grants/human_verification.py", "tests/fixtures/real_grant_sample.json", "agents/grant_research_agent.py", "endpoints/grants_intelligence.py", "scripts/verify_hgis.py", "scripts/migrate_hgis_schema.py", "tests/test_hgis.py", "tests/test_research_snapshot.py", "tests/test_wyszukiwarka_pipeline.py", "docs/FILE_ANALYSIS.md", "docs/HYBRID_GRANT_INTELLIGENCE.md", ] files = [] for rel in rel_paths: p = BACKEND / rel assert p.exists(), f"shipped file missing: {p}" files.append({ "absolute_path": str(p.resolve()), "relative_path": rel, "bytes": p.stat().st_size, "sha256": hashlib.sha256(p.read_bytes()).hexdigest(), }) fa_src = BACKEND / "docs" / "FILE_ANALYSIS.md" fa_dst = SCRATCH / "FILE_ANALYSIS.md" fa_dst.write_text(fa_src.read_text(encoding="utf-8"), encoding="utf-8") out = {"backend_root": str(BACKEND.resolve()), "file_count": len(files), "files": files} _write("shipped_code_manifest.json", out) _write("CHANGED_FILES.log", "\n".join(f["absolute_path"] for f in files)) return out def main() -> int: print("HGIS verification →", SCRATCH) failure_log = SCRATCH / "verification_failure.log" init_models() migrate() db = SessionLocal() try: step0_shipped_code_manifest() print("step0 OK") step1_model_enrichment(db) print("step1 OK") step2_live_research(db) print("step2 OK") step3_agent(db) print("step3 OK") step4_diagnostics(db) print("step4 OK") step5_docs() print("step5 OK") db.close() step6_tests() print("step6 OK") detection = _PIPELINE_STATE.get("detection", {}) pytest_log = (SCRATCH / "pytest_and_verify.log").read_text(encoding="utf-8") import re as _re m = _re.search(r"(\d+) passed", pytest_log) summary = { "status": "PASS", "goal_scratch": str(SCRATCH), "pytest_passed": int(m.group(1)) if m else None, "change_detection_changed": detection.get("changed"), "change_detection_count": detection.get("change_count"), "hybrid_note": "fts_only without PINECONE_API_KEY; fts+pinecone when key set", "organic_refresh": bool(_PIPELINE_STATE.get("organic", {}).get("ok")), } _write("verification_summary.json", summary) if failure_log.exists(): failure_log.unlink() bundle = { "status": "PASS", "scratch": str(SCRATCH), "backend_root": str(BACKEND.resolve()), "summary": summary, "artifacts": [ "shipped_code_manifest.json", "CHANGED_FILES.log", "FILE_ANALYSIS.md", "model_enrichment.log", "live_research.log", "grant_research_agent.log", "diagnostics_filters_recs.log", "pytest_and_verify.log", "verification_summary.json", ], "limitations": { "regulation_url_pct_visible": "see regulation_completeness in model_enrichment.log (~26% visible; source JSON ~23%)", "regulation_links_working_pct": "see regulation_links_sample in model_enrichment.log", "pinecone": "fts_only without PINECONE_API_KEY; fts+pinecone when key set (catalog_vector_index)", "change_detection": "organic aggregator refresh dotacje-2026-06-20 → latest (no scratch mutations)", "pytest_db": "shared verify_hgis.db when PYTEST_USE_VERIFY_DB=1 (default)", }, } _write("evidence_bundle.json", bundle) print("ALL GATING STEPS PASS") return 0 except Exception as e: import traceback _write("verification_failure.log", {"error": str(e), "traceback": traceback.format_exc()}) print("FAIL:", e) traceback.print_exc() return 1 finally: db.close() if __name__ == "__main__": raise SystemExit(main())