Feat/listado 69

.gitignore

@@ -148,6 +148,7 @@ dmypy.json
 /examples/_*_.*
 
 
+/satcfdi/pacs/listado_69.db
 /examples/_OÑO120726RX30DOTAAL1LCMBB10371.dec
 /tests/test_portal/session
 /tests/test_certifica/test_csd/

docs/pages/getting_started/50_pac_guide.rst

@@ -91,6 +91,19 @@ ______________________
     print(res)
 
 
+Listado 69
+______________________
+
+.. code-block:: python
+
+    from satcfdi.pacs import TaxpayerStatusList69
+    from satcfdi.pacs.sat import SAT
+
+    sat_service = SAT()
+    res = sat_service.list_69('AAL081211JP0')
+    assert res == { TaxpayerStatusList69.CANCELADOS: 'CANCELADOS POR INSOLVENCIA' }
+
+
 Listado 69B
 ______________________
 
@@ -102,4 +115,4 @@ ______________________
     sat_service = SAT()
     res = sat_service.list_69b('AAL081211JP0')
     assert res == TaxpayerStatus.DEFINITIVO
-
+

readme.rst

@@ -75,6 +75,7 @@ ____________________
   * SW Sapien
 * Descarga Masiva
 * Validación de Comprobantes
+* Listado 69
 * Listado 69B
 * Exportar Comprobantes a Excel
 * Descarga de Constancia de Situación Fiscal

satcfdi/pacs/__init__.py

@@ -1,5 +1,6 @@
 from dataclasses import dataclass
 from enum import Flag, auto, Enum
+from ..utils import StrEnum
 from ..cfdi import CFDI
 from ..models import Signer
 from ..create.cancela import cancelacion
