elmau
/
cfdi-pac
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Cancel xml in Finkok

This commit is contained in:
Mauricio Baeza 2021-01-01 22:44:27 -06:00
parent e432200a35
commit f56a44dc0a
4 changed files with 110 additions and 138 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

84
source/finkok/finkok.py
View File

@ -16,7 +16,7 @@
# ~ You should have received a copy of the GNU General Public License
# ~ along with this program. If not, see <http://www.gnu.org/licenses/>.
import base64
import datetime
import logging
import os
@ -47,7 +47,7 @@ logging.getLogger('zeep').setLevel(logging.ERROR)
TIMEOUT = 10
DEBUG_SOAP = False
DEBUG_SOAP = True
class DebugPlugin(Plugin):
@ -102,6 +102,9 @@ class PACFinkok(object):
def _validate_result(self, result):
if hasattr(result, 'CodEstatus'):
ce = result.CodEstatus
if ce is None:
return result
if ce == self.CODE['IP']:
self._error = self.CODE['IPMSG']
return {}
@ -163,7 +166,8 @@ class PACFinkok(object):
'xml': cfdi.encode('utf-8'),
}
result = self._get_result(client, 'stamp', args)
if self._error:
if self.error:
log.error(self.error)
return ''
data = {
@ -172,3 +176,77 @@ class PACFinkok(object):
'date': result.Fecha,
}
return data
def _get_data_cancel(self, cfdi):
NS_CFDI = {
'cfdi': 'http://www.sat.gob.mx/cfd/3',
'tdf': 'http://www.sat.gob.mx/TimbreFiscalDigital',
}
tree = ET.fromstring(cfdi.encode())
rfc_emisor = tree.xpath(
'string(//cfdi:Comprobante/cfdi:Emisor/@Rfc)',
namespaces=NS_CFDI)
cfdi_uuid = tree.xpath(
'string(//cfdi:Complemento/tdf:TimbreFiscalDigital/@UUID)',
namespaces=NS_CFDI)
return rfc_emisor, cfdi_uuid
def cancel(self, cfdi, info, auth={}):
if not auth:
auth = AUTH
rfc_emisor, cfdi_uuid = self._get_data_cancel(cfdi)
method = 'cancel'
client = Client(self.URL[method],
transport=self._transport, plugins=self._plugins)
uuid_type = client.get_type('ns1:UUIDS')
sa = client.get_type('ns0:stringArray')
args = {
'UUIDS': uuid_type(uuids=sa(string=cfdi_uuid)),
'username': auth['user'],
'password': auth['pass'],
'taxpayer_id': rfc_emisor,
'cer': info['cer'],
'key': info['key'],
'store_pending': False,
}
result = self._get_result(client, 'cancel', args)
if self.error:
log.error(self.error)
return ''
folio = result['Folios']['Folio'][0]
status = folio['EstatusUUID']
if status != '201':
log.debug(f'Cancel status: {status} - {cfdi_uuid}')
data = {
'acuse': result['Acuse'],
'date': result['Fecha'],
}
return data
def cancel_xml(self, xml, auth={}):
if not auth:
auth = AUTH
method = 'cancel'
client = Client(self.URL[method],
transport=self._transport, plugins=self._plugins)
client.set_ns_prefix('can', 'http://facturacion.finkok.com/cancel')
xml = f'<?xml version="1.0" encoding="UTF-8" standalone="yes"?>\n{xml}'
args = {
'xml': base64.b64encode(xml.encode()),
'username': auth['user'],
'password': auth['pass'],
'store_pending': False,
}
result = self._get_result(client, 'cancel_signature', args)
if self.error:
log.error(self.error)
return ''
return result

26
source/finkok/finkok1.py
View File

@ -40,32 +40,6 @@ DEBUG_SOAP = False
class PACFinkok(object):
def cfdi_cancel(self, rfc, uuid, cer, key, auth={}):
if not auth:
auth = FINKOK['AUTH']
method = 'cancel'
client = Client(
self.URL[method], transport=self._transport, plugins=self._plugins)
uuid_type = client.get_type('ns1:UUIDS')
sa = client.get_type('ns0:stringArray')
args = {
'UUIDS': uuid_type(uuids=sa(string=uuid)),
'username': auth['USER'],
'password': auth['PASS'],
'taxpayer_id': rfc,
'cer': cer,
'key': key,
'store_pending': False,
}
result = self._get_result(client, 'cancel', args)
if self.error:
return {}
return result
def cfdi_status(self, uuid, auth={}):
if not auth:
auth = FINKOK['AUTH']

58
source/finkok/finkok2.py
View File

