242 lines
9.7 KiB
Python
242 lines
9.7 KiB
Python
#!/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()
|