importacion historico legacy completo: citas con estados reales + 24,497 ventas con pagos (cuadre al centavo 48/48 meses); method points en skeen_pagos + sin acumular al canjear; scripts y doc de migracion
This commit is contained in:
241
migracion/x_full_concentrado.py
Normal file
241
migracion/x_full_concentrado.py
Normal file
@@ -0,0 +1,241 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""Descarga mensual del reporte concentrado del legacy SKEEN (SOLO LECTURA).
|
||||
|
||||
Por cada mes 2022-11 → 2026-10 hace POST a /reportes/concentrado/excel
|
||||
(mismo formulario que la vista HTML del reporte; no modifica datos) y guarda
|
||||
el Excel en extraccion/concentrado/concentrado_mensual/concentrado_YYYY-MM.xlsx.
|
||||
Luego combina todo en concentrado_full.csv (17 columnas del reporte + mes).
|
||||
|
||||
Quirk del Excel generado por el legacy: <pageMargins left="" right="" .../>
|
||||
con cadenas vacías → openpyxl 3.1.5 revienta ("expected <class 'float'>").
|
||||
Se parchea el XML al guardar (márgenes por defecto), quedando el .xlsx en
|
||||
disco ya abrible.
|
||||
|
||||
El Excel trae 4 filas de encabezado (título, rango, vacía, cabeceras de las
|
||||
17 columnas) y desde la fila 5 los datos. No trae fila de totales (las
|
||||
totales solo salen en el HTML, tabla resumen).
|
||||
|
||||
Uso: /root/odoo-venv/bin/python /root/migracion/x_full_concentrado.py
|
||||
"""
|
||||
import csv
|
||||
import io
|
||||
import json
|
||||
import re
|
||||
import sys
|
||||
import time
|
||||
import zipfile
|
||||
from collections import defaultdict
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
import openpyxl
|
||||
import requests
|
||||
from bs4 import BeautifulSoup
|
||||
|
||||
sys.path.insert(0, '/root/migracion')
|
||||
import extraer_skeen as ex # login(), MESES_ES
|
||||
|
||||
OUTDIR = Path('/root/migracion/extraccion/concentrado')
|
||||
MENSUAL = OUTDIR / 'concentrado_mensual'
|
||||
MENSUAL.mkdir(parents=True, exist_ok=True)
|
||||
CSV_PATH = OUTDIR / 'concentrado_full.csv'
|
||||
ERROR_PATH = OUTDIR / 'errores.txt'
|
||||
|
||||
MESES = ['2022-11', '2022-12'] + [f'{y}-{m:02d}' for y in range(2023, 2027) for m in range(1, 13)]
|
||||
MESES = [m for m in MESES if '2022-11' <= m <= '2026-10']
|
||||
SLEEP = 1.2
|
||||
|
||||
COLS = ['folio', 'fecha', 'nombre_paciente', 'folio_dr', 'atendido_por', 'vendedor',
|
||||
'concepto', 'efectivo_usd', 'efectivo_mn', 'tarjeta_usd', 'tarjeta_mn',
|
||||
'transferencia_usd', 'transferencia_mn', 'saldo_a_favor_usd', 'saldo_a_favor_mn',
|
||||
'monedero_usd', 'monedero_mn', 'mes']
|
||||
|
||||
MARG_RE = re.compile(r'<pageMargins left="" right="" top="" bottom=""')
|
||||
|
||||
|
||||
def patch_xlsx(data: bytes) -> bytes:
|
||||
zin = zipfile.ZipFile(io.BytesIO(data))
|
||||
out = io.BytesIO()
|
||||
with zipfile.ZipFile(out, 'w', zipfile.ZIP_DEFLATED) as zout:
|
||||
for item in zin.infolist():
|
||||
buf = zin.read(item.filename)
|
||||
if item.filename.startswith('xl/worksheets/') and buf.lstrip().startswith(b'<'):
|
||||
txt = buf.decode('utf-8')
|
||||
txt = MARG_RE.sub('<pageMargins left="0.7" right="0.7" top="0.75" bottom="0.75"', txt)
|
||||
buf = txt.encode('utf-8')
|
||||
zout.writestr(item, buf)
|
||||
return out.getvalue()
|
||||
|
||||
|
||||
def get_token():
|
||||
r = ex.session.get(f'{ex.BASE}/reportes/concentrado', timeout=30)
|
||||
if '/login' in r.url:
|
||||
ex.login()
|
||||
r = ex.session.get(f'{ex.BASE}/reportes/concentrado', timeout=30)
|
||||
soup = BeautifulSoup(r.text, 'html.parser')
|
||||
inp = soup.find('input', {'name': '_token'})
|
||||
if not inp:
|
||||
raise RuntimeError('no se encontró _token en /reportes/concentrado')
|
||||
return inp['value']
|
||||
|
||||
|
||||
def download_month(ym, token, max_attempts=5):
|
||||
y, m = map(int, ym.split('-'))
|
||||
import calendar
|
||||
ini = f'{ym}-01'
|
||||
fin = f'{ym}-{calendar.monthrange(y, m)[1]:02d}'
|
||||
payload = {'_token': token, 'fecha_pago': json.dumps({'start': ini, 'end': fin})}
|
||||
for attempt in range(1, max_attempts + 1):
|
||||
try:
|
||||
r = ex.session.post(f'{ex.BASE}/reportes/concentrado/excel', data=payload, timeout=120)
|
||||
if '/login' in r.url:
|
||||
ex.log(f'{ym}: sesión expirada, re-login...')
|
||||
ex.login()
|
||||
token = get_token()
|
||||
payload['_token'] = token
|
||||
continue
|
||||
if r.status_code >= 500:
|
||||
raise requests.HTTPError(f'{r.status_code}')
|
||||
ct = r.headers.get('Content-Type') or ''
|
||||
if 'spreadsheetml' not in ct and 'octet-stream' not in ct:
|
||||
raise ValueError(f'respuesta no-Excel: ct={ct} url={r.url} bytes={len(r.content)}')
|
||||
return patch_xlsx(r.content)
|
||||
except Exception as e:
|
||||
wait = 2 ** attempt
|
||||
ex.log(f'{ym}: intento {attempt}/{max_attempts} falló: {e} (espera {wait}s)')
|
||||
time.sleep(wait)
|
||||
raise RuntimeError(f'{ym}: imposible descargar tras {max_attempts} intentos')
|
||||
|
||||
|
||||
def parse_xlsx(path, ym):
|
||||
wb = openpyxl.load_workbook(path, read_only=True)
|
||||
ws = wb.active
|
||||
rows = list(ws.iter_rows(values_only=True))
|
||||
wb.close()
|
||||
# localizar fila de cabecera (col A == 'Folio')
|
||||
hdr_idx = None
|
||||
for i, row in enumerate(rows[:10]):
|
||||
if row and str(row[0]).strip() == 'Folio':
|
||||
hdr_idx = i
|
||||
break
|
||||
if hdr_idx is None:
|
||||
raise ValueError(f'{path}: fila de cabecera no encontrada')
|
||||
header = [str(h).strip() if h is not None else '' for h in rows[hdr_idx]]
|
||||
out = []
|
||||
for row in rows[hdr_idx + 1:]:
|
||||
if not row or row[0] is None or str(row[0]).strip() == '':
|
||||
continue
|
||||
vals = list(row) + [None] * (17 - len(row))
|
||||
d = {COLS[i]: vals[i] for i in range(17)}
|
||||
a = str(d['folio']).strip().lower()
|
||||
if a.startswith('total'):
|
||||
continue # por si algún mes trae fila de totales
|
||||
d['mes'] = ym
|
||||
out.append(d)
|
||||
return out
|
||||
|
||||
|
||||
def parse_fecha_es(s):
|
||||
m = re.match(r'(\d{1,2})/([A-Za-zÁÉÍÓÚáéíóúñ]+)/(\d{4})', str(s or '').strip())
|
||||
if not m:
|
||||
return None
|
||||
dia, mes, anio = m.groups()
|
||||
mn = ex.MESES_ES.get(mes.lower())
|
||||
if not mn:
|
||||
return None
|
||||
return f'{anio}-{mn}-{int(dia):02d}'
|
||||
|
||||
|
||||
def main():
|
||||
ex.login()
|
||||
token = get_token()
|
||||
|
||||
failed = []
|
||||
row_counts = {}
|
||||
all_rows = []
|
||||
for ym in MESES:
|
||||
dest = MENSUAL / f'concentrado_{ym}.xlsx'
|
||||
if dest.exists() and dest.stat().st_size > 2000:
|
||||
ex.log(f'{ym}: ya existe {dest.name} ({dest.stat().st_size} bytes), reutilizo')
|
||||
else:
|
||||
try:
|
||||
data = download_month(ym, token)
|
||||
except Exception as e:
|
||||
with open(ERROR_PATH, 'a', encoding='utf-8') as f:
|
||||
f.write(f'[concentrado] {ym}: {e}\n')
|
||||
ex.log(f'ERROR {ym}: {e} → registrado en errores.txt, continúo')
|
||||
failed.append(ym)
|
||||
time.sleep(SLEEP)
|
||||
continue
|
||||
dest.write_bytes(data)
|
||||
ex.log(f'{ym}: guardado {dest.name} ({len(data)} bytes)')
|
||||
time.sleep(SLEEP)
|
||||
try:
|
||||
rows = parse_xlsx(dest, ym)
|
||||
row_counts[ym] = len(rows)
|
||||
all_rows.extend(rows)
|
||||
except Exception as e:
|
||||
with open(ERROR_PATH, 'a', encoding='utf-8') as f:
|
||||
f.write(f'[concentrado-parse] {ym}: {e}\n')
|
||||
ex.log(f'ERROR parse {ym}: {e}')
|
||||
failed.append(ym)
|
||||
|
||||
with open(CSV_PATH, 'w', newline='', encoding='utf-8') as f:
|
||||
w = csv.DictWriter(f, fieldnames=COLS, extrasaction='ignore')
|
||||
w.writeheader()
|
||||
for d in all_rows:
|
||||
w.writerow({k: ('' if d.get(k) is None else d.get(k)) for k in COLS})
|
||||
|
||||
ex.log('========== RESUMEN CONCENTRADO ==========')
|
||||
ex.log(f'Meses pedidos: {len(MESES)} | descargados/parseados: {len(row_counts)} | fallados: {failed}')
|
||||
ex.log(f'Filas totales concentrado_full.csv: {len(all_rows)}')
|
||||
by_year = defaultdict(int)
|
||||
for d in all_rows:
|
||||
by_year[str(d['mes'])[:4]] += 1
|
||||
ex.log(f'Filas por año: {dict(sorted(by_year.items()))}')
|
||||
ex.log(f'CSV: {CSV_PATH}')
|
||||
|
||||
# ---- Verificación cruzada: semana 2026-08-01..07 del Excel vs HTML ya guardado ----
|
||||
probe = OUTDIR / 'probe_concentrado.html'
|
||||
if probe.exists():
|
||||
soup = BeautifulSoup(probe.read_text(encoding='utf-8'), 'html.parser')
|
||||
tabs = soup.find_all('table')
|
||||
sums_html = {}
|
||||
if len(tabs) >= 1:
|
||||
for cell in tabs[0].find_all('tr')[1].find_all('td'):
|
||||
txt = cell.get_text(strip=True)
|
||||
mm = re.match(r'(Efectivo|Tarjeta|Transferencia|Saldo a favor|Monedero electrónico)\$\s*([\d,]+\.\d{2})\s*USD', txt)
|
||||
if mm:
|
||||
sums_html[mm.group(1)] = float(mm.group(2).replace(',', ''))
|
||||
for cell in tabs[0].find_all('tr')[2].find_all('td'):
|
||||
txt = cell.get_text(strip=True)
|
||||
mm = re.match(r'(Efectivo|Tarjeta|Transferencia|Saldo a favor|Monedero electrónico)\$\s*([\d,]+\.\d{2})\s*M\.N\.', txt)
|
||||
if mm:
|
||||
sums_html[mm.group(1) + ' MN'] = float(mm.group(2).replace(',', ''))
|
||||
mine = defaultdict(float)
|
||||
for d in all_rows:
|
||||
f = parse_fecha_es(d['fecha'])
|
||||
if f and '2026-08-01' <= f <= '2026-08-07':
|
||||
mine['Efectivo'] += float(d['efectivo_usd'] or 0)
|
||||
mine['Efectivo MN'] += float(d['efectivo_mn'] or 0)
|
||||
mine['Tarjeta'] += float(d['tarjeta_usd'] or 0)
|
||||
mine['Tarjeta MN'] += float(d['tarjeta_mn'] or 0)
|
||||
mine['Transferencia'] += float(d['transferencia_usd'] or 0)
|
||||
mine['Transferencia MN'] += float(d['transferencia_mn'] or 0)
|
||||
mine['Saldo a favor'] += float(d['saldo_a_favor_usd'] or 0)
|
||||
mine['Saldo a favor MN'] += float(d['saldo_a_favor_mn'] or 0)
|
||||
mine['Monedero electrónico'] += float(d['monedero_usd'] or 0)
|
||||
mine['Monedero electrónico MN'] += float(d['monedero_mn'] or 0)
|
||||
ex.log('---- Cross-check semana 2026-08-01..07 (Excel mensual vs HTML) ----')
|
||||
ok = True
|
||||
for k in sorted(set(sums_html) | set(mine)):
|
||||
a, b = sums_html.get(k, 0.0), round(mine.get(k, 0.0), 2)
|
||||
match = abs(a - b) < 0.01
|
||||
ok = ok and match
|
||||
ex.log(f' {k}: HTML={a:>12,.2f} Excel={b:>12,.2f} {"OK" if match else "DIFF"}')
|
||||
ex.log(f' RESULTADO: {"TODO COINCIDE" if ok else "HAY DIFERENCIAS"}')
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
Reference in New Issue
Block a user