satcfdi.portal.SATPortalConstancia.generar_constancia() - Code Metrics - SAT-CFDI/python-satcfdi - Measure and Improve Code Quality continuously with Scrutinizer

SATPortalConstancia.generar_constancia() A
last analyzed 2025-11-02 19:33 UTC

↳ Parent: satcfdi.portal

Complexity

Conditions

Size

Total Lines	44
Code Lines	34

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	10.7163

Importance

Changes

Metric	Value
cc	3
eloc	34
nop	1
dl	0
loc	44
ccs	1
cts	20
cp	0.05
crap	10.7163
rs	9.064
c	0
b	0
f	0

import json
import pickle
from time import time

import requests
from requests.structures import CaseInsensitiveDict

from .utils import get_form, generate_token, request_ref_headers, request_verification_token, random_ajax_id
from ..exceptions import ResponseError
from ..models import Signer
from ..sat_requests_utils import SSLAdapter


class PortalManager(requests.Session):
    def __init__(self, signer: Signer):
        super().__init__()
        self.mount('https://', SSLAdapter())
        self.signer = signer

        self.headers = CaseInsensitiveDict(
            {
                "User-Agent": 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/107.0.0.0 Safari/537.36',
                "Accept-Encoding": 'gzip, deflate, br',
                "Accept": 'text/html,application/xhtml+xml,application/xml',
                "Connection": "keep-alive",
                'Pragma': 'no-cache',
                'Cache-Control': 'no-cache',
            }
        )

    def save_session(self, target):
        pickle.dump(self.cookies, target)

    def load_session(self, source):
        self.cookies.update(pickle.load(source))

    def form(self, action, referer_url, data):
        res = self.post(
            url=action,
            headers=request_ref_headers(referer_url),
            data=data
        )
        assert res.status_code == 200
        return res

    def fiel_login(self, login_response):
        action, data = get_form(login_response, id='certform')
        return self.form(
            action,
            login_response.request.url,
            data | {
                'token': generate_token(self.signer, code=data['guid']),
                'fert': self.signer.certificate.get_notAfter()[2:].decode(),
            }
        )


class SATPortal(PortalManager):
    BASE_URL = 'https://loginda.siat.sat.gob.mx'

    def login(self):
        res = self.get(
            url=f'{self.BASE_URL}/nidp/app/login?id=fiel'
        )
        assert res.status_code == 200
        action, data = get_form(res)

        return self.fiel_login(
            login_response=self.form(action, res.request.url, data)
        )

    def home_page(self):
        return self.get(
            url=f'{self.BASE_URL}/nidp/app?sid=0'
        )

    def logout(self):
        return self.get(
            url=f'{self.BASE_URL}/nidp/app/logout',
            headers={
                'referer': f'{self.BASE_URL}/nidp/app?sid=0'
            },
            allow_redirects=False
        )

    def declaraciones_provisionales_login(self):
        res = self.get(
            url='https://ptscdecprov.clouda.sat.gob.mx',
            allow_redirects=True
        )
        assert res.status_code == 200

        action, data = get_form(res)
        res = self.form(action, res.request.url, data)
        return res


