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()