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/
416 lines
14 KiB
Python
416 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""Importador rápido de servicios, citas, ventas y monederos SKEEN a Odoo."""
|
|
import csv
|
|
import sys
|
|
import re
|
|
import html
|
|
import hashlib
|
|
from pathlib import Path
|
|
|
|
ODOO_PATH = Path('/root/skeen-odoo')
|
|
sys.path.insert(0, str(ODOO_PATH))
|
|
|
|
import odoo
|
|
from odoo import api, SUPERUSER_ID
|
|
|
|
DATA_DIR = Path('/root/migracion/datos')
|
|
|
|
|
|
def normalize_bool(v):
|
|
return str(v).lower() in ('1', 'true', 'si', 'sí')
|
|
|
|
|
|
def parse_time(value):
|
|
if not value:
|
|
return 9.0
|
|
value = str(value).strip()
|
|
if ':' in value:
|
|
h, m = value.split(':')[:2]
|
|
return float(h) + float(m) / 60.0
|
|
try:
|
|
return float(value)
|
|
except ValueError:
|
|
return 9.0
|
|
|
|
|
|
def extract_service(text, patient_name):
|
|
"""Extrae el nombre del servicio del campo servicio_nombre."""
|
|
text = html.unescape(text or '')
|
|
if patient_name:
|
|
text = re.sub(r'^' + re.escape(patient_name) + r'\s*[-–—]\s*', '', text, flags=re.IGNORECASE)
|
|
text = re.sub(r'\s*\(\d+\s*(sesión|sesiones|unidad|unidades)\s*\)\s*$', '', text, flags=re.IGNORECASE)
|
|
text = ' '.join(text.split())
|
|
return text.strip()
|
|
|
|
|
|
def normalize_name(name):
|
|
return ' '.join(str(name).split()).lower()
|
|
|
|
|
|
def main():
|
|
odoo.tools.config.parse_config(['-c', '/root/skeen-odoo/odoo.conf'])
|
|
db = odoo.sql_db.db_connect('skeen_odoo')
|
|
cr = db.cursor()
|
|
env = api.Environment(cr, SUPERUSER_ID, {})
|
|
|
|
Partner = env['res.partner'].sudo()
|
|
Servicio = env['skeen.servicio'].sudo()
|
|
Employee = env['hr.employee'].sudo()
|
|
CitaModel = env.registry['skeen.cita']
|
|
Cita = env['skeen.cita'].sudo()
|
|
Venta = env['skeen.venta'].sudo()
|
|
Monedero = env['skeen.monedero'].sudo()
|
|
MonederoTx = env['skeen.monedero.transaction'].sudo()
|
|
|
|
# Cachés en memoria (SQL directo para velocidad)
|
|
print('Cargando cachés...')
|
|
partner_by_phone = {}
|
|
partner_by_name = {}
|
|
partner_by_legacy = {}
|
|
cr.execute("SELECT id, phone, name, legacy_id FROM res_partner WHERE is_patient = true")
|
|
for pid, phone, name, legacy_id in cr.fetchall():
|
|
if phone:
|
|
partner_by_phone.setdefault(phone, pid)
|
|
if name:
|
|
partner_by_name.setdefault(normalize_name(name), pid)
|
|
if legacy_id:
|
|
partner_by_legacy[legacy_id] = pid
|
|
|
|
service_by_code = {}
|
|
service_by_name = {}
|
|
cr.execute("SELECT id, code, name FROM skeen_servicio")
|
|
for sid, code, name in cr.fetchall():
|
|
if code:
|
|
service_by_code[code] = sid
|
|
if name:
|
|
service_by_name[normalize_name(name)] = sid
|
|
|
|
doctor_by_name = {}
|
|
cr.execute("SELECT id, name FROM hr_employee")
|
|
for eid, name in cr.fetchall():
|
|
if name:
|
|
doctor_by_name[normalize_name(name)] = eid
|
|
print(f' Pacientes: {len(partner_by_phone)} por teléfono, {len(partner_by_name)} por nombre')
|
|
print(f' Servicios: {len(service_by_name)}')
|
|
print(f' Doctores: {len(doctor_by_name)}')
|
|
|
|
skipped_reasons = {'no_patient': 0, 'no_service': 0, 'no_date': 0}
|
|
|
|
def resolve_partner(row, phone_field='paciente_telefono', name_field='paciente_nombre', legacy_field='expedientes_id'):
|
|
legacy_id = (row.get(legacy_field) or '').strip()
|
|
if legacy_id and legacy_id in partner_by_legacy:
|
|
return partner_by_legacy[legacy_id]
|
|
phone = (row.get(phone_field) or '').strip()
|
|
if phone and phone in partner_by_phone:
|
|
return partner_by_phone[phone]
|
|
name = normalize_name(row.get(name_field, ''))
|
|
if name and name in partner_by_name:
|
|
return partner_by_name[name]
|
|
return False
|
|
|
|
# Desactivar validación de solapamiento de citas
|
|
original_check = CitaModel._check_availability
|
|
CitaModel._check_availability = lambda self: None
|
|
|
|
resultados = {}
|
|
|
|
# 1. Servicios
|
|
print('Importando servicios...')
|
|
category_map = {
|
|
'consulta': 'consulta',
|
|
'tratamiento': 'tratamiento',
|
|
'procedimiento': 'procedimiento',
|
|
'paquete': 'paquete',
|
|
'servicios_medicos': 'consulta',
|
|
'procedimientos_cosmeticos': 'procedimiento',
|
|
'tecnologia_medica': 'tratamiento',
|
|
'cosmetologia_basica': 'tratamiento',
|
|
'cosmetologia_avanzada': 'tratamiento',
|
|
}
|
|
seen_codes = set()
|
|
batch = []
|
|
count = 0
|
|
with open(DATA_DIR / 'servicios.csv', newline='', encoding='utf-8') as f:
|
|
reader = csv.DictReader(f)
|
|
for row in reader:
|
|
code = (row.get('codigo') or '').strip()
|
|
name = (row.get('nombre') or '').strip()
|
|
if not code or not name or code in seen_codes:
|
|
continue
|
|
seen_codes.add(code)
|
|
if code in service_by_code:
|
|
continue
|
|
cat = category_map.get((row.get('categoria') or '').strip().lower(), 'tratamiento')
|
|
batch.append({
|
|
'name': name,
|
|
'code': code,
|
|
'category': cat,
|
|
'price': float(row.get('precio', 0) or 0),
|
|
'duration_min': int(row.get('duracion_min', 30) or 30),
|
|
'description': row.get('descripcion', ''),
|
|
'color': row.get('color', '#1abc9c'),
|
|
'service_group': row.get('grupo', ''),
|
|
'is_favorite': normalize_bool(row.get('favorito')),
|
|
'active': normalize_bool(row.get('activo', '1')),
|
|
})
|
|
if len(batch) >= 100:
|
|
created = Servicio.create(batch)
|
|
count += len(created)
|
|
cr.commit()
|
|
batch = []
|
|
if batch:
|
|
created = Servicio.create(batch)
|
|
count += len(created)
|
|
cr.commit()
|
|
# Recargar caché de servicios (SQL directo)
|
|
service_by_code = {}
|
|
service_by_name = {}
|
|
cr.execute("SELECT id, code, name FROM skeen_servicio")
|
|
for sid, code, name in cr.fetchall():
|
|
if code:
|
|
service_by_code[code] = sid
|
|
if name:
|
|
service_by_name[normalize_name(name)] = sid
|
|
resultados['servicios'] = count
|
|
print(f' {count} servicios creados')
|
|
|
|
# Helper: obtener o crear servicio por nombre
|
|
service_name_cache = dict(service_by_name)
|
|
|
|
def get_or_create_service_by_name(name):
|
|
name = name.strip()
|
|
if not name:
|
|
name = 'Servicio genérico'
|
|
key = normalize_name(name)
|
|
if key in service_name_cache:
|
|
return service_name_cache[key]
|
|
# Crear nuevo servicio
|
|
code_hash = hashlib.md5(key.encode('utf-8')).hexdigest()[:12]
|
|
code = f'HIST_{code_hash}'
|
|
new_s = Servicio.create({
|
|
'name': name,
|
|
'code': code,
|
|
'category': 'tratamiento',
|
|
'price': 0,
|
|
'duration_min': 30,
|
|
})
|
|
service_name_cache[key] = new_s.id
|
|
return new_s.id
|
|
|
|
# 2. Citas
|
|
print('Importando citas...')
|
|
state_map = {
|
|
'pendiente': 'pending',
|
|
'confirmada': 'confirmed',
|
|
'llego': 'arrived',
|
|
'en_progreso': 'in_progress',
|
|
'completada': 'done',
|
|
'cancelada': 'cancelled',
|
|
'no_show': 'no_show',
|
|
}
|
|
payment_state_map = {
|
|
'no_pagado': 'not_paid',
|
|
'parcial': 'partial',
|
|
'pagado': 'paid',
|
|
}
|
|
batch = []
|
|
count = 0
|
|
skipped = 0
|
|
BATCH_SIZE = 500
|
|
with open(DATA_DIR / 'citas.csv', newline='', encoding='utf-8') as f:
|
|
reader = csv.DictReader(f)
|
|
for row in reader:
|
|
partner_id = resolve_partner(row)
|
|
date_str = (row.get('fecha') or '').strip()
|
|
if not partner_id:
|
|
skipped += 1
|
|
skipped_reasons['no_patient'] += 1
|
|
continue
|
|
if not date_str:
|
|
skipped += 1
|
|
skipped_reasons['no_date'] += 1
|
|
continue
|
|
|
|
service_name = extract_service(row.get('servicio_nombre', ''), row.get('paciente_nombre', ''))
|
|
if not service_name:
|
|
service_name = row.get('servicio_nombre', '').strip() or 'Servicio genérico'
|
|
servicio_id = get_or_create_service_by_name(service_name)
|
|
|
|
doctor_name = (row.get('doctor_nombre') or '').strip()
|
|
doctor_id = doctor_by_name.get(normalize_name(doctor_name)) if doctor_name else False
|
|
|
|
branch = 'rosarito'
|
|
if (row.get('sucursal') or '').strip().lower() in ('tijuana', 'tj'):
|
|
branch = 'tijuana'
|
|
|
|
medium = 'onsite'
|
|
if (row.get('medio') or '').strip().lower() in ('videollamada', 'video', 'remota'):
|
|
medium = 'videocall'
|
|
|
|
state = state_map.get((row.get('estado') or '').strip().lower(), 'confirmed')
|
|
payment_state = payment_state_map.get((row.get('estado_pago') or '').strip().lower(), 'not_paid')
|
|
|
|
batch.append({
|
|
'partner_id': partner_id,
|
|
'servicio_id': servicio_id,
|
|
'date': date_str,
|
|
'time': parse_time(row.get('hora')),
|
|
'state': state,
|
|
'payment_state': payment_state,
|
|
'amount_paid': float(row.get('monto_pagado', 0) or 0),
|
|
'doctor_id': doctor_id,
|
|
'branch': branch,
|
|
'medium': medium,
|
|
'notes': row.get('notas', ''),
|
|
})
|
|
|
|
if len(batch) >= BATCH_SIZE:
|
|
Cita.create(batch)
|
|
count += len(batch)
|
|
batch = []
|
|
cr.commit()
|
|
print(f' {count} citas creadas')
|
|
if batch:
|
|
Cita.create(batch)
|
|
count += len(batch)
|
|
cr.commit()
|
|
resultados['citas'] = count
|
|
print(f' {count} citas creadas, {skipped} omitidas ({skipped_reasons})')
|
|
|
|
# 3. Ventas
|
|
print('Importando ventas...')
|
|
state_map_venta = {
|
|
'borrador': 'draft',
|
|
'confirmada': 'confirmed',
|
|
'pagada': 'paid',
|
|
'parcial': 'partial',
|
|
'cancelada': 'cancelled',
|
|
}
|
|
count = 0
|
|
skipped = 0
|
|
batch = []
|
|
BATCH_SIZE = 500
|
|
with open(DATA_DIR / 'ventas.csv', newline='', encoding='utf-8') as f:
|
|
reader = csv.DictReader(f)
|
|
for row in reader:
|
|
partner_id = resolve_partner(row)
|
|
if not partner_id:
|
|
skipped += 1
|
|
skipped_reasons['no_patient'] += 1
|
|
continue
|
|
|
|
service_name = extract_service(row.get('servicio_nombre', ''), row.get('paciente_nombre', ''))
|
|
if not service_name:
|
|
service_name = row.get('servicio_nombre', '').strip() or 'Servicio genérico'
|
|
servicio_id = get_or_create_service_by_name(service_name)
|
|
|
|
qty = float(row.get('cantidad', 1) or 1)
|
|
price = float(row.get('precio_unitario', 0) or 0)
|
|
discount = float(row.get('descuento', 0) or 0)
|
|
tax = float(row.get('impuesto', 0) or 0)
|
|
state = state_map_venta.get((row.get('estado') or '').strip().lower(), 'confirmed')
|
|
date = row.get('fecha') or odoo.fields.Date.today()
|
|
|
|
batch.append({
|
|
'partner_id': partner_id,
|
|
'date': date,
|
|
'discount': discount,
|
|
'tax': tax,
|
|
'state': state,
|
|
'notes': row.get('notas', ''),
|
|
'line_ids': [(0, 0, {
|
|
'service_id': servicio_id,
|
|
'description': service_name or 'Servicio',
|
|
'quantity': qty,
|
|
'price_unit': price,
|
|
})],
|
|
})
|
|
|
|
if len(batch) >= BATCH_SIZE:
|
|
Venta.create(batch)
|
|
count += len(batch)
|
|
batch = []
|
|
cr.commit()
|
|
print(f' {count} ventas creadas')
|
|
if batch:
|
|
Venta.create(batch)
|
|
count += len(batch)
|
|
cr.commit()
|
|
resultados['ventas'] = count
|
|
print(f' {count} ventas creadas, {skipped} omitidas')
|
|
|
|
# 4. Monederos
|
|
print('Importando monederos...')
|
|
count = 0
|
|
skipped = 0
|
|
batch_monedero = []
|
|
batch_tx = []
|
|
BATCH_SIZE = 500
|
|
with open(DATA_DIR / 'monederos.csv', newline='', encoding='utf-8') as f:
|
|
reader = csv.DictReader(f)
|
|
for row in reader:
|
|
phone = (row.get('telefono') or '').strip()
|
|
points = int(float(row.get('puntos', 0) or 0))
|
|
if not phone or phone not in partner_by_phone:
|
|
skipped += 1
|
|
continue
|
|
partner_id = partner_by_phone[phone]
|
|
existing = Monedero.search([('partner_id', '=', partner_id)], limit=1)
|
|
if existing:
|
|
existing.write({'points': points})
|
|
count += 1
|
|
continue
|
|
|
|
batch_monedero.append({
|
|
'partner_id': partner_id,
|
|
'points': points,
|
|
})
|
|
|
|
if len(batch_monedero) >= BATCH_SIZE:
|
|
created = Monedero.create(batch_monedero)
|
|
for m in created:
|
|
batch_tx.append({
|
|
'monedero_id': m.id,
|
|
'type': 'adjustment',
|
|
'points': m.points,
|
|
'description': 'Saldo inicial migrado de SKEEN',
|
|
'reference': 'MIGRACION',
|
|
})
|
|
if batch_tx:
|
|
MonederoTx.create(batch_tx)
|
|
batch_tx = []
|
|
count += len(batch_monedero)
|
|
batch_monedero = []
|
|
cr.commit()
|
|
print(f' {count} monederos creados')
|
|
if batch_monedero:
|
|
created = Monedero.create(batch_monedero)
|
|
for m in created:
|
|
batch_tx.append({
|
|
'monedero_id': m.id,
|
|
'type': 'adjustment',
|
|
'points': m.points,
|
|
'description': 'Saldo inicial migrado de SKEEN',
|
|
'reference': 'MIGRACION',
|
|
})
|
|
if batch_tx:
|
|
MonederoTx.create(batch_tx)
|
|
count += len(batch_monedero)
|
|
cr.commit()
|
|
resultados['monederos'] = count
|
|
print(f' {count} monederos creados, {skipped} omitidos')
|
|
|
|
# Restaurar validación
|
|
CitaModel._check_availability = original_check
|
|
cr.commit()
|
|
cr.close()
|
|
|
|
print('Importación completada:')
|
|
for k, v in resultados.items():
|
|
print(f' {k}: {v} registros')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|