backend/core/grants/source_research.py

"""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",
    }