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/
703 lines
26 KiB
Python
Executable File
703 lines
26 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
Extracción read-only de datos del sistema legacy SKEEN.
|
|
Uso autorizado por el responsable del proyecto con credenciales proporcionadas.
|
|
Extrae pacientes (con expediente clínico completo), citas, ventas, monederos, servicios, etc.
|
|
"""
|
|
|
|
import requests
|
|
import csv
|
|
import json
|
|
import re
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
from datetime import datetime, timedelta
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
from bs4 import BeautifulSoup
|
|
|
|
BASE = 'https://sistema.skeenmx.app'
|
|
USERNAME = 'JoseSkeen'
|
|
PASSWORD = 'SKr#902_88'
|
|
OUTDIR = Path('/root/migracion/extraccion')
|
|
OUTDIR.mkdir(parents=True, exist_ok=True)
|
|
CACHE = OUTDIR / 'detalle_pacientes.json'
|
|
|
|
session = requests.Session()
|
|
|
|
|
|
def log(msg):
|
|
print(f'[{datetime.now().strftime("%H:%M:%S")}] {msg}')
|
|
|
|
|
|
def login():
|
|
log('Iniciando sesión...')
|
|
r = session.get(f'{BASE}/login', timeout=30)
|
|
r.raise_for_status()
|
|
soup = BeautifulSoup(r.text, 'html.parser')
|
|
token = soup.find('input', {'name': '_token'})
|
|
token = token['value'] if token else ''
|
|
data = {
|
|
'_token': token,
|
|
'username': USERNAME,
|
|
'password': PASSWORD,
|
|
}
|
|
r = session.post(f'{BASE}/login', data=data, timeout=30, allow_redirects=True)
|
|
r.raise_for_status()
|
|
if '/login' in r.url or 'estas credenciales' in r.text.lower():
|
|
raise Exception('Login fallido. Verifica credenciales.')
|
|
log('Login exitoso')
|
|
return True
|
|
|
|
|
|
def dt_params(start=0, length=100, search=''):
|
|
return {
|
|
'draw': 1,
|
|
'columns[0][data]': 'id',
|
|
'columns[0][name]': '',
|
|
'columns[0][searchable]': 'true',
|
|
'columns[0][orderable]': 'true',
|
|
'columns[0][search][value]': '',
|
|
'columns[0][search][regex]': 'false',
|
|
'start': start,
|
|
'length': length,
|
|
'search[value]': search,
|
|
'search[regex]': 'false',
|
|
'order[0][column]': '0',
|
|
'order[0][dir]': 'asc',
|
|
'_': int(time.time() * 1000),
|
|
}
|
|
|
|
|
|
def fetch_json(url, params=None):
|
|
headers = {'X-Requested-With': 'XMLHttpRequest'}
|
|
r = session.get(f'{BASE}{url}', params=params or {}, headers=headers, timeout=120)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
|
|
def paginate(url, params_builder, label, page_size=500, max_pages=None):
|
|
log(f'Extrayendo {label} desde {url}...')
|
|
all_rows = []
|
|
start = 0
|
|
page = 0
|
|
while True:
|
|
params = params_builder(start, page_size)
|
|
try:
|
|
data = fetch_json(url, params)
|
|
except Exception as e:
|
|
log(f' ERROR página {page}: {e}')
|
|
break
|
|
rows = data.get('data', [])
|
|
if not rows:
|
|
break
|
|
all_rows.extend(rows)
|
|
total = data.get('recordsTotal') or data.get('recordsFiltered') or 0
|
|
log(f' página {page + 1}: {len(rows)} filas (total acumulado: {len(all_rows)}/{total})')
|
|
if len(rows) < page_size:
|
|
break
|
|
start += page_size
|
|
page += 1
|
|
if max_pages and page >= max_pages:
|
|
break
|
|
time.sleep(0.3)
|
|
log(f'{label}: {len(all_rows)} registros extraídos')
|
|
return all_rows
|
|
|
|
|
|
def save_csv(name, rows, fieldnames):
|
|
path = OUTDIR / f'{name}.csv'
|
|
with open(path, 'w', newline='', encoding='utf-8') as f:
|
|
writer = csv.DictWriter(f, fieldnames=fieldnames, extrasaction='ignore')
|
|
writer.writeheader()
|
|
writer.writerows(rows)
|
|
log(f'Guardado {path}: {len(rows)} registros')
|
|
|
|
|
|
MESES_ES = {
|
|
'enero': '01', 'febrero': '02', 'marzo': '03', 'abril': '04',
|
|
'mayo': '05', 'junio': '06', 'julio': '07', 'agosto': '08',
|
|
'septiembre': '09', 'octubre': '10', 'noviembre': '11', 'diciembre': '12',
|
|
}
|
|
|
|
|
|
def parse_spanish_date(text):
|
|
text = (text or '').strip().lower()
|
|
if not text:
|
|
return ''
|
|
m = re.search(r'(\d{1,2})\s+de\s+([a-záéíóúñ]+)\s+(?:del|de)\s+(\d{4})', text)
|
|
if not m:
|
|
return ''
|
|
dia, mes, anio = m.groups()
|
|
mes_num = MESES_ES.get(mes)
|
|
if not mes_num:
|
|
return ''
|
|
return f'{anio}-{mes_num}-{int(dia):02d}'
|
|
|
|
|
|
def normalize_phone(phone):
|
|
phone = re.sub(r'\D', '', (phone or ''))
|
|
if phone and not phone.startswith('52') and len(phone) == 10:
|
|
phone = '52' + phone
|
|
return phone
|
|
|
|
|
|
def parse_patient_detail(html, fallback_name=''):
|
|
"""Extrae datos personales e historia clínica completa del HTML del expediente."""
|
|
soup = BeautifulSoup(html, 'html.parser')
|
|
text = soup.get_text(separator='\n')
|
|
lines = [l.strip() for l in text.split('\n') if l.strip()]
|
|
|
|
def find_line_after(keywords):
|
|
for i, line in enumerate(lines):
|
|
low = line.lower()
|
|
if any(k in low for k in keywords):
|
|
if i + 1 < len(lines):
|
|
return lines[i + 1]
|
|
return ''
|
|
|
|
def find_pair(keywords):
|
|
for i, line in enumerate(lines):
|
|
low = line.lower()
|
|
if any(k in low for k in keywords):
|
|
# Si la respuesta está en la misma línea separada por ':' o espacio
|
|
if ':' in line:
|
|
return line.split(':', 1)[1].strip()
|
|
if i + 1 < len(lines):
|
|
return lines[i + 1]
|
|
return ''
|
|
|
|
def yes_no(value):
|
|
v = (value or '').lower()
|
|
if v.startswith('s') or v == 'yes':
|
|
return True
|
|
return False
|
|
|
|
# Extraer email con regex del texto completo
|
|
emails = re.findall(r'[\w.\-]+@[\w.\-]+', text)
|
|
email = emails[0] if emails else ''
|
|
if email.lower() in ('no@no.com', 'no@no'):
|
|
email = ''
|
|
|
|
data = {
|
|
'celular': '',
|
|
'whatsapp': '',
|
|
'email': email,
|
|
'fecha_nacimiento': '',
|
|
'sexo': '',
|
|
'lugar_nacimiento': '',
|
|
'empleo': '',
|
|
'estado_civil': '',
|
|
'contacto_emergencia': '',
|
|
'telefono_emergencia': '',
|
|
'telefono_casa': '',
|
|
'direccion': '',
|
|
'recomendado_por': '',
|
|
'comentarios': '',
|
|
'hijos': 0,
|
|
'embarazada': False,
|
|
'lactancia': False,
|
|
'anticonceptivos': False,
|
|
'ejercicio': False,
|
|
'alimentacion': False,
|
|
'peso_normal': False,
|
|
'enfermedad_seria': False,
|
|
'alcohol': False,
|
|
'problemas_rinon': False,
|
|
'problemas_espalda': False,
|
|
'problemas_cardiacos': False,
|
|
'problemas_respiratorios': False,
|
|
'alta_presion': False,
|
|
'baja_presion': False,
|
|
'diabetes': False,
|
|
'hipertiroidismo': False,
|
|
'hipotiroidismo': False,
|
|
'colitis': False,
|
|
'estrenimiento': False,
|
|
'problemas_higado': False,
|
|
'enfermedades_cronicas': False,
|
|
'cirugias': False,
|
|
'varices': False,
|
|
'migrana': False,
|
|
'desmaya_agujas': False,
|
|
'alergias': '',
|
|
'medicamentos': '',
|
|
'tratamientos_previos': '',
|
|
}
|
|
|
|
# Recorrer líneas buscando pares pregunta/respuesta
|
|
i = 0
|
|
while i < len(lines):
|
|
line = lines[i]
|
|
low = line.lower()
|
|
nxt = lines[i + 1] if i + 1 < len(lines) else ''
|
|
|
|
# Datos personales
|
|
if 'celular' in low and len(low) < 20:
|
|
data['celular'] = re.sub(r'\D', '', nxt)
|
|
elif 'whatsapp' in low and len(low) < 20:
|
|
data['whatsapp'] = re.sub(r'\D', '', nxt)
|
|
elif 'correo electrónico' in low or 'email' in low:
|
|
data['email'] = nxt if '@' in nxt else data['email']
|
|
elif 'fecha de nacimiento' in low:
|
|
data['fecha_nacimiento'] = parse_spanish_date(nxt or line)
|
|
elif 'lugar de nacimiento' in low:
|
|
data['lugar_nacimiento'] = nxt
|
|
elif low == 'sexo':
|
|
data['sexo'] = 'female' if 'femenino' in nxt.lower() else ('male' if 'masculino' in nxt.lower() else '')
|
|
elif 'empleo' in low:
|
|
data['empleo'] = nxt
|
|
elif 'estado civil' in low:
|
|
data['estado_civil'] = nxt
|
|
elif 'en caso de emergencia llamar a' in low:
|
|
parts = nxt.split()
|
|
if parts:
|
|
data['contacto_emergencia'] = parts[0]
|
|
data['telefono_emergencia'] = normalize_phone(' '.join(parts[1:]))
|
|
elif 'teléfono casa' in low or 'telefono casa' in low:
|
|
data['telefono_casa'] = re.sub(r'\D', '', nxt)
|
|
elif low == 'dirección' or low == 'direccion':
|
|
data['direccion'] = nxt
|
|
elif 'recomendado por' in low:
|
|
data['recomendado_por'] = nxt
|
|
elif 'comentarios adicionales' in low:
|
|
data['comentarios'] = nxt
|
|
|
|
# Historia clínica
|
|
elif 'tiene hijos' in low:
|
|
val = nxt.lower()
|
|
if val.isdigit():
|
|
data['hijos'] = int(val)
|
|
elif val.startswith('s'):
|
|
data['hijos'] = 1
|
|
elif 'embarazada' in low:
|
|
data['embarazada'] = yes_no(nxt)
|
|
elif 'lactancia' in low:
|
|
data['lactancia'] = yes_no(nxt)
|
|
elif 'anticonceptivo' in low:
|
|
data['anticonceptivos'] = yes_no(nxt)
|
|
elif 'hace ejercicio' in low:
|
|
data['ejercicio'] = yes_no(nxt)
|
|
elif 'come en forma adecuada' in low:
|
|
data['alimentacion'] = yes_no(nxt)
|
|
elif 'peso corporal normal' in low:
|
|
data['peso_normal'] = yes_no(nxt)
|
|
elif 'enfermedad seria' in low or 'incapacidad física' in low:
|
|
data['enfermedad_seria'] = yes_no(nxt)
|
|
elif 'bebidas alcohólicas' in low:
|
|
data['alcohol'] = yes_no(nxt)
|
|
elif 'riñones' in low or 'rinon' in low:
|
|
data['problemas_rinon'] = yes_no(nxt)
|
|
elif 'espalda' in low:
|
|
data['problemas_espalda'] = yes_no(nxt)
|
|
elif 'cardiacos' in low or 'cardíacos' in low:
|
|
data['problemas_cardiacos'] = yes_no(nxt)
|
|
elif 'respiratorios' in low:
|
|
data['problemas_respiratorios'] = yes_no(nxt)
|
|
elif 'alta presión' in low or 'alta presion' in low:
|
|
data['alta_presion'] = yes_no(nxt)
|
|
elif 'baja presión' in low or 'baja presion' in low:
|
|
data['baja_presion'] = yes_no(nxt)
|
|
elif 'diabetes' in low:
|
|
data['diabetes'] = yes_no(nxt)
|
|
elif 'hipertiroidismo' in low:
|
|
data['hipertiroidismo'] = yes_no(nxt)
|
|
elif 'hipotiroidismo' in low:
|
|
data['hipotiroidismo'] = yes_no(nxt)
|
|
elif 'colitis' in low:
|
|
data['colitis'] = yes_no(nxt)
|
|
elif 'estreñimiento' in low:
|
|
data['estrenimiento'] = yes_no(nxt)
|
|
elif 'hígado' in low or 'higado' in low or 'vesícula' in low:
|
|
data['problemas_higado'] = yes_no(nxt)
|
|
elif 'enfermedades crónicas' in low or 'enfermedades cronicas' in low:
|
|
data['enfermedades_cronicas'] = yes_no(nxt)
|
|
elif 'realizado cirugías' in low or 'realizado cirugias' in low:
|
|
data['cirugias'] = yes_no(nxt)
|
|
elif 'varicosas' in low or 'varices' in low:
|
|
data['varices'] = yes_no(nxt)
|
|
elif 'migraña' in low or 'migrana' in low:
|
|
data['migrana'] = yes_no(nxt)
|
|
elif 'desmaya' in low and 'agujas' in low:
|
|
data['desmaya_agujas'] = yes_no(nxt)
|
|
elif low == 'alergias':
|
|
data['alergias'] = nxt if nxt.lower() != 'no' else ''
|
|
elif 'medicamentos actualmente' in low:
|
|
data['medicamentos'] = nxt if nxt.lower() != 'no' else ''
|
|
elif 'tratamientos médico' in low or 'tratamientos medico' in low or 'tratamientos cosméticos previos' in low:
|
|
data['tratamientos_previos'] = nxt if nxt.lower() != 'no' else ''
|
|
|
|
i += 1
|
|
|
|
return data
|
|
|
|
|
|
def load_detail_cache():
|
|
if CACHE.exists():
|
|
try:
|
|
with open(CACHE, 'r', encoding='utf-8') as f:
|
|
return json.load(f)
|
|
except Exception as e:
|
|
log(f'No se pudo cargar caché: {e}')
|
|
return {}
|
|
|
|
|
|
def save_detail_cache(cache):
|
|
with open(CACHE, 'w', encoding='utf-8') as f:
|
|
json.dump(cache, f, ensure_ascii=False, indent=2)
|
|
|
|
|
|
def fetch_one_detail(legacy_id, fallback_name):
|
|
try:
|
|
resp = session.get(f'{BASE}/expediente/ver/{legacy_id}', timeout=30)
|
|
resp.raise_for_status()
|
|
return str(legacy_id), parse_patient_detail(resp.text, fallback_name)
|
|
except Exception as e:
|
|
return str(legacy_id), {'_error': str(e)}
|
|
|
|
|
|
def extract_paciente_details(rows, cache, max_workers=10):
|
|
to_fetch = []
|
|
for r in rows:
|
|
legacy_id = r.get('id')
|
|
if not legacy_id:
|
|
continue
|
|
sid = str(legacy_id)
|
|
if sid in cache and cache[sid].get('celular'):
|
|
continue
|
|
to_fetch.append((legacy_id, r.get('nombre', '')))
|
|
|
|
log(f'Extrayendo detalles de {len(to_fetch)} pacientes con {max_workers} workers...')
|
|
total = len(to_fetch)
|
|
completed = 0
|
|
with ThreadPoolExecutor(max_workers=max_workers) as ex:
|
|
future_to_id = {ex.submit(fetch_one_detail, lid, name): lid for lid, name in to_fetch}
|
|
for future in as_completed(future_to_id):
|
|
lid, detail = future.result()
|
|
if '_error' not in detail:
|
|
cache[lid] = detail
|
|
completed += 1
|
|
if completed % 500 == 0:
|
|
log(f' {completed}/{total} detalles procesados')
|
|
save_detail_cache(cache)
|
|
time.sleep(0.01)
|
|
save_detail_cache(cache)
|
|
log('Detalles de pacientes completados')
|
|
|
|
|
|
def extract_expedientes():
|
|
rows = paginate('/expedientes/json', lambda s, l: dt_params(s, l), 'pacientes')
|
|
cache = load_detail_cache()
|
|
extract_paciente_details(rows, cache)
|
|
|
|
out = []
|
|
for r in rows:
|
|
nombre = (r.get('nombre') or '').strip()
|
|
if not nombre or nombre.upper() == '. . NO CITAR':
|
|
continue
|
|
detail = cache.get(str(r.get('id', '')), {})
|
|
phone = detail.get('celular') or detail.get('whatsapp') or (r.get('telefono') or '').strip()
|
|
phone = normalize_phone(phone)
|
|
email = detail.get('email', '')
|
|
if email.lower() in ('no@no.com', 'no@no'):
|
|
email = ''
|
|
|
|
out.append({
|
|
'legacy_id': r.get('id', ''),
|
|
'folio': r.get('folio', ''),
|
|
'nombre': nombre,
|
|
'apellido_paterno': '',
|
|
'apellido_materno': '',
|
|
'telefono': phone,
|
|
'email': email,
|
|
'fecha_nacimiento': detail.get('fecha_nacimiento', ''),
|
|
'sexo': detail.get('sexo', ''),
|
|
'lugar_nacimiento': detail.get('lugar_nacimiento', ''),
|
|
'empleo': detail.get('empleo', ''),
|
|
'estado_civil': detail.get('estado_civil', ''),
|
|
'contacto_emergencia': detail.get('contacto_emergencia', ''),
|
|
'telefono_emergencia': detail.get('telefono_emergencia', ''),
|
|
'telefono_casa': detail.get('telefono_casa', ''),
|
|
'direccion': detail.get('direccion', ''),
|
|
'recomendado_por': detail.get('recomendado_por', ''),
|
|
'comentarios': detail.get('comentarios', ''),
|
|
'medico_principal': r.get('medico_principal') or '',
|
|
'etiquetas': ','.join(str(c) for c in (r.get('condiciones') or [])) if r.get('condiciones') else '',
|
|
'vip': '1' if r.get('vip') else '0',
|
|
'fuente': 'legacy',
|
|
# Historia clínica
|
|
'hijos': detail.get('hijos', 0),
|
|
'embarazada': '1' if detail.get('embarazada') else '0',
|
|
'lactancia': '1' if detail.get('lactancia') else '0',
|
|
'anticonceptivos': '1' if detail.get('anticonceptivos') else '0',
|
|
'ejercicio': '1' if detail.get('ejercicio') else '0',
|
|
'alimentacion': '1' if detail.get('alimentacion') else '0',
|
|
'peso_normal': '1' if detail.get('peso_normal') else '0',
|
|
'enfermedad_seria': '1' if detail.get('enfermedad_seria') else '0',
|
|
'alcohol': '1' if detail.get('alcohol') else '0',
|
|
'problemas_rinon': '1' if detail.get('problemas_rinon') else '0',
|
|
'problemas_espalda': '1' if detail.get('problemas_espalda') else '0',
|
|
'problemas_cardiacos': '1' if detail.get('problemas_cardiacos') else '0',
|
|
'problemas_respiratorios': '1' if detail.get('problemas_respiratorios') else '0',
|
|
'alta_presion': '1' if detail.get('alta_presion') else '0',
|
|
'baja_presion': '1' if detail.get('baja_presion') else '0',
|
|
'diabetes': '1' if detail.get('diabetes') else '0',
|
|
'hipertiroidismo': '1' if detail.get('hipertiroidismo') else '0',
|
|
'hipotiroidismo': '1' if detail.get('hipotiroidismo') else '0',
|
|
'colitis': '1' if detail.get('colitis') else '0',
|
|
'estrenimiento': '1' if detail.get('estrenimiento') else '0',
|
|
'problemas_higado': '1' if detail.get('problemas_higado') else '0',
|
|
'enfermedades_cronicas': '1' if detail.get('enfermedades_cronicas') else '0',
|
|
'cirugias': '1' if detail.get('cirugias') else '0',
|
|
'varices': '1' if detail.get('varices') else '0',
|
|
'migrana': '1' if detail.get('migrana') else '0',
|
|
'desmaya_agujas': '1' if detail.get('desmaya_agujas') else '0',
|
|
'alergias': detail.get('alergias', ''),
|
|
'medicamentos': detail.get('medicamentos', ''),
|
|
'tratamientos_previos': detail.get('tratamientos_previos', ''),
|
|
})
|
|
|
|
save_csv('pacientes', out, [
|
|
'legacy_id', 'folio', 'nombre', 'apellido_paterno', 'apellido_materno',
|
|
'telefono', 'email', 'fecha_nacimiento', 'sexo', 'lugar_nacimiento',
|
|
'empleo', 'estado_civil', 'contacto_emergencia', 'telefono_emergencia',
|
|
'telefono_casa', 'direccion', 'recomendado_por', 'comentarios',
|
|
'medico_principal', 'etiquetas', 'vip', 'fuente',
|
|
'hijos', 'embarazada', 'lactancia', 'anticonceptivos', 'ejercicio',
|
|
'alimentacion', 'peso_normal', 'enfermedad_seria', 'alcohol',
|
|
'problemas_rinon', 'problemas_espalda', 'problemas_cardiacos',
|
|
'problemas_respiratorios', 'alta_presion', 'baja_presion', 'diabetes',
|
|
'hipertiroidismo', 'hipotiroidismo', 'colitis', 'estrenimiento',
|
|
'problemas_higado', 'enfermedades_cronicas', 'cirugias', 'varices',
|
|
'migrana', 'desmaya_agujas', 'alergias', 'medicamentos', 'tratamientos_previos'
|
|
])
|
|
return out
|
|
|
|
|
|
def extract_monederos():
|
|
rows = paginate('/monederos/json', lambda s, l: dt_params(s, l), 'monederos')
|
|
out = []
|
|
for r in rows:
|
|
saldo = r.get('saldo') or {}
|
|
phone = normalize_phone(r.get('celular') or '')
|
|
out.append({
|
|
'telefono': phone,
|
|
'puntos': float(saldo.get('mn') or 0),
|
|
'activo': '1' if r.get('activado') else '0',
|
|
})
|
|
save_csv('monederos', out, ['telefono', 'puntos', 'activo'])
|
|
return out
|
|
|
|
|
|
def extract_crm():
|
|
rows = paginate('/crm/prospectos/json', lambda s, l: dt_params(s, l), 'crm prospectos')
|
|
out = []
|
|
for r in rows:
|
|
out.append({
|
|
'nombre': r.get('nombre') or '',
|
|
'folio': r.get('folio') or '',
|
|
'status': r.get('status') or 0,
|
|
'total_citas': r.get('total_citas') or 0,
|
|
'siguiente_cita': r.get('siguiente_cita') or '',
|
|
})
|
|
save_csv('crm_prospectos', out, ['nombre', 'folio', 'status', 'total_citas', 'siguiente_cita'])
|
|
return out
|
|
|
|
|
|
def extract_servicios():
|
|
rows = paginate('/servicios/json', lambda s, l: dt_params(s, l), 'servicios')
|
|
out = []
|
|
for r in rows:
|
|
precio_str = (r.get('precio_minimo') or '0').replace('$', '').replace('m.n.', '').replace(',', '').strip()
|
|
try:
|
|
precio = float(precio_str.split()[0])
|
|
except Exception:
|
|
precio = 0
|
|
out.append({
|
|
'codigo': f"S{r.get('id', 0):04d}",
|
|
'nombre': r.get('servicio') or '',
|
|
'categoria': (r.get('categoria') or '').lower().replace(' ', '_'),
|
|
'precio': precio,
|
|
'duracion_min': 30,
|
|
'descripcion': r.get('grupo_texto') or '',
|
|
'color': r.get('color') or '#1abc9c',
|
|
'grupo': r.get('grupo_texto') or '',
|
|
'favorito': '1' if r.get('favorito') else '0',
|
|
'activo': '1' if r.get('activado') else '0',
|
|
})
|
|
save_csv('servicios', out, [
|
|
'codigo', 'nombre', 'categoria', 'precio', 'duracion_min',
|
|
'descripcion', 'color', 'grupo', 'favorito', 'activo'
|
|
])
|
|
return out
|
|
|
|
|
|
def extract_ventas(start_date='2022-01-01', end_date='2026-07-06'):
|
|
log(f'Extrayendo ventas de {start_date} a {end_date}...')
|
|
phone_by_legacy = {}
|
|
pac_path = OUTDIR / 'pacientes.csv'
|
|
if pac_path.exists():
|
|
with open(pac_path, newline='', encoding='utf-8') as f:
|
|
for row in csv.DictReader(f):
|
|
phone_by_legacy[row.get('legacy_id', '')] = row.get('telefono', '')
|
|
|
|
all_rows = []
|
|
current = datetime.strptime(start_date, '%Y-%m-%d').date()
|
|
end = datetime.strptime(end_date, '%Y-%m-%d').date()
|
|
while current <= end:
|
|
chunk_end = min(current + timedelta(days=30), end)
|
|
url = f"/ventas/json/{current.strftime('%Y-%m-%d')}/{chunk_end.strftime('%Y-%m-%d')}"
|
|
try:
|
|
data = fetch_json(url, dt_params(0, 10000))
|
|
rows = data.get('data', [])
|
|
all_rows.extend(rows)
|
|
log(f' {current} - {chunk_end}: {len(rows)} ventas (total: {len(all_rows)})')
|
|
except Exception as e:
|
|
log(f' ERROR {current}-{chunk_end}: {e}')
|
|
current = chunk_end + timedelta(days=1)
|
|
time.sleep(0.5)
|
|
|
|
out = []
|
|
for r in all_rows:
|
|
expediente_id = r.get('expedientes_id') or ''
|
|
phone = normalize_phone(phone_by_legacy.get(str(expediente_id), ''))
|
|
conceptos = r.get('conceptos') or []
|
|
if not conceptos:
|
|
conceptos = [{'titulo': r.get('motivo') or 'Servicio', 'subtotal': r.get('subtotal') or 0}]
|
|
for c in conceptos:
|
|
precio = 0
|
|
try:
|
|
precio = float(c.get('subtotal') or c.get('pagado') or 0)
|
|
except Exception:
|
|
pass
|
|
out.append({
|
|
'legacy_id': r.get('id', ''),
|
|
'expedientes_id': expediente_id,
|
|
'paciente_telefono': phone,
|
|
'paciente_nombre': r.get('nombre') or '',
|
|
'fecha': (r.get('created_at') or '')[:10],
|
|
'servicio_codigo': '',
|
|
'servicio_nombre': c.get('titulo') or '',
|
|
'cantidad': c.get('cantidad', 1) or 1,
|
|
'precio_unitario': precio,
|
|
'descuento': 0,
|
|
'impuesto': 0,
|
|
'estado': 'confirmed' if not r.get('cancelada') else 'cancelled',
|
|
'notas': r.get('motivo') or '',
|
|
})
|
|
save_csv('ventas', out, [
|
|
'legacy_id', 'expedientes_id', 'paciente_telefono', 'paciente_nombre', 'fecha',
|
|
'servicio_codigo', 'servicio_nombre', 'cantidad', 'precio_unitario',
|
|
'descuento', 'impuesto', 'estado', 'notas'
|
|
])
|
|
return out
|
|
|
|
|
|
def extract_citas(start_date='2022-11-01', end_date='2026-07-06'):
|
|
log(f'Extrayendo citas de {start_date} a {end_date}...')
|
|
phone_by_legacy = {}
|
|
pac_path = OUTDIR / 'pacientes.csv'
|
|
if pac_path.exists():
|
|
with open(pac_path, newline='', encoding='utf-8') as f:
|
|
for row in csv.DictReader(f):
|
|
phone_by_legacy[row.get('legacy_id', '')] = row.get('telefono', '')
|
|
|
|
medicos_map = {
|
|
'32': 'Dra. Alejandra Ramos',
|
|
'33': 'Dra. Lidia Martinez',
|
|
'46': 'Dra. Fernanda Cerecer',
|
|
'49': 'RODRIGUEZ FRIDA',
|
|
'54': 'Dr. Benjamín Adrián',
|
|
'79': 'XIMENA',
|
|
'80': 'ELY',
|
|
}
|
|
|
|
all_citas = []
|
|
current = datetime.strptime(start_date, '%Y-%m-%d').date()
|
|
end = datetime.strptime(end_date, '%Y-%m-%d').date()
|
|
days_with = 0
|
|
while current <= end:
|
|
url = f"/agenda/citas/{current.strftime('%Y-%m-%d')}"
|
|
try:
|
|
data = fetch_json(url)
|
|
citas = data.get('citas', [])
|
|
if citas:
|
|
days_with += 1
|
|
all_citas.extend(citas)
|
|
if current.day == 1 or len(citas) > 0:
|
|
log(f' {current}: {len(citas)} citas (total acumulado: {len(all_citas)})')
|
|
except Exception as e:
|
|
log(f' ERROR {current}: {e}')
|
|
current += timedelta(days=1)
|
|
time.sleep(0.05)
|
|
|
|
out = []
|
|
for c in all_citas:
|
|
nombre = (c.get('nombre') or '').strip()
|
|
if not nombre or nombre.upper().startswith('. . NO CITAR'):
|
|
continue
|
|
expediente_id = c.get('expedientes_id') or ''
|
|
phone = normalize_phone(c.get('telefono') or phone_by_legacy.get(str(expediente_id), ''))
|
|
if not phone and c.get('whatsapp'):
|
|
phone = normalize_phone(c.get('whatsapp'))
|
|
email = (c.get('correo') or '').strip()
|
|
if email.lower() in ('no@no.com', 'no@no'):
|
|
email = ''
|
|
inicio = c.get('inicio') or ''
|
|
fin = c.get('fin') or ''
|
|
hora_str = ''
|
|
if len(inicio) == 4:
|
|
hora_str = f"{inicio[:2]}:{inicio[2:]}"
|
|
elif len(inicio) == 3:
|
|
hora_str = f"0{inicio[0]}:{inicio[1:]}"
|
|
medico_id = str(c.get('medicos_id') or '')
|
|
doctor_nombre = medicos_map.get(medico_id, medico_id if medico_id and medico_id != '0' else '')
|
|
out.append({
|
|
'legacy_id': c.get('id', ''),
|
|
'expedientes_id': expediente_id,
|
|
'paciente_telefono': phone,
|
|
'paciente_nombre': nombre,
|
|
'servicio_codigo': '',
|
|
'servicio_nombre': c.get('titulo') or '',
|
|
'fecha': c.get('fecha', ''),
|
|
'hora': hora_str,
|
|
'hora_fin': f"{fin[:2]}:{fin[2:]}" if len(fin) == 4 else '',
|
|
'estado': 'confirmed',
|
|
'doctor_nombre': doctor_nombre,
|
|
'sucursal': 'rosarito',
|
|
'medio': 'onsite',
|
|
'notas': c.get('observaciones') or '',
|
|
'estado_pago': 'not_paid',
|
|
'monto_pagado': 0,
|
|
})
|
|
save_csv('citas', out, [
|
|
'legacy_id', 'expedientes_id', 'paciente_telefono', 'paciente_nombre',
|
|
'servicio_codigo', 'servicio_nombre', 'fecha', 'hora', 'hora_fin',
|
|
'estado', 'doctor_nombre', 'sucursal', 'medio', 'notas',
|
|
'estado_pago', 'monto_pagado'
|
|
])
|
|
log(f'Citas: {len(out)} registros útiles en {days_with} días con citas')
|
|
return out
|
|
|
|
|
|
def main():
|
|
try:
|
|
login()
|
|
except Exception as e:
|
|
log(f'ERROR login: {e}')
|
|
sys.exit(1)
|
|
|
|
extract_expedientes()
|
|
extract_monederos()
|
|
extract_crm()
|
|
extract_servicios()
|
|
extract_ventas()
|
|
extract_citas()
|
|
|
|
log('Extracción completada. Archivos en /root/migracion/extraccion/')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|