""" GrantForge bootstrap — jawne uruchomienie z server lifespan: grantforge_bootstrap.bootstrap_startup() """ import asyncio import logging import os import sys import threading from typing import Any import core.langchain_compat # noqa: F401 logger = logging.getLogger(__name__) PATCH_SET_VERSION = "2026.07.09-functional-audit" _PATCH_FLAG_NAMES = ( "ENABLE_SEARCH_PATCHES", "ENABLE_MATCH_PHASE1_PATCHES", "ENABLE_NEO4J_SCHEMA_PATCH", "ENABLE_HGIS_MIGRATION", "ENABLE_GRANTS_MATCH_FIX", "ENABLE_NABORY_ENHANCEMENTS", "ENABLE_GRANTS_INTELLIGENCE", "ENABLE_PROJECTS_MATCH_PATCH", ) _BOOTSTRAP_APPLIED = False _last_bootstrap_status: dict[str, Any] | None = None def _env_flag(name: str, default: bool = True) -> bool: val = os.environ.get(name) if val is None: return default return val.lower() not in ("0", "false", "no", "off") def get_patch_flags() -> dict[str, bool]: """Zwraca aktualny stan flag patchy (domyślnie włączone dla kompatybilności wstecznej).""" return {name: _env_flag(name, default=True) for name in _PATCH_FLAG_NAMES} def _running_under_pytest() -> bool: """Wykrywa środowisko testowe (pytest).""" return ( "PYTEST_CURRENT_TEST" in os.environ or os.environ.get("ENV", "").lower() == "test" or "pytest" in sys.modules ) def _acquire_bootstrap_leader() -> bool: """Only one worker runs catalog bootstrap thread (Gunicorn multi-worker).""" if os.environ.get("RUN_BOOTSTRAP_CATALOG", "true").lower() in ("0", "false", "no"): logger.info("[grantforge_bootstrap] RUN_BOOTSTRAP_CATALOG=false — pomijam wątek katalogu.") return False lock_path = os.environ.get( "BOOTSTRAP_LOCK_PATH", os.path.join(os.environ.get("TMPDIR", "/tmp"), "grantforge_bootstrap.lock"), ) try: import fcntl lock_file = open(lock_path, "w") fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) lock_file.write(str(os.getpid())) lock_file.flush() return True except OSError: logger.info("[grantforge_bootstrap] Inny worker już uruchomił bootstrap katalogu — pomijam.") return False except ImportError: return True from core.subscription.db import init_models init_models() def _run_async(coro): try: loop = asyncio.get_event_loop() if loop.is_running(): loop.create_task(coro) else: loop.run_until_complete(coro) except RuntimeError: asyncio.run(coro) def _startup_catalog(): try: from bootstrap_wyszukiwarka import ensure_catalog_on_empty_db _run_async(ensure_catalog_on_empty_db()) except Exception as e: logger.warning("[grantforge_bootstrap] Katalog startup: %s", e) def _apply_search_patches(flags: dict[str, bool]) -> None: if not flags["ENABLE_SEARCH_PATCHES"]: return from core.search.grant_search_catalog import apply_search_patches apply_search_patches() def _apply_match_phase1_patches(flags: dict[str, bool]) -> None: if not flags["ENABLE_MATCH_PHASE1_PATCHES"]: return from core.match.apply_phase1_patches import apply_match_phase1_patches apply_match_phase1_patches() def _apply_neo4j_schema_patch(flags: dict[str, bool]) -> None: if not flags["ENABLE_NEO4J_SCHEMA_PATCH"]: return from core.graph_db import neo4j_client from core.graph_db.neo4j_schema import ensure_company_graph_schema _orig_neo4j_connect = neo4j_client.Neo4jClient.connect def _neo4j_connect_with_schema(self): _orig_neo4j_connect(self) if self.driver: ensure_company_graph_schema(self.driver) neo4j_client.Neo4jClient.connect = _neo4j_connect_with_schema def _apply_hgis_migration(flags: dict[str, bool]) -> None: if not flags["ENABLE_HGIS_MIGRATION"]: return from scripts.migrate_hgis_schema import migrate as migrate_hgis mig = migrate_hgis() if mig.get("columns_added") or mig.get("tables_created"): logger.info("[grantforge_bootstrap] HGIS migration: %s", mig) elif mig.get("errors"): logger.warning("[grantforge_bootstrap] HGIS migration errors: %s", mig) else: logger.debug("[grantforge_bootstrap] HGIS schema up to date") def _apply_endpoint_patches(flags: dict[str, bool]) -> None: from endpoints.patches.registry import apply_endpoint_patches apply_endpoint_patches(flags) def get_bootstrap_status() -> dict[str, Any]: """Ostatni status bootstrapu (flagi + wersja patch-set).""" flags = get_patch_flags() return { "patch_set_version": PATCH_SET_VERSION, "applied": _BOOTSTRAP_APPLIED, "flags": flags, "enabled_patches": [k for k, v in flags.items() if v], "disabled_patches": [k for k, v in flags.items() if not v], } def reset_bootstrap_for_tests() -> None: """Reset idempotency guard — tylko testy.""" global _BOOTSTRAP_APPLIED, _last_bootstrap_status _BOOTSTRAP_APPLIED = False _last_bootstrap_status = None def bootstrap_startup() -> dict[str, Any]: """Jawne uruchomienie patchy i wątku katalogu (idempotentne).""" global _BOOTSTRAP_APPLIED, _last_bootstrap_status if _BOOTSTRAP_APPLIED: return _last_bootstrap_status or get_bootstrap_status() flags = get_patch_flags() enabled = [k for k, v in flags.items() if v] disabled = [k for k, v in flags.items() if not v] try: _apply_search_patches(flags) except Exception as e: logger.warning("[grantforge_bootstrap] Search patches: %s", e) try: _apply_match_phase1_patches(flags) except Exception as e: logger.warning("[grantforge_bootstrap] Match phase1 patches: %s", e) try: _apply_neo4j_schema_patch(flags) except Exception as e: logger.warning("[grantforge_bootstrap] Neo4j schema patch: %s", e) try: _apply_hgis_migration(flags) except Exception as e: logger.warning("[grantforge_bootstrap] HGIS migration: %s", e) if _running_under_pytest(): logger.info( "[grantforge_bootstrap] Środowisko testowe — pomijam startowy wątek synchronizacji katalogu." ) elif _acquire_bootstrap_leader(): threading.Thread(target=_startup_catalog, daemon=True).start() try: _apply_endpoint_patches(flags) except Exception as e: logger.warning("[grantforge_bootstrap] Endpoint loaders: %s", e) logger.info( "[grantforge_bootstrap] patch-set version=%s enabled=%s disabled=%s", PATCH_SET_VERSION, enabled, disabled, ) _BOOTSTRAP_APPLIED = True _last_bootstrap_status = get_bootstrap_status() return _last_bootstrap_status