backend/core/grants/sync_service.py

import logging
import asyncio
import os
from datetime import datetime, timezone
from sqlalchemy.orm import Session

from core.grants.models import Grant
from core.search.grant_search_service import grant_search_service
from core.subscription.db import SessionLocal

logger = logging.getLogger(__name__)


async def _auto_vectorize_grant(grant_data: dict):
    from core.grants.regulation_url_quality import collect_all_regulation_urls, infer_doc_role
    from core.search.regulation_snapshot import regulation_snapshot_store

    urls = collect_all_regulation_urls(grant_data, max_urls=6)
    if not urls:
        url = grant_data.get('precise_regulation_url') or grant_data.get('url')
        if url:
            urls = [url]
    if not urls:
        return

    program = grant_data.get('program') or grant_data.get('source', 'UNKNOWN')
    name = grant_data.get('name', 'regulamin')
    grant_id = str(grant_data.get('id') or '')
    snap_key = f"{program}|{name}"[:80].upper()

    import requests
    for url in urls:
        try:
            jina_url = f"https://r.jina.ai/{url}"
            headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"}
            response = await asyncio.to_thread(requests.get, jina_url, headers=headers, timeout=30)
            if response.status_code == 200 and len(response.text) > 500:
                md = response.text
                doc_role = infer_doc_role(url)
                regulation_snapshot_store.create_snapshot(
                    program=snap_key,
                    call_name=name[:60],
                    source_url=url,
                    raw_text=md,
                    module=doc_role,
                )
                logger.info(f"[AutoVectorize] Snapshot RAG dla {program} ({doc_role}) z {url}")
                try:
                    from rag_pipeline.ingest import process_and_ingest
                    ingested = process_and_ingest(
                        md,
                        source_url=url,
                        priority="high",
                        extra_metadata={
                            "grant_id": grant_id,
                            "grant_program_key": snap_key,
                            "doc_role": doc_role,
                            "program_name": program,
                        },
                    )
                    if ingested:
                        logger.info(f"[AutoVectorize] Wektoryzacja Pinecone OK dla {url}")
                except Exception as ingest_err:
                    logger.warning(f"[AutoVectorize] Ingest wektorowy nieudany dla {url}: {ingest_err}")
        except Exception as e:
            logger.warning(f"[AutoVectorize] Błąd auto-wektoryzacji dla {url}: {e}")


async def _vectorize_grants_safe(grants: list, concurrency: int = 4) -> None:
    """Wektoryzuje nabory z ograniczoną współbieżnością i obsługą wyjątków.

    Zastępuje wcześniejsze fire-and-forget `asyncio.create_task(...)` bez referencji
    (ryzyko GC zadania, brak obsługi błędów, brak gwarancji ukończenia).
    """
    if not grants:
        return
    sem = asyncio.Semaphore(concurrency)

    async def _run(g):
        async with sem:
            try:
                await _auto_vectorize_grant(g)
            except Exception as e:
                logger.warning(
                    "[SyncService] Auto-wektoryzacja nieudana dla %s: %s",
                    g.get("name", "?"),
                    e,
                )

    await asyncio.gather(*[_run(g) for g in grants], return_exceptions=True)


def _upsert_grant_rows(db: Session, live_grants: list) -> dict:
    updated = inserted = 0
    to_vectorize: list = []
    existing = {g.source_id: g for g in db.query(Grant).all()}

    for grant_data in live_grants:
        source_id = str(grant_data.get('id', ''))
        if not source_id:
            continue

        db_grant = existing.get(source_id)
        if db_grant:
            db_grant.name = grant_data.get('name', db_grant.name)
            db_grant.program = grant_data.get('program', db_grant.program)
            db_grant.status = grant_data.get('status', db_grant.status)
            db_grant.url = grant_data.get('url', db_grant.url)
            db_grant.deadline = grant_data.get('deadline', db_grant.deadline)
            db_grant.instrument_type = grant_data.get('instrument_type', db_grant.instrument_type)
            db_grant.is_financial_instrument = grant_data.get('is_financial_instrument', db_grant.is_financial_instrument)
            # Odświeżanie pełnego zestawu pól kwalifikowalności/opisu (poprzednio pomijane → utrata świeżości)
            db_grant.eligible_company_sizes = grant_data.get('eligible_company_sizes', db_grant.eligible_company_sizes)
            db_grant.eligible_regions = grant_data.get('eligible_regions', db_grant.eligible_regions)
            db_grant.description = grant_data.get('description', db_grant.description)
            db_grant.precise_regulation_url = grant_data.get('precise_regulation_url', db_grant.precise_regulation_url)
            db_grant.is_outdated_warning = grant_data.get('is_outdated_warning', False)
            db_grant.url_warning = grant_data.get('url_warning', None)
            db_grant.last_verified = datetime.now(timezone.utc)
            db_grant.raw_data = grant_data
            updated += 1
            if grant_data.get('status') in ('active', 'planned') and not db_grant.is_outdated_warning:
                to_vectorize.append(grant_data)
        else:
            new_grant = Grant(
                source_id=source_id,
                name=grant_data.get('name', 'Brak nazwy'),
                program=grant_data.get('program', ''),
                status=grant_data.get('status', 'active'),
                url=grant_data.get('url', ''),
                deadline=grant_data.get('deadline', ''),
                source=grant_data.get('source', 'unknown'),
                instrument_type=grant_data.get('instrument_type', 'grant'),
                is_financial_instrument=grant_data.get('is_financial_instrument', False),
                eligible_company_sizes=grant_data.get('eligible_company_sizes', []),
                eligible_regions=grant_data.get('eligible_regions', []),
                description=grant_data.get('description', ''),
                precise_regulation_url=grant_data.get('precise_regulation_url', ''),
                is_outdated_warning=grant_data.get('is_outdated_warning', False),
                url_warning=grant_data.get('url_warning', None),
                last_verified=datetime.now(timezone.utc),
                raw_data=grant_data
            )
            db.add(new_grant)
            inserted += 1
            if grant_data.get('status') in ('active', 'planned') and not grant_data.get('is_outdated_warning'):
                to_vectorize.append(grant_data)

    db.commit()
    return {"inserted": inserted, "updated": updated, "to_vectorize": to_vectorize}


