Coverage for src/jtech_installer/installer/asset_copier.py: 56%
111 statements
« prev ^ index » next coverage.py v7.8.0, created at 2025-08-20 15:10 -0300
1"""
2Asset copier for JTECH™ Installer - handles copying framework files
3"""
5import hashlib
6import shutil
7from pathlib import Path
8from typing import Callable, Dict, List, Optional
10from jtech_installer.core.exceptions import FileOperationError
11from jtech_installer.core.models import (
12 AssetInfo,
13 InstallationConfig,
14 InstallationProgress,
15 TeamType,
16)
19class AssetCopier:
20 """Gerencia cópia de assets do framework JTECH™ Core"""
22 # Mapeamento de arquivos por tipo de equipe
23 TEAM_AGENT_MAPPING = {
24 TeamType.ALL: [
25 "jtech-master.md",
26 "jtech-orchestrator.md",
27 "analyst.md",
28 "pm.md",
29 "po.md",
30 "architect.md",
31 "dev.md",
32 "qa.md",
33 "ux-expert.md",
34 "sm.md",
35 ],
36 TeamType.FULLSTACK: [
37 "jtech-orchestrator.md",
38 "analyst.md",
39 "pm.md",
40 "ux-expert.md",
41 "architect.md",
42 "po.md",
43 "dev.md",
44 ],
45 TeamType.NO_UI: [
46 "jtech-orchestrator.md",
47 "analyst.md",
48 "pm.md",
49 "architect.md",
50 "dev.md",
51 "qa.md",
52 ],
53 TeamType.IDE_MINIMAL: ["pm.md", "architect.md", "dev.md"],
54 }
56 def __init__(
57 self,
58 config: InstallationConfig,
59 dry_run: bool = False,
60 progress_callback: Optional[
61 Callable[[InstallationProgress], None]
62 ] = None,
63 ):
64 self.config = config
65 self.dry_run = dry_run
66 self.progress_callback = progress_callback
67 self._determine_source_path()
69 def _determine_source_path(self) -> None:
70 """Determina o caminho fonte do framework"""
71 if self.config.framework_source_path:
72 self.framework_source = self.config.framework_source_path
73 else:
74 # Usar o framework do projeto atual como fonte
75 current_project = Path(__file__).parent.parent.parent.parent.parent
76 self.framework_source = current_project / ".jtech-core"
78 if not self.framework_source.exists():
79 raise FileOperationError(
80 f"Framework source não encontrado: {self.framework_source}"
81 )
83 def copy_agents(self) -> List[AssetInfo]:
84 """Copia agentes especializados baseado no tipo de equipe"""
85 agent_files = self.TEAM_AGENT_MAPPING.get(
86 self.config.team_type, self.TEAM_AGENT_MAPPING[TeamType.FULLSTACK]
87 )
89 assets_copied = []
90 source_agents_dir = self.framework_source / "agents"
91 target_agents_dir = self.config.project_path / ".jtech-core" / "agents"
93 if not source_agents_dir.exists():
94 raise FileOperationError(
95 f"Diretório de agentes não encontrado: {source_agents_dir}"
96 )
98 # Criar diretório de destino se não existir
99 if not self.dry_run:
100 target_agents_dir.mkdir(parents=True, exist_ok=True)
102 total_files = len(agent_files)
104 for i, agent_file in enumerate(agent_files):
105 source_file = source_agents_dir / agent_file
106 target_file = target_agents_dir / agent_file
108 if source_file.exists():
109 # Calcular progresso
110 if self.progress_callback:
111 progress = InstallationProgress(
112 current_phase="Copiando agentes",
113 total_phases=5,
114 current_phase_number=2,
115 files_processed=i,
116 total_files=total_files,
117 current_file=agent_file,
118 )
119 self.progress_callback(progress)
121 # Copiar arquivo
122 asset_info = self._copy_file(source_file, target_file)
123 assets_copied.append(asset_info)
124 else:
125 # Log warning mas continue
126 print(f"⚠️ Agente não encontrado: {source_file}")
128 return assets_copied
130 def copy_chatmodes(self) -> List[AssetInfo]:
131 """Copia arquivos chatmode para .github/chatmodes/"""
132 assets_copied = []
133 source_chatmodes_dir = self.framework_source / "chatmodes"
134 target_chatmodes_dir = (
135 self.config.project_path / ".github" / "chatmodes"
136 )
138 if not source_chatmodes_dir.exists():
139 # Tentar localização alternativa
140 source_chatmodes_dir = (
141 self.framework_source.parent / ".github" / "chatmodes"
142 )
144 if not source_chatmodes_dir.exists():
145 print(f"⚠️ Diretório chatmodes não encontrado, pulando...")
146 return assets_copied
148 # Criar diretório de destino
149 if not self.dry_run:
150 target_chatmodes_dir.mkdir(parents=True, exist_ok=True)
152 # Copiar todos os arquivos .chatmode.md
153 chatmode_files = list(source_chatmodes_dir.glob("*.chatmode.md"))
155 for i, source_file in enumerate(chatmode_files):
156 target_file = target_chatmodes_dir / source_file.name
158 if self.progress_callback:
159 progress = InstallationProgress(
160 current_phase="Copiando chatmodes",
161 total_phases=5,
162 current_phase_number=3,
163 files_processed=i,
164 total_files=len(chatmode_files),
165 current_file=source_file.name,
166 )
167 self.progress_callback(progress)
169 asset_info = self._copy_file(source_file, target_file)
170 assets_copied.append(asset_info)
172 return assets_copied
174 def copy_templates_and_workflows(self) -> List[AssetInfo]:
175 """Copia templates e workflows"""
176 assets_copied = []
178 # Diretórios a copiar
179 directories_to_copy = [
180 ("templates", "templates"),
181 ("workflows", "workflows"),
182 ("tasks", "tasks"),
183 ("checklists", "checklists"),
184 ("utils", "utils"),
185 ("data", "data"),
186 ]
188 for source_dir, target_dir in directories_to_copy:
189 source_path = self.framework_source / source_dir
190 target_path = self.config.project_path / ".jtech-core" / target_dir
192 if source_path.exists():
193 assets = self._copy_directory(source_path, target_path)
194 assets_copied.extend(assets)
196 return assets_copied
198 def copy_core_config(self) -> Optional[AssetInfo]:
199 """Copia e adapta core-config.yml"""
200 source_config = self.framework_source / "core-config.yml"
201 target_config = (
202 self.config.project_path / ".jtech-core" / "core-config.yml"
203 )
205 if source_config.exists():
206 return self._copy_file(source_config, target_config)
207 return None
209 def _copy_file(self, source: Path, target: Path) -> AssetInfo:
210 """Copia um arquivo individual com verificação"""
211 try:
212 # Calcular checksum do arquivo fonte
213 checksum = (
214 self._calculate_checksum(source) if source.exists() else None
215 )
217 if not self.dry_run:
218 # Criar diretório pai se necessário
219 target.parent.mkdir(parents=True, exist_ok=True)
221 # Copiar arquivo
222 shutil.copy2(source, target)
224 # Verificar se foi copiado corretamente
225 if not target.exists():
226 raise FileOperationError(
227 f"Falha ao copiar {source} -> {target}"
228 )
230 return AssetInfo(
231 source_path=source,
232 target_path=target,
233 file_type=source.suffix,
234 checksum=checksum,
235 )
237 except Exception as e:
238 raise FileOperationError(f"Erro ao copiar {source}: {e}")
240 def _copy_directory(self, source: Path, target: Path) -> List[AssetInfo]:
241 """Copia um diretório completo"""
242 assets_copied = []
244 if not self.dry_run:
245 target.mkdir(parents=True, exist_ok=True)
247 # Copiar todos os arquivos do diretório
248 for source_file in source.rglob("*"):
249 if source_file.is_file():
250 relative_path = source_file.relative_to(source)
251 target_file = target / relative_path
253 asset_info = self._copy_file(source_file, target_file)
254 assets_copied.append(asset_info)
256 return assets_copied
258 def _calculate_checksum(self, file_path: Path) -> str:
259 """Calcula checksum SHA256 de um arquivo"""
260 sha256_hash = hashlib.sha256()
261 with open(file_path, "rb") as f:
262 for chunk in iter(lambda: f.read(4096), b""):
263 sha256_hash.update(chunk)
264 return sha256_hash.hexdigest()
266 def copy_all(self) -> Dict[str, List[AssetInfo]]:
267 """Copia todos os assets necessários"""
268 all_assets = {
269 "agents": self.copy_agents(),
270 "chatmodes": self.copy_chatmodes(),
271 "templates_workflows": self.copy_templates_and_workflows(),
272 }
274 # Copiar core-config separadamente
275 core_config = self.copy_core_config()
276 if core_config:
277 all_assets["core_config"] = [core_config]
279 return all_assets