|
|
|
|
| """
|
| Scanner — Inspección del entorno al boot.
|
|
|
| Escanea el workspace, paquetes instalados, servicios disponibles y
|
| capacidades del sistema. El resultado se usa para generar el AGENTS.md
|
| contextual de cada despliegue.
|
|
|
| Principios:
|
| - Solo lectura: ast.parse, no exec/eval
|
| - Sin dependencias externas: usa stdlib pura
|
| - Timeout en detección de servicios
|
| - Resultado como dataclass serializable
|
| """
|
|
|
| import ast
|
| import importlib.metadata
|
| import platform
|
| import shutil
|
| import socket
|
| from dataclasses import dataclass, field
|
| from pathlib import Path
|
| from typing import Dict, List, Optional
|
|
|
|
|
| @dataclass
|
| class InfoArchivo:
|
| """Metadata de un archivo Python del workspace."""
|
|
|
| ruta: str
|
| funciones: List[str] = field(default_factory=list)
|
| clases: List[str] = field(default_factory=list)
|
| imports: List[str] = field(default_factory=list)
|
| lineas: int = 0
|
|
|
|
|
| @dataclass
|
| class InfoSistema:
|
| """Información del sistema operativo y hardware."""
|
|
|
| os: str = ""
|
| os_version: str = ""
|
| python_version: str = ""
|
| arquitectura: str = ""
|
| gpu: Optional[str] = None
|
| vram_mb: Optional[int] = None
|
| ram_gb: Optional[float] = None
|
|
|
|
|
| @dataclass
|
| class ResultadoScan:
|
| """Resultado completo del scan del entorno."""
|
|
|
| workspace_root: str = ""
|
| archivos: List[InfoArchivo] = field(default_factory=list)
|
| lenguajes: Dict[str, int] = field(default_factory=dict)
|
| paquetes: Dict[str, str] = field(default_factory=dict)
|
| servicios: Dict[str, bool] = field(default_factory=dict)
|
| sistema: InfoSistema = field(default_factory=InfoSistema)
|
| voz: List[str] = field(default_factory=list)
|
|
|
| @property
|
| def resumen(self) -> str:
|
| """Genera resumen legible del scan para inyectar en contexto."""
|
| partes: list[str] = []
|
|
|
| partes.append(f"## Entorno detectado\n")
|
| partes.append(f"- **OS**: {self.sistema.os} {self.sistema.os_version}")
|
| partes.append(f"- **Python**: {self.sistema.python_version}")
|
| if self.sistema.gpu:
|
| partes.append(f"- **GPU**: {self.sistema.gpu} ({self.sistema.vram_mb} MB)")
|
| if self.sistema.ram_gb:
|
| partes.append(f"- **RAM**: {self.sistema.ram_gb:.1f} GB")
|
|
|
| if self.lenguajes:
|
| langs = ", ".join(f"{k}: {v}" for k, v in sorted(
|
| self.lenguajes.items(), key=lambda x: -x[1],
|
| ))
|
| partes.append(f"- **Archivos**: {langs}")
|
|
|
| if self.paquetes:
|
| top = list(self.paquetes.items())[:20]
|
| pkgs = ", ".join(f"{n}=={v}" for n, v in top)
|
| partes.append(f"- **Paquetes** ({len(self.paquetes)} total): {pkgs}")
|
|
|
| if self.servicios:
|
| activos = [k for k, v in self.servicios.items() if v]
|
| if activos:
|
| partes.append(f"- **Servicios activos**: {', '.join(activos)}")
|
|
|
| if self.voz:
|
| partes.append(f"- **Voz**: {', '.join(self.voz)}")
|
|
|
| return "\n".join(partes)
|
|
|
|
|
| class Scanner:
|
| """
|
| Inspecciona el entorno donde PAMPAr se despliega.
|
|
|
| Solo lectura — usa ast.parse (no exec), importlib.metadata,
|
| socket con timeout, platform, shutil.which.
|
|
|
| Args:
|
| workspace_root: Directorio raíz del workspace a inspeccionar.
|
| scan_depth: Profundidad máxima de directorios a escanear.
|
| service_timeout: Timeout en segundos para detectar servicios.
|
| """
|
|
|
| EXTENSIONES = {
|
| ".py": "Python",
|
| ".js": "JavaScript",
|
| ".ts": "TypeScript",
|
| ".json": "JSON",
|
| ".md": "Markdown",
|
| ".yaml": "YAML",
|
| ".yml": "YAML",
|
| ".toml": "TOML",
|
| ".sh": "Shell",
|
| ".sql": "SQL",
|
| }
|
|
|
| SERVICIOS = {
|
| "PostgreSQL": ("127.0.0.1", 5432),
|
| "Redis": ("127.0.0.1", 6379),
|
| "HTTP-8000": ("127.0.0.1", 8000),
|
| "HTTP-3000": ("127.0.0.1", 3000),
|
| }
|
|
|
| MOTORES_VOZ = ["espeak", "espeak-ng", "say", "festival"]
|
|
|
|
|
| _IGNORAR = {
|
| "__pycache__", ".git", ".venv", "venv", "node_modules",
|
| ".mypy_cache", ".pytest_cache", ".tox", "dist", "build",
|
| "egg-info",
|
| }
|
|
|
| def __init__(
|
| self,
|
| workspace_root: str = ".",
|
| scan_depth: int = 5,
|
| service_timeout: float = 0.3,
|
| ):
|
| self.root = Path(workspace_root).resolve()
|
| self.scan_depth = scan_depth
|
| self.service_timeout = service_timeout
|
|
|
| def scan(self) -> ResultadoScan:
|
| """
|
| Ejecuta un scan completo del entorno.
|
|
|
| Returns:
|
| ResultadoScan con toda la información detectada.
|
| """
|
| resultado = ResultadoScan(workspace_root=str(self.root))
|
| resultado.sistema = self._scan_sistema()
|
| resultado.archivos, resultado.lenguajes = self._scan_workspace()
|
| resultado.paquetes = self._scan_paquetes()
|
| resultado.servicios = self._scan_servicios()
|
| resultado.voz = self._scan_voz()
|
| return resultado
|
|
|
| def _scan_sistema(self) -> InfoSistema:
|
| """Detecta OS, Python, GPU, RAM."""
|
| info = InfoSistema(
|
| os=platform.system(),
|
| os_version=platform.version(),
|
| python_version=platform.python_version(),
|
| arquitectura=platform.machine(),
|
| )
|
|
|
|
|
| try:
|
| import torch
|
| if torch.cuda.is_available():
|
| info.gpu = torch.cuda.get_device_name(0)
|
| info.vram_mb = torch.cuda.get_device_properties(0).total_memory // (1024 * 1024)
|
| except ImportError:
|
| pass
|
|
|
|
|
| try:
|
| import psutil
|
| info.ram_gb = psutil.virtual_memory().total / (1024 ** 3)
|
| except ImportError:
|
| pass
|
|
|
| return info
|
|
|
| def _scan_workspace(self) -> tuple[list[InfoArchivo], dict[str, int]]:
|
| """Inspecciona archivos del workspace con ast.parse para .py."""
|
| archivos: list[InfoArchivo] = []
|
| lenguajes: dict[str, int] = {}
|
|
|
| if not self.root.is_dir():
|
| return archivos, lenguajes
|
|
|
| for path in self._iter_files():
|
| ext = path.suffix.lower()
|
| lang = self.EXTENSIONES.get(ext)
|
| if lang:
|
| lenguajes[lang] = lenguajes.get(lang, 0) + 1
|
|
|
| if ext == ".py":
|
| info = self._analizar_python(path)
|
| if info:
|
| archivos.append(info)
|
|
|
| return archivos, lenguajes
|
|
|
| def _iter_files(self):
|
| """Itera archivos del workspace respetando profundidad y exclusiones."""
|
| for path in self.root.rglob("*"):
|
| if not path.is_file():
|
| continue
|
|
|
|
|
| try:
|
| rel = path.relative_to(self.root)
|
| except ValueError:
|
| continue
|
| if len(rel.parts) > self.scan_depth:
|
| continue
|
|
|
|
|
| if any(
|
| p in self._IGNORAR or p.endswith(".egg-info")
|
| for p in rel.parts
|
| ):
|
| continue
|
|
|
| yield path
|
|
|
| def _analizar_python(self, path: Path) -> Optional[InfoArchivo]:
|
| """Analiza un archivo .py con ast.parse (no exec)."""
|
| try:
|
| source = path.read_text(encoding="utf-8", errors="ignore")
|
| tree = ast.parse(source, filename=str(path))
|
| except (SyntaxError, UnicodeDecodeError):
|
| return None
|
|
|
| rel_path = str(path.relative_to(self.root)).replace("\\", "/")
|
| info = InfoArchivo(ruta=rel_path, lineas=len(source.splitlines()))
|
|
|
| for node in ast.iter_child_nodes(tree):
|
| if isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef):
|
| info.funciones.append(node.name)
|
| elif isinstance(node, ast.ClassDef):
|
| info.clases.append(node.name)
|
| elif isinstance(node, ast.Import):
|
| for alias in node.names:
|
| info.imports.append(alias.name)
|
| elif isinstance(node, ast.ImportFrom):
|
| if node.module:
|
| info.imports.append(node.module)
|
|
|
| return info
|
|
|
| def _scan_paquetes(self) -> dict[str, str]:
|
| """Detecta paquetes instalados via importlib.metadata."""
|
| paquetes: dict[str, str] = {}
|
| try:
|
| for dist in importlib.metadata.distributions():
|
| name = dist.metadata.get("Name", "")
|
| version = dist.metadata.get("Version", "")
|
| if name:
|
| paquetes[name] = version
|
| except Exception:
|
| pass
|
| return paquetes
|
|
|
| def _scan_servicios(self) -> dict[str, bool]:
|
| """Detecta servicios activos con socket connect + timeout."""
|
| servicios: dict[str, bool] = {}
|
| for nombre, (host, port) in self.SERVICIOS.items():
|
| servicios[nombre] = self._puerto_abierto(host, port)
|
| return servicios
|
|
|
| def _puerto_abierto(self, host: str, port: int) -> bool:
|
| """Verifica si un puerto está abierto (solo localhost)."""
|
| try:
|
| with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
|
| s.settimeout(self.service_timeout)
|
| return s.connect_ex((host, port)) == 0
|
| except OSError:
|
| return False
|
|
|
| def _scan_voz(self) -> list[str]:
|
| """Detecta motores de voz disponibles en el sistema."""
|
| disponibles: list[str] = []
|
| for motor in self.MOTORES_VOZ:
|
| if shutil.which(motor):
|
| disponibles.append(motor)
|
|
|
|
|
| if platform.system() == "Windows":
|
| try:
|
| import winreg
|
| key = winreg.OpenKey(
|
| winreg.HKEY_LOCAL_MACHINE,
|
| r"SOFTWARE\Microsoft\Speech\Voices",
|
| )
|
| winreg.CloseKey(key)
|
| disponibles.append("SAPI")
|
| except (ImportError, OSError):
|
| pass
|
|
|
| return disponibles
|
|
|
