from __future__ import annotations import json from pathlib import Path from typing import Any from agente_contas_tim.workflows.contracts import WorkflowDef from agente_contas_tim.workflows.exceptions import ( WorkflowConfigurationError, WorkflowNotFoundError, ) class FileWorkflowRepository: """Repositório de workflow baseado em arquivos no disco. Convenções suportadas: - Versão explícita: ``.v.json|yaml|yml`` - Ativo por ponteiro: ``.active.json|yaml|yml`` - Pode conter um inteiro: ``{"version": 3}`` - Ou conter o workflow completo. - Sem ponteiro ativo: usa maior versão disponível. """ def __init__(self, base_dir: str | Path) -> None: self._base_dir = Path(base_dir) def get_active(self, name: str) -> WorkflowDef: self._ensure_base_dir() active = self._find_active_file(name) if active is not None: loaded = self._load_dict(active) if self._is_workflow_dict(loaded): return WorkflowDef.model_validate(loaded) version = loaded.get("version") if not isinstance(version, int): raise WorkflowConfigurationError( f"Arquivo ativo inválido: {active}. " "Esperado {'version': } ou workflow completo." ) return self.get_version(name, version) versions = self._list_versions(name) if not versions: raise WorkflowNotFoundError(f"Workflow {name!r} não encontrado") return self.get_version(name, max(versions)) def get_version(self, name: str, version: int) -> WorkflowDef: self._ensure_base_dir() candidate = self._find_version_file(name, version) if candidate is not None: return WorkflowDef.model_validate(self._load_dict(candidate)) raise WorkflowNotFoundError( f"Workflow {name!r} v{version} não encontrado em {self._base_dir}" ) def _ensure_base_dir(self) -> None: if not self._base_dir.exists(): raise WorkflowConfigurationError( f"Diretório de workflows não existe: {self._base_dir}" ) def _find_active_file(self, name: str) -> Path | None: return self._find_unique_file( [f"{name}.active.json", f"{name}.active.yaml", f"{name}.active.yml"], not_found_message=None, ambiguous_message=( "Múltiplos ponteiros ativos encontrados para " f"{name!r} em {self._base_dir}. " "Mantenha apenas um arquivo .active.." ), ) def _list_versions(self, name: str) -> list[int]: versions: list[int] = [] for file in self._base_dir.rglob(f"{name}.v*.*"): suffixes = file.suffixes if not suffixes: continue ext = suffixes[-1] if ext not in {".json", ".yaml", ".yml"}: continue stem = file.stem # stem ex.: vas_decision.v1 if ".v" not in stem: continue version_str = stem.rsplit(".v", maxsplit=1)[-1] if version_str.isdigit(): versions.append(int(version_str)) return versions def _find_version_file(self, name: str, version: int) -> Path | None: return self._find_unique_file( [ f"{name}.v{version}.json", f"{name}.v{version}.yaml", f"{name}.v{version}.yml", ], not_found_message=None, ambiguous_message=( f"Múltiplos workflows encontrados para {name!r} v{version} " f"em {self._base_dir}. Mantenha apenas um arquivo por versão." ), ) def _find_unique_file( self, names: list[str], *, not_found_message: str | None, ambiguous_message: str, ) -> Path | None: matches: list[Path] = [] for file_name in names: matches.extend(self._base_dir.rglob(file_name)) if not matches: if not_found_message is None: return None raise WorkflowConfigurationError(not_found_message) unique_matches = sorted({path.resolve() for path in matches}) if len(unique_matches) > 1: raise WorkflowConfigurationError(ambiguous_message) return unique_matches[0] def _load_dict(self, path: Path) -> dict[str, Any]: raw = path.read_text(encoding="utf-8") try: loaded = json.loads(raw) if not isinstance(loaded, dict): raise WorkflowConfigurationError( f"Workflow em {path} deve ser um objeto/dict" ) return loaded except json.JSONDecodeError: pass if path.suffix not in {".yaml", ".yml"}: raise WorkflowConfigurationError(f"Arquivo inválido: {path}") try: import yaml # type: ignore except ModuleNotFoundError as exc: raise WorkflowConfigurationError( f"Arquivo YAML detectado em {path}, mas PyYAML não está instalado. " "Instale a dependência 'pyyaml' ou use JSON." ) from exc loaded_yaml = yaml.safe_load(raw) if not isinstance(loaded_yaml, dict): raise WorkflowConfigurationError( f"Workflow YAML em {path} deve ser um objeto/dict" ) return loaded_yaml @staticmethod def _is_workflow_dict(data: dict[str, Any]) -> bool: return {"name", "version", "start", "nodes", "edges"} <= set(data.keys())