@@ -134,6 +135,53 @@ class TaxpayerStatus(Enum):
     """
 
 
+class TaxpayerStatusList69(StrEnum):
+    FIRMES = 'firmes'
+    """Contribuyentes con créditos fiscales firmes (ya no admiten medio de defensa o fueron consentidos)."""
+
+    EXIGIBLES = 'exigibles'
+    """Créditos fiscales determinados, exigibles, que no se encuentran pagados ni garantizados."""
+
+    NO_LOCALIZADOS = 'no_localizados'
+    """Contribuyentes inscritos en el RFC que se encuentran como no localizados en su domicilio fiscal."""
+
+    SENTENCIAS = 'sentencias'
+    """Contribuyentes con sentencia condenatoria ejecutoria por la comisión de un delito fiscal."""
+
+    CANCELADOS = 'cancelados'
+    """Créditos fiscales cancelados por incosteabilidad en el cobro o por insolvencia del deudor.
+    El SUPUESTO distingue "CANCELADOS" y "CANCELADOS POR INSOLVENCIA"."""
+
+    CONDONADOS_146B = 'condonados_146b'
+    """Contribuyentes a los que se les condonó algún crédito fiscal en concurso mercantil (Art. 146-B CFF)."""
+
+    REDUCCION_ART21 = 'reduccion_art21'
+    """Reducción de recargos conforme al Art. 21 CFF."""
+
+    REDUCCION_ART74 = 'reduccion_art74'
+    """Reducción de multas conforme al Art. 74 CFF."""
+
+    CONDONADOS_DECRETO = 'condonados_decreto'
+    """Condonados por decreto presidencial (22 de enero y 26 de marzo de 2015)."""
+
+    RETORNO_INVERSIONES = 'retorno_inversiones'
+    """Programa de retorno de inversiones (repatriación de capitales, 2016)."""
+
+    CSD_SIN_EFECTOS = 'csd_sin_efectos'
+    """Contribuyentes a los que se les dejó sin efectos el Certificado de Sello Digital (Arts. 17-H y 17-H Bis CFF)."""
+
+    ENTES_OMISOS = 'entes_omisos'
+    """Entes públicos y de gobierno omisos en la presentación de declaraciones. El SUPUESTO indica la fracción."""
+
+    CONDONADOS_07_15 = 'condonados_07_15'
+    """Histórico: créditos condonados del 1 de enero de 2007 al 4 de mayo de 2015.
+    El archivo no tiene columna SUPUESTO; tiene "Motivo de condonación"."""
+
+    CANCELADOS_07_15 = 'cancelados_07_15'
+    """Histórico: créditos cancelados conforme al Art. 146-A CFF del 1 de enero de 2007 al 4 de mayo de 2015.
+    El archivo no tiene columna SUPUESTO; tiene "MOTIVO"."""
+
+
 class Environment(Enum):
     PRODUCTION = auto()
     TEST = auto()

satcfdi/pacs/sat.py

@@ -1,37 +1,59 @@
-import base64
+import io
+import os
 import csv
+import sqlite3
+import time
 import json
+import base64
 import logging
-import os
-import time
+import requests
 import warnings
-from abc import abstractmethod
-from collections.abc import Sequence
-from datetime import date, datetime, timedelta, timezone
-from enum import IntEnum, Enum
-from functools import cache
-from itertools import islice
+import unicodedata
 from uuid import UUID
-
-import requests
-from bs4 import BeautifulSoup
 from lxml import etree
+from ..cfdi import CFDI
+from .. import __version__
+from functools import cache
+from itertools import islice
 from lxml.etree import QName
 from packaging import version
-
-from satcfdi.create.cfd.catalogos import TipoDeComprobante
-from . import PAC, Environment, TaxpayerStatus
-from .. import __version__
-from ..cfdi import CFDI
-from ..create.w3.signature import signature_c14n_sha1, _digest, _tobytes
+from bs4 import BeautifulSoup
+from abc import abstractmethod
+from enum import IntEnum, Enum
+from collections.abc import Sequence
 from ..exceptions import ResponseError
 from ..models import Signer, Certificate
-from ..transform import MEXICO_TZ, get_timezone, verify_certificate
 from ..utils import iterate, StrEnum, parser
+from datetime import date, datetime, timedelta, timezone
+from satcfdi.create.cfd.catalogos import TipoDeComprobante
+from ..transform import MEXICO_TZ, get_timezone, verify_certificate
+from . import PAC, Environment, TaxpayerStatus, TaxpayerStatusList69
+from ..create.w3.signature import signature_c14n_sha1, _digest, _tobytes
 
 logger = logging.getLogger(__name__)
 current_dir = os.path.dirname(__file__)
 
+# Listado 69
+LISTADO_69_DB = os.path.join(current_dir, 'Listado_69.db')
+LISTADO_69_BASE_URL = "https://wu1agsprosta001.blob.core.windows.net/agsc-publicaciones/Datos_abiertos/"
+LISTADO_69_SOURCES = {
+    TaxpayerStatusList69.FIRMES:              ('Documents_AGR/Firmes.csv',                          'RFC', 'SUPUESTO'),
+    TaxpayerStatusList69.EXIGIBLES:           ('Documents_AGR/Exigibles.csv',                       'RFC', 'SUPUESTO'),
+    TaxpayerStatusList69.NO_LOCALIZADOS:      ('Documents_AGR/No_localizados.csv',                  'RFC', 'SUPUESTO'),
+    TaxpayerStatusList69.SENTENCIAS:          ('Documents_AGR/Sentencias.csv',                      'RFC', 'SUPUESTO'),
+    TaxpayerStatusList69.CANCELADOS:          ('Documents_AGR/Cancelados.csv',                      'RFC', 'SUPUESTO'),
+    TaxpayerStatusList69.CSD_SIN_EFECTOS:     ('Documents_AGR/CSDsinefectos.csv',                   'RFC', 'SUPUESTO DE CANCELACION'),
+    TaxpayerStatusList69.REDUCCION_ART74:     ('Documents_AGR/ReduccionArt74CFF.csv',               'RFC', 'SUPUESTO'),
+    TaxpayerStatusList69.ENTES_OMISOS:        ('Documents_AGR/EntespublicosydeGobiernoomisos.csv',  'RFC', 'SUPUESTO'),
+    TaxpayerStatusList69.CONDONADOS_146B:     ('Cifras_SAT/Indef/Condonadosart146BCFF.csv',         'RFC', 'SUPUESTO'),
+    TaxpayerStatusList69.REDUCCION_ART21:     ('Cifras_SAT/Indef/Condonadosart21CFF.csv',           'RFC', 'SUPUESTO'),
+    TaxpayerStatusList69.CONDONADOS_DECRETO:  ('Cifras_SAT/Indef/CondonadosporDecreto.csv',         'RFC', 'SUPUESTO'),
+    TaxpayerStatusList69.CONDONADOS_07_15:    ('Cifras_SAT/Indef/Condonados_07_15.csv',             'RFC', None),
+    TaxpayerStatusList69.CANCELADOS_07_15:    ('Cifras_SAT/Indef/Cancelados_07_15.csv',             'RFC', None),
+    TaxpayerStatusList69.RETORNO_INVERSIONES: ('Cifras_SAT/Indef/Retornoinversiones.csv',           'RFC', 'SUPUESTO'),
+}
+
+# Listado 69B
 LISTADO_COMPLETO_69B_JSON = os.path.join(current_dir, 'Listado_Completo_69-B.json')
 REFRESH_TIME = (15 * 86400)  # 15 Days
 UTC = timezone.utc
@@ -107,6 +129,114 @@ def _certificate_path(certificate_no):
     return f"{certificate_no[0:6]}/{certificate_no[6:12]}/{certificate_no[12:14]}/{certificate_no[14:16]}/{certificate_no[16:18]}/{certificate_no}"
 
 
+
+def _normalizar_header(value: str) -> str:
+    value = unicodedata.normalize('NFKD', value).encode('ascii', 'ignore').decode()
+    return ' '.join(value.upper().split())
+
+
+def _parse_listado_69_csv(content: bytes, rfc_col: str, supuesto_col: str | None):
+    text = content.decode('windows-1252', errors='replace')
+    reader = csv.DictReader(io.StringIO(text))
+
+    if not reader.fieldnames:
+        raise ValueError('Empty CSV')
+
+    headers = {}
+
+    for header in reader.fieldnames:
+        normalized = _normalizar_header(header)
+        headers[normalized] = header
+
+    if rfc_col not in headers:
+        raise ValueError(f"Column '{rfc_col}' not found in {reader.fieldnames}")
+
+    rfc_key = headers[rfc_col]
+    sup_key = headers.get(supuesto_col) if supuesto_col else None
+
+    if supuesto_col and not sup_key:
+        logger.warning("Column '%s' not found in %s, supuesto will be None", supuesto_col, reader.fieldnames)
+
+    for row in reader:
+        rfc = (row.get(rfc_key) or '').strip().upper()
+        if not rfc:
+            continue
+        supuesto = (row.get(sup_key) or '').strip() if sup_key else None
+        yield rfc, (supuesto or None)
+
+
+def _init_listado_69_db(connection: sqlite3.Connection):
+    connection.executescript("""
+        CREATE TABLE IF NOT EXISTS listado_69 (
+        lista       TEXT NOT NULL,
+        rfc         TEXT NOT NULL,
+        supuesto    TEXT,
+        PRIMARY KEY(lista, rfc)
+        );
+        CREATE INDEX IF NOT EXISTS ix_listado_69_rfc ON listado_69(rfc);
+        CREATE TABLE IF NOT EXISTS listado_69_meta(
+        lista       TEXT PRIMARY KEY,
+        updated_at  REAL NOT NULL
+        );
+    """)
+
+
+def _refresh_listado_69(connection: sqlite3.Connection, lista: TaxpayerStatusList69) -> bool:
+    path, rfc_col, supuesto_col = LISTADO_69_SOURCES[lista]
+
+    try:
+        request = requests.get(
+            url = LISTADO_69_BASE_URL + path,
+            headers = {"User-Agent": __version__.__user_agent__}
+        )
+
+        if request.status_code != 200:
+            logger.error("Unable to get Listado 69 '%s', status code: %s", lista, request.status_code)
+            return False
+
+        rows = list(_parse_listado_69_csv(request.content, rfc_col, supuesto_col))
+    except (requests.RequestException, ValueError, csv.Error) as ex:
+        logger.error("Unable to get Listado 69 '%s': %s", lista, ex)
+        return False
+
+    if not rows:
+        logger.error("Listado 69 '%s' came back empty, keeping previous data", lista)
+        return False
+
+    with connection:
+        connection.execute("DELETE FROM listado_69 WHERE lista = ?", (lista.value,))
+        connection.executemany(
+            "INSERT OR IGNORE INTO listado_69 (lista, rfc, supuesto) VALUES (?, ?, ?)",
+            ((lista.value, rfc, sup) for rfc, sup in rows)
+        )
+        connection.execute(
+            "INSERT OR REPLACE INTO listado_69_meta (lista, updated_at) VALUES (?, ?)",
+            (lista.value, time.time())
+        )
+    return True
+
+
+@cache
+def _get_listado_69(refresh_time=REFRESH_TIME):
+    """
+    Retrieve the csvs that, altogether, make the lista 69. Update the list if
+    enough time has happened between the first fetch. This data will be stored 
+    in sqlite
+    """
+    connection = sqlite3.connect(LISTADO_69_DB, check_same_thread = False)
+    _init_listado_69_db(connection)
+
+    meta = dict(connection.execute("SELECT lista, updated_at FROM listado_69_meta"))
+    for lista in TaxpayerStatusList69:
+        t = meta.get(lista.value, -refresh_time)
+        if time.time() > t + refresh_time:
+            ok = _refresh_listado_69(connection, lista)
+            if not ok and lista.value not in meta:
+                raise ResponseError(f"Unable to load Listado 69 '{lista}'")
+
+    return connection
+
+
 @cache
 def _get_listado_69b(refresh_time=REFRESH_TIME):
     try:
@@ -597,6 +727,33 @@ def _validate(self, cfdi, certificate, num_certificate, rfc, seal, date: datetim
 
         return True
 
+
+    def list_69(self, rfc: str) -> dict[TaxpayerStatusList69, str | None]:
+        """
+        Look up an RFC in the lists published by the SAT under Article 69 of the CFF.
+        The lists are downloaded on first use and cached locally for 15 days.
+
+        :param rfc: RFC to look up, case and surrounding spaces are ignored
+        :return: dict with each list where the RFC appears as key and its SUPUESTO text,
+            as published by the SAT, as value, or None if the list does not publish one.
+            Empty dict if the RFC is not in any list.
+        :raises ResponseError: if a list cannot be downloaded and there is no previous data
+        """
+        connection = _get_listado_69()
+
+        rows = connection.execute(
+            "SELECT lista, supuesto FROM listado_69 WHERE rfc = ?",
+            (rfc.strip().upper(),)
+        ).fetchall()
+
+        response = {}
+
+        for lista, supuesto in rows:
+            response[TaxpayerStatusList69(lista)] = supuesto
+
+        return response
+
+
     def list_69b(self, rfc: str) -> TaxpayerStatus | None:
         listado = _get_listado_69b()
         r = listado.get(rfc)

tests/pac_sat_responses/listado_69/CSDsinefectos.csv

@@ -0,0 +1,2 @@
+rfc,Razon Social,  Supuesto   de cancelacion
+DDD040404DDD,EMPRESA CUATRO SA DE CV,SUPUESTO CSD

tests/pac_sat_responses/listado_69/Cancelados.csv

@@ -0,0 +1,4 @@
+RFC,RAZON SOCIAL,SUPUESTO
+AAA010101AAA,EMPRESA UNO SA DE CV,CANCELADOS POR INSOLVENCIA
+CCC030303CCC,EMPRESA TRES SA DE CV,CANCELADOS
+CCC030303CCC,EMPRESA TRES SA DE CV,CANCELADOS POR INSOLVENCIA

tests/pac_sat_responses/listado_69/Condonados_07_15.csv

@@ -0,0 +1,3 @@
+RFC,RAZON SOCIAL,MOTIVO DE CONDONACION
+AAA010101AAA,EMPRESA UNO SA DE CV,MOTIVO A
+AAA010101AAA,EMPRESA UNO SA DE CV,MOTIVO B

tests/pac_sat_responses/listado_69/Firmes.csv

@@ -0,0 +1,4 @@
+RFC,RAZON SOCIAL,SUPUESTO
+AAA010101AAA,EMPRESA UNO SA DE CV,FIRMES
+ bbb020202bbb ,EMPRESA DOS SA DE CV,FIRMES
+,REGISTRO SIN RFC,FIRMES

tests/pac_sat_responses/listado_69/default.csv

@@ -0,0 +1,2 @@
+RFC,RAZON SOCIAL,SUPUESTO
+ZZZ999999ZZZ,EMPRESA RELLENO SA DE CV,SUPUESTO