from __future__ import annotations import argparse import json import shutil import sqlite3 from pathlib import Path from core_db import CORE_ROOT, DB_PATH, SOURCE_ROOT, SECRETS_ROOT, connect from core_schema import init_core_schema from secret_store import save_secret from source_store import init_source_schema ROOT = Path(__file__).resolve().parent.parent SHARED = ROOT / "shared" LEGACY_DB = SHARED / "quality_rules.db" LEGACY_SOURCE_ROOT = SHARED / "quality_sources" LEGACY_CRED = SHARED / ".quality_techexpert_credentials.enc" LEGACY_KEY = SHARED / ".quality_techexpert_key" def _legacy_secret() -> dict: if not LEGACY_CRED.exists() or not LEGACY_KEY.exists(): return {} try: from cryptography.fernet import Fernet raw=Fernet(LEGACY_KEY.read_bytes().strip()).decrypt(LEGACY_CRED.read_bytes()) obj=json.loads(raw.decode("utf-8")) return obj if isinstance(obj,dict) else {} except Exception as exc: print("WARN: старые credentials не мигрированы:",exc) return {} def _copy_sources() -> None: if not LEGACY_SOURCE_ROOT.exists(): return target=SOURCE_ROOT/"techexpert" target.mkdir(parents=True,exist_ok=True) for item in LEGACY_SOURCE_ROOT.iterdir(): if item.name.startswith("_"): # Diagnostics/runtime logs are intentionally not migrated. continue dst=target/item.name if item.is_dir(): shutil.copytree(item,dst,dirs_exist_ok=True) elif item.is_file(): shutil.copy2(item,dst) def _rewrite_path(raw: str|None) -> str|None: if not raw: return raw p=Path(raw) try: rel=p.resolve().relative_to(LEGACY_SOURCE_ROOT.resolve()) except Exception: return raw return str((SOURCE_ROOT/"techexpert"/rel).resolve()) def _import_legacy_db() -> None: if not LEGACY_DB.exists(): return old=sqlite3.connect(LEGACY_DB) old.row_factory=sqlite3.Row with connect() as new: tables={r[0] for r in old.execute("SELECT name FROM sqlite_master WHERE type='table'")} if "tracked_legal_sources" in tables: for r in old.execute("SELECT * FROM tracked_legal_sources WHERE source_key='techexpert'"): d=dict(r) new.execute(""" INSERT INTO tracked_documents( source_key,document_slug,document_code,search_query,discovered_url,enabled, last_checked_at,last_changed_at,last_hash,last_snapshot_id,last_error ) VALUES(?,?,?,?,?,?,?,?,?,?,?) ON CONFLICT(source_key,document_slug) DO UPDATE SET document_code=excluded.document_code, search_query=excluded.search_query, discovered_url=COALESCE(excluded.discovered_url,tracked_documents.discovered_url), last_checked_at=COALESCE(excluded.last_checked_at,tracked_documents.last_checked_at), last_changed_at=COALESCE(excluded.last_changed_at,tracked_documents.last_changed_at), last_hash=COALESCE(excluded.last_hash,tracked_documents.last_hash), last_error=excluded.last_error """,( d["source_key"],d["document_slug"],d["document_code"],d["search_query"], d["discovered_url"],d["enabled"],d["last_checked_at"],d["last_changed_at"], d["last_hash"],None,d["last_error"] )) id_map={} if "source_snapshots" in tables: for r in old.execute("SELECT * FROM source_snapshots WHERE source_key='techexpert' ORDER BY id"): d=dict(r) # Avoid duplicate import by semantic identity. existing=new.execute(""" SELECT id FROM source_snapshots WHERE source_key=? AND document_slug=? AND checked_at=? AND content_hash=? """,(d["source_key"],d["document_slug"],d["checked_at"],d["content_hash"])).fetchone() if existing: new_id=existing["id"] else: cur=new.execute(""" INSERT INTO source_snapshots( source_key,document_slug,document_code,source_url,checked_at,content_hash,title, text_path,html_path,downloaded_file_path,manifest_path,is_current, extraction_status,extraction_meta_json ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?) """,( d["source_key"],d["document_slug"],d["document_code"],d["source_url"], d["checked_at"],d["content_hash"],d["title"],_rewrite_path(d["text_path"]), _rewrite_path(d["html_path"]),_rewrite_path(d["downloaded_file_path"]), _rewrite_path(d["manifest_path"]),d["is_current"],"legacy", json.dumps({"migrated_from":"quality_rules.db"},ensure_ascii=False) )) new_id=cur.lastrowid id_map[d["id"]]=new_id for old_id,new_id in id_map.items(): # Re-link tracked document if this was its current snapshot. row=old.execute(""" SELECT document_slug FROM tracked_legal_sources WHERE source_key='techexpert' AND last_snapshot_id=? """,(old_id,)).fetchone() if "tracked_legal_sources" in tables else None if row: new.execute("""UPDATE tracked_documents SET last_snapshot_id=? WHERE source_key='techexpert' AND document_slug=?""", (new_id,row["document_slug"])) if "external_sources" in tables: r=old.execute("SELECT * FROM external_sources WHERE source_key='techexpert'").fetchone() if r: d=dict(r) new.execute("""UPDATE connectors SET base_url=?,status=?,last_test_at=?,last_sync_at=?,last_error=? WHERE connector_key='techexpert'""", (d.get("base_url"),d.get("status") or "configured",d.get("last_login_at"), d.get("last_sync_at"),d.get("last_error"))) new.commit() old.close() def main(): ap=argparse.ArgumentParser(description="Миграция QUALITY source layer в ORIGUS Core") ap.add_argument("--apply",action="store_true",help="выполнить миграцию") args=ap.parse_args() print("Legacy DB:",LEGACY_DB,LEGACY_DB.exists()) print("Legacy sources:",LEGACY_SOURCE_ROOT,LEGACY_SOURCE_ROOT.exists()) print("Legacy credentials:",LEGACY_CRED.exists() and LEGACY_KEY.exists()) print("Core root:",CORE_ROOT) if not args.apply: print("DRY RUN. Для миграции: python migrate_from_quality.py --apply") return init_core_schema() init_source_schema() secret=_legacy_secret() if secret: save_secret("techexpert",secret) print("OK: credentials Техэксперта перенесены в Core.") _copy_sources() print("OK: нормативные файлы скопированы в Core (старые не удалены).") _import_legacy_db() print("OK: история источников импортирована.") print("ВАЖНО: старые QUALITY файлы/таблицы не удалены. Удалять их только после проверки Core.") if __name__=="__main__": main()