Skip to content

GitLab

Commit 7917cd20 authored by Cuauhtémoc Díaz Minor's avatar Cuauhtémoc Díaz Minor
Browse files
Options

feat(xpd_tu_factura): nuevo esquema de cancelacion 

se crean las funciones:
\t`consult_xpd_cfdi()`: para consultar el estado de un
cfdi
\t`cancel_xpd_cfdi()`: para cancelar un cfdi
parent 086653fd
Hide whitespace changes
Inline Side-by-side
Showing with 197 additions and 215 deletions
l10n_mx_params_pac/lib/__init__.py
View file @ 7917cd20
# -*- coding: utf-8 -*-
from . import pac_answer
from . import xpd_envelopes
\ No newline at end of file
l10n_mx_params_pac/lib/xpd_envelopes.py 0 → 100644
View file @ 7917cd20
# -*- coding: utf-8 -*-
_SOAPENV = """<?xml version="1.0" encoding="UTF-8"?>
<soapenv:Envelope
xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/"
xmlns:ser="%s">
<soapenv:Header/>
<soapenv:Body>
<ser:timbrar>
<usuario>%s</usuario>
<contrasena>%s</contrasena>
<cfdi>%s</cfdi>
</ser:timbrar>
</soapenv:Body>
</soapenv:Envelope>
"""
_SOAPENV_CANCEL = '''<?xml version="1.0" encoding="UTF-8"?>
<soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/"
xmlns:soap="http://concretepage.com/soap">
<soapenv:Header/>
<soapenv:Body>
<soap:cancelarCfdiRequest>
<soap:usuario>%s</soap:usuario>
<soap:contrasena>%s</soap:contrasena>
<soap:parametros>
<soap:noCertificado>%s</soap:noCertificado>
<soap:rfcEmisor>%s</soap:rfcEmisor>
<soap:rfcReceptor>%s</soap:rfcReceptor>
<soap:total>%s</soap:total>
<soap:uuid>%s</soap:uuid>
<soap:xmlB64></soap:xmlB64>
</soap:parametros>
</soap:cancelarCfdiRequest>
</soapenv:Body>
</soapenv:Envelope>
'''
_SOAPENV_CONSULTAR = """<?xml version="1.0" encoding="UTF-8"?>
<soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/"
xmlns:soap="http://concretepage.com/soap">
<soapenv:Header/>
<soapenv:Body>
<soap:consultarCfdiRequest>
<soap:usuario>%s</soap:usuario>
<soap:contrasena>%s</soap:contrasena>
<!--Zero or more repetitions:-->
<soap:parametros>
<!--Optional:-->
<soap:noCertificado>%s</soap:noCertificado>
<!--Optional:-->
<soap:rfcEmisor>%s</soap:rfcEmisor>
<!--Optional:-->
<soap:rfcReceptor>%s</soap:rfcReceptor>
<!--Optional:-->
<soap:total>%s</soap:total>
<!--Optional:-->
<soap:uuid>%s</soap:uuid>
<!--Optional:-->
<soap:xmlB64></soap:xmlB64>
</soap:parametros>
</soap:consultarCfdiRequest>
</soapenv:Body>
</soapenv:Envelope>
"""
l10n_mx_params_pac/pacs/expide_tu_factura.py
View file @ 7917cd20
......@@ -12,37 +12,25 @@ from pytz import timezone, utc
from requests import Request, Session
import xmltodict
from SOAPpy import WSDL
from SOAPpy import Types, WSDL
from openerp import api, fields, models
from openerp.addons.l10n_mx_ir_attachment_facturae.lib import pyxmldsig
from openerp.exceptions import Warning as UserError
from openerp.tools.translate import _
from ..lib.pac_answer import PacCancelAnswer
from ..lib.pac_answer import PacCancelAnswer, PacConsultAnswer
from ..models import exceptions as pac_exceptions
from ..lib import xpd_envelopes
_logger = logging.getLogger(__name__)
TIMEOUT = 60
_ACTIONS = {"timbrar": "timbrar"}
_SOAPENV = """<?xml version="1.0" encoding="UTF-8"?>
<soapenv:Envelope
xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/"
xmlns:ser="%s">
<soapenv:Header/>
<soapenv:Body>
<ser:timbrar>
<usuario>%s</usuario>
<contrasena>%s</contrasena>
<cfdi>%s</cfdi>
</ser:timbrar>
</soapenv:Body>
</soapenv:Envelope>
"""
_ACTIONS = {"timbrar": "timbrar",
"cancelar": "",
"consultar": ""}
class ParamsPac(models.Model):
......@@ -80,9 +68,20 @@ class ParamsPac(models.Model):
factura_mx_type__fc = super(ParamsPac, self).get_driver_fc_cancel()
if factura_mx_type__fc is None:
factura_mx_type__fc = {}
factura_mx_type__fc.update({"pac_facturalo": self.cancel_file})
factura_mx_type__fc.update({"pac_facturalo": self.cancel_xpd_cfdi})
return factura_mx_type__fc
def get_driver_fc_consult(self):
""" Implement driver for consult cfdi's"""
factura_mx_type__fc = super(ParamsPac, self).get_driver_fc_cancel()
if factura_mx_type__fc is None:
factura_mx_type__fc = {}
factura_mx_type__fc.update({"pac_facturalo": self.consult_xpd_cfdi})
return factura_mx_type__fc
def create_header(self, method):
return {"SOAPAction": "%s" % (_ACTIONS[method]), "Content-Type": "text/xml;charset=UTF-8"}
@api.one
def sign_file(self, fdata=None):
"""
......@@ -94,10 +93,11 @@ class ParamsPac(models.Model):
if self.user == "pruebas":
msg += _("WARNING, SIGNED IN TEST!!!!\n\n")
data = _SOAPENV % (self.namespace_sign, self.user, self.password, fdata)
data = xpd_envelopes._SOAPENV % (
self.namespace_sign, self.user, self.password, fdata
)
_logger.debug(data)
headers = {"SOAPAction": "%s" % (_ACTIONS[method]), "Content-Type": "text/xml"}
headers =self.create_header(method)
s = Session()
req = Request("POST", self.url_webservice_sign, data=data, headers=headers)
prepped = req.prepare()
......@@ -155,209 +155,125 @@ class ParamsPac(models.Model):
raise Exception(" ".join(["Code", resultado["codigo"], mensaje]))
@api.model
def cancel_file(self, uuid):
"""
Cancel xml file onto PAC webservice
@params
id:s Id from invoice to validate
def consult_xpd_cfdi(self, uuid):
""" Consul the cfdi status with XPD pac
:param uuid: UUID of CFDI to consult
"""
method = "consultar"
msg = ""
xml_data = self.cancel_xml(uuid)
# Send file to PAC for cancelation
if self.user == "pruebas":
msg += _("WARNING, CANCEL IN TEST!!!!\n\n")
cfdi = self.env['ir.attachment.facturae.mx'].search([["uuid", "=", uuid]])
try:
from SOAPpy import Types
wsdl = WSDL.SOAPProxy(self.url_webservice_cancel, self.namespace_cancel)
# Webservice configuration
wsdl.soapproxy.config.dumpSOAPOut = 0
wsdl.soapproxy.config.dumpSOAPIn = 0
wsdl.soapproxy.config.debug = 0
wsdl.config.preventescape = 1
wsdl.soapproxy.config.dict_encoding = "UTF-8"
# SUBIR factura al PAC
resultado = wsdl.cancelar(
usuario=self.user,
token=self.password,
xmlBytes=Types.base64BinaryType(xml_data),
data = xpd_envelopes._SOAPENV_CONSULTAR % (
self.user,
self.password,
cfdi.no_certificado.encode("utf-8"),
cfdi.rfc_emisor.encode("utf-8"),
cfdi.rfc_receptor.encode("utf-8"),
cfdi.total,
cfdi.uuid.encode("utf-8").lower(),
)
except Exception as e:
e.message="Error al crear ENVELOPE"
_logger.info(e)
raise e
else:
_logger.debug(data)
headers =self.create_header(method)
s = Session()
req = Request("POST", self.url_webservice_cancel, data=data, headers=headers)
prepped = req.prepare()
try:
# Timbrar factura
response = s.send(prepped, timeout=TIMEOUT)
res = xmltodict.parse(response.text)
_logger.debug(res)
except Exception as e:
# Error al establecer comunicación con el PAC
raise UserError(_("PAC communication failed. \n%s") % str(e))
cancel_answer = PacCancelAnswer(
uuid, resultado["codEstatus"], resultado["estatusUUIDs"]
)
_logger.debug(e)
raise e
# Procesar los resultados obtenidos del PAC
if cancel_answer.estatus not in "200":
raise Exception(
" ".join(["Code", cancel_answer.estatus, resultado["codMensaje"]])
)
elif cancel_answer.estatus_cancelacion in pac_exceptions.cancel_error_list:
raise pac_exceptions.CancelError(
code=cancel_answer.estatus_cancelacion,
message=resultado["codMensaje"],
)
# If code result is 200 then all was okay
else:
return cancel_answer
def binary2file(
self, cr, uid, ids, binary_data=False, file_prefix="", file_suffix=""
):
"""
@param binary_data : Field binary with the information of certificate
of the company
@param file_prefix : Name to be used for create the file with the
information of certificate
@file_suffix : Sufix to be used for the file that create in
this function
"""
import os
try:
resultado = res["soapenv:Envelope"]["soapenv:Body"]["ns2:consultarCfdiResponse"][
"ns2:consultarCfdiResponse"] # noqa
except KeyError as e:
resultado = res["soapenv:Envelope"]["soapenv:Body"]["ns2:consultarCfdiResponse"][
"ns2:consultarCfdiResponse"
] # noqa
_logger.debug(e)
except Exception as e:
_logger.debug(e)
raise UserError(_("Unknow answer.\n%s") % resultado)
(fileno, fname) = tempfile.mkstemp(file_suffix, file_prefix)
f = open(fname, "wb")
f.write(binary_data)
f.close()
os.close(fileno)
return fname
res = resultado["ns2:responseGenericoQr"]["ns2:servicio"]
res = res.popitem(True)
res = res[1]
res = xmltodict.parse(res.encode("utf-8"))
resultado = res["s:Envelope"]["s:Body"]["ConsultaResponse"]["ConsultaResult"]
estatus_cancelacion = resultado["a:EstatusCancelacion"]
estado_cfdi = resultado["a:Estado"]
consult_answer = PacConsultAnswer(uuid, estado_cfdi, estatus_cancelacion)
return consult_answer
@api.model
def cancel_xml(self, uuid):
""" This function call at cancel_invoice_dict to cancel a invoice.
@params
default params
def cancel_xpd_cfdi(self, uuid):
""" Cancel CFDI
:param uuid: uuid of cfdi to cancel
:type uuid: string
"""
self.ensure_one()
data_xml = self.cancel_dict(uuid)
new_file = tempfile.NamedTemporaryFile(delete=False)
new_file.write(data_xml)
new_file.close()
cert = self.company_id.certificate_id
# Get certificate files for sign the xml
fname_cer_pem = fname_key_pem = False
name = "openerp_" + (cert.serial_number or "") + "__certificate__"
if cert:
try:
fname_cer_pem = self.binary2file(
cert.certificate_file_pem, name, ".cer.pem"
)
except: # noqa
raise UserError(_("Not captured a CERTIFICATE file in the company!"))
try:
fname_key_pem = self.binary2file(
cert.certificate_key_file_pem, name, ".key.pem"
)
except: # noqa
raise UserError(
_("Not captured a KEY file in format PEM, in the " "company!")
)
method = "cancelar"
msg = ""
cfdi = self.env['ir.attachment.facturae.mx'].search([["uuid", "=", uuid]])
try:
data = xpd_envelopes._SOAPENV_CANCEL % (
self.user,
self.password,
cfdi.no_certificado.encode("utf-8"),
cfdi.rfc_emisor.encode("utf-8"),
cfdi.rfc_receptor.encode("utf-8"),
cfdi.total,
cfdi.uuid.encode("utf-8").lower(),
)
except Exception as e:
e.message="Error while create ENVELOPE"
_logger.info(e)
raise e
else:
raise UserError(
_(
"There is no VALID certificate for company: {company}!\n"
"It is possible that the CERTIFICATE is already expired.\n"
"Go to Configuration >> Companies >> Companies to verify "
"it."
).format(company=self.company_id.name)
_logger.debug(data)
headers =self.create_header(method)
s = Session()
req = Request("POST", self.url_webservice_cancel, data=data, headers=headers)
prepped = req.prepare()
try:
# Timbrar factura
response = s.send(prepped, timeout=TIMEOUT)
res = xmltodict.parse(response.text)
_logger.debug(res)
except Exception as e:
# Error al establecer comunicación con el PAC
_logger.debug(e)
raise e
# Procesar los resultados obtenidos del PAC
try:
resultado = res["soapenv:Envelope"]["soapenv:Body"]["ns2:cancelarCfdiResponse"][
"ns2:cancelarCfdiResponse"
] # noqa
except KeyError as e:
resultado = res["soapenv:Envelope"]["soapenv:Body"]["ns2:cancelarCfdiResponse"][
"ns2:canselarCfdiResponse"
] # noqa
_logger.debug(e)
except Exception as e:
_logger.debug(e)
raise UserError(_("Unknow answer.\n%s") % resultado)
mensaje = resultado["ns2:mensaje"]
codigo = resultado["ns2:codigo"]
# Procesar los resultados obtenidos del PAC
if codigo not in "200":
raise Exception(
" ".join(["Code", codigo, mensaje])
)
signed_xml = pyxmldsig.sign_file(
template_file=new_file.name,
cert_file=fname_cer_pem.encode("ascii", "ignore"),
key_file=fname_key_pem.encode("ascii", "ignore"),
password=cert.certificate_password.encode("ascii", "ignore"),
)
_logger.debug("Signed file %s", signed_xml)
return signed_xml # data_dict
@api.model
def cancel_dict(self, uuid):
""" This function create a dictionary with the necesary
data to create the DigestValue.
"""
# get user's timezone
tz = timezone(self.env.user.partner_id.tz) or utc
# get localized dates
date = datetime.now(utc).astimezone(tz)
xmlns = "http://cancelacfd.sat.gob.mx"
xsd = "http://www.w3.org/2001/XMLSchema"
xsi = "http://www.w3.org/2001/XMLSchema-instance"
signed_xmlns = "http://www.w3.org/2000/09/xmldsig#"
nsmap = {None: xmlns, "xsd": xsd, "xsi": xsi}
signed_nsmap = {None: signed_xmlns, "xsd": xsd, "xsi": xsi}
# section: Cancelation
# create XML
cancelacion = etree.Element("Cancelacion", nsmap=nsmap)
cancelacion.set("Fecha", date.strftime("%Y-%m-%dT%H:%M:%S"))
cancelacion.set("RfcEmisor", self.company_id.partner_id.vat_split)
folios = etree.Element("Folios")
uuid_to_cancel = etree.Element("UUID")
uuid_to_cancel.text = uuid
folios.append(uuid_to_cancel)
cancelacion.append(folios)
signature = etree.Element("Signature", nsmap={None: signed_xmlns})
signedinfo = etree.Element("SignedInfo", nsmap=signed_nsmap)
canonicalizacion = etree.Element("CanonicalizationMethod")
canonicalizacion.set(
"Algorithm", "http://www.w3.org/TR/2001/REC-xml-c14n-20010315"
)
signaturemethod = etree.Element("SignatureMethod")
signaturemethod.set("Algorithm", "http://www.w3.org/2000/09/xmldsig#rsa-sha1")
reference = etree.Element("Reference")
reference.set("URI", "")
transforms = etree.Element("Transforms")
transform = etree.Element("Transform")
transform.set(
"Algorithm", "http://www.w3.org/2000/09/xmldsig#enveloped-signature"
)
digestme = etree.Element("DigestMethod")
digestme.set("Algorithm", "http://www.w3.org/2000/09/xmldsig#sha1")
digestva = etree.Element("DigestValue")
digestva.text = "TEMPLATE"
signedinfo.append(canonicalizacion)
signedinfo.append(signaturemethod)
signedinfo.append(reference)
transforms.append(transform)
reference.append(transforms)
reference.append(digestme)
reference.append(digestva)
signature.append(signedinfo)
signaturevalue = etree.Element("SignatureValue")
signaturevalue.text = "TEMPLATE"
signature.append(signaturevalue)
keyinfo = etree.Element("KeyInfo")
# Issuername & Serialnumber not added because library
# auto fill the fields on X509IssuerSerial
# see www.aleksey.com/pipermail/xmlsec/2007/008125.html
x509 = etree.Element("X509Data")
xiserial = etree.Element("X509IssuerSerial")
xiserial.text = ""
xcertificate = etree.Element("X509Certificate")
xcertificate.text = ""
x509.append(xiserial)
x509.append(xcertificate)
keyinfo.append(x509)
signature.append(keyinfo)
cancelacion.append(signature)
# pretty string
_logger.debug("XML file %s", etree.tostring(cancelacion))
return etree.tostring(cancelacion)
consulta = self.consult_xpd_cfdi(uuid)
answer = PacCancelAnswer(uuid, codigo, consulta.estatus_cancelacion)
return answer
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment