| from __future__ import annotations | |
| import argparse | |
| import difflib | |
| import sqlite3 | |
| from pathlib import Path | |
| import yaml | |
| from engine.refinement.multiagent_refinement import execute_refinement | |
| BASE_DEMO = Path(__file__).resolve().parent.parent | |
| DB_PATH = BASE_DEMO / "temp" / "audiodescriptions.db" | |
| CONFIG_PATH = BASE_DEMO / "temp" / "introspection_only.yaml" | |
| def ensure_introspection_only_config() -> Path: | |
| """Crea (o sobrescribe) un config.yaml mínimo con solo introspection activado.""" | |
| cfg = { | |
| "refinement": { | |
| "reflection_enabled": False, | |
| "reflexion_enabled": False, | |
| "introspection_enabled": True, | |
| } | |
| } | |
| CONFIG_PATH.parent.mkdir(parents=True, exist_ok=True) | |
| with CONFIG_PATH.open("w", encoding="utf-8") as f: | |
| yaml.safe_dump(cfg, f, allow_unicode=True) | |
| return CONFIG_PATH | |
| def load_srt_from_db(sha1sum: str | None = None, version: str | None = None) -> tuple[str, str, str]: | |
| """Carga un UNE SRT desde audiodescriptions.db. | |
| Si no se especifican sha1sum/version, toma la primera fila con une_ad no nulo. | |
| Devuelve (sha1sum, version, une_ad). | |
| """ | |
| if not DB_PATH.exists(): | |
| raise SystemExit(f"❌ No s'ha trobat la base de dades: {DB_PATH}") | |
| conn = sqlite3.connect(str(DB_PATH)) | |
| conn.row_factory = sqlite3.Row | |
| try: | |
| cur = conn.cursor() | |
| if sha1sum and version: | |
| row = cur.execute( | |
| "SELECT sha1sum, version, une_ad FROM audiodescriptions WHERE sha1sum=? AND version=?", | |
| (sha1sum, version), | |
| ).fetchone() | |
| else: | |
| row = cur.execute( | |
| "SELECT sha1sum, version, une_ad FROM audiodescriptions WHERE une_ad IS NOT NULL LIMIT 1", | |
| ).fetchone() | |
| if not row or not row["une_ad"]: | |
| raise SystemExit("❌ No s'ha trobat cap registre amb une_ad a audiodescriptions.db") | |
| return row["sha1sum"], row["version"], row["une_ad"] | |
| finally: | |
| conn.close() | |
| def show_diff(initial_srt: str, refined_srt: str) -> None: | |
| initial_lines = initial_srt.splitlines() | |
| refined_lines = refined_srt.splitlines() | |
| diff = difflib.unified_diff( | |
| initial_lines, | |
| refined_lines, | |
| fromfile="initial_une_ad.srt", | |
| tofile="introspected_une_ad.srt", | |
| lineterm="", | |
| ) | |
| for line in diff: | |
| print(line) | |
| def main() -> None: | |
| parser = argparse.ArgumentParser( | |
| description="Prova del pas d'introspection sobre un SRT de audiodescriptions.db (sense reflection/reflexion).", | |
| ) | |
| parser.add_argument("--sha1sum", type=str, default=None, help="SHA1 del vídeo (opcional)") | |
| parser.add_argument("--version", type=str, default=None, help="Versió de la AD (opcional)") | |
| args = parser.parse_args() | |
| cfg_path = ensure_introspection_only_config() | |
| sha1sum, version, une_ad = load_srt_from_db(args.sha1sum, args.version) | |
| print(f"Usant registre sha1sum={sha1sum}, version={version} de {DB_PATH}") | |
| refined = execute_refinement(une_ad, config_path=cfg_path) | |
| print("\n===== DIFF entre SRT inicial i SRT després d'introspection =====") | |
| show_diff(une_ad, refined) | |
| if __name__ == "__main__": | |
| main() | |