#!/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 base64 import logging import lxml.etree as ET from . import mureq from conf import DEBUG LOG_FORMAT = '%(asctime)s - %(levelname)s - %(message)s' LOG_DATE = '%d/%m/%Y %H:%M:%S' logging.addLevelName(logging.ERROR, '\033[1;41mERROR\033[1;0m') logging.addLevelName(logging.DEBUG, '\x1b[33mDEBUG\033[1;0m') logging.addLevelName(logging.INFO, '\x1b[32mINFO\033[1;0m') logging.basicConfig(level=logging.DEBUG, format=LOG_FORMAT, datefmt=LOG_DATE) log = logging.getLogger(__name__) logging.getLogger('requests').setLevel(logging.ERROR) TIMEOUT = 10 def pretty_print_POST(req): """ At this point it is completely built and ready to be fired; it is "prepared". However pay attention at the formatting used in this function because it is programmed to be pretty printed and may differ from the actual request. """ print('{}\n{}\r\n{}\r\n\r\n{}'.format( '-----------START-----------', req.method + ' ' + req.url, '\r\n'.join('{}: {}'.format(k, v) for k, v in req.headers.items()), req.body, )) class PACComercioDigital(object): ws = 'https://{}.comercio-digital.mx/{}' api = 'https://app2.comercio-digital.mx/{}' URL = { 'timbra': ws.format('ws', 'timbre4/timbrarV5'), 'cancel': ws.format('cancela', 'cancela4/cancelarUuid'), 'cancelxml': ws.format('cancela', 'cancela4/cancelarXml'), 'status': ws.format('cancela', 'arws/consultaEstatus'), 'client': api.format('x3/altaEmpresa'), 'saldo': api.format('x3/saldo'), 'timbres': api.format('x3/altaTimbres'), } CODES = { '000': '000 Exitoso', '004': '004 RFC {} ya esta dado de alta con Estatus=A', '704': '704 Usuario Invalido', '702': '702 Error rfc/empresa invalido', } NS_CFDI = { 'cfdi': 'http://www.sat.gob.mx/cfd/4', 'tdf': 'http://www.sat.gob.mx/TimbreFiscalDigital', } if DEBUG: ws = 'https://pruebas.comercio-digital.mx/{}' ws6 = 'https://pruebas6.comercio-digital.mx/arws/{}' URL = { 'timbra': ws.format('timbre4/timbrarV5'), 'cancel': ws.format('cancela4/cancelarUuid'), 'cancelxml': ws.format('cancela4/cancelarXml'), 'status': ws6.format('consultaEstatus'), 'client': api.format('x3/altaEmpresa'), 'saldo': api.format('x3/saldo'), 'timbres': api.format('x3/altaTimbres'), } def __init__(self): self.error = '' def _error(self, msg): self.error = str(msg) # ~ log.error(msg) return def _post(self, url, data, headers={}): result = None headers['host'] = url.split('/')[2] headers['Content-type'] = 'text/plain' headers['Connection'] = 'Keep-Alive' headers['Expect'] = '100-continue' # ~ if DEBUG: # ~ req = requests.Request('POST', url, headers=headers, data=data) # ~ prepared = req.prepare() # ~ pretty_print_POST(prepared) try: result = mureq.post(url, body=data, headers=headers, timeout=TIMEOUT) except Exception as e: self._error(e) return result def _validate_cfdi(self, xml): """ Comercio Digital solo soporta la declaración con doble comilla """ # ~ tree = ET.fromstring(xml.encode()) # ~ xml = ET.tostring(tree, # ~ pretty_print=True, doctype='') return xml.encode('utf-8') def stamp(self, cfdi, auth): url = self.URL['timbra'] headers = { 'usrws': auth['user'], 'pwdws': auth['pass'], 'tipo': 'XML', } cfdi = self._validate_cfdi(cfdi) result = self._post(url, cfdi, headers) if result is None: return '' if result.status_code != 200: return '' if 'errmsg' in result.headers: self._error(result.headers['errmsg']) return '' xml = result.content tree = ET.fromstring(xml) cfdi_uuid = tree.xpath( 'string(//cfdi:Complemento/tdf:TimbreFiscalDigital/@UUID)', namespaces=self.NS_CFDI) date_stamped = tree.xpath( 'string(//cfdi:Complemento/tdf:TimbreFiscalDigital/@FechaTimbrado)', namespaces=self.NS_CFDI) data = { 'xml': xml.decode(), 'uuid': cfdi_uuid, 'date': date_stamped, } return data def _get_data_cancel(self, cfdi, info, auth): info['tipo'] = 'cfdi' info['key'] = base64.b64encode(info['key_enc']).decode() info['cer'] = base64.b64encode(info['cer_ori']).decode() NS_CFDI = { 'cfdi': 'http://www.sat.gob.mx/cfd/3', 'tdf': 'http://www.sat.gob.mx/TimbreFiscalDigital', } tree = ET.fromstring(cfdi.encode()) tipo = tree.xpath( 'string(//cfdi:Comprobante/@TipoDeComprobante)', namespaces=NS_CFDI) total = tree.xpath( 'string(//cfdi:Comprobante/@Total)', namespaces=NS_CFDI) rfc_emisor = tree.xpath( 'string(//cfdi:Comprobante/cfdi:Emisor/@Rfc)', namespaces=NS_CFDI) rfc_receptor = tree.xpath( 'string(//cfdi:Comprobante/cfdi:Receptor/@Rfc)', namespaces=NS_CFDI) uid = tree.xpath( 'string(//cfdi:Complemento/tdf:TimbreFiscalDigital/@UUID)', namespaces=NS_CFDI) data = ( f"USER={auth['user']}", f"PWDW={auth['pass']}", f"RFCE={rfc_emisor}", f"UUID={uid}", f"PWDK={info['pass']}", f"KEYF={info['key']}", f"CERT={info['cer']}", f"TIPO1={info['tipo']}", f"ACUS=SI", f"RFCR={rfc_receptor}", f"TIPOC={tipo}", f"TOTAL={total}", f"UUIDREL={info['args']['uuid']}", f"MOTIVO={info['args']['reason']}", ) return '\n'.join(data) def cancel(self, cfdi, info, auth): # ~ if DEBUG or not auth: # ~ auth = AUTH url = self.URL['cancel'] data = self._get_data_cancel(cfdi, info, auth) result = self._post(url, data) if result is None: return '' if result.status_code != 200: return '' if result.headers['codigo'] != '000': self._error(result.headers['errmsg']) return '' tree = ET.fromstring(result.text) date_cancel = tree.xpath('string(//Acuse/@Fecha)')[:19] data = { 'acuse': result.text, 'date': date_cancel, } return data def _get_headers_cancel_xml(self, cfdi, info, auth): NS_CFDI = { 'cfdi': 'http://www.sat.gob.mx/cfd/3', 'tdf': 'http://www.sat.gob.mx/TimbreFiscalDigital', } tree = ET.fromstring(cfdi.encode()) tipocfdi = tree.xpath( 'string(//cfdi:Comprobante/@TipoDeComprobante)', namespaces=NS_CFDI) total = tree.xpath( 'string(//cfdi:Comprobante/@Total)', namespaces=NS_CFDI) rfc_receptor = tree.xpath( 'string(//cfdi:Comprobante/cfdi:Receptor/@Rfc)', namespaces=NS_CFDI) headers = { 'usrws': auth['user'], 'pwdws': auth['pass'], 'rfcr': rfc_receptor, 'total': total, 'tipocfdi': tipocfdi, } headers.update(info) return headers def cancel_xml(self, xml, auth, cfdi='', info={'tipo': 'cfdi'}): # ~ if DEBUG or not auth: # ~ auth = AUTH url = self.URL['cancelxml'] headers = self._get_headers_cancel_xml(cfdi, info, auth) result = self._post(url, xml, headers) if result is None: return '' if result.status_code != 200: return '' if result.headers['codigo'] != '000': self._error(result.headers['errmsg']) return '' tree = ET.fromstring(result.text) date_cancel = tree.xpath('string(//Acuse/@Fecha)')[:19] data = { 'acuse': result.text, 'date': date_cancel, } return data def status(self, data, auth): # ~ if not auth: # ~ auth = AUTH url = self.URL['status'] data = ( f"USER={auth['user']}", f"PWDW={auth['pass']}", f"RFCR={data['rfc_receptor']}", f"RFCE={data['rfc_emisor']}", f"TOTAL={data['total']}", f"UUID={data['uuid']}", ) data = '\n'.join(data) result = self._post(url, data) if result is None: return '' if result.status_code != 200: self._error(result.status_code) return self.error return result.text def _get_data_client(self, auth, values): data = [f"usr_ws={auth['user']}", f"pwd_ws={auth['pass']}"] fields = ( 'rfc_contribuyente', 'nombre_contribuyente', 'calle', 'noExterior', 'noInterior', 'colonia', 'localidad', 'municipio', 'estado', 'pais', 'cp', 'contacto', 'telefono', 'email', 'rep_nom', 'rep_rfc', 'email_fact', 'pwd_asignado', ) data += [f"{k}={values[k]}" for k in fields] return '\n'.join(data) def client_add(self, data, auth): # ~ auth = AUTH url = self.URL['client'] data = self._get_data_client(auth, data) result = self._post(url, data) if result is None: return False if result.status_code != 200: self._error(f'Code: {result.status_code}') return False if result.text != self.CODES['000']: self._error(result.text) return False return True def client_balance(self, data, rfc=''): url = self.URL['saldo'] host = url.split('/')[2] headers = { 'Content-type': 'text/plain', 'Host': host, 'Connection' : 'Keep-Alive', } data = {'usr': data['user'], 'pwd': data['pass']} try: result = requests.get(url, params=data, headers=headers, timeout=TIMEOUT) except ConnectionError as e: self._error(e) return '' if result.status_code != 200: return '' if result.text == self.CODES['704']: self._error(result.text) return '' if result.text == self.CODES['702']: self._error(result.text) return '' return result.text def client_add_timbres(self, data, auth): # ~ if not auth: # ~ auth = AUTH url = self.URL['timbres'] data = '\n'.join(( f"usr_ws={auth['user']}", f"pwd_ws={auth['pass']}", f"rfc_recibir={data['rfc']}", f"num_timbres={data['timbres']}" )) result = self._post(url, data) if result is None: return False if result.status_code != 200: self._error(f'Code: {result.status_code}') return False if result.text != self.CODES['000']: self._error(result.text) return False return True