@ -90,64 +90,6 @@ class Finkok(object):
self.error = str(e)
return ''
def cancel_xml(self, rfc, uuid, cer, key):
# ~ for u in uuids:
# ~ if not self._validate_uuid(u):
# ~ return ''
method = 'cancel'
client = Client(
URL[method], transport=self._transport, plugins=self._plugins)
uuid_type = client.get_type('ns1:UUIDS')
sa = client.get_type('ns0:stringArray')
args = {
'UUIDS': uuid_type(uuids=sa(string=uuid)),
'username': self._auth['USER'],
'password': self._auth['PASS'],
'taxpayer_id': rfc,
'cer': cer,
'key': key,
'store_pending': False,
}
try:
result = client.service.cancel(**args)
except Fault as e:
self.error = str(e)
return ''
if result.CodEstatus and self.codes['205'] in result.CodEstatus:
self.error = result.CodEstatus
return ''
return result
def cancel_signature(self, file_xml):
method = 'cancel'
if os.path.isfile(file_xml):
root = etree.parse(file_xml).getroot()
else:
root = etree.fromstring(file_xml.encode())
xml = etree.tostring(root)
client = Client(
URL[method], transport=self._transport, plugins=self._plugins)
args = {
'username': self._auth['USER'],
'password': self._auth['PASS'],
'xml': xml,
'store_pending': False,
}
try:
result = client.service.cancel_signature(**args)
return result
except Fault as e:
self.error = str(e)
return ''
def get_acuse(self, rfc, uuids, type_acuse='C'):
for u in uuids:
if not self._validate_uuid(u):

80
source/tests/tests_finkok.py
View File

@ -121,6 +121,14 @@ class TestCfdi(object):
key = base64.b64encode(self._key_ori).decode()
return key, cer
@property
def cer_pem(self):
return self._cert.cer_pem
@property
def key_pem(self):
return self._cert.key_pem
class TestStamp(unittest.TestCase):
@ -140,34 +148,36 @@ class TestStamp(unittest.TestCase):
expected = '201'
cfdi = TestCfdi()
result = self.pac.stamp(cfdi.xml)
cfdi_uuid = self.pac.cfdi_uuid
cfdi_xml = result['xml']
cfdi_uuid = result['uuid']
self.assertFalse(bool(self.pac.error))
self.assertTrue(bool(uuid.UUID(cfdi_uuid)))
time.sleep(1)
cert = cfdi.cert
time.sleep(3)
info = {
'key': cert[0],
'cer': cert[1],
'pass': '12345678a',
'tipo': 'cfdi3.3',
'cer': cfdi.cer_pem,
'key': cfdi.key_pem,
}
result = self.pac.cancel(result, info)
result = self.pac.cancel(cfdi_xml, info)
self.assertFalse(bool(self.pac.error))
tree = ET.fromstring(result)
cancel_uuid = tree.xpath('string(//Acuse/Folios/UUID)')
status = tree.xpath('string(//Acuse/Folios/EstatusUUID)')
tree = ET.fromstring(result['acuse'].encode())
NS = {'s': 'http://schemas.xmlsoap.org/soap/envelope/'}
cancel_uuid = tree.xpath('string(//s:Body/*/*/*[1]/*[1])', namespaces=NS)
status = tree.xpath('string(//s:Body/*/*/*[1]/*[2])', namespaces=NS)
self.assertEqual(cfdi_uuid, cancel_uuid)
self.assertEqual(status, expected)
return
def test_cfdi_cancel_xml(self):
expected = '201'
cfdi = TestCfdi()
result = self.pac.stamp(cfdi.xml)
cfdi_uuid = self.pac.cfdi_uuid
cfdi_uuid = result['uuid']
self.assertFalse(bool(self.pac.error))
self.assertTrue(bool(uuid.UUID(cfdi_uuid)))
@ -176,7 +186,7 @@ class TestStamp(unittest.TestCase):
'cfdi': 'http://www.sat.gob.mx/cfd/3',
'tdf': 'http://www.sat.gob.mx/TimbreFiscalDigital',
}
tree = ET.fromstring(result.encode())
tree = ET.fromstring(result['xml'].encode())
rfc_emisor = tree.xpath(
'string(//cfdi:Comprobante/cfdi:Emisor/@Rfc)',
namespaces=NS_CFDI)
@ -189,10 +199,12 @@ class TestStamp(unittest.TestCase):
}
template = TEMPLATE_CANCEL.format(**data)
sign_xml = cfdi.sign_xml(template)
info = {
'tipo': 'cfdi3.3',
}
result = self.pac.cancel_xml(result, sign_xml, info)
time.sleep(5)
result = self.pac.cancel_xml(sign_xml)
self.assertFalse(bool(self.pac.error))
tree = ET.fromstring(result)
uid = tree.xpath('string(//Acuse/Folios/UUID)')
status = tree.xpath('string(//Acuse/Folios/EstatusUUID)')
@ -200,40 +212,6 @@ class TestStamp(unittest.TestCase):
self.assertEqual(cfdi_uuid, uid)
self.assertEqual(status, expected)
def test_cfdi_status(self):
expected = ''
cfdi = TestCfdi()
result = self.pac.stamp(cfdi.xml)
cfdi_uuid = self.pac.cfdi_uuid
self.assertFalse(bool(self.pac.error))
self.assertTrue(bool(uuid.UUID(cfdi_uuid)))
NS_CFDI = {
'cfdi': 'http://www.sat.gob.mx/cfd/3',
'tdf': 'http://www.sat.gob.mx/TimbreFiscalDigital',
}
tree = ET.fromstring(result.encode())
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)
total = tree.xpath(
'string(//cfdi:Comprobante/@Total)',
namespaces=NS_CFDI)
time.sleep(3)
data = {
'rfc_receptor': rfc_receptor,
'rfc_emisor': rfc_emisor,
'total': total,
'uuid': cfdi_uuid,
}
result = self.pac.status(data)
self.assertEqual(result, expected)
if __name__ == '__main__':
unittest.main()