Spaces:
Running
Running
| #!/usr/bin/env python3 | |
| """Idempotentna migracja schematu Hybrid Grant Intelligence System.""" | |
| from __future__ import annotations | |
| import sys | |
| from pathlib import Path | |
| sys.path.insert(0, str(Path(__file__).resolve().parents[1])) | |
| from sqlalchemy import inspect, text | |
| from core.subscription.db import engine, init_models, Base | |
| NEW_GRANT_COLUMNS = { | |
| "eligible_beneficiaries": "JSON", | |
| "eligible_pkd": "JSON", | |
| "program_goals": "TEXT", | |
| "eligibility_criteria": "TEXT", | |
| "operator": "VARCHAR", | |
| "program_year": "INTEGER", | |
| "budget_min_pln": "FLOAT", | |
| "budget_max_pln": "FLOAT", | |
| "regulation_url": "VARCHAR", | |
| "official_page_url": "VARCHAR", | |
| "application_url": "VARCHAR", | |
| "catalog_hidden": "BOOLEAN DEFAULT 0", | |
| "completeness_score": "INTEGER DEFAULT 0", | |
| "source_credibility_score": "FLOAT DEFAULT 0.5", | |
| "search_document": "TEXT", | |
| } | |
| def _is_sqlite() -> bool: | |
| return "sqlite" in str(engine.url) | |
| def _column_ddl(col: str, col_type: str) -> str: | |
| """SQLite vs PostgreSQL — BOOLEAN DEFAULT 0 nie działa na Neon.""" | |
| if _is_sqlite(): | |
| return col_type | |
| if col == "catalog_hidden": | |
| return "BOOLEAN DEFAULT false" | |
| return col_type.replace("BOOLEAN DEFAULT 0", "BOOLEAN DEFAULT false") | |
| def migrate() -> dict: | |
| init_models() | |
| inspector = inspect(engine) | |
| existing_tables = set(inspector.get_table_names()) | |
| stats: dict = {"columns_added": 0, "tables_created": 0, "errors": []} | |
| if "grants" not in existing_tables: | |
| Base.metadata.tables["grants"].create(bind=engine, checkfirst=True) | |
| stats["tables_created"] += 1 | |
| existing_tables.add("grants") | |
| if "grants" in existing_tables: | |
| existing_cols = {c["name"] for c in inspector.get_columns("grants")} | |
| for col, col_type in NEW_GRANT_COLUMNS.items(): | |
| if col in existing_cols: | |
| continue | |
| ddl = _column_ddl(col, col_type) | |
| try: | |
| with engine.begin() as conn: | |
| conn.execute(text(f"ALTER TABLE grants ADD COLUMN {col} {ddl}")) | |
| stats["columns_added"] += 1 | |
| existing_cols.add(col) | |
| except Exception as exc: | |
| msg = str(exc).lower() | |
| if "duplicatecolumn" in msg or "already exists" in msg: | |
| existing_cols.add(col) | |
| continue | |
| stats["errors"].append(f"{col}: {exc}") | |
| for table in ("grant_versions", "research_jobs", "human_verification_queue"): | |
| if table not in existing_tables: | |
| Base.metadata.tables[table].create(bind=engine, checkfirst=True) | |
| stats["tables_created"] += 1 | |
| return stats | |
| if __name__ == "__main__": | |
| result = migrate() | |
| print(f"Migration OK: {result}") | |