"""Organic live research — real aggregator snapshots + URL content hashing.""" from __future__ import annotations import logging import os from pathlib import Path from typing import Any, Dict, List, Optional from sqlalchemy.orm import Session from core.grants.completeness import grant_dict_from_row from core.grants.live_research import detect_grant_changes, establish_research_baselines, enqueue_research_job from core.grants.models import Grant from core.grants.wyszukiwarka_import import import_wyszukiwarka_file, resolve_wyszukiwarka_path logger = logging.getLogger(__name__) _ORGANIC_OLD_CANDIDATES = [ os.environ.get("WYSZUKIWARKA_JSON_OLD", ""), "/home/bmazu/WYSZUKIWARKA/dotacje-2026-06-20.json", os.path.expanduser("~/WYSZUKIWARKA/dotacje-2026-06-20.json"), ] def resolve_older_aggregator_path() -> Optional[Path]: for candidate in _ORGANIC_OLD_CANDIDATES: if not candidate: continue p = Path(candidate) if p.exists(): return p return None def collect_representative_sources(db: Session, *, limit: int = 20) -> List[Dict[str, Any]]: """Top catalog sources (operator/zrodlo) for dashboard research state.""" from core.grants.catalog_service import list_catalog_sources return list_catalog_sources(db, limit=limit) def detect_url_content_changes( db: Session, *, limit: int = 30, store_path: Optional[str] = None, ) -> Dict[str, Any]: """Incremental hash diff on official page content (ChangeDetector per URL).""" from rag_pipeline.change_detector import ChangeDetector scratch = os.environ.get("CATALOG_SCRATCH", "/tmp/grok-goal-hgis") default_store = Path(scratch) / "source_url_hashes.json" detector = ChangeDetector(store_path=store_path or str(default_store)) rows = ( db.query(Grant) .filter(Grant.status.in_(["active", "planned"])) .filter(Grant.official_page_url.isnot(None)) .filter(Grant.official_page_url != "") .limit(limit * 2) .all() ) stats = {"checked": 0, "changed": 0, "baselined": 0, "jobs_enqueued": 0, "mode": "url_content_hash"} changes: List[Dict[str, Any]] = [] for row in rows: item = grant_dict_from_row(row) url = str(item.get("official_page_url") or item.get("url") or "").strip() if not url.startswith("http"): continue stats["checked"] += 1 content = "|".join( [ item.get("name", ""), item.get("description", "")[:2000], item.get("deadline", ""), item.get("status", ""), ] ) rec = detector.record(url, content) if rec.get("is_new"): stats["baselined"] += 1 enqueue_research_job( db, "url_check", target_id=row.source_id, target_url=url, payload={"action": "baseline_url_content"}, priority=4, ) stats["jobs_enqueued"] += 1 elif rec.get("changed"): stats["changed"] += 1 changes.append({ "grant_source_id": row.source_id, "url": url, "change_type": "url_content_hash", }) enqueue_research_job( db, "analyze_change", target_id=row.source_id, target_url=url, payload={"url_content_changed": True}, priority=8, ) stats["jobs_enqueued"] += 1 db.commit() stats["changes"] = changes return stats def run_organic_aggregator_refresh( db: Session, *, older_path: Optional[Path] = None, newer_path: Optional[Path] = None, ) -> Dict[str, Any]: """ Real aggregator refresh: import older snapshot → baseline → import newer snapshot → detect. Uses two dated WYSZUKIWARKA JSON files (no scratch mutations). """ older = older_path or resolve_older_aggregator_path() newer = newer_path or resolve_wyszukiwarka_path() if not older or not newer: return { "ok": False, "reason": "missing_aggregator_snapshots", "older": str(older) if older else None, "newer": str(newer) if newer else None, } r_old = import_wyszukiwarka_file(older, db=db) baseline = establish_research_baselines(db, limit=5000) r_new = import_wyszukiwarka_file(newer, db=db) detection = detect_grant_changes(db, limit=5000) return { "ok": True, "older_file": str(older), "newer_file": str(newer), "import_older": r_old, "baseline": baseline, "import_newer": r_new, "detection": detection, "organic": True, "note": "Two real aggregator JSON snapshots — incremental field diff on shared source_ids", }