|
import asyncio |
|
import base64 |
|
from datetime import datetime, timedelta |
|
|
|
import httpx |
|
|
|
BASE_URL = "https://mifactura.napse.global/mf/pq1rt7" |
|
FIXED_PREFIX = "cencosud_434_1_6_0434001" |
|
|
|
# FIXME[privacy]: este código construye identificadores de comprobantes. No |
|
# ejecutarlo contra un servicio real salvo con autorización explícita y URLs |
|
# propias; para investigación usar un transporte/mock local. |
|
|
|
START_HOUR = 8 |
|
MAX_HOUR = 21 |
|
WINDOW_HOURS = 4 |
|
CONCURRENCY = 20 |
|
EMPTY_RESPONSE = b"<html><body></body></html>" |
|
|
|
|
|
def make_payload(receipt_number: int, dt: datetime) -> str: |
|
""" |
|
Ej: |
|
cencosud_434_1_6_0434001 + 0122 + 2405041552 |
|
""" |
|
# FIXME[structure]: los anchos y el significado de estos campos son una |
|
# hipótesis basada en ejemplos de Disco y Vea, no un formato documentado. |
|
receipt = f"{receipt_number:04d}" |
|
timestamp = dt.strftime("%y%m%d%H%M") |
|
|
|
return f"{FIXED_PREFIX}{receipt}{timestamp}" |
|
|
|
|
|
def make_url(receipt_number: int, dt: datetime) -> str: |
|
payload = make_payload(receipt_number, dt) |
|
encoded = base64.b64encode(payload.encode()).decode() |
|
|
|
return f"{BASE_URL}/{encoded}" |
|
|
|
|
|
def iter_minutes(start: datetime, end: datetime): |
|
current = start |
|
|
|
while current <= end: |
|
yield current |
|
current += timedelta(minutes=1) |
|
|
|
|
|
async def probe(client, url, semaphore): |
|
async with semaphore: |
|
try: |
|
# FIXME[validation]: status 200 + tamaño del body no distingue un |
|
# comprobante de una página de error, proxy o anti-bot. Validar |
|
# una estructura HTML conocida en respuestas guardadas/locales. |
|
async with client.stream("GET", url) as r: |
|
if r.status_code != 200: |
|
return None |
|
|
|
body = bytearray() |
|
|
|
async for chunk in r.aiter_bytes(): |
|
body.extend(chunk) |
|
|
|
if len(body) > len(EMPTY_RESPONSE): |
|
return url |
|
|
|
if body and bytes(body).strip() != EMPTY_RESPONSE: |
|
return url |
|
|
|
# FIXME[errors]: descartar timeout/5xx como "no encontrado" puede |
|
# cortar la búsqueda por una falla transitoria. Separar NOT_FOUND, |
|
# TRANSIENT_ERROR y MATCH antes de usar este flujo. |
|
except httpx.HTTPError: |
|
pass |
|
|
|
return None |
|
|
|
|
|
async def find_receipt( |
|
receipt_number: int, |
|
start: datetime, |
|
window_hours: int = WINDOW_HOURS, |
|
): |
|
end = start + timedelta(hours=window_hours) |
|
|
|
# No buscar después de las 21:00. |
|
day_limit = start.replace( |
|
hour=MAX_HOUR, |
|
minute=0, |
|
second=0, |
|
microsecond=0, |
|
) |
|
|
|
# FIXME[timezone]: estos datetimes son ingenuos. La hora del payload debe |
|
# estar asociada explícitamente a America/Argentina/Cordoba. |
|
end = min(end, day_limit) |
|
|
|
print(f"Buscando {receipt_number:04d}: {start:%H:%M} -> {end:%H:%M}") |
|
candidates = [(dt, make_url(receipt_number, dt)) for dt in iter_minutes(start, end)] |
|
semaphore = asyncio.Semaphore(CONCURRENCY) |
|
|
|
async with httpx.AsyncClient( |
|
timeout=5, |
|
follow_redirects=True, |
|
) as client: |
|
# FIXME[load]: 20 conexiones simultáneas y ventanas minuto a minuto |
|
# pueden generar una carga considerable sobre el servicio. Para tests |
|
# usar MockTransport; si hubiera autorización, aplicar rate limit y |
|
# backoff respetando las condiciones del proveedor. |
|
results = await asyncio.gather( |
|
*(probe(client, url, semaphore) for _, url in candidates) |
|
) |
|
|
|
for (dt, _), url in zip(candidates, results): |
|
if url is not None: |
|
print( |
|
f"Encontrado | Comprobante: {receipt_number:04d} " |
|
f"| Hora: {dt:%H:%M} | URL: {url}" |
|
) |
|
return dt, url |
|
|
|
return None |
|
|
|
|
|
async def main(): |
|
date = datetime(2024, 5, 4) |
|
|
|
current_start = date.replace( |
|
hour=START_HOUR, |
|
minute=0, |
|
second=0, |
|
microsecond=0, |
|
) |
|
|
|
# FIXME[search]: empezar en 0001 y detenerse ante el primer faltante supone |
|
# que la numeración diaria es contigua y empieza en 1. Los ejemplos del |
|
# repo de Disco usan secuencias 0199 y 0255, por lo que esta suposición no |
|
# está respaldada por la evidencia disponible. |
|
receipt_number = 1 |
|
|
|
while current_start.hour < MAX_HOUR: |
|
result = await find_receipt( |
|
receipt_number, |
|
current_start, |
|
) |
|
|
|
if result is None: |
|
# FIXME[search]: un resultado vacío no demuestra que no existan |
|
# comprobantes posteriores; puede ser un hueco o un error de red. |
|
print( |
|
f"No encontré {receipt_number:04d} " |
|
f"en las siguientes {WINDOW_HOURS} horas." |
|
) |
|
break |
|
|
|
found_dt, _ = result |
|
|
|
# El comprobante siguiente no puede ser anterior |
|
# al comprobante que acabamos de encontrar. |
|
# FIXME[boundary]: reutilizar found_dt permite volver a consultar el |
|
# mismo minuto para el siguiente comprobante. La regla correcta depende |
|
# de si la hora pertenece a la caja, a la transacción o al servidor. |
|
current_start = found_dt |
|
receipt_number += 1 |
|
|
|
|
|
if __name__ == "__main__": |
|
asyncio.run(main()) |
