diff --git a/.gitignore b/.gitignore index e9578f1..999a919 100644 --- a/.gitignore +++ b/.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/ diff --git a/docs/pages/getting_started/50_pac_guide.rst b/docs/pages/getting_started/50_pac_guide.rst index b29fd72..969bc79 100644 --- a/docs/pages/getting_started/50_pac_guide.rst +++ b/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 - \ No newline at end of file + diff --git a/readme.rst b/readme.rst index 0ca5187..06809ed 100644 --- a/readme.rst +++ b/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 diff --git a/satcfdi/pacs/__init__.py b/satcfdi/pacs/__init__.py index 292e6a6..50d1b96 100644 --- a/satcfdi/pacs/__init__.py +++ b/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() diff --git a/satcfdi/pacs/sat.py b/satcfdi/pacs/sat.py index 775676a..b958b62 100644 --- a/satcfdi/pacs/sat.py +++ b/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) diff --git a/tests/pac_sat_responses/listado_69/CSDsinefectos.csv b/tests/pac_sat_responses/listado_69/CSDsinefectos.csv new file mode 100644 index 0000000..39a75fb --- /dev/null +++ b/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 diff --git a/tests/pac_sat_responses/listado_69/Cancelados.csv b/tests/pac_sat_responses/listado_69/Cancelados.csv new file mode 100644 index 0000000..52abc91 --- /dev/null +++ b/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 diff --git a/tests/pac_sat_responses/listado_69/Condonados_07_15.csv b/tests/pac_sat_responses/listado_69/Condonados_07_15.csv new file mode 100644 index 0000000..bf92e3a --- /dev/null +++ b/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 diff --git a/tests/pac_sat_responses/listado_69/Firmes.csv b/tests/pac_sat_responses/listado_69/Firmes.csv new file mode 100644 index 0000000..7083a37 --- /dev/null +++ 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 diff --git a/tests/pac_sat_responses/listado_69/default.csv b/tests/pac_sat_responses/listado_69/default.csv new file mode 100644 index 0000000..80469db --- /dev/null +++ b/tests/pac_sat_responses/listado_69/default.csv @@ -0,0 +1,2 @@ +RFC,RAZON SOCIAL,SUPUESTO +ZZZ999999ZZZ,EMPRESA RELLENO SA DE CV,SUPUESTO diff --git a/tests/test_pac_sat.py b/tests/test_pac_sat.py index 6e644d9..9794fb4 100644 --- a/tests/test_pac_sat.py +++ b/tests/test_pac_sat.py @@ -1,4 +1,5 @@ import os.path +import sqlite3 import types from datetime import date, datetime, timedelta, timezone from decimal import Decimal @@ -6,6 +7,8 @@ from pprint import PrettyPrinter from unittest import mock +import requests + from satcfdi.utils import StrEnum from satcfdi.create.catalogos import EstadoComprobante @@ -14,12 +17,15 @@ from satcfdi import __version__ from satcfdi.cfdi import CFDI +from satcfdi.exceptions import ResponseError from satcfdi.models import Code from satcfdi.pacs import Environment from satcfdi.pacs import TaxpayerStatus +from satcfdi.pacs import TaxpayerStatusList69 from satcfdi.pacs.sat import _CFDISolicitaDescargaEmitidos, _CFDIAutenticacion, EstadoSolicitud, TipoDescargaMasivaTerceros from satcfdi.pacs.sat import SAT from satcfdi.pacs.sat import _get_listado_69b +from satcfdi.pacs.sat import _get_listado_69, _refresh_listado_69, _init_listado_69_db, _parse_listado_69_csv from satcfdi.pacs.sat import _set_arguments from .utils import get_signer, verify_result @@ -120,6 +126,102 @@ def test_sat_list_69b_notfound(): assert res is None +def _listado_69_request(url, **kwargs): + path = os.path.join(current_dir, 'pac_sat_responses', 'listado_69', os.path.basename(url)) + if not os.path.exists(path): + path = os.path.join(current_dir, 'pac_sat_responses', 'listado_69', 'default.csv') + + with open(path, 'rb') as f: + data = f.read() + + a = types.SimpleNamespace() + a.status_code = 200 + a.content = data + return a + + +def test_listado69(): + with mock.patch(f'requests.get', _listado_69_request) as mk: + connection = _get_listado_69(refresh_time=-100000) + + meta = dict(connection.execute("SELECT lista, updated_at FROM listado_69_meta")) + assert set(meta) == {e.value for e in TaxpayerStatusList69} + + rows = connection.execute("SELECT lista, supuesto FROM listado_69 WHERE rfc = 'CCC030303CCC'").fetchall() + assert rows == [('cancelados', 'CANCELADOS')] + + rows = connection.execute("SELECT lista, supuesto FROM listado_69 WHERE rfc = 'DDD040404DDD'").fetchall() + assert rows == [('csd_sin_efectos', 'SUPUESTO CSD')] + + rows = connection.execute("SELECT count(*) FROM listado_69 WHERE rfc = ''").fetchone() + assert rows == (0,) + + +def test_sat_list_69(): + with mock.patch(f'requests.get', _listado_69_request) as mk: + sat_service = SAT(environment=Environment.TEST) + + res = sat_service.list_69('AAA010101AAA') + assert res == { + TaxpayerStatusList69.FIRMES: 'FIRMES', + TaxpayerStatusList69.CANCELADOS: 'CANCELADOS POR INSOLVENCIA', + TaxpayerStatusList69.CONDONADOS_07_15: None, + } + + res = sat_service.list_69(' bbb020202bbb ') + assert res == {TaxpayerStatusList69.FIRMES: 'FIRMES'} + + +def test_sat_list_69_notfound(): + with mock.patch(f'requests.get', _listado_69_request) as mk: + sat_service = SAT(environment=Environment.TEST) + res = sat_service.list_69('xxxx') + assert res == {} + + +def test_listado69_csv_windows_1252(): + data = 'RFC,Razón Social,Supuesto de Cancelación\r\nEEE050505EEE,CAMIÓN SA DE CV,ARTÍCULO 17-H\r\n'.encode('windows-1252') + data += b'FFF060606FFF,EMPRESA \x81 SA DE CV,X\r\n' + + rows = list(_parse_listado_69_csv(data, 'RFC', 'SUPUESTO DE CANCELACION')) + assert rows == [('EEE050505EEE', 'ARTÍCULO 17-H'), ('FFF060606FFF', 'X')] + + +def test_listado69_refresh_failure_keeps_data(): + connection = sqlite3.connect(':memory:') + _init_listado_69_db(connection) + + with mock.patch(f'requests.get', _listado_69_request) as mk: + assert _refresh_listado_69(connection, TaxpayerStatusList69.FIRMES) + + failures = [ + types.SimpleNamespace(status_code=500, content=b''), + types.SimpleNamespace(status_code=200, content=b''), + types.SimpleNamespace(status_code=200, content=b''), + types.SimpleNamespace(status_code=200, content=b'RFC,SUPUESTO\r\n'), + ] + for response in failures: + with mock.patch(f'requests.get', return_value=response) as mk: + assert not _refresh_listado_69(connection, TaxpayerStatusList69.FIRMES) + + with mock.patch(f'requests.get', side_effect=requests.ConnectionError) as mk: + assert not _refresh_listado_69(connection, TaxpayerStatusList69.FIRMES) + + rows = connection.execute("SELECT rfc FROM listado_69 WHERE lista = 'firmes' ORDER BY rfc").fetchall() + assert rows == [('AAA010101AAA',), ('BBB020202BBB',)] + + +def test_listado69_never_loaded(): + import pytest + + _get_listado_69.cache_clear() + response = types.SimpleNamespace(status_code=500, content=b'') + + with mock.patch(f'{module}.pacs.sat.LISTADO_69_DB', ':memory:'), mock.patch(f'requests.get', return_value=response): + with pytest.raises(ResponseError): + _get_listado_69(refresh_time=-100000) + + def test_sat_service_authentication(): signer = get_signer('xiqb891116qe4')