class SATFacturaElectronica(PortalManager):
    BASE_URL = 'https://portal.facturaelectronica.sat.gob.mx'
    REQUEST_CONTEXT = 'appId=cid-v1:20ff76f4-0bca-495f-b7fd-09ca520e39f7'

    def __init__(self, signer: Signer):
        super().__init__(signer)
        self._ajax_id = random_ajax_id()
        self._request_verification_token = None

    def login(self):
        res = self.get(
            url=self.BASE_URL
        )
        assert res.status_code == 200
        try:
            action, data = get_form(res)
        except IndexError as ex:
            raise ValueError("Login form not found, please try again") from ex

        if action.startswith('https://cfdiau.sat.gob.mx/'):
            assert 'nidp/wsfed/ep?id=SATUPCFDiCon' in action

            res = self.fiel_login(
                login_response=self.form(
                    action.replace('nidp/wsfed/ep?id=SATUPCFDiCon', 'nidp/app/login?id=SATx509Custom'),
                    res.request.url,
                    data
                )
            )

            action, data = get_form(res)
            res = self.form(action, res.request.url, data)

            action, data = get_form(res)

        res = self.form(action, res.request.url, data)

        self._request_verification_token = request_verification_token(res)
        self._ajax_id = random_ajax_id()
        return res

    def _reload_verification_token(self):
        res = self.get(
            url=f'{self.BASE_URL}/Factura/GeneraFactura',
            allow_redirects=False
        )
        if res.status_code == 200:
            self._request_verification_token = request_verification_token(res)
        else:
            raise ValueError('Please Login Again')

    def reactivate_session(self):
        res = self.post(
            url=f'{self.BASE_URL}/Home/ReActiveSession',
            headers={
                'Origin': self.BASE_URL,
                'Request-Context': self.REQUEST_CONTEXT,
                'Request-Id': f'|{self._ajax_id}.{random_ajax_id()}'
            },
            allow_redirects=False
        )
        return res

    def _request(self, method, path, data=None, params=None):
        if self._request_verification_token is None:
            self._reload_verification_token()

        res = self.request(
            method=method,
            url=f'{self.BASE_URL}/{path}',
            headers={
                'Origin': self.BASE_URL,
                'Authority': self.BASE_URL,
                'Request-Context': self.REQUEST_CONTEXT,
                '__RequestVerificationToken': self._request_verification_token,
                'Request-Id': f'|{self._ajax_id}.{random_ajax_id()}'  # |pR4Px.o0yAS
            },
            data=data,
            params=params,
            allow_redirects=False
        )
        if res.status_code == 200:
            return res.json()
        else:
            raise ResponseError(res)

    def legal_name_valid(self, rfc, legal_name):
        res = self._request(
            method='POST',
            path='Clientes/ValidaRazonSocialRFC',
            data={
                'rfcValidar': rfc.upper(),
                'razonSocial': legal_name.upper(),
            })
        if not res['exitoso']:
            raise ResponseError(res)
        return res['resultado']

    def rfc_valid(self, rfc):
        res = self._request(
            method='POST',
            path='Clientes/ExisteLrfc',
            data={
                'rfcValidar': rfc.upper()
            }
        )
        if not res['exitoso']:
            raise ResponseError(res)
        return res['resultado']

    def lco_details(self, rfc, apply_border_region=True):
        res = self._request(
            method='GET',
            path='Clientes/ValidaLco',
            params={
                'rfcValidar': rfc.upper(),
                'aplicaRegionFronteriza': apply_border_region,
                "_": int(time() * 1000)
            }
        )
        return json.loads(res)


class SATPortalConstancia(PortalManager):
    BASE_URL = 'https://login.siat.sat.gob.mx'

    def login(self):
        res = self.get(
            url=f'{self.BASE_URL}/nidp/idff/sso?id=fiel_Aviso'
        )
        assert res.status_code == 200
        action, data = get_form(res)
        if action is None:
            return res

        return self.fiel_login(
            login_response=self.form(action, res.request.url, data)
        )

    def generar_constancia(self):
        self.login()
        res = self.get(
            url="https://rfcampc.siat.sat.gob.mx/app/seg/SessionBroker?url=/PTSC/IdcSiat/autc/ReimpresionTramite/ConsultaTramite.jsf&parametro=c&idSessionBit=&idSessionBit=null",
            allow_redirects=True
        )
        assert res.status_code == 200

        # Execute Authentication Request
        action, data = get_form(res)
        if action == "https://login.siat.sat.gob.mx/nidp/saml2/sso":
            res = self.form(action, res.request.url, data)
            assert res.status_code == 200

            # Execute Authentication Response
            action, data = get_form(res)
            assert action == "https://rfcampc.siat.sat.gob.mx/saml2/sp/acs/post"
            res = self.form(action, res.request.url, data)
            assert res.status_code == 200

        # Execute formReimpAcuse
        action, data = get_form(res)
        if action == "https://rfcampc.siat.sat.gob.mx/PTSC/IdcSiat/autc/ReimpresionTramite/ConsultaTramite.jsf":
            data = {
                'javax.faces.partial.ajax': "true",
                'javax.faces.source': 'formReimpAcuse:j_idt50',
                'javax.faces.partial.execute': '@all',
                'formReimpAcuse:j_idt50': 'formReimpAcuse:j_idt50',
                'formReimpAcuse': 'formReimpAcuse',
                'formReimpAcuse:tipoTramite_input': '0',
                'formReimpAcuse:tipoTramite_focus': '',
                'formReimpAcuse:fechaInicio_input:': '',
                'formReimpAcuse:fechaFin_input': '',
                'formReimpAcuse:folio': '',
                'javax.faces.ViewState': data['javax.faces.ViewState']
            }
            res = self.form(action, res.request.url, data)
            assert res.status_code == 200

        res = self.get(
            url='https://rfcampc.siat.sat.gob.mx/PTSC/IdcSiat/IdcGeneraConstancia.jsf'
        )
        assert res.status_code == 200
        return res.content


