323 lines
9.2 KiB
Python
323 lines
9.2 KiB
Python
#!/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 <http://www.gnu.org/licenses/>.
|
|
|
|
|
|
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 {}
|