diff --git a/.github/workflows/python-tests.yml b/.github/workflows/python-tests.yml new file mode 100644 index 0000000..86ad41e --- /dev/null +++ b/.github/workflows/python-tests.yml @@ -0,0 +1,27 @@ +name: Python tests + +on: + push: + branches: ["main", "fix/**"] + pull_request: + workflow_dispatch: + +jobs: + test: + runs-on: ubuntu-latest + defaults: + run: + working-directory: dataforge-cli + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.11" + cache: pip + cache-dependency-path: dataforge-cli/requirements.txt + - name: Install dependencies + run: python -m pip install -r requirements.txt + - name: Compile Python files + run: python -m compileall -q . + - name: Run regression tests + run: python -m unittest discover -s tests -v diff --git a/dataforge-cli/dataforge/api_handler.py b/dataforge-cli/dataforge/api_handler.py index 7d279d0..ef56cdf 100644 --- a/dataforge-cli/dataforge/api_handler.py +++ b/dataforge-cli/dataforge/api_handler.py @@ -1,56 +1,116 @@ -import os -import json -import urllib.request -from groq import Groq -from dataforge.config_manager import get_config - -def ask_enterprise_ai(prompt): - """Enruta el prompt al proveedor de IA configurado (Groq, OpenAI, Gemini, Claude u Ollama local)""" - config = get_config() - if not config: - config = {} - - # Soporta tanto 'ai_provider' como 'provider' según la versión del diccionario de configuración - provider = config.get("ai_provider", config.get("provider", "groq")).lower() - - # 1. Proveedor Ollama (100% Offline / Local) - if provider == "ollama": - url = "http://localhost:11434/api/generate" - payload = { - "model": "llama3", # O el modelo local que prefiera el usuario - "prompt": prompt, - "stream": False - } - data = json.dumps(payload).encode("utf-8") - req = urllib.request.Request( - url, - data=data, - headers={"Content-Type": "application/json"} - ) - try: - with urllib.request.urlopen(req, timeout=60) as response: - result = json.loads(response.read().decode("utf-8")) - return result.get("response", "Error: Respuesta vacía de Ollama.") - except Exception as e: - return f"⚠️ Error de conexión con Ollama local (Verifica que esté abierto y el modelo instalado): {e}" - - # 2. Proveedor Groq (Nube Ultra Rápida) - elif provider == "groq": - # Revisa tanto 'groq_api_key' como 'api_key' - api_key = config.get("groq_api_key", config.get("api_key")) - if not api_key: - raise ValueError("Falta la API Key de Groq. Ejecuta 'python main.py setup'") - client = Groq(api_key=api_key) - chat_completion = client.chat.completions.create( - messages=[{"role": "user", "content": prompt}], - model="llama-3.3-70b-versatile", - ) - return chat_completion.choices[0].message.content - - # 3. Otros proveedores en nube (OpenAI, Gemini, Claude) - else: - raise ValueError(f"Proveedor de IA '{provider}' no soportado o pendiente de credenciales.") - -def ask_groq(prompt, api_key=None): - """Puente de compatibilidad para llamadas antiguas a ask_groq""" - return ask_enterprise_ai(prompt) \ No newline at end of file +import json +import urllib.error +import urllib.request +import urllib.parse + +from groq import Groq +from dataforge.config_manager import get_config + + +def _post_json(url, payload, headers=None, timeout=60): + request_headers = {"Content-Type": "application/json"} + request_headers.update(headers or {}) + request = urllib.request.Request( + url, + data=json.dumps(payload).encode("utf-8"), + headers=request_headers, + method="POST", + ) + with urllib.request.urlopen(request, timeout=timeout) as response: + return json.loads(response.read().decode("utf-8")) + + +def ask_enterprise_ai(prompt): + """Route a prompt to the provider selected in ~/.dataforge/config.json.""" + config = get_config() or {} + provider = str(config.get("ai_provider", config.get("provider", "groq"))).strip().lower() + api_key = config.get(f"{provider}_api_key") or config.get("api_key") + + if provider == "ollama": + model = config.get("ollama_model", "llama3") + try: + result = _post_json( + config.get("ollama_url", "http://localhost:11434/api/generate"), + {"model": model, "prompt": prompt, "stream": False}, + timeout=60, + ) + except Exception as exc: + raise RuntimeError( + "Could not connect to Ollama. Start Ollama and verify the configured model." + ) from exc + answer = result.get("response") + if not answer: + raise RuntimeError("Ollama returned an empty response.") + return answer + + if not api_key: + raise ValueError( + f"Missing API key for {provider}. Run 'python main.py setup' and configure credentials." + ) + + if provider == "groq": + client = Groq(api_key=api_key) + completion = client.chat.completions.create( + messages=[{"role": "user", "content": prompt}], + model=config.get("groq_model", "llama-3.3-70b-versatile"), + ) + return completion.choices[0].message.content + + if provider == "openai": + result = _post_json( + "https://api.openai.com/v1/chat/completions", + { + "model": config.get("openai_model", "gpt-4o-mini"), + "messages": [{"role": "user", "content": prompt}], + }, + {"Authorization": f"Bearer {api_key}"}, + ) + try: + return result["choices"][0]["message"]["content"] + except (KeyError, IndexError, TypeError) as exc: + raise RuntimeError("OpenAI returned an unexpected response format.") from exc + + if provider == "gemini": + url = ( + "https://generativelanguage.googleapis.com/v1beta/models/" + + config.get("gemini_model", "gemini-2.5-flash") + + ":generateContent?key=" + + urllib.parse.quote(api_key, safe="") + ) + result = _post_json( + url, + {"contents": [{"parts": [{"text": prompt}]}]}, + ) + try: + return "".join(part["text"] for part in result["candidates"][0]["content"]["parts"]) + except (KeyError, IndexError, TypeError) as exc: + raise RuntimeError("Gemini returned an unexpected response format.") from exc + + if provider == "claude": + result = _post_json( + "https://api.anthropic.com/v1/messages", + { + "model": config.get("claude_model", "claude-3-5-haiku-latest"), + "max_tokens": 4096, + "messages": [{"role": "user", "content": prompt}], + }, + { + "x-api-key": api_key, + "anthropic-version": "2023-06-01", + }, + ) + try: + return "".join( + block["text"] for block in result["content"] if block.get("type") == "text" + ) + except (KeyError, TypeError) as exc: + raise RuntimeError("Claude returned an unexpected response format.") from exc + + raise ValueError( + f"Unsupported AI provider '{provider}'. Supported providers: groq, openai, gemini, claude, ollama." + ) + + +def ask_groq(prompt, api_key=None): + """Backward-compatible alias; routing follows the configured provider.""" + return ask_enterprise_ai(prompt) diff --git a/dataforge-cli/dataforge/core.py b/dataforge-cli/dataforge/core.py index fe06c22..c5cd338 100644 --- a/dataforge-cli/dataforge/core.py +++ b/dataforge-cli/dataforge/core.py @@ -1,69 +1,110 @@ -import os -from datetime import datetime - -# Extensiones que nos interesan -VALID_EXTENSIONS = {'.py', '.js', '.ts', '.txt', '.md', '.html', '.css', '.json', '.go', '.rs', '.java', '.cpp'} - -# Directorios excluidos por seguridad y rendimiento -EXCLUDED_DIRS = {'.git', '__pycache__', 'node_modules', '.venv', 'venv', 'dataforge-reports', 'dist', 'build', '.idea', '.vscode', 'target', 'bin', 'obj', '.dataforge'} - -def get_all_files(root_path): - files_to_process = [] - for root, dirs, files in os.walk(root_path): - # Ignorar carpetas conflictivas de forma segura in-place - dirs[:] = [d for d in dirs if d not in EXCLUDED_DIRS] - for file in files: - ext = os.path.splitext(file)[1].lower() - if ext in VALID_EXTENSIONS: - full_path = os.path.join(root, file) - files_to_process.append(full_path) - return files_to_process - -def read_file_content(file_path): - try: - # Usamos errors='ignore' para evitar bloqueos por codificación de caracteres raros - with open(file_path, 'r', encoding='utf-8', errors='ignore') as f: - return f.read() - except Exception as e: - return f"Error al leer {file_path}: {e}" - -def get_single_file_content(file_path): - if not os.path.exists(file_path): - return None - return read_file_content(file_path) - -def prepare_code_payload(files_list, max_chars=35000): - """Combina el contenido de los archivos y aplica un recorte estricto para evitar el Error 413 de tokens.""" - project_data = [] - total_chars = 0 - - for file_path in files_list: - content = read_file_content(file_path) - file_header = f"--- FILE: {file_path} ---\n" - file_block = f"{file_header}{content}\n\n" - - # Si agregar este archivo rebasa el límite seguro de caracteres, recortamos o detenemos - if total_chars + len(file_block) > max_chars: - remaining_chars = max_chars - total_chars - if remaining_chars > 100: - project_data.append(file_block[:remaining_chars] + "\n[... Archivo recortado por límite de presupuesto de tokens ...]") - project_data.append("\n⚠️ [Aviso del Sistema]: El proyecto excede el presupuesto seguro de tokens. Se omitieron algunos archivos restantes.") - break - - project_data.append(file_block) - total_chars += len(file_block) - - return "".join(project_data) - -def save_report(content, target_path): - report_dir = os.path.join(target_path, "dataforge-reports") - if not os.path.exists(report_dir): - os.makedirs(report_dir) - - timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S") - filename = f"reporte_{timestamp}.txt" - file_path = os.path.join(report_dir, filename) - - with open(file_path, 'w', encoding='utf-8') as f: - f.write(content) +import os +import fnmatch +from datetime import datetime + +# Extensiones que nos interesan +VALID_EXTENSIONS = {'.py', '.js', '.ts', '.txt', '.md', '.html', '.css', '.json', '.go', '.rs', '.java', '.cpp'} + +# Directorios excluidos por seguridad y rendimiento +EXCLUDED_DIRS = {'.git', '__pycache__', 'node_modules', '.venv', 'venv', 'dataforge-reports', 'dist', 'build', '.idea', '.vscode', 'target', 'bin', 'obj', '.dataforge'} +EXCLUDED_FILES = {'.env', '.env.local', '.env.production', 'id_rsa', 'id_ed25519', 'credentials.json', 'secrets.json'} +MAX_FILE_BYTES = 1_000_000 + + +def _load_gitignore_patterns(root_path): + """Read simple .gitignore patterns; unsupported negation patterns are ignored safely.""" + patterns = [] + ignore_path = os.path.join(root_path, ".gitignore") + try: + with open(ignore_path, "r", encoding="utf-8", errors="ignore") as handle: + for raw in handle: + pattern = raw.strip() + if pattern and not pattern.startswith("#") and not pattern.startswith("!"): + patterns.append(pattern.rstrip("/")) + except OSError: + pass + return patterns + + +def _is_ignored(relative_path, patterns): + normalized = relative_path.replace(os.sep, "/") + parts = normalized.split("/") + for pattern in patterns: + if "/" in pattern: + if fnmatch.fnmatch(normalized, pattern) or normalized == pattern: + return True + elif any(fnmatch.fnmatch(part, pattern) for part in parts): + return True + return False + +def get_all_files(root_path): + root_path = os.path.abspath(root_path) + patterns = _load_gitignore_patterns(root_path) + files_to_process = [] + for root, dirs, files in os.walk(root_path): + dirs[:] = [ + d for d in dirs + if d not in EXCLUDED_DIRS + and not _is_ignored(os.path.relpath(os.path.join(root, d), root_path), patterns) + ] + for file in files: + if file.lower() in EXCLUDED_FILES or file.lower().startswith(".env."): + continue + relative_path = os.path.relpath(os.path.join(root, file), root_path) + if _is_ignored(relative_path, patterns): + continue + ext = os.path.splitext(file)[1].lower() + if ext in VALID_EXTENSIONS: + files_to_process.append(os.path.join(root, file)) + return files_to_process + +def read_file_content(file_path, max_bytes=MAX_FILE_BYTES): + try: + with open(file_path, 'r', encoding='utf-8', errors='ignore') as f: + content = f.read(max_bytes + 1) + if len(content) > max_bytes: + return content[:max_bytes] + "\\n[DataForge: file truncated at configured safety limit.]" + return content + except Exception as e: + return f"Error al leer {file_path}: {e}" + +def get_single_file_content(file_path): + if not os.path.exists(file_path): + return None + return read_file_content(file_path) + +def prepare_code_payload(files_list, max_chars=35000): + """Combina el contenido de los archivos y aplica un recorte estricto para evitar el Error 413 de tokens.""" + project_data = [] + total_chars = 0 + + for file_path in files_list: + content = read_file_content(file_path) + file_header = f"--- FILE: {file_path} ---\n" + file_block = f"{file_header}{content}\n\n" + + # Si agregar este archivo rebasa el límite seguro de caracteres, recortamos o detenemos + if total_chars + len(file_block) > max_chars: + remaining_chars = max_chars - total_chars + if remaining_chars > 100: + project_data.append(file_block[:remaining_chars] + "\n[... Archivo recortado por límite de presupuesto de tokens ...]") + project_data.append("\n⚠️ [Aviso del Sistema]: El proyecto excede el presupuesto seguro de tokens. Se omitieron algunos archivos restantes.") + break + + project_data.append(file_block) + total_chars += len(file_block) + + return "".join(project_data) + +def save_report(content, target_path): + report_dir = os.path.join(target_path, "dataforge-reports") + if not os.path.exists(report_dir): + os.makedirs(report_dir) + + timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S") + filename = f"reporte_{timestamp}.txt" + file_path = os.path.join(report_dir, filename) + + with open(file_path, 'w', encoding='utf-8') as f: + f.write(content) return file_path \ No newline at end of file diff --git a/dataforge-cli/main.py b/dataforge-cli/main.py index ffe9c4e..f396e99 100644 --- a/dataforge-cli/main.py +++ b/dataforge-cli/main.py @@ -1,465 +1,489 @@ -import sys -import os -import json -import urllib.request -import urllib.parse -import webbrowser -import platform -import uuid -from datetime import datetime -from dataforge.config_manager import init_config, get_config, setup_config, save_license_advanced, LICENSE_FILE -from dataforge.api_handler import ask_groq -from dataforge.core import get_all_files, read_file_content, save_report, get_single_file_content, prepare_code_payload - -WELCOME_URL = "https://rabbitgamesdev.github.io/DataForge-CLI/" -SUPABASE_FUNCTION_URL = "https://yultjjqcxtqfakppxyqv.supabase.co/functions/v1/validate-license" - -def get_theme_color(): - config = get_config() or {} - theme = config.get("theme", "dark") - if theme == "tokyo_night": return "\033[38;5;189m" - elif theme == "dracula": return "\033[38;5;212m" - elif theme == "light": return "\033[30m" - return "\033[38;5;248m" - -def cprint(text, style="normal"): - reset = "\033[0m" - color = get_theme_color() - if style == "success": color = "\033[32m" - elif style == "error": color = "\033[31m" - elif style == "warning": color = "\033[33m" - elif style == "pro": color = "\033[38;5;220m" - elif style == "teams": color = "\033[38;5;141m" - print(f"{color}{text}{reset}") - -def get_language_prefix(): - config = get_config() or {} - lang = config.get("language", "es") - if lang == "en": - return "[SYSTEM DIRECTIVE: You MUST respond entirely and strictly in English.]\n\n" - return "[DIRECTIVA DEL SISTEMA: Debes responder entera y estrictamente en Español.]\n\n" - -def get_license_data(): - if os.path.exists(LICENSE_FILE): - try: - with open(LICENSE_FILE, "r", encoding="utf-8") as f: - return json.load(f) - except Exception: - return None - return None - -def verify_license_hybrid() -> bool: - """Motor de Licenciamiento Híbrido: Supabase Cloud + Offline Grace Period + Expiration Check""" - data = get_license_data() - if not data or not data.get("active", False): - return False - - key = data.get("key") - if not key: - return False - - now = datetime.now() - - # Validar expiración local inmediata si existe el campo - if data.get("expires_at"): - try: - if now > datetime.fromisoformat(data["expires_at"]): - data["active"] = False - with open(LICENSE_FILE, "w", encoding="utf-8") as f: - json.dump(data, f, indent=4) - return False - except Exception: - pass - - last_checked_str = data.get("last_checked") - should_check = True - - if last_checked_str: - try: - last_checked = datetime.fromisoformat(last_checked_str) - if (now - last_checked).days < 7: - should_check = False - except Exception: - pass - - if should_check: - machine_id = str(uuid.uuid5(uuid.NAMESPACE_DNS, platform.node())) - payload = json.dumps({ - "license_key": key, - "machine_id": machine_id - }).encode("utf-8") - - try: - req = urllib.request.Request( - SUPABASE_FUNCTION_URL, - data=payload, - headers={"Content-Type": "application/json"}, - method="POST" - ) - with urllib.request.urlopen(req, timeout=5) as response: - res_data = json.loads(response.read().decode("utf-8")) - if not res_data.get("success", False): - data["active"] = False - with open(LICENSE_FILE, "w", encoding="utf-8") as f: - json.dump(data, f, indent=4) - return False - - # Actualizar datos con sincronización exitosa de servidor - data["tier"] = res_data.get("tier", data.get("tier")) - data["expires_at"] = res_data.get("expires_at", data.get("expires_at")) - data["last_checked"] = now.isoformat() - with open(LICENSE_FILE, "w", encoding="utf-8") as f: - json.dump(data, f, indent=4) - except Exception: - # Modo Offline Grace Period (si falla la red, permite acceso si se verificó hace menos de 7 días) - pass - - return True - -def is_pro_active() -> bool: - data = get_license_data() - if not data or not data.get("active", False): - return False - if not verify_license_hybrid(): - return False - return data.get("tier", "").lower() in ["pro", "teams"] - -def is_teams_active() -> bool: - data = get_license_data() - if not data or not data.get("active", False): - return False - if not verify_license_hybrid(): - return False - return data.get("tier", "").lower() == "teams" - -def ask_enterprise_ai(prompt): - config = get_config() or {} - provider = config.get("provider", "groq").lower() - - if provider == "ollama": - try: - url = "http://localhost:11434/api/generate" - payload = json.dumps({ - "model": "llama3", - "prompt": prompt, - "stream": False - }).encode("utf-8") - req = urllib.request.Request(url, data=payload, headers={"Content-Type": "application/json"}) - with urllib.request.urlopen(req, timeout=45) as response: - res_data = json.loads(response.read().decode("utf-8")) - return res_data.get("response", "Error: Respuesta vacía de Ollama.") - except Exception as e: - return f"⚠️ Error de conexión con Ollama local: {e}" - - if provider == "claude": - system_directive = "[EXPERT CLAUDE ENGINE: Prioritize deep architectural nuance, pristine production documentation structure.]\n" - return ask_groq(system_directive + prompt) - elif provider == "openai": - system_directive = "[EXPERT OPENAI ENGINE: Prioritize absolute logical precision, strict code correctness.]\n" - return ask_groq(system_directive + prompt) - elif provider == "gemini": - system_directive = "[EXPERT GEMINI ENGINE: Prioritize exhaustive global context handling, multi-step structural breakdown.]\n" - return ask_groq(system_directive + prompt) - - return ask_groq(prompt) - -def get_footer_branding(): - config = get_config() or {} - brand = config.get("brand_name", "Developer") - - if is_teams_active(): - license_data = get_license_data() or {} - team_name = license_data.get("team_name", brand) - return f"\n\n---\n*Generated by {team_name}*\n>_ Enterprise Architecture Documentation Pipeline" - elif is_pro_active(): - return f"\n\n---\n*Created by {brand}*\n>_ [Powered by DataForge CLI](https://rabbitgamesdev.github.io/DataForge-CLI/)" - else: - return ( - f"\n\n---\n" - f"\n" - f"*Created by {brand}*\n" - f">_ **Powered by DataForge CLI by RGS Labs™**\n" - f">_ [https://rabbitgamesdev.github.io/DataForge-CLI/](https://rabbitgamesdev.github.io/DataForge-CLI/)" - ) - -def open_welcome_page(): - try: webbrowser.open(WELCOME_URL, new=2) - except Exception: cprint(f"👉 Carta de bienvenida: {WELCOME_URL}", "warning") - -def show_help_table(): - status = "💜 TEAMS (Enterprise Branding)" if is_teams_active() else ("⭐ PRO (Developer Suite)" if is_pro_active() else "🔓 Free Tier") - config = get_config() or {} - - cprint(f"\n======================================================================", "pro") - cprint(f" 🚀 DataForge CLI v2.0 - RGS Labs™ | Estado: [{status}]", "pro") - cprint(f" 🤖 IA Activa: {config.get('provider', 'groq').upper()} | Idioma: {config.get('language', 'es').upper()}", "success") - cprint(f"======================================================================", "pro") - - print("\n📋 TABLA DE COMANDOS DISPONIBLES:\n") - print(f"{'Comando':<18} | {'Tipo':<10} | {'Descripción detallada de la función'}") - print("-" * 75) - print(f"{'scan':<18} | {'Free/Pro':<10} | Escanea el proyecto completo y genera un reporte de análisis de IA.") - print(f"{'explain':<18} | {'Free/Pro':<10} | Explica a detalle un archivo de código específico (admite perfiles).") - print(f"{'map':<18} | {'Free/Pro':<10} | Genera un diagrama ASCII visual de la arquitectura del proyecto.") - print(f"{'onboard':<18} | {'Free/Pro':<10} | Crea una guía de incorporación técnica corporativa automatizada.") - print(f"{'doc++':<18} | {'PRO / TEAMS':<10} | Genera documentación técnica avanzada lista para producción.") - print(f"{'architecture++':<18} | {'PRO / TEAMS':<10} | Genera diagramas estructurales Mermaid y análisis profundo.") - print(f"{'export':<18} | {'PRO / TEAMS':<10} | Exporta los reportes generados a formatos limpios (--md, --json, --html).") - print(f"{'preset':<18} | {'PRO / TEAMS':<10} | Ejecuta un pipeline completo automatizado de desarrollo.") - print(f"{'setup':<18} | {'Global':<10} | Reconfigura el motor de IA, API Keys, idioma y tema visual.") - print(f"{'license':<18} | {'Global':<10} | Activa licencias de categoría PRO o TEAMS (con branding corporativo).") - print(f"{'license status':<18} | {'Global':<10} | Activa o consulta el estado de licencias (ej. license status).") - print("-" * 75) - -def run_scan(target_path): - cprint(f"🔍 Escaneando proyecto en: {target_path}...") - files = get_all_files(target_path) - if not files: return cprint("No se encontraron archivos válidos.", "error") - - file_list_str = "\n".join([f"- {f}" for f in files]) - # Payload optimizado con presupuesto seguro de tokens para evitar Error 413 - full_context = prepare_code_payload(files, max_chars=35000) - - prompt = f"{get_language_prefix()}Analiza la arquitectura y código de este proyecto:\n{full_context}" - respuesta = ask_enterprise_ai(prompt) - - final_content = f"# ARCHIVO DE REPORTE - DataForge CLI\n\n**Archivos analizados:**\n{file_list_str}\n\n{respuesta}{get_footer_branding()}" - report_path = save_report(final_content, target_path) - cprint(f"\n✅ Reporte guardado en: {report_path}\n", "success") - print(respuesta) - -def run_explain(file_path, audience=""): - content = get_single_file_content(file_path) - if content is None: return cprint("Error: El archivo no existe.", "error") - - modifier = f"Adapta la explicación para un perfil: {audience.upper()}." if audience else "" - if audience and not is_pro_active(): - return cprint("🔒 Requiere licencia PRO.", "warning") - - prompt = f"{get_language_prefix()}Explica detalladamente qué hace este archivo. {modifier}\n{content}" - respuesta = ask_enterprise_ai(prompt) - cprint(f"\n--- EXPLICACIÓN ({audience.upper() if audience else 'GENERAL'}) ---") - print(respuesta) - -def run_map(target_path): - cprint(f"🗺️ Generando mapa de dependencias ASCII...") - files = get_all_files(target_path) - file_names = "\n".join([os.path.relpath(f, target_path) for f in files]) - prompt = f"{get_language_prefix()}Basado en:\n{file_names}\n\nGenera un mapa visual (ASCII) limpio de la arquitectura." - respuesta = ask_enterprise_ai(prompt) - cprint(f"\n--- MAPA DE ARQUITECTURA ---\n{respuesta}") - -def run_onboard(target_path): - cprint(f"🚀 Generando guía de onboarding...") - files = get_all_files(target_path) - if not files: return cprint("No se encontraron archivos válidos.", "error") - - full_context = prepare_code_payload(files, max_chars=35000) - prompt = f"{get_language_prefix()}Actúa como Líder Técnico. Genera Onboarding corporativo:\n{full_context}" - respuesta = ask_enterprise_ai(prompt) - - final_content = f"# ONBOARDING GUIDE\n\n{respuesta}{get_footer_branding()}" - report_path = save_report(final_content, target_path) - cprint(f"✅ Onboarding generado: {report_path}", "success") - print(f"\n{respuesta}") - -def run_doc_pro(target_path): - if not is_pro_active(): return cprint("🔒 Exclusivo PRO/TEAMS.", "warning") - cprint(f"🚀 Generando documentación técnica de producción (Documentation++)...", "teams" if is_teams_active() else "pro") - files = get_all_files(target_path) - if not files: return - - file_list = "\n".join([os.path.relpath(f, target_path) for f in files[:50]]) - prompt = f"{get_language_prefix()}Arquitecto Senior. Genera README profesional y documentación técnica basada en:\n{file_list}" - respuesta = ask_enterprise_ai(prompt) - - output = os.path.join(target_path, "PRODUCTION_DOCS.md") - with open(output, "w", encoding="utf-8") as f: f.write(respuesta + get_footer_branding()) - cprint(f"✅ Documentación guardada en: {output}", "success") - -def run_architecture_pro(target_path): - if not is_pro_active(): return cprint("🔒 Exclusivo PRO/TEAMS.", "warning") - cprint(f"📐 Generando diagramas Mermaid y análisis profundo...", "pro") - files = get_all_files(target_path) - if not files: return - file_names = "\n".join([os.path.relpath(f, target_path) for f in files[:50]]) - prompt = f"{get_language_prefix()}Arquitecto Experto. Genera código Mermaid (graph TD) y explicación basada en:\n{file_names}" - respuesta = ask_enterprise_ai(prompt) - - output = os.path.join(target_path, "ARCHITECTURE_REPORT.md") - with open(output, "w", encoding="utf-8") as f: f.write(respuesta + get_footer_branding()) - cprint(f"✅ Arquitectura guardada en: {output}", "success") - -def run_export(target_path, fmt): - if not is_pro_active(): return cprint("🔒 Exclusivo PRO/TEAMS.", "warning") - fmt = fmt.replace("--", "").lower() - reports_dir = os.path.join(target_path, "dataforge-reports") - if not os.path.exists(reports_dir): return cprint("⚠️ Sin reportes previos.", "warning") - reports = [os.path.join(reports_dir, f) for f in os.listdir(reports_dir) if f.endswith(".txt") or f.endswith(".md")] - if not reports: return cprint("⚠️ No hay reportes para exportar.", "warning") - - target_file = max(reports, key=os.path.getmtime) - with open(target_file, "r", encoding="utf-8") as f: content = f.read() - output_path = f"{os.path.splitext(target_file)[0]}.{fmt}" - - if fmt == "json": - with open(output_path, "w", encoding="utf-8") as out: json.dump({"content": content}, out, indent=4) - elif fmt == "html": - with open(output_path, "w", encoding="utf-8") as out: out.write(f"
{content}
") - else: - with open(output_path, "w", encoding="utf-8") as out: out.write(content) - cprint(f"✅ Exportado a: {output_path}", "success") - -def run_preset(name, target_path): - if not is_pro_active(): return cprint("🔒 Exclusivo PRO/TEAMS.", "warning") - if name == "onboarding": - cprint(f"⚡ Ejecutando Pipeline Automático de Ingeniería...", "pro") - run_scan(target_path) - run_onboard(target_path) - run_doc_pro(target_path) - run_architecture_pro(target_path) - run_export(target_path, "--html") - cprint("🎉 ¡Preset completado exitosamente!", "success") - else: - cprint(f"⚠️ Preset desconocido: {name}.", "warning") - -def run_license_activation(key: str): - machine_id = str(uuid.uuid5(uuid.NAMESPACE_DNS, platform.node())) - owner = input("Introduce tu correo de compra (ej. tu@email.com): ").strip() - if not owner: owner = "cliente@dataforge.dev" - - upper_key = key.upper() - is_team = "TEAM" in upper_key or "TEAMS" in upper_key - - team_name = "" - if is_team: - print("\n🏢 [TEAMS] Licencia corporativa detectada.") - team_name = input("Introduce el Nombre de tu Estudio / Empresa para el Team Branding: ").strip() - if not team_name: team_name = "Enterprise Team" - - payload = json.dumps({ - "license_key": key, - "machine_id": machine_id, - "email": owner, - "team_branding": team_name - }).encode("utf-8") - - try: - req = urllib.request.Request( - SUPABASE_FUNCTION_URL, - data=payload, - headers={"Content-Type": "application/json"}, - method="POST" - ) - with urllib.request.urlopen(req, timeout=10) as response: - res_data = json.loads(response.read().decode("utf-8")) - if not res_data.get("success", False): - return cprint(f"❌ Error de activación: {res_data.get('message', 'Clave inválida')}", "error") - - tier = res_data.get("tier", "pro").lower() - saved_owner = owner - saved_branding = res_data.get("team_branding", team_name) - expires_at = res_data.get("expires_at") - except Exception as e: - return cprint(f"⚠️ Error conectando con el servidor de licencias: {e}", "error") - - # Guardar licencia avanzada - save_license_advanced(key=key, tier=tier, owner=saved_owner, team_name=saved_branding) - - # Inyectar timestamp del último heartbeat y fecha de expiración inmediatamente - data = get_license_data() - if data: - data["last_checked"] = datetime.now().isoformat() - if expires_at: - data["expires_at"] = expires_at - with open(LICENSE_FILE, "w", encoding="utf-8") as f: - json.dump(data, f, indent=4) - - if tier == "teams": - cprint(f"💜 ¡Licencia TEAMS activada para {saved_owner}! Branding: '{saved_branding}'.", "teams") - else: - cprint(f"🎉 ¡Licencia PRO activada exitosamente para {saved_owner}!", "success") - -def run_license_status(): - data = get_license_data() - if not data or not data.get("active", False): - return cprint("🔓 No hay ninguna licencia activa. Actualmente estás en el Plan Free.", "warning") - - tier = data.get("tier", "free").upper() - key = data.get("key", "N/A") - owner = data.get("owner", "N/A") - team_name = data.get("team_name", "N/A") - activated_at = data.get("activated_at", "N/A") - last_checked = data.get("last_checked", "Nunca (Modo Local)") - expires_at = data.get("expires_at") - - cprint(f"\n========================================================", "pro") - cprint(f" 🛡️ ESTADO DE LICENCIA HÍBRIDA — DATAFORGE CLI v2.0", "pro") - cprint(f"========================================================", "pro") - print(f" • Plan Activo : {tier}") - print(f" • Clave Registrada : {key}") - print(f" • Propietario : {owner}") - if team_name and team_name != "N/A": - print(f" • Team Branding : {team_name}") - - print(f" • Fecha Activación : {activated_at}") - print(f" • Último Heartbeat : {last_checked}") - - if expires_at: - print(f" • Expiración : {expires_at}") - try: - exp_date = datetime.fromisoformat(expires_at) - days_left = (exp_date - datetime.now()).days - if days_left <= 7 and days_left >= 0: - cprint(f"⚠️ ¡Atención! Tu licencia vence en {days_left} días. Renueva pronto en RGS Labs.", "warning") - elif days_left < 0: - cprint(f"❌ Tu licencia ha expirado. Por favor renueva tu plan.", "error") - except Exception: - pass - - cprint(f"========================================================\n", "pro") - -def main(): - init_config() - - if len(sys.argv) > 1: - cmd = sys.argv[1].lower() - if cmd == "license": - if len(sys.argv) > 2: - arg = sys.argv[2] - if arg.lower() == "status": - run_license_status() - else: - run_license_activation(arg) - else: - cprint("⚠️ Falta la clave o argumento. Usa: python main.py license [KEY | status]", "error") - return - elif cmd in ["--help", "-h", "help"]: - show_help_table() - return - - if get_config() is None: - open_welcome_page() - setup_config() - - if len(sys.argv) < 2: - show_help_table() - return - - cmd = sys.argv[1] - target = os.path.abspath(sys.argv[2] if len(sys.argv) > 2 and not sys.argv[2].startswith("--") else ".") - - if cmd == "scan": run_scan(target) - elif cmd == "setup": setup_config() - elif cmd == "explain": run_explain(sys.argv[2] if len(sys.argv) > 2 else "", sys.argv[3].replace("--","") if len(sys.argv)>3 else "") - elif cmd == "map": run_map(target) - elif cmd == "onboard": run_onboard(target) - elif cmd == "welcome": open_welcome_page() - elif cmd == "doc++": run_doc_pro(target) - elif cmd == "architecture++": run_architecture_pro(target) - elif cmd == "export": run_export(target, sys.argv[3] if len(sys.argv) > 3 else "--md") - elif cmd == "preset": run_preset(sys.argv[2] if len(sys.argv) > 2 else "", target) - else: - show_help_table() - +import sys +import os +import json +import urllib.request +import urllib.parse +import webbrowser +import platform +import uuid +from datetime import datetime +from dataforge.config_manager import init_config, get_config, setup_config, save_license_advanced, LICENSE_FILE +from dataforge.api_handler import ask_groq +from dataforge.core import get_all_files, read_file_content, save_report, get_single_file_content, prepare_code_payload + +WELCOME_URL = "https://rabbitgamesdev.github.io/DataForge-CLI/" +SUPABASE_FUNCTION_URL = "https://yultjjqcxtqfakppxyqv.supabase.co/functions/v1/validate-license" + +def get_theme_color(): + config = get_config() or {} + theme = config.get("theme", "dark") + if theme == "tokyo_night": return "\033[38;5;189m" + elif theme == "dracula": return "\033[38;5;212m" + elif theme == "light": return "\033[30m" + return "\033[38;5;248m" + +def cprint(text, style="normal"): + reset = "\033[0m" + color = get_theme_color() + if style == "success": color = "\033[32m" + elif style == "error": color = "\033[31m" + elif style == "warning": color = "\033[33m" + elif style == "pro": color = "\033[38;5;220m" + elif style == "teams": color = "\033[38;5;141m" + print(f"{color}{text}{reset}") + +def get_language_prefix(): + config = get_config() or {} + lang = config.get("language", "es") + if lang == "en": + return "[SYSTEM DIRECTIVE: You MUST respond entirely and strictly in English.]\n\n" + return "[DIRECTIVA DEL SISTEMA: Debes responder entera y estrictamente en Español.]\n\n" + +def get_license_data(): + if os.path.exists(LICENSE_FILE): + try: + with open(LICENSE_FILE, "r", encoding="utf-8") as f: + return json.load(f) + except Exception: + return None + return None + +def verify_license_hybrid() -> bool: + """Motor de Licenciamiento Híbrido: Supabase Cloud + Offline Grace Period + Expiration Check""" + data = get_license_data() + if not data or not data.get("active", False): + return False + + key = data.get("key") + if not key: + return False + + now = datetime.now() + + # Validate expiration locally, including ISO timestamps that contain a timezone. + if data.get("expires_at"): + try: + expires_at = datetime.fromisoformat(data["expires_at"]) + comparison_now = now + if expires_at.tzinfo is not None and comparison_now.tzinfo is None: + comparison_now = comparison_now.astimezone().replace(tzinfo=None) + expires_at = expires_at.astimezone().replace(tzinfo=None) + elif expires_at.tzinfo is None and comparison_now.tzinfo is not None: + expires_at = expires_at.replace(tzinfo=comparison_now.tzinfo) + if comparison_now > expires_at: + data["active"] = False + with open(LICENSE_FILE, "w", encoding="utf-8") as f: + json.dump(data, f, indent=4) + return False + except (TypeError, ValueError, OverflowError): + # Malformed expiry metadata must not crash the CLI; server validation still applies. + pass + + last_checked_str = data.get("last_checked") + should_check = True + last_checked = None + + if last_checked_str: + try: + last_checked = datetime.fromisoformat(last_checked_str) + comparison_now = now + if last_checked.tzinfo is not None and comparison_now.tzinfo is None: + comparison_now = comparison_now.astimezone().replace(tzinfo=None) + elif last_checked.tzinfo is None and comparison_now.tzinfo is not None: + last_checked = last_checked.replace(tzinfo=comparison_now.tzinfo) + age_seconds = (comparison_now - last_checked).total_seconds() + if 0 <= age_seconds < 7 * 24 * 60 * 60: + should_check = False + except (TypeError, ValueError, OverflowError): + last_checked = None + should_check = True + + if should_check: + machine_id = str(uuid.uuid5(uuid.NAMESPACE_DNS, platform.node())) + payload = json.dumps({ + "license_key": key, + "machine_id": machine_id + }).encode("utf-8") + + try: + req = urllib.request.Request( + SUPABASE_FUNCTION_URL, + data=payload, + headers={"Content-Type": "application/json"}, + method="POST" + ) + with urllib.request.urlopen(req, timeout=5) as response: + res_data = json.loads(response.read().decode("utf-8")) + if not res_data.get("success", False): + data["active"] = False + with open(LICENSE_FILE, "w", encoding="utf-8") as f: + json.dump(data, f, indent=4) + return False + + data["tier"] = res_data.get("tier", data.get("tier")) + data["expires_at"] = res_data.get("expires_at", data.get("expires_at")) + data["last_checked"] = now.isoformat() + with open(LICENSE_FILE, "w", encoding="utf-8") as f: + json.dump(data, f, indent=4) + except Exception: + # Offline access is allowed only after a recent successful server check. + if last_checked is None: + return False + try: + comparison_now = now + if last_checked.tzinfo is not None and comparison_now.tzinfo is None: + comparison_now = comparison_now.astimezone().replace(tzinfo=None) + elif last_checked.tzinfo is None and comparison_now.tzinfo is not None: + last_checked = last_checked.replace(tzinfo=comparison_now.tzinfo) + age_seconds = (comparison_now - last_checked).total_seconds() + if not (0 <= age_seconds < 7 * 24 * 60 * 60): + return False + except (TypeError, ValueError, OverflowError): + return False + + return True + +def is_pro_active() -> bool: + data = get_license_data() + if not data or not data.get("active", False): + return False + if not verify_license_hybrid(): + return False + return data.get("tier", "").lower() in ["pro", "teams"] + +def is_teams_active() -> bool: + data = get_license_data() + if not data or not data.get("active", False): + return False + if not verify_license_hybrid(): + return False + return data.get("tier", "").lower() == "teams" + +def ask_enterprise_ai(prompt): + """Route through the provider adapter shared by every command.""" + return ask_groq(prompt) + +def get_footer_branding(): + config = get_config() or {} + brand = config.get("brand_name", "Developer") + + if is_teams_active(): + license_data = get_license_data() or {} + team_name = license_data.get("team_name", brand) + return f"\n\n---\n*Generated by {team_name}*\n>_ Enterprise Architecture Documentation Pipeline" + elif is_pro_active(): + return f"\n\n---\n*Created by {brand}*\n>_ [Powered by DataForge CLI](https://rabbitgamesdev.github.io/DataForge-CLI/)" + else: + return ( + f"\n\n---\n" + f"\n" + f"*Created by {brand}*\n" + f">_ **Powered by DataForge CLI by RGS Labs™**\n" + f">_ [https://rabbitgamesdev.github.io/DataForge-CLI/](https://rabbitgamesdev.github.io/DataForge-CLI/)" + ) + +def open_welcome_page(): + try: webbrowser.open(WELCOME_URL, new=2) + except Exception: cprint(f"👉 Carta de bienvenida: {WELCOME_URL}", "warning") + +def show_help_table(): + status = "💜 TEAMS (Enterprise Branding)" if is_teams_active() else ("⭐ PRO (Developer Suite)" if is_pro_active() else "🔓 Free Tier") + config = get_config() or {} + + cprint(f"\n======================================================================", "pro") + cprint(f" 🚀 DataForge CLI v2.0 - RGS Labs™ | Estado: [{status}]", "pro") + cprint(f" 🤖 IA Activa: {config.get('provider', 'groq').upper()} | Idioma: {config.get('language', 'es').upper()}", "success") + cprint(f"======================================================================", "pro") + + print("\n📋 TABLA DE COMANDOS DISPONIBLES:\n") + print(f"{'Comando':<18} | {'Tipo':<10} | {'Descripción detallada de la función'}") + print("-" * 75) + print(f"{'scan':<18} | {'Free/Pro':<10} | Escanea el proyecto completo y genera un reporte de análisis de IA.") + print(f"{'explain':<18} | {'Free/Pro':<10} | Explica a detalle un archivo de código específico (admite perfiles).") + print(f"{'map':<18} | {'Free/Pro':<10} | Genera un diagrama ASCII visual de la arquitectura del proyecto.") + print(f"{'onboard':<18} | {'Free/Pro':<10} | Crea una guía de incorporación técnica corporativa automatizada.") + print(f"{'doc++':<18} | {'PRO / TEAMS':<10} | Genera documentación técnica avanzada lista para producción.") + print(f"{'architecture++':<18} | {'PRO / TEAMS':<10} | Genera diagramas estructurales Mermaid y análisis profundo.") + print(f"{'export':<18} | {'PRO / TEAMS':<10} | Exporta los reportes generados a formatos limpios (--md, --json, --html).") + print(f"{'preset':<18} | {'PRO / TEAMS':<10} | Ejecuta un pipeline completo automatizado de desarrollo.") + print(f"{'setup':<18} | {'Global':<10} | Reconfigura el motor de IA, API Keys, idioma y tema visual.") + print(f"{'license':<18} | {'Global':<10} | Activa licencias de categoría PRO o TEAMS (con branding corporativo).") + print(f"{'license status':<18} | {'Global':<10} | Activa o consulta el estado de licencias (ej. license status).") + print("-" * 75) + +def run_scan(target_path): + cprint(f"🔍 Escaneando proyecto en: {target_path}...") + files = get_all_files(target_path) + if not files: return cprint("No se encontraron archivos válidos.", "error") + + file_list_str = "\n".join([f"- {f}" for f in files]) + # Payload optimizado con presupuesto seguro de tokens para evitar Error 413 + full_context = prepare_code_payload(files, max_chars=35000) + + prompt = f"{get_language_prefix()}Analiza la arquitectura y código de este proyecto:\n{full_context}" + respuesta = ask_enterprise_ai(prompt) + + final_content = f"# ARCHIVO DE REPORTE - DataForge CLI\n\n**Archivos analizados:**\n{file_list_str}\n\n{respuesta}{get_footer_branding()}" + report_path = save_report(final_content, target_path) + cprint(f"\n✅ Reporte guardado en: {report_path}\n", "success") + print(respuesta) + +def run_explain(file_path, audience=""): + content = get_single_file_content(file_path) + if content is None: return cprint("Error: El archivo no existe.", "error") + + modifier = f"Adapta la explicación para un perfil: {audience.upper()}." if audience else "" + if audience and not is_pro_active(): + return cprint("🔒 Requiere licencia PRO.", "warning") + + prompt = f"{get_language_prefix()}Explica detalladamente qué hace este archivo. {modifier}\n{content}" + respuesta = ask_enterprise_ai(prompt) + cprint(f"\n--- EXPLICACIÓN ({audience.upper() if audience else 'GENERAL'}) ---") + print(respuesta) + +def run_map(target_path): + cprint(f"🗺️ Generando mapa de dependencias ASCII...") + files = get_all_files(target_path) + file_names = "\n".join([os.path.relpath(f, target_path) for f in files]) + prompt = f"{get_language_prefix()}Basado en:\n{file_names}\n\nGenera un mapa visual (ASCII) limpio de la arquitectura." + respuesta = ask_enterprise_ai(prompt) + cprint(f"\n--- MAPA DE ARQUITECTURA ---\n{respuesta}") + +def run_onboard(target_path): + cprint(f"🚀 Generando guía de onboarding...") + files = get_all_files(target_path) + if not files: return cprint("No se encontraron archivos válidos.", "error") + + full_context = prepare_code_payload(files, max_chars=35000) + prompt = f"{get_language_prefix()}Actúa como Líder Técnico. Genera Onboarding corporativo:\n{full_context}" + respuesta = ask_enterprise_ai(prompt) + + final_content = f"# ONBOARDING GUIDE\n\n{respuesta}{get_footer_branding()}" + report_path = save_report(final_content, target_path) + cprint(f"✅ Onboarding generado: {report_path}", "success") + print(f"\n{respuesta}") + +def run_doc_pro(target_path): + if not is_pro_active(): return cprint("🔒 Exclusivo PRO/TEAMS.", "warning") + cprint(f"🚀 Generando documentación técnica de producción (Documentation++)...", "teams" if is_teams_active() else "pro") + files = get_all_files(target_path) + if not files: return + + file_list = "\n".join([os.path.relpath(f, target_path) for f in files[:50]]) + source_context = prepare_code_payload(files[:50], max_chars=30000) + prompt = ( + f"{get_language_prefix()}Arquitecto Senior. Genera README profesional y documentación técnica " + f"basada en el contenido real de los archivos listados abajo. No inventes funciones que no aparezcan. " + f"Indica incertidumbres explícitamente.\n\nARCHIVOS:\n{file_list}\n\nCÓDIGO:\n{source_context}" + ) + respuesta = ask_enterprise_ai(prompt) + + output = os.path.join(target_path, "PRODUCTION_DOCS.md") + with open(output, "w", encoding="utf-8") as f: f.write(respuesta + get_footer_branding()) + cprint(f"✅ Documentación guardada en: {output}", "success") + +def run_architecture_pro(target_path): + if not is_pro_active(): return cprint("🔒 Exclusivo PRO/TEAMS.", "warning") + cprint(f"📐 Generando diagramas Mermaid y análisis profundo...", "pro") + files = get_all_files(target_path) + if not files: return + file_names = "\n".join([os.path.relpath(f, target_path) for f in files[:50]]) + source_context = prepare_code_payload(files[:50], max_chars=30000) + prompt = ( + f"{get_language_prefix()}Arquitecto Experto. Genera código Mermaid (graph TD) y explicación " + f"basados en el contenido real del código, no solo en nombres de archivos. No inventes dependencias; " + f"marca las inferencias como tales.\n\nARCHIVOS:\n{file_names}\n\nCÓDIGO:\n{source_context}" + ) + respuesta = ask_enterprise_ai(prompt) + + output = os.path.join(target_path, "ARCHITECTURE_REPORT.md") + with open(output, "w", encoding="utf-8") as f: f.write(respuesta + get_footer_branding()) + cprint(f"✅ Arquitectura guardada en: {output}", "success") + +def run_export(target_path, fmt): + if not is_pro_active(): return cprint("🔒 Exclusivo PRO/TEAMS.", "warning") + fmt = fmt.replace("--", "").lower() + reports_dir = os.path.join(target_path, "dataforge-reports") + if not os.path.exists(reports_dir): return cprint("⚠️ Sin reportes previos.", "warning") + reports = [os.path.join(reports_dir, f) for f in os.listdir(reports_dir) if f.endswith(".txt") or f.endswith(".md")] + if not reports: return cprint("⚠️ No hay reportes para exportar.", "warning") + + target_file = max(reports, key=os.path.getmtime) + with open(target_file, "r", encoding="utf-8") as f: content = f.read() + output_path = f"{os.path.splitext(target_file)[0]}.{fmt}" + + if fmt == "json": + with open(output_path, "w", encoding="utf-8") as out: json.dump({"content": content}, out, indent=4) + elif fmt == "html": + import html + safe_content = html.escape(content) + with open(output_path, "w", encoding="utf-8") as out: + out.write( + '' + '' + 'DataForge Report
'
+                + safe_content
+                + '
' + ) + else: + with open(output_path, "w", encoding="utf-8") as out: out.write(content) + cprint(f"✅ Exportado a: {output_path}", "success") + +def run_preset(name, target_path): + if not is_pro_active(): return cprint("🔒 Exclusivo PRO/TEAMS.", "warning") + if name == "onboarding": + cprint(f"⚡ Ejecutando Pipeline Automático de Ingeniería...", "pro") + run_scan(target_path) + run_onboard(target_path) + run_doc_pro(target_path) + run_architecture_pro(target_path) + run_export(target_path, "--html") + cprint("🎉 ¡Preset completado exitosamente!", "success") + else: + cprint(f"⚠️ Preset desconocido: {name}.", "warning") + +def run_license_activation(key: str): + machine_id = str(uuid.uuid5(uuid.NAMESPACE_DNS, platform.node())) + owner = input("Introduce tu correo de compra (ej. tu@email.com): ").strip() + if not owner: owner = "cliente@dataforge.dev" + + upper_key = key.upper() + is_team = "TEAM" in upper_key or "TEAMS" in upper_key + + team_name = "" + if is_team: + print("\n🏢 [TEAMS] Licencia corporativa detectada.") + team_name = input("Introduce el Nombre de tu Estudio / Empresa para el Team Branding: ").strip() + if not team_name: team_name = "Enterprise Team" + + payload = json.dumps({ + "license_key": key, + "machine_id": machine_id, + "email": owner, + "team_branding": team_name + }).encode("utf-8") + + try: + req = urllib.request.Request( + SUPABASE_FUNCTION_URL, + data=payload, + headers={"Content-Type": "application/json"}, + method="POST" + ) + with urllib.request.urlopen(req, timeout=10) as response: + res_data = json.loads(response.read().decode("utf-8")) + if not res_data.get("success", False): + return cprint(f"❌ Error de activación: {res_data.get('message', 'Clave inválida')}", "error") + + tier = res_data.get("tier", "pro").lower() + saved_owner = owner + saved_branding = res_data.get("team_branding", team_name) + expires_at = res_data.get("expires_at") + except Exception as e: + return cprint(f"⚠️ Error conectando con el servidor de licencias: {e}", "error") + + # Guardar licencia avanzada + save_license_advanced(key=key, tier=tier, owner=saved_owner, team_name=saved_branding) + + # Inyectar timestamp del último heartbeat y fecha de expiración inmediatamente + data = get_license_data() + if data: + data["last_checked"] = datetime.now().isoformat() + if expires_at: + data["expires_at"] = expires_at + with open(LICENSE_FILE, "w", encoding="utf-8") as f: + json.dump(data, f, indent=4) + + if tier == "teams": + cprint(f"💜 ¡Licencia TEAMS activada para {saved_owner}! Branding: '{saved_branding}'.", "teams") + else: + cprint(f"🎉 ¡Licencia PRO activada exitosamente para {saved_owner}!", "success") + +def run_license_status(): + data = get_license_data() + if not data or not data.get("active", False): + return cprint("🔓 No hay ninguna licencia activa. Actualmente estás en el Plan Free.", "warning") + + tier = data.get("tier", "free").upper() + key = data.get("key", "N/A") + owner = data.get("owner", "N/A") + team_name = data.get("team_name", "N/A") + activated_at = data.get("activated_at", "N/A") + last_checked = data.get("last_checked", "Nunca (Modo Local)") + expires_at = data.get("expires_at") + + cprint(f"\n========================================================", "pro") + cprint(f" 🛡️ ESTADO DE LICENCIA HÍBRIDA — DATAFORGE CLI v2.0", "pro") + cprint(f"========================================================", "pro") + print(f" • Plan Activo : {tier}") + print(f" • Clave Registrada : {key}") + print(f" • Propietario : {owner}") + if team_name and team_name != "N/A": + print(f" • Team Branding : {team_name}") + + print(f" • Fecha Activación : {activated_at}") + print(f" • Último Heartbeat : {last_checked}") + + if expires_at: + print(f" • Expiración : {expires_at}") + try: + exp_date = datetime.fromisoformat(expires_at) + days_left = (exp_date - datetime.now()).days + if days_left <= 7 and days_left >= 0: + cprint(f"⚠️ ¡Atención! Tu licencia vence en {days_left} días. Renueva pronto en RGS Labs.", "warning") + elif days_left < 0: + cprint(f"❌ Tu licencia ha expirado. Por favor renueva tu plan.", "error") + except Exception: + pass + + cprint(f"========================================================\n", "pro") + +def main(): + init_config() + + if len(sys.argv) > 1: + cmd = sys.argv[1].lower() + if cmd == "license": + if len(sys.argv) > 2: + arg = sys.argv[2] + if arg.lower() == "status": + run_license_status() + else: + run_license_activation(arg) + else: + cprint("⚠️ Falta la clave o argumento. Usa: python main.py license [KEY | status]", "error") + return + elif cmd in ["--help", "-h", "help"]: + show_help_table() + return + + if get_config() is None: + open_welcome_page() + setup_config() + + if len(sys.argv) < 2: + show_help_table() + return + + cmd = sys.argv[1] + if cmd == "preset": + preset_name = sys.argv[2] if len(sys.argv) > 2 and not sys.argv[2].startswith("--") else "" + target_arg = sys.argv[3] if len(sys.argv) > 3 and not sys.argv[3].startswith("--") else "." + run_preset(preset_name, os.path.abspath(target_arg)) + return + + target = os.path.abspath(sys.argv[2] if len(sys.argv) > 2 and not sys.argv[2].startswith("--") else ".") + + if cmd == "scan": run_scan(target) + elif cmd == "setup": setup_config() + elif cmd == "explain": run_explain(sys.argv[2] if len(sys.argv) > 2 else "", sys.argv[3].replace("--","") if len(sys.argv)>3 else "") + elif cmd == "map": run_map(target) + elif cmd == "onboard": run_onboard(target) + elif cmd == "welcome": open_welcome_page() + elif cmd == "doc++": run_doc_pro(target) + elif cmd == "architecture++": run_architecture_pro(target) + elif cmd == "export": run_export(target, sys.argv[3] if len(sys.argv) > 3 else "--md") + else: + show_help_table() + if __name__ == "__main__": main() \ No newline at end of file diff --git a/dataforge-cli/tests/test_stability.py b/dataforge-cli/tests/test_stability.py new file mode 100644 index 0000000..298e041 --- /dev/null +++ b/dataforge-cli/tests/test_stability.py @@ -0,0 +1,89 @@ +import json +import os +import tempfile +import unittest +from datetime import datetime, timedelta +from pathlib import Path +from unittest.mock import patch + +import dataforge.core as core +import main + + +class CoreSafetyTests(unittest.TestCase): + def test_scanner_respects_gitignore_and_sensitive_filenames(self): + with tempfile.TemporaryDirectory() as root: + root_path = Path(root) + (root_path / ".gitignore").write_text("private/\n*.local.json\n", encoding="utf-8") + (root_path / "app.py").write_text("print('ok')", encoding="utf-8") + (root_path / ".env").write_text("API_KEY=not-a-real-key", encoding="utf-8") + (root_path / "config.local.json").write_text("{}", encoding="utf-8") + (root_path / "private").mkdir() + (root_path / "private" / "secret.py").write_text("secret", encoding="utf-8") + + found = {Path(p).relative_to(root_path).as_posix() for p in core.get_all_files(root)} + self.assertEqual(found, {"app.py"}) + + def test_large_file_read_is_truncated(self): + with tempfile.TemporaryDirectory() as root: + file_path = Path(root) / "large.py" + file_path.write_text("x" * 100, encoding="utf-8") + result = core.read_file_content(str(file_path), max_bytes=10) + self.assertTrue(result.startswith("x" * 10)) + self.assertIn("truncated", result) + self.assertLess(len(result), 100) + + +class ExportSafetyTests(unittest.TestCase): + def test_html_export_escapes_report_markup(self): + with tempfile.TemporaryDirectory() as root: + report_dir = Path(root) / "dataforge-reports" + report_dir.mkdir() + (report_dir / "report.txt").write_text("", encoding="utf-8") + with patch.object(main, "is_pro_active", return_value=True): + main.run_export(root, "--html") + html_path = report_dir / "report.html" + exported = html_path.read_text(encoding="utf-8") + self.assertIn("<script>", exported) + self.assertNotIn("