class SATPortalOpinionCumplimiento(PortalManager):
    BASE_URL = 'https://login.mat.sat.gob.mx'

    def login(self):
        res = self.get(
            url=f'{self.BASE_URL}/nidp/app/login?id=contr-dual-eFirma-totp'
        )
        assert res.status_code == 200
        action, data = get_form(res)

        res = self.form(action, res.request.url, data)
        assert res.status_code == 200

        res = self.form(action, res.request.url, data)
        assert res.status_code == 200

        res = self.form(action, res.request.url, data)
        assert res.status_code == 200

        return res

    def generar_opinion_cumplimiento_login(self):
        res = self.get(
            url="https://ptsc32d.clouda.sat.gob.mx/?/reporteOpinion32DContribuyente",
            allow_redirects=True
        )
        assert res.status_code == 200

        # Execute Authentication Request
        action, data = get_form(res)
        if action.startswith("https://login.mat.sat.gob.mx/nidp//app/login"):
            res = self.form(action, res.request.url, data)
            assert res.status_code == 200

            # Execute Authentication Response
            action, data = get_form(res)
            res = self.form(action, res.request.url, data)
            assert res.status_code == 200


1	1	import json
2	1	import pickle
3	1	from time import time
4
5	1	import requests
6	1	from requests.structures import CaseInsensitiveDict
7
8	1	from .utils import get_form, generate_token, request_ref_headers, request_verification_token, random_ajax_id
9	1	from ..exceptions import ResponseError
10	1	from ..models import Signer
11	1	from ..sat_requests_utils import SSLAdapter
12
13
14	1	class PortalManager(requests.Session):
15	1	def __init__(self, signer: Signer):
16	1	super().__init__()
17	1	self.mount('https://', SSLAdapter())
18	1	self.signer = signer
19
20	1	self.headers = CaseInsensitiveDict(
21		{
22		"User-Agent": 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/107.0.0.0 Safari/537.36',
23		"Accept-Encoding": 'gzip, deflate, br',
24		"Accept": 'text/html,application/xhtml+xml,application/xml',
25		"Connection": "keep-alive",
26		'Pragma': 'no-cache',
27		'Cache-Control': 'no-cache',
28		}
29		)
30
31	1	def save_session(self, target):
32	1	pickle.dump(self.cookies, target)
33
34	1	def load_session(self, source):
35		self.cookies.update(pickle.load(source))
36
37	1	def form(self, action, referer_url, data):
38		res = self.post(
39		url=action,
40		headers=request_ref_headers(referer_url),
41		data=data
42		)
43		assert res.status_code == 200
44		return res
45
46	1	def fiel_login(self, login_response):
47		action, data = get_form(login_response, id='certform')
48		return self.form(
49		action,
50		login_response.request.url,
51		data \| {
52		'token': generate_token(self.signer, code=data['guid']),
53		'fert': self.signer.certificate.get_notAfter()[2:].decode(),
54		}
55		)
56
57
58	1	class SATPortal(PortalManager):
59	1	BASE_URL = 'https://loginda.siat.sat.gob.mx'
60
61	1	def login(self):
62		res = self.get(
63		url=f'{self.BASE_URL}/nidp/app/login?id=fiel'
64		)
65		assert res.status_code == 200
66		action, data = get_form(res)
67
68		return self.fiel_login(
69		login_response=self.form(action, res.request.url, data)
70		)
71
72	1	def home_page(self):
73		return self.get(
74		url=f'{self.BASE_URL}/nidp/app?sid=0'
75		)
76
77	1	def logout(self):
78		return self.get(
79		url=f'{self.BASE_URL}/nidp/app/logout',
80		headers={
81		'referer': f'{self.BASE_URL}/nidp/app?sid=0'
82		},
83		allow_redirects=False
84		)
85
86	1	def declaraciones_provisionales_login(self):
87		res = self.get(
88		url='https://ptscdecprov.clouda.sat.gob.mx',
89		allow_redirects=True
90		)
91		assert res.status_code == 200
92
93		action, data = get_form(res)
94		res = self.form(action, res.request.url, data)
95		return res
96
97
98	1	class SATFacturaElectronica(PortalManager):
99	1	BASE_URL = 'https://portal.facturaelectronica.sat.gob.mx'
100	1	REQUEST_CONTEXT = 'appId=cid-v1:20ff76f4-0bca-495f-b7fd-09ca520e39f7'
101
102	1	def __init__(self, signer: Signer):
103		super().__init__(signer)
104		self._ajax_id = random_ajax_id()
105		self._request_verification_token = None
106
107	1	def login(self):
108		res = self.get(
109		url=self.BASE_URL
110		)
111		assert res.status_code == 200
112		try:
113		action, data = get_form(res)
114		except IndexError as ex:
115		raise ValueError("Login form not found, please try again") from ex
116
117		if action.startswith('https://cfdiau.sat.gob.mx/'):
118		assert 'nidp/wsfed/ep?id=SATUPCFDiCon' in action
119
120		res = self.fiel_login(
121		login_response=self.form(
122		action.replace('nidp/wsfed/ep?id=SATUPCFDiCon', 'nidp/app/login?id=SATx509Custom'),
123		res.request.url,
124		data
125		)
126		)
127
128		action, data = get_form(res)
129		res = self.form(action, res.request.url, data)
130
131		action, data = get_form(res)
132
133		res = self.form(action, res.request.url, data)
134
135		self._request_verification_token = request_verification_token(res)
136		self._ajax_id = random_ajax_id()
137		return res
138
139	1	def _reload_verification_token(self):
140		res = self.get(
141		url=f'{self.BASE_URL}/Factura/GeneraFactura',
142		allow_redirects=False
143		)
144		if res.status_code == 200:
145		self._request_verification_token = request_verification_token(res)
146		else:
147		raise ValueError('Please Login Again')
148
149	1	def reactivate_session(self):
150		res = self.post(
151		url=f'{self.BASE_URL}/Home/ReActiveSession',
152		headers={
153		'Origin': self.BASE_URL,
154		'Request-Context': self.REQUEST_CONTEXT,
155		'Request-Id': f'\|{self._ajax_id}.{random_ajax_id()}'
156		},
157		allow_redirects=False
158		)
159		return res
160
161	1	def _request(self, method, path, data=None, params=None):
162		if self._request_verification_token is None:
163		self._reload_verification_token()
164
165		res = self.request(
166		method=method,
167		url=f'{self.BASE_URL}/{path}',
168		headers={
169		'Origin': self.BASE_URL,
170		'Authority': self.BASE_URL,
171		'Request-Context': self.REQUEST_CONTEXT,
172		'__RequestVerificationToken': self._request_verification_token,
173		'Request-Id': f'\|{self._ajax_id}.{random_ajax_id()}' # \|pR4Px.o0yAS
174		},
175		data=data,
176		params=params,
177		allow_redirects=False
178		)
179		if res.status_code == 200:
180		return res.json()
181		else:
182		raise ResponseError(res)
183
184	1	def legal_name_valid(self, rfc, legal_name):
185		res = self._request(
186		method='POST',
187		path='Clientes/ValidaRazonSocialRFC',
188		data={
189		'rfcValidar': rfc.upper(),
190		'razonSocial': legal_name.upper(),
191		})
192		if not res['exitoso']:
193		raise ResponseError(res)
194		return res['resultado']
195
196	1	def rfc_valid(self, rfc):
197		res = self._request(
198		method='POST',
199		path='Clientes/ExisteLrfc',
200		data={
201		'rfcValidar': rfc.upper()
202		}
203		)
204		if not res['exitoso']:
205		raise ResponseError(res)
206		return res['resultado']
207
208	1	def lco_details(self, rfc, apply_border_region=True):
209		res = self._request(
210		method='GET',
211		path='Clientes/ValidaLco',
212		params={
213		'rfcValidar': rfc.upper(),
214		'aplicaRegionFronteriza': apply_border_region,
215		"_": int(time() * 1000)
216		}
217		)
218		return json.loads(res)
219
220
221	1	class SATPortalConstancia(PortalManager):
222	1	BASE_URL = 'https://login.siat.sat.gob.mx'
223
224	1	def login(self):
225		res = self.get(
226		url=f'{self.BASE_URL}/nidp/idff/sso?id=fiel_Aviso'
227		)
228		assert res.status_code == 200
229		action, data = get_form(res)
230		if action is None:
231		return res
232
233		return self.fiel_login(
234		login_response=self.form(action, res.request.url, data)
235		)
236
237	1	def generar_constancia(self):
238		self.login()
239		res = self.get(
240		url="https://rfcampc.siat.sat.gob.mx/app/seg/SessionBroker?url=/PTSC/IdcSiat/autc/ReimpresionTramite/ConsultaTramite.jsf&parametro=c&idSessionBit=&idSessionBit=null",
241		allow_redirects=True
242		)
243		assert res.status_code == 200
244
245		# Execute Authentication Request
246		action, data = get_form(res)
247		if action == "https://login.siat.sat.gob.mx/nidp/saml2/sso":
248		res = self.form(action, res.request.url, data)
249		assert res.status_code == 200
250
251		# Execute Authentication Response
252		action, data = get_form(res)
253		assert action == "https://rfcampc.siat.sat.gob.mx/saml2/sp/acs/post"
254		res = self.form(action, res.request.url, data)
255		assert res.status_code == 200
256
257		# Execute formReimpAcuse
258		action, data = get_form(res)
259		if action == "https://rfcampc.siat.sat.gob.mx/PTSC/IdcSiat/autc/ReimpresionTramite/ConsultaTramite.jsf":
260		data = {
261		'javax.faces.partial.ajax': "true",
262		'javax.faces.source': 'formReimpAcuse:j_idt50',
263		'javax.faces.partial.execute': '@all',
264		'formReimpAcuse:j_idt50': 'formReimpAcuse:j_idt50',
265		'formReimpAcuse': 'formReimpAcuse',
266		'formReimpAcuse:tipoTramite_input': '0',
267		'formReimpAcuse:tipoTramite_focus': '',
268		'formReimpAcuse:fechaInicio_input:': '',
269		'formReimpAcuse:fechaFin_input': '',
270		'formReimpAcuse:folio': '',
271		'javax.faces.ViewState': data['javax.faces.ViewState']
272		}
273		res = self.form(action, res.request.url, data)
274		assert res.status_code == 200
275
276		res = self.get(
277		url='https://rfcampc.siat.sat.gob.mx/PTSC/IdcSiat/IdcGeneraConstancia.jsf'
278		)
279		assert res.status_code == 200
280		return res.content
281
282
283	1	class SATPortalOpinionCumplimiento(PortalManager):
284	1	BASE_URL = 'https://login.mat.sat.gob.mx'
285
286	1	def login(self):
287		res = self.get(
288		url=f'{self.BASE_URL}/nidp/app/login?id=contr-dual-eFirma-totp'
289		)
290		assert res.status_code == 200
291		action, data = get_form(res)
292
293		res = self.form(action, res.request.url, data)
294		assert res.status_code == 200
295
296		res = self.form(action, res.request.url, data)
297		assert res.status_code == 200
298
299		res = self.form(action, res.request.url, data)
300		assert res.status_code == 200
301
302		return res
303
304	1	def generar_opinion_cumplimiento_login(self):
305		res = self.get(
306		url="https://ptsc32d.clouda.sat.gob.mx/?/reporteOpinion32DContribuyente",
307		allow_redirects=True
308		)
309		assert res.status_code == 200
310
311		# Execute Authentication Request
312		action, data = get_form(res)
313		if action.startswith("https://login.mat.sat.gob.mx/nidp//app/login"):
314		res = self.form(action, res.request.url, data)
315		assert res.status_code == 200
316
317		# Execute Authentication Response
318		action, data = get_form(res)
319		res = self.form(action, res.request.url, data)
320		assert res.status_code == 200
321

SAT-CFDI / python-satcfdi

SATPortalConstancia.generar_constancia() A last analyzed 2025-11-02 19:33 UTC

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

SATPortalConstancia.generar_constancia() A
last analyzed 2025-11-02 19:33 UTC