Spaces:
Paused
Paused
| # aduc_framework/director.py | |
| # | |
| # Copyright (C) August 4, 2025 Carlos Rodrigues dos Santos | |
| # | |
| # Versão 4.0.0 (Gerenciamento de Estado Persistente) | |
| # | |
| # - O Diretor agora gerencia um arquivo `dna.json` dentro de um diretório de projeto. | |
| # - No `__init__`, ele tenta carregar um estado existente do arquivo ou cria um novo. | |
| # - O método `save_state()` permite que o Maestro ADUC comande a persistência | |
| # do estado atual em disco em marcos importantes do processo. | |
| import logging | |
| import os | |
| import json | |
| from pathlib import Path | |
| from typing import List, Dict, Any | |
| from .types import GenerationState, PreProductionParams, Scene, Ato, MediaRef | |
| logger = logging.getLogger(__name__) | |
| class AducDirector: | |
| """ | |
| Representa o Diretor de Cena, responsável por gerenciar o estado | |
| persistente de um único projeto (`dna.json`). | |
| """ | |
| def __init__(self, project_path: str): | |
| self.project_path = Path(project_path) | |
| self.dna_file_path = self.project_path / "dna.json" | |
| self.project_path.mkdir(parents=True, exist_ok=True) | |
| self.state: GenerationState = self._load_or_initialize_state() | |
| logger.info(f"AducDirector inicializado para o projeto em '{self.project_path}'.") | |
| def _load_or_initialize_state(self) -> GenerationState: | |
| """Carrega o estado do dna.json ou cria um novo se não existir.""" | |
| if self.dna_file_path.exists(): | |
| try: | |
| logger.info(f"Encontrado dna.json existente. Carregando estado...") | |
| with open(self.dna_file_path, 'r', encoding='utf-8') as f: | |
| state_dict = json.load(f) | |
| return GenerationState(**state_dict) | |
| except (json.JSONDecodeError, TypeError) as e: | |
| logger.error(f"Falha ao carregar ou validar dna.json: {e}. Criando um novo estado.") | |
| return GenerationState(workspace_dir=str(self.project_path)) | |
| else: | |
| logger.info("Nenhum dna.json encontrado. Criando novo estado de geração.") | |
| return GenerationState(workspace_dir=str(self.project_path)) | |
| def save_state(self): | |
| """Salva o estado atual (self.state) no arquivo dna.json.""" | |
| logger.info(f"Persistindo estado atual para '{self.dna_file_path}'...") | |
| with open(self.dna_file_path, 'w', encoding='utf-8') as f: | |
| f.write(self.state.model_dump_json(indent=2)) | |
| logger.info("Estado salvo com sucesso.") | |
| def get_full_state(self) -> GenerationState: | |
| """Retorna o objeto de estado Pydantic completo.""" | |
| return self.state | |
| def load_state_from_dict(self, state_dict: Dict[str, Any]): | |
| """Carrega o estado a partir de um dicionário (ex: vindo de um planner).""" | |
| try: | |
| self.state = GenerationState(**state_dict) | |
| logger.info("Diretor: Estado em memória atualizado com sucesso a partir de um dicionário.") | |
| except Exception as e: | |
| logger.error(f"Diretor: Falha ao carregar estado a partir do dicionário: {e}", exc_info=True) | |
| def update_parameters(self, stage: str, params: Any): | |
| """Atualiza o nó de parâmetros no estado de geração.""" | |
| if hasattr(self.state.parametros_geracao, stage): | |
| setattr(self.state.parametros_geracao, stage, params) | |
| else: | |
| logger.warning(f"Tentativa de atualizar parâmetros para um estágio desconhecido: '{stage}'") | |
| def update_state_from_pre_production_dna(self, params: PreProductionParams, final_dna_json: Dict[str, Any]): | |
| """Analisa o dicionário bruto do Composer e o usa para popular o estado.""" | |
| logger.info("Diretor: Recebendo DNA da pré-produção para popular o estado.") | |
| self.state.parametros_geracao.pre_producao = params | |
| self.state.prompt_geral = final_dna_json.get("global_prompt", "") | |
| ref_paths = final_dna_json.get("initial_media_paths", []) | |
| self.state.midias_referencia = [MediaRef(id=i, caminho=path) for i, path in enumerate(ref_paths)] | |
| self.state.scenes = [] | |
| scenes_data = final_dna_json.get("scenes", []) | |
| for scene_dict in scenes_data: | |
| ato_objects = [ | |
| Ato(id=ato_dict.get("act_id", i), resumo_ato=ato_dict.get("context", "")) | |
| for i, ato_dict in enumerate(scene_dict.get("acts", [])) | |
| ] | |
| scene_object = Scene(id=scene_dict.get("scene_id"), atos=ato_objects) | |
| self.state.scenes.append(scene_object) | |
| logger.info(f"Diretor: Estado populado com sucesso. {len(self.state.scenes)} cenas carregadas.") |