async def sync_wyszukiwarka_catalog(db: Session = None, json_path: str = None) -> dict:
    """Importuje katalog z pliku JSON agregatora (WYSZUKIWARKA)."""
    from core.grants.wyszukiwarka_import import import_wyszukiwarka_file, resolve_wyszukiwarka_path

    path = resolve_wyszukiwarka_path(json_path)
    if path is None:
        logger.warning("[SyncService] Brak pliku WYSZUKIWARKA — pomijam import katalogu.")
        return {"inserted": 0, "updated": 0, "error": "file_not_found"}

    close_db = False
    if db is None:
        db = SessionLocal()
        close_db = True
    try:
        result = import_wyszukiwarka_file(path, db=db)
    finally:
        if close_db:
            db.close()

    if result.get("inserted", 0) or result.get("updated", 0):
        try:
            from core.grants.post_import_pipeline import run_post_import_pipeline
            from core.subscription.db import refresh_engine_pool

            refresh_engine_pool()
            pipeline_db = SessionLocal()
            try:
                pipeline = await run_post_import_pipeline(pipeline_db)
                result["post_import"] = pipeline
            finally:
                pipeline_db.close()
        except Exception as pipe_err:
            logger.warning("[SyncService] Post-import pipeline: %s", pipe_err)
    return result


async def sync_grants_to_db(db: Session = None):
    """
    Aktualizuje PostgreSQL: najpierw katalog WYSZUKIWARKA (jeśli dostępny),
    potem opcjonalnie live scrape (gdy WYSZUKIWARKA_ONLY != true).
    """
    logger.info("[SyncService] Rozpoczynam synchronizację naborów dotacyjnych...")

    close_db = False
    if db is None:
        db = SessionLocal()
        close_db = True

    wysz_result = {"inserted": 0, "updated": 0}
    try:
        wysz_result = await sync_wyszukiwarka_catalog()
        logger.info("[SyncService] Import WYSZUKIWARKA: %s", wysz_result)
    except Exception as e:
        logger.warning("[SyncService] Import WYSZUKIWARKA nieudany: %s", e)

    wysz_only = os.environ.get("WYSZUKIWARKA_ONLY", "true").lower() == "true"
    if wysz_only:
        logger.info("[SyncService] WYSZUKIWARKA_ONLY=true — pomijam live scrape.")
        return wysz_result

    try:
        live_grants = await grant_search_service.get_all_grants(force_refresh=True)
        logger.info(f"[SyncService] Pobrano {len(live_grants)} naborów z live scrape.")
        scrape_result = _upsert_grant_rows(db, live_grants)
        to_vectorize = scrape_result.pop("to_vectorize", [])
        # Kontrolowana wektoryzacja (z referencją + obsługą błędów) zamiast fire-and-forget
        await _vectorize_grants_safe(to_vectorize)
        logger.info(
            "[SyncService] Live scrape: dodano %s, zaktualizowano %s",
            scrape_result["inserted"],
            scrape_result["updated"],
        )
        return {
            "wyszukiwarka": wysz_result,
            "live_scrape": scrape_result,
        }
    except Exception as e:
        db.rollback()
        logger.error(f"[SyncService] Błąd podczas zapisu live scrape: {e}")
        if wysz_result.get("inserted", 0) or wysz_result.get("updated", 0):
            return {"wyszukiwarka": wysz_result, "live_scrape_error": str(e)}
        raise
    finally:
        if close_db:
            db.close()