""" Motor de Sincronização Incremental do Painel Visual Central (Zero-Token-Waste). =============================================================================== Varre automaticamente STATUS_PROJETO.md, docs/painel_visual/pranchas/*.json, guias/*/ e workspace/output/*_manifest.json para gerar docs/painel_visual/pranchas_registry.js. Uso de arquivo .js (window.VGP_HUB_DATA = {...}) garante funcionamento 100% offline via protocolo file:/// no Windows (Chrome/Edge) sem bloqueio de CORS e sem exigir que agentes de IA reescrevam o index.html a cada nova prancha entregue. """ from __future__ import annotations import datetime as dt import json import re import shutil from pathlib import Path from typing import Any, Dict, List REFERENCE_FILES_TO_MIRROR: List[str] = [ "specs/rules/FONTE_DA_VERDADE_CLINICA.md", "specs/rules/clinical_ground_truth.json", "specs/ESCOPO_CANONICO_MVP_PILOTO.md", "specs/METODOLOGIA_PILOTO_PRANCHAS.md", "STATUS_PROJETO.md", "REGRAS_GERAIS.md", "AGENTS.md", "PREFERENCIAS.md", "orchestrator.py", "core/linters.py", "core/visual_hub_sync.py", ] MVP_MODULES_CATALOG: List[Dict[str, Any]] = [ {"module": 1, "layer": "Camada 1", "title": "Abertura, Enquadramento & Blindagem Legal", "format": "A4 Simples"}, {"module": 2, "layer": "Camada 1", "title": "Triagem em 5s, Semáforo & Sólido vs. Líquido", "format": "A4 Simples / Dupla"}, {"module": 3, "layer": "Camada 1", "title": "Protocolo Central de Desengasgo (AHA 2025)", "format": "Spread Duplo (420×297 mm)"}, {"module": 4, "layer": "Camada 1", "title": "Parada, RCP Neonatal (30:2) & Script 192", "format": "Spread / A4 Contínuo"}, {"module": 5, "layer": "Camada 2", "title": "Mamada, Refluxo & Lavagem Nasal Segura", "format": "Expansão Livre"}, {"module": 6, "layer": "Camada 2", "title": "Introdução Alimentar Segura, BLW & Alérgenos", "format": "Expansão Livre"}, {"module": 7, "layer": "Camada 2", "title": "Pilha-Botão, Ímãs & Disque-Intoxicação 0800", "format": "Expansão Livre"}, {"module": 8, "layer": "Camada 2", "title": "Sono Seguro (SMSL), Sling, Banho & Queimaduras", "format": "Expansão Livre"}, {"module": 9, "layer": "Camada 2", "title": "Quedas, Febre, Convulsão, Coto & Ficha Médica", "format": "Expansão Livre"}, {"module": 10, "layer": "Camada 2", "title": "Anexo Destacável: Cartaz Rápido de Geladeira", "format": "A4 Alta Densidade"}, ] def _parse_status_markdown(status_path: Path) -> Dict[str, Any]: """Extrai o estado vivo atual diretamente de STATUS_PROJETO.md.""" result: Dict[str, Any] = { "updated_date": "", "phase": "indefinida", "last_activity": "", "result_obtained": "", "next_activity": "", "pending_tasks": [], } if not status_path.exists(): return result text = status_path.read_text(encoding="utf-8") m_date = re.search(r"Atualizado em ([0-9\-]+)", text) if m_date: result["updated_date"] = m_date.group(1) patterns = { "phase": r"-\s*\*\*Fase Atual:\*\*\s*(.+)", "last_activity": r"-\s*\*\*Última Atividade Concluída:\*\*\s*(.+)", "result_obtained": r"-\s*\*\*Resultado Obtido:\*\*\s*(.+)", "next_activity": r"-\s*\*\*Próxima Atividade:\*\*\s*(.+)", } for key, pat in patterns.items(): m = re.search(pat, text) if m: result[key] = m.group(1).strip() for line in text.splitlines(): m_task = re.match(r"^\s*-\s*\[([ xX])\]\s*(.+)$", line) if m_task: result["pending_tasks"].append( { "done": m_task.group(1).lower() == "x", "text": m_task.group(2).strip(), } ) return result def _collect_visual_boards(hub_dir: Path) -> List[Dict[str, Any]]: """Coleta todas as pranchas visuais incrementais registradas em docs/painel_visual/pranchas/*.json.""" boards_dir = hub_dir / "pranchas" boards: List[Dict[str, Any]] = [] if not boards_dir.exists(): return boards for jf in sorted(boards_dir.glob("*.json")): try: data = json.loads(jf.read_text(encoding="utf-8")) data["_source_file"] = f"pranchas/{jf.name}" boards.append(data) except Exception: continue boards.sort(key=lambda b: (int(b.get("module", 99)), int(b.get("page_start", 99)))) return boards def _collect_pipeline_manifests(base_dir: Path) -> List[Dict[str, Any]]: """Coleta manifestos JSON gerados pelo orquestrador em workspace/output/.""" out_dir = base_dir / "workspace" / "output" manifests: List[Dict[str, Any]] = [] if not out_dir.exists(): return manifests for mf in sorted(out_dir.glob("page_*_manifest.json")): try: raw = json.loads(mf.read_text(encoding="utf-8")) spec = raw.get("page_specification", {}) blocks = spec.get("text_blocks", []) word_counts = [len(str(b.get("text", "")).split()) for b in blocks] manifests.append( { "page_number": spec.get("page_number", 0), "title": spec.get("title", mf.stem), "layout_model": spec.get("layout_model", ""), "template_type": spec.get("template_type", ""), "status": raw.get("status", "UNKNOWN"), "domain": raw.get("domain", ""), "word_counts": word_counts, "manifest_rel_path": f"acervo/workspace/output/{mf.name}", } ) except Exception: continue return manifests def _mirror_reference_files(base_dir: Path, hub_dir: Path) -> None: """ Espelha documentos e manifestos referenciados para docs/painel_visual/acervo/ garantindo zero links quebrados tanto offline (file:///) quanto online no Cloudflare Pages. """ acervo_dir = hub_dir / "acervo" for rel_str in REFERENCE_FILES_TO_MIRROR: src = base_dir / rel_str if src.exists(): dst = acervo_dir / rel_str dst.parent.mkdir(parents=True, exist_ok=True) shutil.copyfile(src, dst) out_dir = base_dir / "workspace" / "output" if out_dir.exists(): dst_out = acervo_dir / "workspace" / "output" dst_out.mkdir(parents=True, exist_ok=True) for jf in out_dir.glob("*.json"): shutil.copyfile(jf, dst_out / jf.name) headers_path = hub_dir / "_headers" if not headers_path.exists(): headers_path.write_text( "/*.md\n Content-Type: text/plain; charset=utf-8\n" "/*.py\n Content-Type: text/plain; charset=utf-8\n" "/*.json\n Content-Type: application/json; charset=utf-8\n" "/*.js\n Content-Type: application/javascript; charset=utf-8\n", encoding="utf-8", ) def sync_visual_hub(base_dir: Path) -> Dict[str, Any]: """ Atualiza de forma determinística e incremental o registro do Painel Visual (docs/painel_visual/pranchas_registry.js) sem tocar no HTML principal. """ base = Path(base_dir) hub_dir = base / "docs" / "painel_visual" hub_dir.mkdir(parents=True, exist_ok=True) (hub_dir / "pranchas").mkdir(parents=True, exist_ok=True) project_status = _parse_status_markdown(base / "STATUS_PROJETO.md") visual_boards = _collect_visual_boards(hub_dir) engine_manifests = _collect_pipeline_manifests(base) _mirror_reference_files(base, hub_dir) # Mapeia cobertura incremental por módulo (1 a 10) boards_by_mod: Dict[int, List[Dict[str, Any]]] = {} for b in visual_boards: mod_num = int(b.get("module", 0)) boards_by_mod.setdefault(mod_num, []).append(b) modules_summary: List[Dict[str, Any]] = [] for mod in MVP_MODULES_CATALOG: m_num = mod["module"] linked = boards_by_mod.get(m_num, []) pdf_ready = any(bool(x.get("pdf_path")) for x in linked) if pdf_ready: m_state = "PDF_300DPI_PRONTO" elif linked: m_state = "PROTOTIPO_VISUAL_NO_PAINEL" else: m_state = "AGUARDANDO_PRANCHA" modules_summary.append( { **mod, "boards_count": len(linked), "module_state": m_state, "board_ids": [x.get("id", "") for x in linked], } ) payload: Dict[str, Any] = { "schema_version": "1.0.0", "generated_at": dt.datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "project_status": project_status, "stats": { "total_mvp_modules": len(MVP_MODULES_CATALOG), "modules_with_visual_boards": sum(1 for m in modules_summary if m["boards_count"] > 0), "modules_with_final_pdf": sum( 1 for m in modules_summary if m["module_state"] == "PDF_300DPI_PRONTO" ), "total_visual_boards": len(visual_boards), "total_engine_manifests": len(engine_manifests), }, "modules_summary": modules_summary, "visual_boards": visual_boards, "engine_manifests": engine_manifests, } registry_js_path = hub_dir / "pranchas_registry.js" js_content = ( "// ARQUIVO GERADO AUTOMATICAMENTE POR core/visual_hub_sync.py — NÃO EDITAR MANUALMENTE\n" "// Carregado via