Consultoría Alcaraz Salazar
a718592291
- Frontend React (SKEEN Brand) con Vite, TypeScript, Tailwind - Frontend Homenest (versión alternativa) - Módulos Odoo 17 custom (citas, pacientes, monedero, pagos, ventas, inventario, whatsapp) - WACRM fork (Next.js 16 + Supabase) - Hermes + Bridge + Skills (Qwen3.6 via Nan Builders) - Scripts de migración y operación - Documentación extensiva en docs/
366 lines
14 KiB
Python
366 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
Importador de servicios e inventario para SKEEN Derma Experts.
|
|
|
|
Uso:
|
|
python3 importar_servicios_inventario.py \
|
|
--servicios servicios.csv \
|
|
--inventario inventario.csv \
|
|
[--purge-citas] \
|
|
[--generar-hermes]
|
|
|
|
Formato CSV servicios:
|
|
name,code,category,price,duration_min,package_price,package_sessions,package_notes,description,service_group,color,is_favorite
|
|
|
|
category: consulta | tratamiento | procedimiento | paquete
|
|
price: precio por sesion (float)
|
|
package_price: precio total del paquete (float, 0 si no aplica)
|
|
package_sessions: numero de sesiones del paquete (int, 0 si no aplica)
|
|
|
|
Formato CSV inventario:
|
|
name,kind,sku,category,unit,qty,qty_optimal,qty_min,cost,expiry_date,notes
|
|
|
|
kind: producto | consumible
|
|
expiry_date: opcional, formato YYYY-MM-DD
|
|
"""
|
|
|
|
import argparse
|
|
import csv
|
|
import json
|
|
import os
|
|
import sys
|
|
import xmlrpc.client
|
|
from datetime import datetime
|
|
|
|
|
|
ODOO_URL = os.environ.get('ODOO_URL', 'http://localhost:8069')
|
|
ODOO_DB = os.environ.get('ODOO_DB', 'skeen_odoo')
|
|
ODOO_USER = os.environ.get('ODOO_USER', 'admin')
|
|
ODOO_PASSWORD = os.environ.get('ODOO_PASSWORD', 'skeen_admin_2026')
|
|
|
|
|
|
def conectar():
|
|
common = xmlrpc.client.ServerProxy(f'{ODOO_URL}/xmlrpc/2/common')
|
|
uid = common.authenticate(ODOO_DB, ODOO_USER, ODOO_PASSWORD, {})
|
|
if not uid:
|
|
raise Exception("Autenticacion fallida. Verifica ODOO_USER / ODOO_PASSWORD.")
|
|
models = xmlrpc.client.ServerProxy(f'{ODOO_URL}/xmlrpc/2/object')
|
|
return uid, models
|
|
|
|
|
|
def leer_csv(path):
|
|
if not path or not os.path.exists(path):
|
|
return []
|
|
with open(path, 'r', encoding='utf-8-sig', newline='') as f:
|
|
reader = csv.DictReader(f)
|
|
rows = [row for row in reader]
|
|
return rows
|
|
|
|
|
|
def _run_sql(command):
|
|
"""Ejecuta un comando SQL via psql como usuario postgres."""
|
|
import subprocess
|
|
full = f"sudo -u postgres psql -d {ODOO_DB} -c \"{command}\""
|
|
result = subprocess.run(full, shell=True, capture_output=True, text=True)
|
|
if result.returncode != 0:
|
|
raise RuntimeError(f"SQL error: {result.stderr}\n{result.stdout}")
|
|
return result.stdout
|
|
|
|
|
|
def limpiar_inventario(models, uid, use_sql=False):
|
|
if use_sql:
|
|
print("[INVENTARIO] Eliminando movimientos e items via SQL...")
|
|
_run_sql("TRUNCATE TABLE skeen_inventario_move RESTART IDENTITY CASCADE")
|
|
_run_sql("TRUNCATE TABLE skeen_inventario_item RESTART IDENTITY CASCADE")
|
|
print(" -> Inventario limpio")
|
|
return
|
|
|
|
print("[INVENTARIO] Eliminando movimientos...")
|
|
move_ids = models.execute_kw(ODOO_DB, uid, ODOO_PASSWORD,
|
|
'skeen.inventario.move', 'search', [[]])
|
|
if move_ids:
|
|
models.execute_kw(ODOO_DB, uid, ODOO_PASSWORD,
|
|
'skeen.inventario.move', 'unlink', [move_ids])
|
|
print(f" -> Eliminados {len(move_ids)} movimientos")
|
|
|
|
print("[INVENTARIO] Eliminando items...")
|
|
item_ids = models.execute_kw(ODOO_DB, uid, ODOO_PASSWORD,
|
|
'skeen.inventario.item', 'search', [[]])
|
|
if item_ids:
|
|
models.execute_kw(ODOO_DB, uid, ODOO_PASSWORD,
|
|
'skeen.inventario.item', 'unlink', [item_ids])
|
|
print(f" -> Eliminados {len(item_ids)} items")
|
|
|
|
|
|
def limpiar_servicios(models, uid, purge_citas=False, use_sql=False):
|
|
if use_sql and purge_citas:
|
|
print("[SERVICIOS] Eliminando citas y servicios via SQL...")
|
|
_run_sql("TRUNCATE TABLE skeen_cita RESTART IDENTITY CASCADE")
|
|
_run_sql("TRUNCATE TABLE skeen_service_tag_skeen_servicio_rel RESTART IDENTITY CASCADE")
|
|
_run_sql("TRUNCATE TABLE skeen_service_tag RESTART IDENTITY CASCADE")
|
|
_run_sql("TRUNCATE TABLE skeen_servicio RESTART IDENTITY CASCADE")
|
|
print(" -> Servicios y citas limpios")
|
|
return
|
|
|
|
if purge_citas:
|
|
print("[SERVICIOS] Eliminando citas existentes...")
|
|
cita_ids = models.execute_kw(ODOO_DB, uid, ODOO_PASSWORD,
|
|
'skeen.cita', 'search', [[]])
|
|
if cita_ids:
|
|
models.execute_kw(ODOO_DB, uid, ODOO_PASSWORD,
|
|
'skeen.cita', 'unlink', [cita_ids])
|
|
print(f" -> Eliminadas {len(cita_ids)} citas")
|
|
|
|
print("[SERVICIOS] Eliminando servicios...")
|
|
servicio_ids = models.execute_kw(ODOO_DB, uid, ODOO_PASSWORD,
|
|
'skeen.servicio', 'search', [[]])
|
|
if servicio_ids:
|
|
if purge_citas:
|
|
models.execute_kw(ODOO_DB, uid, ODOO_PASSWORD,
|
|
'skeen.servicio', 'unlink', [servicio_ids])
|
|
print(f" -> Eliminados {len(servicio_ids)} servicios")
|
|
else:
|
|
# Intentar eliminar uno a uno; los que tengan citas fallaran
|
|
eliminados = 0
|
|
fallidos = []
|
|
for sid in servicio_ids:
|
|
try:
|
|
models.execute_kw(ODOO_DB, uid, ODOO_PASSWORD,
|
|
'skeen.servicio', 'unlink', [[sid]])
|
|
eliminados += 1
|
|
except Exception as e:
|
|
serv = models.execute_kw(ODOO_DB, uid, ODOO_PASSWORD,
|
|
'skeen.servicio', 'read',
|
|
[[sid], ['name', 'code']])
|
|
fallidos.append((sid, serv[0] if serv else {}, str(e)))
|
|
print(f" -> Eliminados {eliminados} servicios")
|
|
if fallidos:
|
|
print(f" -> No se pudieron eliminar {len(fallidos)} servicios (tienen citas/historial):")
|
|
for sid, info, err in fallidos[:10]:
|
|
print(f" - {info.get('name', sid)} ({info.get('code', '')}): {err[:120]}")
|
|
|
|
print("[SERVICIOS] Eliminando tags de servicios...")
|
|
tag_ids = models.execute_kw(ODOO_DB, uid, ODOO_PASSWORD,
|
|
'skeen.service.tag', 'search', [[]])
|
|
if tag_ids:
|
|
models.execute_kw(ODOO_DB, uid, ODOO_PASSWORD,
|
|
'skeen.service.tag', 'unlink', [tag_ids])
|
|
print(f" -> Eliminados {len(tag_ids)} tags")
|
|
|
|
|
|
def parse_float(val, default=0.0):
|
|
if val is None:
|
|
return default
|
|
val = str(val).strip().replace(',', '').replace('$', '')
|
|
if val == '':
|
|
return default
|
|
try:
|
|
return float(val)
|
|
except ValueError:
|
|
return default
|
|
|
|
|
|
def parse_int(val, default=0):
|
|
if val is None:
|
|
return default
|
|
val = str(val).strip()
|
|
if val == '':
|
|
return default
|
|
try:
|
|
return int(float(val))
|
|
except ValueError:
|
|
return default
|
|
|
|
|
|
def importar_servicios(models, uid, rows):
|
|
print(f"[SERVICIOS] Importando {len(rows)} servicios...")
|
|
creados = 0
|
|
for row in rows:
|
|
name = (row.get('name') or '').strip()
|
|
code = (row.get('code') or '').strip()
|
|
if not name or not code:
|
|
print(f" -> Saltando fila sin nombre o codigo: {row}")
|
|
continue
|
|
|
|
category = (row.get('category') or 'tratamiento').strip().lower()
|
|
if category not in ('consulta', 'tratamiento', 'procedimiento', 'paquete'):
|
|
category = 'tratamiento'
|
|
|
|
vals = {
|
|
'name': name,
|
|
'code': code,
|
|
'category': category,
|
|
'price': parse_float(row.get('price')),
|
|
'duration_min': parse_int(row.get('duration_min'), 30),
|
|
'package_price': parse_float(row.get('package_price')),
|
|
'package_sessions': parse_int(row.get('package_sessions')),
|
|
'package_notes': (row.get('package_notes') or '').strip(),
|
|
'description': (row.get('description') or '').strip(),
|
|
'service_group': (row.get('service_group') or '').strip(),
|
|
'color': (row.get('color') or '#1abc9c').strip(),
|
|
'is_favorite': str(row.get('is_favorite', '')).strip().lower() in ('1', 'true', 'si', 'sí', 'yes'),
|
|
}
|
|
|
|
# Si es categoria paquete pero no tiene sesiones, forzar consulta
|
|
if category == 'paquete' and vals['package_sessions'] <= 0:
|
|
vals['category'] = 'tratamiento'
|
|
|
|
existing = models.execute_kw(ODOO_DB, uid, ODOO_PASSWORD,
|
|
'skeen.servicio', 'search',
|
|
[[['code', '=', code]]])
|
|
if existing:
|
|
models.execute_kw(ODOO_DB, uid, ODOO_PASSWORD,
|
|
'skeen.servicio', 'write', [existing, vals])
|
|
print(f" -> Actualizado: {name} ({code})")
|
|
else:
|
|
models.execute_kw(ODOO_DB, uid, ODOO_PASSWORD,
|
|
'skeen.servicio', 'create', [vals])
|
|
creados += 1
|
|
print(f" -> Creado: {name} ({code})")
|
|
print(f"[SERVICIOS] {creados} nuevos creados.")
|
|
|
|
|
|
def importar_inventario(models, uid, rows):
|
|
print(f"[INVENTARIO] Importando {len(rows)} items...")
|
|
creados = 0
|
|
for row in rows:
|
|
name = (row.get('name') or '').strip()
|
|
if not name:
|
|
continue
|
|
kind = (row.get('kind') or 'producto').strip().lower()
|
|
if kind not in ('producto', 'consumible'):
|
|
kind = 'producto'
|
|
|
|
sku = (row.get('sku') or '').strip()
|
|
expiry = (row.get('expiry_date') or '').strip()
|
|
|
|
vals = {
|
|
'name': name,
|
|
'kind': kind,
|
|
'sku': sku,
|
|
'category': (row.get('category') or '').strip(),
|
|
'unit': (row.get('unit') or 'pieza').strip(),
|
|
'qty': parse_float(row.get('qty')),
|
|
'qty_optimal': parse_float(row.get('qty_optimal')),
|
|
'qty_min': parse_float(row.get('qty_min')),
|
|
'cost': parse_float(row.get('cost')),
|
|
'notes': (row.get('notes') or '').strip(),
|
|
}
|
|
if expiry:
|
|
try:
|
|
datetime.strptime(expiry, '%Y-%m-%d')
|
|
vals['expiry_date'] = expiry
|
|
except ValueError:
|
|
pass
|
|
|
|
existing = models.execute_kw(ODOO_DB, uid, ODOO_PASSWORD,
|
|
'skeen.inventario.item', 'search',
|
|
[[['sku', '=', sku]]] if sku else [[['name', '=', name]]])
|
|
if existing:
|
|
models.execute_kw(ODOO_DB, uid, ODOO_PASSWORD,
|
|
'skeen.inventario.item', 'write', [existing, vals])
|
|
print(f" -> Actualizado: {name}")
|
|
else:
|
|
models.execute_kw(ODOO_DB, uid, ODOO_PASSWORD,
|
|
'skeen.inventario.item', 'create', [vals])
|
|
creados += 1
|
|
print(f" -> Creado: {name}")
|
|
print(f"[INVENTARIO] {creados} nuevos items creados.")
|
|
|
|
|
|
def generar_hermes_services(models, uid):
|
|
print("[HERMES] Generando catalogo para skills...")
|
|
servicios = models.execute_kw(ODOO_DB, uid, ODOO_PASSWORD,
|
|
'skeen.servicio', 'search_read',
|
|
[[['active', '=', True]]],
|
|
{'fields': ['name', 'code', 'category', 'price',
|
|
'duration_min', 'package_price',
|
|
'package_sessions', 'package_notes',
|
|
'description']})
|
|
|
|
payload = {
|
|
"clinica": {
|
|
"nombre": "SKEEN Derma Experts",
|
|
"direccion": "Blvd. Popotla 123, Playas de Rosarito, BC, Mexico",
|
|
"telefono": "+52 (664) 123-4567",
|
|
"horario": {
|
|
"lunes_viernes": "9:00 - 18:00",
|
|
"sabado": "9:00 - 14:00",
|
|
"domingo": "Cerrado"
|
|
}
|
|
},
|
|
"servicios": []
|
|
}
|
|
|
|
for s in servicios:
|
|
payload['servicios'].append({
|
|
"id": s['code'].lower().replace(' ', '-'),
|
|
"nombre": s['name'],
|
|
"categoria": s['category'],
|
|
"precio": s['price'],
|
|
"duracion_min": s['duration_min'],
|
|
"precio_paquete": s['package_price'] or 0,
|
|
"sesiones_paquete": s['package_sessions'] or 0,
|
|
"notas_paquete": s['package_notes'] or '',
|
|
"descripcion": s['description'] or '',
|
|
"contraindicaciones": []
|
|
})
|
|
|
|
paths = [
|
|
'/root/.hermes/skills/skeen-services.json',
|
|
'/root/skeen-roadmap-hermes/hermes/skills/skeen-services.json',
|
|
]
|
|
for p in paths:
|
|
try:
|
|
with open(p, 'w', encoding='utf-8') as f:
|
|
json.dump(payload, f, ensure_ascii=False, indent=2)
|
|
print(f" -> Escrito: {p}")
|
|
except Exception as e:
|
|
print(f" -> Error escribiendo {p}: {e}")
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description='Importar servicios e inventario a Odoo')
|
|
parser.add_argument('--servicios', help='CSV de servicios')
|
|
parser.add_argument('--inventario', help='CSV de inventario')
|
|
parser.add_argument('--purge-citas', action='store_true',
|
|
help='Eliminar citas existentes antes de borrar servicios')
|
|
parser.add_argument('--purge-sql', action='store_true',
|
|
help='Usar TRUNCATE SQL directo para limpieza rapida (miles de registros)')
|
|
parser.add_argument('--generar-hermes', action='store_true',
|
|
help='Generar skeen-services.json para Hermes')
|
|
parser.add_argument('--solo-hermes', action='store_true',
|
|
help='Solo generar skeen-services.json desde Odoo actual')
|
|
args = parser.parse_args()
|
|
|
|
uid, models = conectar()
|
|
print(f"[OK] Conectado a Odoo {ODOO_URL} como {ODOO_USER} (uid={uid})")
|
|
|
|
if args.solo_hermes:
|
|
generar_hermes_services(models, uid)
|
|
return
|
|
|
|
if not args.servicios and not args.inventario:
|
|
print("[INFO] No se especificaron CSVs. Usa --servicios y/o --inventario.")
|
|
return
|
|
|
|
if args.inventario:
|
|
limpiar_inventario(models, uid, use_sql=args.purge_sql)
|
|
rows_inv = leer_csv(args.inventario)
|
|
importar_inventario(models, uid, rows_inv)
|
|
|
|
if args.servicios:
|
|
limpiar_servicios(models, uid, purge_citas=args.purge_citas, use_sql=args.purge_sql)
|
|
rows_srv = leer_csv(args.servicios)
|
|
importar_servicios(models, uid, rows_srv)
|
|
|
|
if args.generar_hermes:
|
|
generar_hermes_services(models, uid)
|
|
|
|
print("[OK] Proceso completado.")
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|