TZZR CLI v0.1.0 - Herramienta de gestión sistema TZZR
- Comandos: search, list, tree, copy, info, add, grupos, stats - Conexión SSH a HST (72.62.2.84) - Estructura autoreferenciada (sin cat/subcat) - Grupos válidos: hst, spe, vsn, vue, flg 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
10
.gitignore
vendored
Normal file
10
.gitignore
vendored
Normal file
@@ -0,0 +1,10 @@
|
||||
venv/
|
||||
__pycache__/
|
||||
*.pyc
|
||||
*.pyo
|
||||
*.egg-info/
|
||||
dist/
|
||||
build/
|
||||
.eggs/
|
||||
*.egg
|
||||
.env
|
||||
89
README.md
Normal file
89
README.md
Normal file
@@ -0,0 +1,89 @@
|
||||
# TZZR CLI
|
||||
|
||||
Herramienta de línea de comandos para gestionar el sistema TZZR.
|
||||
|
||||
## Instalación
|
||||
|
||||
```bash
|
||||
cd /home/architect/captain-claude/apps/tzzr-cli
|
||||
pip install -e .
|
||||
```
|
||||
|
||||
## Uso
|
||||
|
||||
### Buscar hashtags
|
||||
|
||||
```bash
|
||||
tzzr search "tecnología"
|
||||
tzzr search "HST" --limit=100
|
||||
```
|
||||
|
||||
### Listar hashtags
|
||||
|
||||
```bash
|
||||
tzzr list # Lista todos
|
||||
tzzr list --grupo=hst # Filtrar por grupo
|
||||
tzzr list -g hst -l 50 # Con límite
|
||||
tzzr list --offset=20 --limit=20 # Paginación
|
||||
```
|
||||
|
||||
### Ver árbol jerárquico
|
||||
|
||||
```bash
|
||||
tzzr tree --root=HST001
|
||||
tzzr tree -r HST001
|
||||
```
|
||||
|
||||
### Copiar h_maestro al clipboard
|
||||
|
||||
```bash
|
||||
tzzr copy HST001
|
||||
```
|
||||
|
||||
### Ver información de un hashtag
|
||||
|
||||
```bash
|
||||
tzzr info HST001
|
||||
tzzr info HST001 --children # Incluir hijos directos
|
||||
```
|
||||
|
||||
### Crear nuevo hashtag
|
||||
|
||||
```bash
|
||||
tzzr add --ref=HST100 --nombre="Nuevo Tag" --grupo=hst
|
||||
tzzr add -r HST100 -n "Nuevo Tag" -g hst -p HST001 # Con padre
|
||||
```
|
||||
|
||||
### Ver grupos disponibles
|
||||
|
||||
```bash
|
||||
tzzr grupos
|
||||
```
|
||||
|
||||
### Ver estadísticas
|
||||
|
||||
```bash
|
||||
tzzr stats
|
||||
```
|
||||
|
||||
## Configuración
|
||||
|
||||
Variables de entorno para la conexión a la base de datos:
|
||||
|
||||
```bash
|
||||
export TZZR_DB_HOST=72.62.2.84
|
||||
export TZZR_DB_PORT=5432
|
||||
export TZZR_DB_NAME=hst_images
|
||||
export TZZR_DB_USER=directus
|
||||
export TZZR_DB_PASSWORD=directus_hst_2024
|
||||
```
|
||||
|
||||
## Desarrollo
|
||||
|
||||
```bash
|
||||
# Instalar en modo desarrollo
|
||||
pip install -e .
|
||||
|
||||
# Ejecutar directamente
|
||||
python -m tzzr.cli search "test"
|
||||
```
|
||||
4
requirements.txt
Normal file
4
requirements.txt
Normal file
@@ -0,0 +1,4 @@
|
||||
click>=8.0.0
|
||||
psycopg2-binary>=2.9.0
|
||||
rich>=13.0.0
|
||||
pyperclip>=1.8.0
|
||||
47
setup.py
Normal file
47
setup.py
Normal file
@@ -0,0 +1,47 @@
|
||||
"""
|
||||
Setup para TZZR CLI
|
||||
"""
|
||||
|
||||
from setuptools import setup, find_packages
|
||||
|
||||
with open("README.md", "r", encoding="utf-8") as f:
|
||||
long_description = f.read()
|
||||
|
||||
with open("requirements.txt", "r", encoding="utf-8") as f:
|
||||
requirements = [line.strip() for line in f if line.strip() and not line.startswith("#")]
|
||||
|
||||
setup(
|
||||
name="tzzr-cli",
|
||||
version="0.1.0",
|
||||
author="ARCHITECT",
|
||||
author_email="architect@tzzr.local",
|
||||
description="CLI para gestionar el sistema TZZR",
|
||||
long_description=long_description,
|
||||
long_description_content_type="text/markdown",
|
||||
url="http://69.62.126.110:3000/architect/tzzr-cli",
|
||||
packages=find_packages(),
|
||||
classifiers=[
|
||||
"Development Status :: 3 - Alpha",
|
||||
"Environment :: Console",
|
||||
"Intended Audience :: Developers",
|
||||
"License :: OSI Approved :: MIT License",
|
||||
"Operating System :: OS Independent",
|
||||
"Programming Language :: Python :: 3",
|
||||
"Programming Language :: Python :: 3.8",
|
||||
"Programming Language :: Python :: 3.9",
|
||||
"Programming Language :: Python :: 3.10",
|
||||
"Programming Language :: Python :: 3.11",
|
||||
"Programming Language :: Python :: 3.12",
|
||||
"Topic :: Database :: Front-Ends",
|
||||
"Topic :: Utilities",
|
||||
],
|
||||
python_requires=">=3.8",
|
||||
install_requires=requirements,
|
||||
entry_points={
|
||||
"console_scripts": [
|
||||
"tzzr=tzzr.cli:main",
|
||||
],
|
||||
},
|
||||
include_package_data=True,
|
||||
zip_safe=False,
|
||||
)
|
||||
6
tzzr/__init__.py
Normal file
6
tzzr/__init__.py
Normal file
@@ -0,0 +1,6 @@
|
||||
"""
|
||||
TZZR CLI - Herramienta de línea de comandos para gestionar el sistema TZZR
|
||||
"""
|
||||
|
||||
__version__ = "0.1.0"
|
||||
__author__ = "ARCHITECT"
|
||||
319
tzzr/cli.py
Normal file
319
tzzr/cli.py
Normal file
@@ -0,0 +1,319 @@
|
||||
"""
|
||||
TZZR CLI - Comandos principales
|
||||
"""
|
||||
|
||||
import sys
|
||||
from typing import Optional
|
||||
|
||||
import click
|
||||
from rich.console import Console
|
||||
|
||||
from . import __version__
|
||||
from .db import (
|
||||
search_hashtags,
|
||||
list_hashtags,
|
||||
get_hashtag_by_ref,
|
||||
get_hashtag_tree,
|
||||
add_hashtag,
|
||||
count_hashtags,
|
||||
get_grupos,
|
||||
get_children,
|
||||
)
|
||||
from .models import build_tree
|
||||
from .utils import (
|
||||
console,
|
||||
print_table,
|
||||
print_hashtag_info,
|
||||
print_tree,
|
||||
print_success,
|
||||
print_error,
|
||||
print_warning,
|
||||
copy_to_clipboard,
|
||||
format_count,
|
||||
)
|
||||
|
||||
|
||||
@click.group()
|
||||
@click.version_option(version=__version__, prog_name="tzzr")
|
||||
def cli():
|
||||
"""
|
||||
TZZR CLI - Herramienta de gestión del sistema TZZR
|
||||
|
||||
Gestiona hashtags y taxonomías del sistema HST.
|
||||
"""
|
||||
pass
|
||||
|
||||
|
||||
@cli.command()
|
||||
@click.argument("query")
|
||||
@click.option("--limit", "-l", default=50, help="Límite de resultados")
|
||||
def search(query: str, limit: int):
|
||||
"""
|
||||
Busca hashtags por nombre, referencia o h_maestro.
|
||||
|
||||
Ejemplo: tzzr search "tecnología"
|
||||
"""
|
||||
try:
|
||||
results = search_hashtags(query, limit)
|
||||
|
||||
if not results:
|
||||
print_warning(f"No se encontraron resultados para '{query}'")
|
||||
return
|
||||
|
||||
print_table(
|
||||
results,
|
||||
title=f"Resultados para '{query}' ({len(results)} encontrados)",
|
||||
columns=["ref", "nombre_es", "grupo", "h_maestro", "padre_h_maestro"]
|
||||
)
|
||||
except Exception as e:
|
||||
print_error(f"Error en búsqueda: {e}")
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
@cli.command("list")
|
||||
@click.option("--grupo", "-g", help="Filtrar por grupo (hst, spe, vsn, vue, flg)")
|
||||
@click.option("--limit", "-l", default=20, help="Límite de resultados")
|
||||
@click.option("--offset", "-o", default=0, help="Offset para paginación")
|
||||
def list_cmd(grupo: Optional[str], limit: int, offset: int):
|
||||
"""
|
||||
Lista hashtags con filtros opcionales.
|
||||
|
||||
Ejemplos:
|
||||
tzzr list --grupo=hst --limit=20
|
||||
tzzr list -g spe -l 50
|
||||
"""
|
||||
try:
|
||||
results = list_hashtags(grupo=grupo, limit=limit, offset=offset)
|
||||
total = count_hashtags(grupo)
|
||||
|
||||
if not results:
|
||||
print_warning("No se encontraron hashtags")
|
||||
return
|
||||
|
||||
title = f"Hashtags"
|
||||
if grupo:
|
||||
title += f" (grupo: {grupo})"
|
||||
title += f" - Mostrando {len(results)} de {total}"
|
||||
|
||||
print_table(
|
||||
results,
|
||||
title=title,
|
||||
columns=["ref", "nombre_es", "grupo", "h_maestro", "padre_h_maestro"]
|
||||
)
|
||||
|
||||
# Mostrar info de paginación
|
||||
if offset + limit < total:
|
||||
remaining = total - offset - limit
|
||||
console.print(f"\n[dim]Usa --offset={offset + limit} para ver los siguientes {min(remaining, limit)} resultados[/dim]")
|
||||
|
||||
except Exception as e:
|
||||
print_error(f"Error listando hashtags: {e}")
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
@cli.command()
|
||||
@click.option("--root", "-r", required=True, help="Referencia del nodo raíz (ref o h_maestro)")
|
||||
@click.option("--depth", "-d", default=10, help="Profundidad máxima")
|
||||
def tree(root: str, depth: int):
|
||||
"""
|
||||
Muestra el árbol jerárquico desde un nodo raíz.
|
||||
|
||||
Ejemplo: tzzr tree --root=hst::tecnologia
|
||||
"""
|
||||
try:
|
||||
# Verificar que existe el nodo raíz
|
||||
root_hashtag = get_hashtag_by_ref(root)
|
||||
if not root_hashtag:
|
||||
print_error(f"No se encontró hashtag con referencia '{root}'")
|
||||
sys.exit(1)
|
||||
|
||||
# Obtener árbol
|
||||
tree_data = get_hashtag_tree(root)
|
||||
|
||||
if not tree_data:
|
||||
print_warning("No se encontraron nodos en el árbol")
|
||||
return
|
||||
|
||||
# Construir y mostrar árbol
|
||||
nodes = build_tree(tree_data, root_hashtag.get('h_maestro'))
|
||||
print_tree(nodes, title=f"Árbol desde {root}")
|
||||
|
||||
console.print(f"\n[dim]Total: {format_count(len(nodes), 'nodo', 'nodos')}[/dim]")
|
||||
|
||||
except Exception as e:
|
||||
print_error(f"Error obteniendo árbol: {e}")
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
@cli.command()
|
||||
@click.argument("ref")
|
||||
def copy(ref: str):
|
||||
"""
|
||||
Copia el h_maestro de un hashtag al clipboard.
|
||||
|
||||
Ejemplo: tzzr copy HST001
|
||||
"""
|
||||
try:
|
||||
hashtag = get_hashtag_by_ref(ref)
|
||||
|
||||
if not hashtag:
|
||||
print_error(f"No se encontró hashtag con referencia '{ref}'")
|
||||
sys.exit(1)
|
||||
|
||||
maestro = hashtag.get("h_maestro")
|
||||
|
||||
if not maestro:
|
||||
print_warning(f"El hashtag '{ref}' no tiene h_maestro definido")
|
||||
return
|
||||
|
||||
if copy_to_clipboard(maestro):
|
||||
print_success(f"Copiado al clipboard: {maestro}")
|
||||
else:
|
||||
# Si no se pudo copiar, al menos mostrar el valor
|
||||
console.print(f"\n[bold]h_maestro:[/bold] {maestro}")
|
||||
|
||||
except Exception as e:
|
||||
print_error(f"Error: {e}")
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
@cli.command()
|
||||
@click.argument("ref")
|
||||
@click.option("--children", "-c", is_flag=True, help="Mostrar también hijos directos")
|
||||
def info(ref: str, children: bool):
|
||||
"""
|
||||
Muestra información detallada de un hashtag.
|
||||
|
||||
Ejemplo: tzzr info hst::tecnologia
|
||||
"""
|
||||
try:
|
||||
hashtag = get_hashtag_by_ref(ref)
|
||||
|
||||
if not hashtag:
|
||||
print_error(f"No se encontró hashtag con referencia '{ref}'")
|
||||
sys.exit(1)
|
||||
|
||||
print_hashtag_info(hashtag)
|
||||
|
||||
# Mostrar hijos si se solicita
|
||||
if children:
|
||||
child_list = get_children(ref)
|
||||
if child_list:
|
||||
console.print(f"\n[bold cyan]Hijos directos ({len(child_list)}):[/bold cyan]")
|
||||
print_table(
|
||||
child_list,
|
||||
columns=["ref", "nombre_es", "h_maestro"]
|
||||
)
|
||||
else:
|
||||
console.print("\n[dim]Sin hijos directos[/dim]")
|
||||
|
||||
except Exception as e:
|
||||
print_error(f"Error: {e}")
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
@cli.command()
|
||||
@click.option("--ref", "-r", required=True, help="Referencia corta del hashtag")
|
||||
@click.option("--nombre", "-n", required=True, help="Nombre del hashtag (español)")
|
||||
@click.option("--grupo", "-g", required=True,
|
||||
type=click.Choice(['hst', 'spe', 'vsn', 'vue', 'flg']),
|
||||
help="Grupo al que pertenece")
|
||||
@click.option("--padre", "-p", default=None, help="h_maestro del padre (opcional)")
|
||||
@click.option("--maestro", "-m", default=None, help="h_maestro personalizado (opcional)")
|
||||
def add(ref: str, nombre: str, grupo: str, padre: Optional[str], maestro: Optional[str]):
|
||||
"""
|
||||
Crea un nuevo hashtag.
|
||||
|
||||
Ejemplo:
|
||||
tzzr add --ref=HST100 --nombre="Nuevo Tag" --grupo=hst --padre=hst::tecnologia
|
||||
"""
|
||||
try:
|
||||
# Verificar si ya existe
|
||||
existing = get_hashtag_by_ref(ref)
|
||||
if existing:
|
||||
print_error(f"Ya existe un hashtag con referencia '{ref}'")
|
||||
sys.exit(1)
|
||||
|
||||
# Verificar padre si se especificó
|
||||
if padre:
|
||||
parent = get_hashtag_by_ref(padre)
|
||||
if not parent:
|
||||
print_error(f"No se encontró el hashtag padre '{padre}'")
|
||||
sys.exit(1)
|
||||
|
||||
# Crear hashtag
|
||||
result = add_hashtag(
|
||||
referencia=ref,
|
||||
nombre=nombre,
|
||||
grupo=grupo,
|
||||
padre=padre,
|
||||
maestro=maestro,
|
||||
)
|
||||
|
||||
if result:
|
||||
print_success(f"Hashtag '{ref}' creado correctamente")
|
||||
print_hashtag_info(result)
|
||||
else:
|
||||
print_error("Error al crear el hashtag")
|
||||
sys.exit(1)
|
||||
|
||||
except Exception as e:
|
||||
print_error(f"Error: {e}")
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
@cli.command()
|
||||
def grupos():
|
||||
"""
|
||||
Lista todos los grupos disponibles.
|
||||
"""
|
||||
try:
|
||||
group_list = get_grupos()
|
||||
|
||||
if not group_list:
|
||||
print_warning("No se encontraron grupos")
|
||||
return
|
||||
|
||||
console.print("\n[bold cyan]Grupos disponibles:[/bold cyan]\n")
|
||||
for g in group_list:
|
||||
count = count_hashtags(g)
|
||||
console.print(f" - [yellow]{g}[/yellow] ({count} hashtags)")
|
||||
|
||||
console.print(f"\n[dim]Total: {format_count(len(group_list), 'grupo', 'grupos')}[/dim]")
|
||||
|
||||
except Exception as e:
|
||||
print_error(f"Error: {e}")
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
@cli.command()
|
||||
def stats():
|
||||
"""
|
||||
Muestra estadísticas generales del sistema.
|
||||
"""
|
||||
try:
|
||||
total = count_hashtags()
|
||||
groups = get_grupos()
|
||||
|
||||
console.print("\n[bold cyan]Estadísticas TZZR (tabla hst)[/bold cyan]\n")
|
||||
console.print(f" Total hashtags: [green]{total}[/green]")
|
||||
console.print(f" Grupos: [green]{len(groups)}[/green]\n")
|
||||
|
||||
console.print("[bold]Por grupo:[/bold]")
|
||||
for g in groups:
|
||||
count = count_hashtags(g)
|
||||
pct = (count / total * 100) if total > 0 else 0
|
||||
console.print(f" - {g}: {count} ({pct:.1f}%)")
|
||||
|
||||
except Exception as e:
|
||||
print_error(f"Error: {e}")
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
def main():
|
||||
"""Punto de entrada principal"""
|
||||
cli()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
287
tzzr/db.py
Normal file
287
tzzr/db.py
Normal file
@@ -0,0 +1,287 @@
|
||||
"""
|
||||
Conexión a PostgreSQL para el sistema TZZR
|
||||
|
||||
La tabla principal es 'hst' con la estructura autoreferenciada:
|
||||
- id: integer (PK)
|
||||
- ref: referencia corta (ej: HST001)
|
||||
- h_maestro: referencia única completa (ej: hst::tecnologia)
|
||||
- grupo: hst (principal), spe, vsn, vue, flg
|
||||
- nombre_es: nombre en español
|
||||
- nombre_en: nombre en inglés
|
||||
- padre_h_maestro: referencia al padre (autoreferenciado a hst.h_maestro)
|
||||
- rootref: referencia raíz
|
||||
- descripcion: descripción extendida
|
||||
- imagen_url: URL de imagen
|
||||
- mrf: referencia MRF
|
||||
- created_at: fecha de creación
|
||||
"""
|
||||
|
||||
import os
|
||||
import subprocess
|
||||
from contextlib import contextmanager
|
||||
from typing import Optional, List, Dict, Any
|
||||
|
||||
import psycopg2
|
||||
from psycopg2.extras import RealDictCursor
|
||||
|
||||
|
||||
# Configuración de conexión - requiere SSH tunnel o ejecución local en hst
|
||||
DB_CONFIG = {
|
||||
"host": os.getenv("TZZR_DB_HOST", "localhost"),
|
||||
"port": int(os.getenv("TZZR_DB_PORT", "5432")),
|
||||
"database": os.getenv("TZZR_DB_NAME", "hst_images"),
|
||||
"user": os.getenv("TZZR_DB_USER", "directus"),
|
||||
"password": os.getenv("TZZR_DB_PASSWORD", "directus_hst_2024"),
|
||||
}
|
||||
|
||||
# Para conexión remota via SSH
|
||||
SSH_CONFIG = {
|
||||
"host": "72.62.2.84",
|
||||
"user": "root",
|
||||
"key": os.path.expanduser("~/.ssh/tzzr"),
|
||||
"container": "postgres_hst",
|
||||
}
|
||||
|
||||
|
||||
class Database:
|
||||
"""Clase para gestionar conexiones a PostgreSQL"""
|
||||
|
||||
def __init__(self, config: Optional[Dict[str, Any]] = None, use_ssh: bool = True):
|
||||
self.config = config or DB_CONFIG
|
||||
self.use_ssh = use_ssh
|
||||
self._connection = None
|
||||
|
||||
def execute_ssh(self, query: str) -> List[Dict]:
|
||||
"""Ejecuta query via SSH al servidor remoto"""
|
||||
cmd = [
|
||||
"ssh", "-i", SSH_CONFIG["key"],
|
||||
f"{SSH_CONFIG['user']}@{SSH_CONFIG['host']}",
|
||||
f"docker exec {SSH_CONFIG['container']} psql -U directus -d hst_images -t -A -F '|' -c \"{query}\""
|
||||
]
|
||||
try:
|
||||
result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
|
||||
if result.returncode != 0:
|
||||
raise Exception(f"Error SSH: {result.stderr}")
|
||||
|
||||
lines = [l.strip() for l in result.stdout.strip().split('\n') if l.strip()]
|
||||
return lines
|
||||
except subprocess.TimeoutExpired:
|
||||
raise Exception("Timeout en conexión SSH")
|
||||
|
||||
@property
|
||||
def connection(self):
|
||||
"""Obtiene o crea una conexión directa (solo si no usa SSH)"""
|
||||
if self.use_ssh:
|
||||
raise Exception("Usando modo SSH, no hay conexión directa")
|
||||
if self._connection is None or self._connection.closed:
|
||||
self._connection = psycopg2.connect(**self.config)
|
||||
return self._connection
|
||||
|
||||
def close(self):
|
||||
"""Cierra la conexión"""
|
||||
if self._connection and not self._connection.closed:
|
||||
self._connection.close()
|
||||
self._connection = None
|
||||
|
||||
@contextmanager
|
||||
def cursor(self, dict_cursor: bool = True):
|
||||
"""Context manager para cursor (solo modo directo)"""
|
||||
if self.use_ssh:
|
||||
raise Exception("Usando modo SSH, no hay cursor disponible")
|
||||
cursor_factory = RealDictCursor if dict_cursor else None
|
||||
cursor = self.connection.cursor(cursor_factory=cursor_factory)
|
||||
try:
|
||||
yield cursor
|
||||
self.connection.commit()
|
||||
except Exception as e:
|
||||
self.connection.rollback()
|
||||
raise e
|
||||
finally:
|
||||
cursor.close()
|
||||
|
||||
def execute(self, query: str, params: tuple = None) -> List[Dict]:
|
||||
"""Ejecuta una query y retorna resultados"""
|
||||
if self.use_ssh:
|
||||
# Formatear query con parámetros para SSH
|
||||
if params:
|
||||
# Escapar parámetros de forma segura
|
||||
safe_params = []
|
||||
for p in params:
|
||||
if p is None:
|
||||
safe_params.append("NULL")
|
||||
elif isinstance(p, str):
|
||||
safe_params.append(f"'{p}'")
|
||||
else:
|
||||
safe_params.append(str(p))
|
||||
query = query.replace("%s", "{}").format(*safe_params)
|
||||
return self._execute_ssh_query(query)
|
||||
else:
|
||||
with self.cursor() as cur:
|
||||
cur.execute(query, params)
|
||||
if cur.description:
|
||||
return cur.fetchall()
|
||||
return []
|
||||
|
||||
def _execute_ssh_query(self, query: str) -> List[Dict]:
|
||||
"""Ejecuta query via SSH y parsea resultados"""
|
||||
# Obtener columnas primero
|
||||
lines = self.execute_ssh(query)
|
||||
if not lines:
|
||||
return []
|
||||
|
||||
# Las columnas vienen en la primera ejecución
|
||||
# Hacemos query con cabeceras
|
||||
query_with_headers = query.rstrip(';')
|
||||
cmd = [
|
||||
"ssh", "-i", SSH_CONFIG["key"],
|
||||
f"{SSH_CONFIG['user']}@{SSH_CONFIG['host']}",
|
||||
f"docker exec {SSH_CONFIG['container']} psql -U directus -d hst_images -F '|' -c \"{query_with_headers}\""
|
||||
]
|
||||
result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
|
||||
|
||||
if result.returncode != 0:
|
||||
raise Exception(f"Error SSH: {result.stderr}")
|
||||
|
||||
lines = [l.strip() for l in result.stdout.strip().split('\n') if l.strip()]
|
||||
if len(lines) < 2:
|
||||
return []
|
||||
|
||||
# Primera línea son headers, segunda es separador
|
||||
headers = [h.strip() for h in lines[0].split('|')]
|
||||
results = []
|
||||
|
||||
for line in lines[2:]:
|
||||
if line.startswith('(') and 'row' in line:
|
||||
break
|
||||
values = [v.strip() for v in line.split('|')]
|
||||
if len(values) == len(headers):
|
||||
row = {}
|
||||
for i, h in enumerate(headers):
|
||||
val = values[i] if values[i] else None
|
||||
row[h] = val
|
||||
results.append(row)
|
||||
|
||||
return results
|
||||
|
||||
def execute_one(self, query: str, params: tuple = None) -> Optional[Dict]:
|
||||
"""Ejecuta una query y retorna un solo resultado"""
|
||||
results = self.execute(query, params)
|
||||
return results[0] if results else None
|
||||
|
||||
|
||||
# Instancia global
|
||||
db = Database(use_ssh=True)
|
||||
|
||||
|
||||
def get_db() -> Database:
|
||||
"""Obtiene la instancia de base de datos"""
|
||||
return db
|
||||
|
||||
|
||||
def search_hashtags(query: str, limit: int = 50) -> List[Dict]:
|
||||
"""Busca hashtags por nombre o referencia"""
|
||||
sql = f"""
|
||||
SELECT id, ref, nombre_es, h_maestro, grupo, padre_h_maestro, rootref
|
||||
FROM hst
|
||||
WHERE nombre_es ILIKE '%{query}%'
|
||||
OR ref ILIKE '%{query}%'
|
||||
OR h_maestro ILIKE '%{query}%'
|
||||
ORDER BY grupo, nombre_es
|
||||
LIMIT {limit}
|
||||
"""
|
||||
return db.execute(sql)
|
||||
|
||||
|
||||
def list_hashtags(grupo: Optional[str] = None, limit: int = 20, offset: int = 0) -> List[Dict]:
|
||||
"""Lista hashtags con filtros opcionales"""
|
||||
where_clause = f"WHERE grupo = '{grupo}'" if grupo else ""
|
||||
|
||||
sql = f"""
|
||||
SELECT id, ref, nombre_es, h_maestro, grupo, padre_h_maestro, rootref
|
||||
FROM hst
|
||||
{where_clause}
|
||||
ORDER BY grupo, nombre_es
|
||||
LIMIT {limit} OFFSET {offset}
|
||||
"""
|
||||
return db.execute(sql)
|
||||
|
||||
|
||||
def get_hashtag_by_ref(ref: str) -> Optional[Dict]:
|
||||
"""Obtiene un hashtag por su referencia (ref o h_maestro)"""
|
||||
sql = f"""
|
||||
SELECT id, ref, nombre_es, nombre_en, h_maestro, grupo,
|
||||
padre_h_maestro, rootref, descripcion, imagen_url, created_at
|
||||
FROM hst
|
||||
WHERE ref = '{ref}' OR h_maestro = '{ref}'
|
||||
"""
|
||||
return db.execute_one(sql)
|
||||
|
||||
|
||||
def get_hashtag_tree(root_ref: str) -> List[Dict]:
|
||||
"""Obtiene el árbol jerárquico desde un nodo raíz"""
|
||||
sql = f"""
|
||||
WITH RECURSIVE tree AS (
|
||||
SELECT id, ref, nombre_es, h_maestro, grupo, padre_h_maestro, rootref, 0 as depth
|
||||
FROM hst
|
||||
WHERE ref = '{root_ref}' OR h_maestro = '{root_ref}'
|
||||
|
||||
UNION ALL
|
||||
|
||||
SELECT h.id, h.ref, h.nombre_es, h.h_maestro, h.grupo, h.padre_h_maestro, h.rootref, t.depth + 1
|
||||
FROM hst h
|
||||
INNER JOIN tree t ON h.padre_h_maestro = t.h_maestro
|
||||
WHERE t.depth < 10
|
||||
)
|
||||
SELECT * FROM tree
|
||||
ORDER BY depth, nombre_es
|
||||
"""
|
||||
return db.execute(sql)
|
||||
|
||||
|
||||
def get_children(parent_ref: str) -> List[Dict]:
|
||||
"""Obtiene los hijos directos de un hashtag"""
|
||||
# Primero obtener el h_maestro del padre
|
||||
parent = get_hashtag_by_ref(parent_ref)
|
||||
if not parent:
|
||||
return []
|
||||
|
||||
h_maestro = parent.get('h_maestro')
|
||||
sql = f"""
|
||||
SELECT id, ref, nombre_es, h_maestro, grupo, padre_h_maestro, rootref
|
||||
FROM hst
|
||||
WHERE padre_h_maestro = '{h_maestro}'
|
||||
ORDER BY nombre_es
|
||||
"""
|
||||
return db.execute(sql)
|
||||
|
||||
|
||||
def add_hashtag(referencia: str, nombre: str, grupo: str, padre: Optional[str] = None,
|
||||
maestro: Optional[str] = None, nivel: int = 1) -> Optional[Dict]:
|
||||
"""Crea un nuevo hashtag"""
|
||||
padre_val = f"'{padre}'" if padre else "NULL"
|
||||
maestro_val = maestro if maestro else f"{grupo}::{nombre.lower().replace(' ', '_')}"
|
||||
|
||||
sql = f"""
|
||||
INSERT INTO hst (ref, nombre_es, grupo, padre_h_maestro, h_maestro)
|
||||
VALUES ('{referencia}', '{nombre}', '{grupo}', {padre_val}, '{maestro_val}')
|
||||
RETURNING id, ref, nombre_es, h_maestro, grupo, padre_h_maestro
|
||||
"""
|
||||
return db.execute_one(sql)
|
||||
|
||||
|
||||
def count_hashtags(grupo: Optional[str] = None) -> int:
|
||||
"""Cuenta hashtags totales o por grupo"""
|
||||
if grupo:
|
||||
sql = f"SELECT COUNT(*) as count FROM hst WHERE grupo = '{grupo}'"
|
||||
else:
|
||||
sql = "SELECT COUNT(*) as count FROM hst"
|
||||
|
||||
result = db.execute_one(sql)
|
||||
return int(result["count"]) if result else 0
|
||||
|
||||
|
||||
def get_grupos() -> List[str]:
|
||||
"""Obtiene la lista de grupos únicos"""
|
||||
sql = "SELECT DISTINCT grupo FROM hst WHERE grupo IS NOT NULL ORDER BY grupo"
|
||||
results = db.execute(sql)
|
||||
return [r["grupo"] for r in results]
|
||||
154
tzzr/models.py
Normal file
154
tzzr/models.py
Normal file
@@ -0,0 +1,154 @@
|
||||
"""
|
||||
Modelos de datos para TZZR
|
||||
|
||||
Representa la tabla hst con su estructura:
|
||||
- id: integer (PK)
|
||||
- ref: referencia corta
|
||||
- h_maestro: referencia única completa
|
||||
- grupo: hst, spe, hsu, msu, cat, subcat
|
||||
- nombre_es, nombre_en: nombres
|
||||
- padre_h_maestro: referencia al padre
|
||||
- rootref, descripcion, imagen_url, mrf
|
||||
"""
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime
|
||||
from typing import Optional, List
|
||||
|
||||
|
||||
@dataclass
|
||||
class Hashtag:
|
||||
"""Representa un hashtag en la tabla hst"""
|
||||
id: int
|
||||
ref: str
|
||||
h_maestro: str
|
||||
grupo: str
|
||||
nombre_es: Optional[str] = None
|
||||
nombre_en: Optional[str] = None
|
||||
padre_h_maestro: Optional[str] = None
|
||||
rootref: Optional[str] = None
|
||||
descripcion: Optional[str] = None
|
||||
imagen_url: Optional[str] = None
|
||||
mrf: Optional[str] = None
|
||||
created_at: Optional[datetime] = None
|
||||
children: List["Hashtag"] = field(default_factory=list)
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict) -> "Hashtag":
|
||||
"""Crea un Hashtag desde un diccionario"""
|
||||
return cls(
|
||||
id=data.get("id"),
|
||||
ref=data.get("ref", ""),
|
||||
h_maestro=data.get("h_maestro", ""),
|
||||
grupo=data.get("grupo", ""),
|
||||
nombre_es=data.get("nombre_es"),
|
||||
nombre_en=data.get("nombre_en"),
|
||||
padre_h_maestro=data.get("padre_h_maestro"),
|
||||
rootref=data.get("rootref"),
|
||||
descripcion=data.get("descripcion"),
|
||||
imagen_url=data.get("imagen_url"),
|
||||
mrf=data.get("mrf"),
|
||||
created_at=data.get("created_at"),
|
||||
)
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
"""Convierte a diccionario"""
|
||||
return {
|
||||
"id": self.id,
|
||||
"ref": self.ref,
|
||||
"h_maestro": self.h_maestro,
|
||||
"grupo": self.grupo,
|
||||
"nombre_es": self.nombre_es,
|
||||
"nombre_en": self.nombre_en,
|
||||
"padre_h_maestro": self.padre_h_maestro,
|
||||
"rootref": self.rootref,
|
||||
"descripcion": self.descripcion,
|
||||
"imagen_url": self.imagen_url,
|
||||
"mrf": self.mrf,
|
||||
"created_at": self.created_at,
|
||||
}
|
||||
|
||||
@property
|
||||
def display_name(self) -> str:
|
||||
"""Nombre para mostrar"""
|
||||
name = self.nombre_es or self.ref
|
||||
return f"{self.ref}: {name}"
|
||||
|
||||
def __str__(self) -> str:
|
||||
return self.display_name
|
||||
|
||||
|
||||
@dataclass
|
||||
class TreeNode:
|
||||
"""Nodo de árbol para visualización jerárquica"""
|
||||
hashtag: Hashtag
|
||||
depth: int = 0
|
||||
is_last: bool = False
|
||||
parent_is_last: List[bool] = field(default_factory=list)
|
||||
|
||||
@property
|
||||
def prefix(self) -> str:
|
||||
"""Genera el prefijo visual para el árbol"""
|
||||
if self.depth == 0:
|
||||
return ""
|
||||
|
||||
parts = []
|
||||
for is_last in self.parent_is_last[:-1]:
|
||||
parts.append(" " if is_last else "| ")
|
||||
|
||||
if self.parent_is_last:
|
||||
parts.append("+-- " if self.is_last else "|-- ")
|
||||
|
||||
return "".join(parts)
|
||||
|
||||
def __str__(self) -> str:
|
||||
return f"{self.prefix}{self.hashtag.display_name}"
|
||||
|
||||
|
||||
def build_tree(hashtags: List[dict], root_h_maestro: str) -> List[TreeNode]:
|
||||
"""
|
||||
Construye una lista de TreeNodes para visualización
|
||||
|
||||
Args:
|
||||
hashtags: Lista de diccionarios con datos de hashtags
|
||||
root_h_maestro: h_maestro del nodo raíz
|
||||
"""
|
||||
# Convertir a Hashtags
|
||||
items = [Hashtag.from_dict(h) for h in hashtags]
|
||||
|
||||
# Crear índice por h_maestro
|
||||
by_maestro = {h.h_maestro: h for h in items}
|
||||
|
||||
# Encontrar hijos de cada nodo
|
||||
children_map = {}
|
||||
for h in items:
|
||||
if h.padre_h_maestro:
|
||||
if h.padre_h_maestro not in children_map:
|
||||
children_map[h.padre_h_maestro] = []
|
||||
children_map[h.padre_h_maestro].append(h)
|
||||
|
||||
# Construir lista de nodos
|
||||
nodes = []
|
||||
|
||||
def add_node(hashtag: Hashtag, depth: int, parent_is_last: List[bool]):
|
||||
children = children_map.get(hashtag.h_maestro, [])
|
||||
children.sort(key=lambda x: x.nombre_es or x.ref)
|
||||
|
||||
for i, child in enumerate(children):
|
||||
is_last = i == len(children) - 1
|
||||
node = TreeNode(
|
||||
hashtag=child,
|
||||
depth=depth,
|
||||
is_last=is_last,
|
||||
parent_is_last=parent_is_last + [is_last]
|
||||
)
|
||||
nodes.append(node)
|
||||
add_node(child, depth + 1, parent_is_last + [is_last])
|
||||
|
||||
# Agregar raíz
|
||||
if root_h_maestro in by_maestro:
|
||||
root = by_maestro[root_h_maestro]
|
||||
nodes.append(TreeNode(hashtag=root, depth=0, is_last=True, parent_is_last=[]))
|
||||
add_node(root, 1, [True])
|
||||
|
||||
return nodes
|
||||
199
tzzr/utils.py
Normal file
199
tzzr/utils.py
Normal file
@@ -0,0 +1,199 @@
|
||||
"""
|
||||
Utilidades para TZZR CLI
|
||||
"""
|
||||
|
||||
import sys
|
||||
from typing import List, Dict, Optional, Any
|
||||
|
||||
from rich.console import Console
|
||||
from rich.table import Table
|
||||
from rich.tree import Tree
|
||||
from rich.panel import Panel
|
||||
from rich.text import Text
|
||||
|
||||
# Intentar importar pyperclip, manejar si no está disponible
|
||||
try:
|
||||
import pyperclip
|
||||
CLIPBOARD_AVAILABLE = True
|
||||
except ImportError:
|
||||
CLIPBOARD_AVAILABLE = False
|
||||
|
||||
|
||||
console = Console()
|
||||
|
||||
|
||||
def copy_to_clipboard(text: str) -> bool:
|
||||
"""
|
||||
Copia texto al clipboard.
|
||||
Retorna True si fue exitoso, False si no.
|
||||
"""
|
||||
if not CLIPBOARD_AVAILABLE:
|
||||
console.print("[yellow]pyperclip no disponible. Instala con: pip install pyperclip[/yellow]")
|
||||
return False
|
||||
|
||||
try:
|
||||
pyperclip.copy(text)
|
||||
return True
|
||||
except Exception as e:
|
||||
console.print(f"[red]Error copiando al clipboard: {e}[/red]")
|
||||
return False
|
||||
|
||||
|
||||
def print_table(data: List[Dict], title: Optional[str] = None,
|
||||
columns: Optional[List[str]] = None) -> None:
|
||||
"""
|
||||
Imprime una tabla formateada con los datos
|
||||
"""
|
||||
if not data:
|
||||
console.print("[yellow]No se encontraron resultados[/yellow]")
|
||||
return
|
||||
|
||||
# Determinar columnas
|
||||
if columns is None:
|
||||
columns = list(data[0].keys())
|
||||
|
||||
# Crear tabla
|
||||
table = Table(title=title, show_header=True, header_style="bold cyan")
|
||||
|
||||
# Nombres de columnas más legibles
|
||||
col_names = {
|
||||
"ref": "Ref",
|
||||
"nombre_es": "Nombre",
|
||||
"nombre_en": "Name (EN)",
|
||||
"h_maestro": "H Maestro",
|
||||
"grupo": "Grupo",
|
||||
"padre_h_maestro": "Padre",
|
||||
"rootref": "Root Ref",
|
||||
"descripcion": "Descripcion",
|
||||
"imagen_url": "Imagen",
|
||||
"created_at": "Creado",
|
||||
"id": "ID",
|
||||
"depth": "Nivel",
|
||||
}
|
||||
|
||||
# Agregar columnas
|
||||
for col in columns:
|
||||
display_name = col_names.get(col, col.replace("_", " ").title())
|
||||
table.add_column(display_name)
|
||||
|
||||
# Agregar filas
|
||||
for row in data:
|
||||
values = []
|
||||
for col in columns:
|
||||
val = row.get(col, "")
|
||||
if val is None:
|
||||
val = "-"
|
||||
# Truncar valores muy largos
|
||||
str_val = str(val)
|
||||
if len(str_val) > 40:
|
||||
str_val = str_val[:37] + "..."
|
||||
values.append(str_val)
|
||||
table.add_row(*values)
|
||||
|
||||
console.print(table)
|
||||
|
||||
|
||||
def print_hashtag_info(hashtag: Dict) -> None:
|
||||
"""
|
||||
Imprime información detallada de un hashtag
|
||||
"""
|
||||
if not hashtag:
|
||||
console.print("[red]Hashtag no encontrado[/red]")
|
||||
return
|
||||
|
||||
# Crear panel con información
|
||||
info_text = Text()
|
||||
|
||||
fields = [
|
||||
("ID", "id"),
|
||||
("Ref", "ref"),
|
||||
("Nombre (ES)", "nombre_es"),
|
||||
("Nombre (EN)", "nombre_en"),
|
||||
("H Maestro", "h_maestro"),
|
||||
("Grupo", "grupo"),
|
||||
("Padre", "padre_h_maestro"),
|
||||
("Root Ref", "rootref"),
|
||||
("Descripcion", "descripcion"),
|
||||
("Imagen URL", "imagen_url"),
|
||||
("Creado", "created_at"),
|
||||
]
|
||||
|
||||
for label, key in fields:
|
||||
value = hashtag.get(key, "-")
|
||||
if value is None:
|
||||
value = "-"
|
||||
info_text.append(f"{label}: ", style="bold cyan")
|
||||
info_text.append(f"{value}\n")
|
||||
|
||||
nombre = hashtag.get('nombre_es') or hashtag.get('ref', 'Hashtag')
|
||||
panel = Panel(
|
||||
info_text,
|
||||
title=f"[bold green]{nombre}[/bold green]",
|
||||
border_style="green"
|
||||
)
|
||||
console.print(panel)
|
||||
|
||||
|
||||
def print_tree(nodes: List, title: Optional[str] = None) -> None:
|
||||
"""
|
||||
Imprime un árbol jerárquico usando rich.tree
|
||||
"""
|
||||
if not nodes:
|
||||
console.print("[yellow]No se encontraron nodos[/yellow]")
|
||||
return
|
||||
|
||||
# El primer nodo es la raíz
|
||||
root_node = nodes[0]
|
||||
tree = Tree(
|
||||
f"[bold green]{root_node.hashtag.display_name}[/bold green]",
|
||||
guide_style="cyan"
|
||||
)
|
||||
|
||||
# Mapa de nodos rich por h_maestro
|
||||
rich_nodes = {root_node.hashtag.h_maestro: tree}
|
||||
|
||||
# Agregar resto de nodos
|
||||
for node in nodes[1:]:
|
||||
parent_ref = node.hashtag.padre_h_maestro
|
||||
if parent_ref in rich_nodes:
|
||||
parent_tree = rich_nodes[parent_ref]
|
||||
style = "yellow" if node.depth == 1 else "white"
|
||||
child = parent_tree.add(f"[{style}]{node.hashtag.display_name}[/{style}]")
|
||||
rich_nodes[node.hashtag.h_maestro] = child
|
||||
|
||||
if title:
|
||||
console.print(f"\n[bold]{title}[/bold]")
|
||||
console.print(tree)
|
||||
|
||||
|
||||
def print_success(message: str) -> None:
|
||||
"""Imprime mensaje de exito"""
|
||||
console.print(f"[green]OK: {message}[/green]")
|
||||
|
||||
|
||||
def print_error(message: str) -> None:
|
||||
"""Imprime mensaje de error"""
|
||||
console.print(f"[red]ERROR: {message}[/red]")
|
||||
|
||||
|
||||
def print_warning(message: str) -> None:
|
||||
"""Imprime mensaje de advertencia"""
|
||||
console.print(f"[yellow]WARN: {message}[/yellow]")
|
||||
|
||||
|
||||
def confirm(message: str, default: bool = False) -> bool:
|
||||
"""Pide confirmacion al usuario"""
|
||||
suffix = " [Y/n]" if default else " [y/N]"
|
||||
response = console.input(f"[cyan]{message}{suffix}[/cyan] ").strip().lower()
|
||||
|
||||
if not response:
|
||||
return default
|
||||
|
||||
return response in ("y", "yes", "s", "si")
|
||||
|
||||
|
||||
def format_count(count: int, singular: str, plural: str) -> str:
|
||||
"""Formatea conteo con singular/plural"""
|
||||
if count == 1:
|
||||
return f"{count} {singular}"
|
||||
return f"{count} {plural}"
|
||||
Reference in New Issue
Block a user