#!/usr/bin/env python3 """ Visual Guide Pipeline Orchestrator & Deterministic Pre-Flight Linter (v3.0.0 — Modular) ======================================================================================== Motor agnóstico de engenharia editorial, copywriting de conversão e SEO Amazon KDP para produção automatizada de Guias Visuais de Consulta Rápida. Arquitetura Modular em 10 Etapas: core/ -> Máquina de estados, geometria KDP, contraste WCAG 2.1 AA e linters Fail-Fast. agents/ -> 5 Agentes Especialistas com ciclo de vida execute() e revise() (Etapa 9). """ from __future__ import annotations import argparse import datetime as dt import json import sys from pathlib import Path from typing import Any, Dict, List, Optional, Tuple from agents import ( CopywriterAgent, InformationArchitectAgent, PipelineContext, PreFlightAndPackagingAgent, TrendHunterAndSEOAgent, VisualDesignerAgent, ) from core import ( ClosureReport, PipelineState, SessionClosureProtocol, StateMachine, compute_page_count_plan, load_smart_context, sync_visual_hub, ) # Catálogo canônico padrão de demonstração (Páginas 01 a 04) DEFAULT_PAGE_CATALOG: List[Tuple[str, str, str, str, str]] = [ ( "Model_B", "summary_grid", "Arquitetura da Rotina Segura", "Domínio Visual em 5 Segundos", "guia visual cuidados recem nascido rotina segura", ), ( "Model_A", "rapid_triage", "Sinais Críticos de Alerta", "Quando Agir Sem Hesitar", "sinais alerta recem nascido engasgo febre emergencia", ), ( "Model_D", "sequential_steps", "Banho Calmo e Higienização", "Proteção Térmica Passo a Passo", "primeiro banho bebe temperatura agua coto umbilical", ), ( "Model_C", "binary_comparison", "Enxoval Inteligente e Essencial", "O Que Comprar vs. O Que Cortar", "lista enxoval bebe economico itens essenciais", ), ] class PipelineOrchestrator: """Orquestrador mestre encarregado de coordenar a máquina de estados e os 5 agentes especialistas.""" def __init__(self, base_dir: Path, cron_cfg: Dict[str, Any]) -> None: self.base_dir = base_dir self.cron_cfg = cron_cfg self.agent_1 = TrendHunterAndSEOAgent() self.agent_2 = InformationArchitectAgent() self.agent_3 = CopywriterAgent() self.agent_4 = VisualDesignerAgent() self.agent_5 = PreFlightAndPackagingAgent() self.closure_protocol = SessionClosureProtocol(base_dir=self.base_dir) def close_session( self, session_data: Dict[str, Any], next_activity: Optional[str] = None, result_summary: Optional[str] = None, ) -> ClosureReport: """Executa transversalmente o Protocolo de Encerramento e Aprendizado Contínuo.""" return self.closure_protocol.execute( session_data=session_data, next_activity=next_activity, result_summary=result_summary, ) def get_smart_context(self, target_phase_or_task: str) -> Dict[str, Any]: """Recupera contexto documental mínimo para a tarefa, otimizando tokens.""" return load_smart_context(self.base_dir, target_phase_or_task) def run_page( self, page_number: int, layout_model: str, template_type: str, title_context: str, title_hook: str, seo_keyword: str, domain: str, authority_sources: List[str], page_plan: Dict[str, Any], kdp_rules: Dict[str, Any], style_tokens: Dict[str, Any], ) -> Dict[str, Any]: """Executa integralmente as 10 etapas da arquitetura v3.0.0 para uma prancha com autocorreção na Etapa 9.""" max_retries = int(self.cron_cfg.get("retry_policy", {}).get("max_retries_per_step", 3)) sm = StateMachine(max_retries=max_retries) context = PipelineContext( base_dir=self.base_dir, page_number=page_number, layout_model=layout_model, template_type=template_type, title_context=title_context, title_hook=title_hook, seo_keyword=seo_keyword, domain=domain, authority_sources=authority_sources, page_plan=page_plan, cron_cfg=self.cron_cfg, kdp_rules=kdp_rules, style_tokens=style_tokens, ) # ------------------------------------------------------------------------- # ETAPA 1: Pesquisa Científica, Demanda & SEO (Agente 1) # ------------------------------------------------------------------------- sm.transition( PipelineState.RESEARCHING_SEO, f"Etapa 1: Pesquisa Científica, Demanda & SEO da página {page_number:02d}", ) res1 = self.agent_1.execute(context) if not res1.success: sm.transition(PipelineState.FAILED, "; ".join(res1.errors)) raise RuntimeError(f"Falha na Etapa 1: {res1.errors}") # ------------------------------------------------------------------------- # ETAPA 2: Briefing Funcional & Gabarito (Agente 1 & Agente 2) # ------------------------------------------------------------------------- sm.transition( PipelineState.BRIEFING, f"Etapa 2: Briefing Funcional & Gabarito ({layout_model} / {template_type})", ) res2 = self.agent_2.execute_briefing(context) if not res2.success: sm.transition(PipelineState.FAILED, "; ".join(res2.errors)) raise RuntimeError(f"Falha na Etapa 2: {res2.errors}") # ------------------------------------------------------------------------- # ETAPA 3: Copywriting Editorial & Redação [Trava Fail-Fast] (Agente 3) # ------------------------------------------------------------------------- sm.transition( PipelineState.DRAFTING_COPY, "Etapa 3: Copywriting Editorial Bicolor & Redação Concisa [Fail-Fast]", ) res3 = self.agent_3.execute(context) if not res3.success: sm.transition(PipelineState.FAILED, "; ".join(res3.errors)) raise RuntimeError(f"Bloqueio Fail-Fast na Etapa 3: {res3.errors}") # ------------------------------------------------------------------------- # ETAPA 4: Planejamento Visual & Wireframe A4 [Trava Fail-Fast] (Agente 2) # ------------------------------------------------------------------------- sm.transition( PipelineState.OUTLINING_GRID, f"Etapa 4: Planejamento Visual & Wireframe A4 ({layout_model}) [Fail-Fast]", ) res4 = self.agent_2.execute_wireframe(context) if not res4.success: sm.transition(PipelineState.FAILED, "; ".join(res4.errors)) raise RuntimeError(f"Bloqueio Fail-Fast na Etapa 4: {res4.errors}") # ------------------------------------------------------------------------- # ETAPAS 5, 6 e 7: Técnica Visual, Prompt Limpo & Diagramação Híbrida (Ag. 4) # ------------------------------------------------------------------------- sm.transition( PipelineState.DESIGNING_ASSETS, "Etapas 5, 6 e 7: Técnica Segura, Prompt Limpo e Prancha Vetorial Híbrida", ) res57 = self.agent_4.execute(context) if not res57.success: sm.transition(PipelineState.FAILED, "; ".join(res57.errors)) raise RuntimeError(f"Falha na geração de ativos (Etapas 5-7): {res57.errors}") # ------------------------------------------------------------------------- # ETAPAS 8 e 9: Auditoria Pré-Voo & Protocolo de Correção Cirúrgica # ------------------------------------------------------------------------- sm.transition(PipelineState.PREFLIGHT_CHECK, "Etapa 8: Auditoria Pré-Voo Integral (KDP + WCAG + SEO)") passed, errors = self.agent_5.validate(context) context.steps_log.append({"step": 8, "name": "Auditoria Pré-Voo Integral", "passed": passed, "errors": errors}) # Loop de Correção Cirúrgica da Etapa 9 com autorrecuperação while not passed: target_agent_desc = self.agent_5.route_deviation(context, errors) sm.transition( PipelineState.REVISION, f"Etapa 9: Acionando {target_agent_desc} -> {'; '.join(errors)}", ) if sm.state == PipelineState.FAILED: context.steps_log.append( {"step": 9, "name": "Protocolo de Correção Cirúrgica", "routed_to": target_agent_desc, "errors": errors} ) raise RuntimeError(f"Pre-Flight reprovado após retentativas ({target_agent_desc}): {errors}") # Despacho cirúrgico para o agente causador if "Agente 1" in target_agent_desc: self.agent_1.revise(context, errors) elif "Agente 2" in target_agent_desc: self.agent_2.revise(context, errors) elif "Agente 3" in target_agent_desc: self.agent_3.revise(context, errors) elif "Agente 4" in target_agent_desc: self.agent_4.revise(context, errors) else: self.agent_3.revise(context, errors) # Revalidação no Gate de Pré-Voo (Etapa 8) sm.transition(PipelineState.PREFLIGHT_CHECK, "Revalidação no Gate de Pré-Voo pós-correção") passed, errors = self.agent_5.validate(context) context.steps_log.append( {"step": 8, "name": "Revalidação Pré-Voo", "passed": passed, "errors": errors} ) context.steps_log.append({"step": 9, "name": "Protocolo de Correção Cirúrgica", "status": "NOT_NEEDED_ZERO_DEFECTS"}) # ------------------------------------------------------------------------- # ETAPA 10: Compilação da Página & Manifesto Auditável # ------------------------------------------------------------------------- sm.transition(PipelineState.PACKAGING_AND_SEO, "Etapa 10: Empacotamento e gravação do manifesto v3.0.0") sm.transition(PipelineState.COMPLETED, "Prancha 100% aprovada e empacotada.") res10 = self.agent_5.package_manifest(context, sm.state.value, sm.history) return res10.data # ============================================================================ # INTERFACE DE COMPATIBILIDADE DIRETA COM A ESPECIFICAÇÃO # ============================================================================ def run_ten_step_pipeline_for_page( base_dir: Path, page_number: int, layout_model: str, template_type: str, title_context: str, title_hook: str, seo_keyword: str, domain: str, authority_sources: List[str], page_plan: Dict[str, Any], cron_cfg: Dict[str, Any], kdp_rules: Dict[str, Any], style_tokens: Dict[str, Any], ) -> Dict[str, Any]: """Função de compatibilidade que instancia o PipelineOrchestrator e processa uma página.""" orchestrator = PipelineOrchestrator(base_dir=base_dir, cron_cfg=cron_cfg) return orchestrator.run_page( page_number=page_number, layout_model=layout_model, template_type=template_type, title_context=title_context, title_hook=title_hook, seo_keyword=seo_keyword, domain=domain, authority_sources=authority_sources, page_plan=page_plan, kdp_rules=kdp_rules, style_tokens=style_tokens, ) def build_and_validate_seo_copy_pack( commercial_title: str, commercial_subtitle: str, domain: str, page_plan: Dict[str, Any], kdp_rules: Dict[str, Any], ) -> Tuple[Dict[str, Any], List[str]]: """Encaminha a geração e validação do pacote comercial SEO para o Agente 3.""" return CopywriterAgent.build_and_validate_seo_copy_pack( commercial_title=commercial_title, commercial_subtitle=commercial_subtitle, domain=domain, page_plan=page_plan, kdp_rules=kdp_rules, ) # ============================================================================ # INTERFACE DE LINHA DE COMANDO (CLI AGNÓSTICA v3.0.0) # ============================================================================ def load_json(path: Path) -> Dict[str, Any]: with path.open("r", encoding="utf-8") as fh: return json.load(fh) def parse_args(argv: Optional[List[str]] = None) -> argparse.Namespace: parser = argparse.ArgumentParser( description="Orquestrador Agnóstico de Guias Visuais v3.0.0 (10 Etapas + SEO KDP + Fail-Fast Modular)" ) parser.add_argument("--guide-title", type=str, default=None, help="Título comercial principal do guia.") parser.add_argument("--subtitle", type=str, default=None, help="Subtítulo comercial/SEO com descritor de utilidade.") parser.add_argument("--domain", type=str, default=None, help="Domínio técnico/editorial do guia.") parser.add_argument("--total-pages", type=int, default=None, help="Quantidade de páginas da edição digital PDF.") parser.add_argument("--kdp-pages", type=int, default=None, help="Meta de páginas da edição física KDP.") parser.add_argument("--page", type=int, default=None, help="Executa uma página específica (ex.: 1).") parser.add_argument("--all", action="store_true", help="Processa todas as páginas do catálogo do guia.") parser.add_argument( "--model", type=str, default="Model_A", choices=["Model_A", "Model_B", "Model_C", "Model_D"], help="Modelo de mancha gráfica para --page.", ) parser.add_argument( "--template", type=str, default="rapid_triage", choices=[ "summary_grid", "central_infographic", "binary_comparison", "sequential_steps", "rapid_triage", ], help="Gabarito estrutural da página.", ) parser.add_argument("--title-context", type=str, default="Protocolo Operacional", help="Contexto do título bicolor (#243832).") parser.add_argument("--title-hook", type=str, default="Ação Segura em 5 Segundos", help="Gancho do título bicolor (#C06845).") parser.add_argument("--seo-keyword", type=str, default="guia visual protocolo passo a passo", help="Keyword SEO alvo da página.") parser.add_argument("--close-session", action="store_true", help="Executa o Protocolo de Encerramento e Aprendizado Contínuo ao término.") parser.add_argument( "--session-phase", type=str, default="execucao_geral", choices=["pesquisa", "briefing", "copywriting", "planejamento_visual", "diagramacao", "preflight", "revisao", "execucao_geral"], help="Fase da sessão para registro e extração de aprendizados.", ) parser.add_argument( "--smart-context", type=str, default=None, help="Exibe o contexto inteligente para a fase informada (otimização de tokens).", ) parser.add_argument( "--sync-hub", action="store_true", help="Sincroniza incrementalmente docs/painel_visual/pranchas_registry.js sem gastar tokens de IA.", ) parser.add_argument( "--deploy-hub", action="store_true", help="Sincroniza o Painel Visual e publica online no Cloudflare Pages (plano gratuito visual-guide-pipeline).", ) return parser.parse_args(argv) def main(argv: Optional[List[str]] = None) -> int: args = parse_args(argv) base_dir = Path(__file__).resolve().parent if (args.sync_hub or args.deploy_hub) and args.page is None and not args.all and not args.close_session and not args.smart_context: payload = sync_visual_hub(base_dir) stats = payload["stats"] print("=== SINCRONIZAÇÃO INCREMENTAL DO PAINEL VISUAL (ZERO-TOKEN) ===") print("Registro atualizado : docs/painel_visual/pranchas_registry.js") print( f"Pranchas no Painel : {stats['total_visual_boards']} | " f"Módulos Cobertos: {stats['modules_with_visual_boards']}/{stats['total_mvp_modules']} | " f"Manifestos Engine: {stats['total_engine_manifests']}" ) if args.deploy_hub: import subprocess hub_dir = base_dir / "docs" / "painel_visual" cmd = [ "npx.cmd" if sys.platform == "win32" else "npx", "wrangler", "pages", "deploy", str(hub_dir), "--project-name", "visual-guide-pipeline", "--branch", "main", "--commit-dirty=true", ] return subprocess.call(cmd, cwd=str(base_dir)) return 0 cron_cfg = load_json(base_dir / "pipeline_cron.json") scope = cron_cfg["scope"] master_spec_path = base_dir / scope["master_spec_file"] if not master_spec_path.exists(): print(f"[ERRO] Documento Mestre ausente: {master_spec_path}", file=sys.stderr) return 1 kdp_rules = load_json(base_dir / scope["rules_file"]) style_tokens = load_json(base_dir / scope["tokens_file"]) commercial_title = args.guide_title or scope["commercial_title"] commercial_subtitle = args.subtitle or scope["commercial_subtitle"] domain = args.domain or scope["domain"] authority_sources = scope.get("authority_sources", ["Diretrizes Clínicas Oficiais"]) digital_pages = args.total_pages or int(scope["page_count_engineering"]["direct_sale_pdf_pages"]) kdp_pages = args.kdp_pages or int(scope["page_count_engineering"]["active_kdp_target_pages"]) page_plan = compute_page_count_plan(digital_pages, kdp_pages, kdp_rules) orchestrator = PipelineOrchestrator(base_dir=base_dir, cron_cfg=cron_cfg) if args.smart_context: ctx = orchestrator.get_smart_context(args.smart_context) print(f"\n=== CONSULTA INTELIGENTE DE DOCUMENTAÇÃO (FASE: {args.smart_context.upper()}) ===") print("Seções Carregadas:") for section, text in ctx["loaded_sections"].items(): print(f" - [{section.upper()}]: {len(text.splitlines())} linhas carregadas") stats = ctx["token_economy_stats"] print(f"Economia Estimada de Tokens: {stats['economy_percentage']}% ({stats['chars_saved']} caracteres poupados)") return 0 print(f"=== {cron_cfg['pipeline_name']} v{cron_cfg['version']} (Modular Engine) ===") print(f"Documento Mestre : {scope['master_spec_file']}") print(f"Título Comercial : {commercial_title}: {commercial_subtitle}") print(f"Domínio : {domain}") print( f"Plano de Páginas : Digital = {page_plan['direct_sale_pdf_pages']} págs | " f"KDP Print = {page_plan['kdp_print_target_pages']} págs ({page_plan['kdp_color_tier']}) | " f"Gutter = {page_plan['required_inside_gutter_mm']} mm | " f"Custo KDP Est. = USD {page_plan['estimated_kdp_print_cost_usd']:.2f}" ) if args.page is not None: pages_to_run = [ ( args.page, args.model, args.template, args.title_context, args.title_hook, args.seo_keyword, ) ] elif args.close_session and not args.all: pages_to_run = [] else: pages_to_run = [ (idx, m, tpl, ctx, hk, kw) for idx, (m, tpl, ctx, hk, kw) in enumerate(DEFAULT_PAGE_CATALOG, start=1) ] for page_num, model_key, tpl_key, ctx_txt, hook_txt, kw_txt in pages_to_run: manifest = orchestrator.run_page( page_number=page_num, layout_model=model_key, template_type=tpl_key, title_context=ctx_txt, title_hook=hook_txt, seo_keyword=kw_txt, domain=domain, authority_sources=authority_sources, page_plan=page_plan, kdp_rules=kdp_rules, style_tokens=style_tokens, ) print( f"[OK] Prancha {page_num:02d} ({model_key} / {tpl_key}) -> {manifest['status']} " f"| {manifest['deliverables']['manifest_json']}" ) out_dir = base_dir / scope["output_directory"] deliverables_list = [f"Prancha {p[0]:02d}" for p in pages_to_run] if pages_to_run: # Etapa 10 (Nível Livro): Compilação e Auditoria do Pacote Comercial & SEO KDP seo_pack, seo_errors = build_and_validate_seo_copy_pack( commercial_title=commercial_title, commercial_subtitle=commercial_subtitle, domain=domain, page_plan=page_plan, kdp_rules=kdp_rules, ) if seo_errors: print(f"[ERRO SEO KDP] Falha na auditoria do pacote comercial: {seo_errors}", file=sys.stderr) return 2 seo_pack_path = out_dir / "guide_seo_copy_pack.json" seo_pack_path.write_text(json.dumps(seo_pack, indent=2, ensure_ascii=False), encoding="utf-8") print(f"[OK] Pacote Comercial & SEO Amazon KDP validado -> {seo_pack_path.name}") deliverables_list.append(seo_pack_path.name) # Execução do Protocolo Obrigatório de Encerramento e Aprendizado Contínuo session_data = { "objective": f"Produção e validação do guia '{commercial_title}'", "phase": args.session_phase, "activities": [f"Processamento de {len(pages_to_run)} prancha(s)", "Auditoria Pré-Voo e Pacote SEO"] if pages_to_run else ["Encerramento operacional e sincronização estratégica"], "deliverables": deliverables_list, "decisions": [f"Adoção de {page_plan['kdp_print_target_pages']} páginas com gutter de {page_plan['required_inside_gutter_mm']} mm"], "instructions": ["Respeitar safe zone de 9.6 mm e limite de 45 palavras por bloco"], "preferences": ["Layout contemporâneo escandinavo com títulos bicolores"], "corrections": [], "successful_techniques": ["Autocorreção cirúrgica na Etapa 9 e validações determinísticas"], "errors": [], "pending_tasks": ["Revisão pré-gráfica e teste piloto das próximas pranchas"] if len(pages_to_run) < page_plan["direct_sale_pdf_pages"] else ["Todas as pranchas do guia concluídas"], "proposed_improvements": [], } closure_report = orchestrator.close_session(session_data=session_data) print("\n" + closure_report.markdown_report) return 0 if __name__ == "__main__": sys.exit(main())