#!/usr/bin/env python # ~ # ~ PAC # ~ Copyright (C) 2018-2019 Mauricio Baeza Servin - public [AT] elmau [DOT] net # ~ # ~ This program is free software: you can redistribute it and/or modify # ~ it under the terms of the GNU General Public License as published by # ~ the Free Software Foundation, either version 3 of the License, or # ~ (at your option) any later version. # ~ # ~ This program is distributed in the hope that it will be useful, # ~ but WITHOUT ANY WARRANTY; without even the implied warranty of # ~ MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # ~ GNU General Public License for more details. # ~ # ~ You should have received a copy of the GNU General Public License # ~ along with this program. If not, see . import datetime import os import re from io import BytesIO from xml.sax.saxutils import unescape import lxml.etree as ET from zeep import Client from zeep.plugins import Plugin from zeep.cache import SqliteCache from zeep.transports import Transport from zeep.exceptions import Fault, TransportError from requests.exceptions import ConnectionError from conf import DEBUG, FINKOK TIMEOUT = 10 DEBUG_SOAP = False class PACFinkok(object): def cfdi_status(self, uuid, auth={}): if not auth: auth = FINKOK['AUTH'] method = 'timbra' client = Client( self.URL[method], transport=self._transport, plugins=self._plugins) args = { 'username': auth['USER'], 'password': auth['PASS'], 'uuid': uuid, } result = self._get_result(client, 'query_pending', args) if self.error: return {} STATUS = { 'C': 'Cancelado', 'S': 'Timbrado, aún no eviado al SAT', 'F': 'Timbrado y enviado al SAT', } data = { 'estatus': STATUS[result.status], 'xml': self._to_string(unescape(result.xml)), 'fecha': result.date, } return data # ~ Send issue to PAC def client_reset_token(self, email): auth = FINKOK['RESELLER'] method = 'util' client = Client( self.URL[method], transport=self._transport, plugins=self._plugins) args = { 'username': auth['USER'], 'password': auth['PASS'], 'token': email, } result = self._get_result(client, 'reset_token', args) if self.error: return '' if not result.success: self.error = result.message return '' return result.token def client_get(self, rfc): """Regresa el estatus del cliente Se requiere cuenta de reseller para usar este método Args: rfc (str): El RFC del emisor Returns: dict 'message': None, 'users': { 'ResellerUser': [ { 'status': 'A', 'counter': 0, 'taxpayer_id': '', 'credit': 0 } ] } or None si no existe """ auth = FINKOK['RESELLER'] method = 'client' client = Client( self.URL[method], transport=self._transport, plugins=self._plugins) args = { 'reseller_username': auth['USER'], 'reseller_password': auth['PASS'], 'taxpayer_id': rfc, } try: self.result = client.service.get(**args) except Fault as e: self.error = str(e) return {} except TransportError as e: self.error = str(e) return {} except ConnectionError: self.error = 'Verifica la conexión a internet' return {} success = bool(self.result.users) if not success: self.error = self.result.message or 'RFC no existe' return {} data = self.result.users.ResellerUser[0] client = { 'status': data.status, 'counter': data.counter, 'credit': data.credit, } return client def get_server_datetime(self): """Regresa la fecha y hora del servidor de timbrado del PAC """ auth = FINKOK['RESELLER'] method = 'util' client = Client( self.URL[method], transport=self._transport, plugins=self._plugins) try: self.result = client.service.datetime(auth['USER'], auth['PASS']) except Fault as e: self.error = str(e) return None except TransportError as e: self.error = str(e) return None except ConnectionError: self.error = 'Verifica la conexión a internet' return None try: dt = datetime.datetime.strptime( self.result.datetime, '%Y-%m-%dT%H:%M:%S') except ValueError: self.error = 'Error al obtener la fecha' return None return dt def get_report_credit(self, rfc): """Obtiene un reporte de los timbres agregados """ auth = FINKOK['RESELLER'] args = { 'username': auth['USER'], 'password': auth['PASS'], 'taxpayer_id': rfc, } method = 'util' client = Client( self.URL[method], transport=self._transport, plugins=self._plugins) try: self.result = client.service.report_credit(**args) except Fault as e: self.error = str(e) return [] except TransportError as e: self.error = str(e) return [] except ConnectionError: self.error = 'Verifica la conexión a internet' return [] if self.result.result is None: # ~ PAC - Debería regresar RFC inexistente o sin registros self.error = 'RFC no existe o no tiene registros' return [] return self.result.result.ReportTotalCredit def get_report_uuid(self, rfc, date_from, date_to, invoice_type='I'): """Obtiene un reporte de los CFDI timbrados """ auth = FINKOK['RESELLER'] args = { 'username': auth['USER'], 'password': auth['PASS'], 'taxpayer_id': rfc, 'date_from': date_from, 'date_to': date_to, 'invoice_type': invoice_type, } method = 'util' client = Client( self.URL[method], transport=self._transport, plugins=self._plugins) try: self.result = client.service.report_uuid(**args) except Fault as e: self.error = str(e) return [] except TransportError as e: self.error = str(e) return [] except ConnectionError: self.error = 'Verifica la conexión a internet' return [] if self.result.invoices is None: # ~ PAC - Debería regresar RFC inexistente o sin registros self.error = 'RFC no existe o no tiene registros' return [] return self.result.invoices.ReportUUID def _to_string(self, data): root = ET.parse(BytesIO(data.encode('utf-8'))).getroot() xml = ET.tostring(root, pretty_print=True, xml_declaration=True, encoding='utf-8') return xml.decode('utf-8') def cfdi_get_by_xml(self, xml, auth): if not auth: auth = FINKOK['AUTH'] method = 'timbra' client = Client( self.URL[method], transport=self._transport, plugins=self._plugins) args = { 'username': auth['USER'], 'password': auth['PASS'], 'xml': xml, } try: result = client.service.stamped(**args) except Fault as e: self.error = str(e) return {} except TransportError as e: self.error = str(e) return {} except ConnectionError as e: msg = '502 - Error de conexión' self.error = msg return {} print(result) error = 'Error: {}\n{}'.format(code_error, msg_error) self.error = self.CODE.get(code_error, error) return {} def cfdi_get_by_uuid(self, uuid, rfc, invoice_type='I', auth={}): if not auth: auth = FINKOK['AUTH'] method = 'util' client = Client( URL[method], transport=self._transport, plugins=self._plugins) args = { 'username': auth['USER'], 'password': auth['PASS'], 'uuid': uuid, 'taxpayer_id': rfc, 'invoice_type': invoice_type, } try: result = client.service.get_xml(**args) except Fault as e: self.error = str(e) return {} except TransportError as e: self.error = str(e) return {} except ConnectionError as e: msg = '502 - Error de conexión' self.error = msg return {} print(result) error = 'Error: {}\n{}'.format(code_error, msg_error) self.error = self.CODE.get(code_error, error) return {}