44 lines
1.6 KiB
Python
44 lines
1.6 KiB
Python
"""Phase 0a: Existing Source Environment Discovery Node."""
|
|
|
|
import logging
|
|
import re
|
|
from typing import Any, Dict
|
|
from app.states.state import GCPArchitectureState
|
|
from app.skills.loader import SkillLoader
|
|
from app.tools.gcp_scanner import GCPEnvironmentScanner
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def source_discover_node(state: GCPArchitectureState, skill_loader: SkillLoader) -> Dict[str, Any]:
|
|
"""Processes Phase 0a Source Environment Discovery: documents existing as-is architecture before target migration."""
|
|
logger.info("Executing source_discover_node")
|
|
skill_prompt = skill_loader.format_skills_for_prompt("source_discover")
|
|
|
|
# Extract target GCP project ID from request if specified
|
|
request_summary = state.get("workflow_request", "")
|
|
project_id = state.get("project_id")
|
|
if not project_id and request_summary:
|
|
match = re.search(r"project\s+([a-z0-9-]+)", request_summary, re.IGNORECASE)
|
|
if match:
|
|
project_id = match.group(1)
|
|
|
|
# Run live GCP environment scanner
|
|
scanner = GCPEnvironmentScanner(project_id=project_id)
|
|
scan_result = scanner.scan_environment()
|
|
report = scanner.format_scan_report(scan_result)
|
|
|
|
source_discovery_doc = report["doc"]
|
|
source_mermaid_diagram = report["mermaid"]
|
|
|
|
active_skills = state.get("active_skills", [])
|
|
if "source_discovery" not in active_skills:
|
|
active_skills.append("source_discovery")
|
|
|
|
return {
|
|
"source_discovery_doc": source_discovery_doc,
|
|
"source_mermaid_diagram": source_mermaid_diagram,
|
|
"current_phase": "source_discover",
|
|
"active_skills": active_skills,
|
|
}
|