265 lines
9.1 KiB
Python
265 lines
9.1 KiB
Python
from trytond.model import fields, ModelView, ModelSQL
|
|
from trytond.pool import PoolMeta, Pool
|
|
from trytond.pyson import Eval
|
|
from trytond.transaction import Transaction
|
|
from trytond.rpc import RPC
|
|
from trytond.exceptions import UserError
|
|
|
|
from facho.fe import form
|
|
from facho import fe
|
|
from facho.fe.client import dian
|
|
|
|
import io
|
|
from datetime import datetime
|
|
|
|
|
|
class Cron(metaclass=PoolMeta):
|
|
__name__ = 'ir.cron'
|
|
|
|
@classmethod
|
|
def __setup__(cls):
|
|
super().__setup__()
|
|
cls.method.selection.extend([
|
|
('account.invoice|fe_delivery', 'FE Delivery')
|
|
])
|
|
|
|
|
|
class Party(metaclass=PoolMeta):
|
|
__name__ = 'party.party'
|
|
|
|
fe_identifiers = fields.One2Many('account_invoice_facho.party.fe_identifier',
|
|
'party', 'Fe Identifiers', required=True)
|
|
fe_organizations = fields.One2Many('account_invoice_facho.party.fe_organization',
|
|
'party', 'Fe Organizations', required=True)
|
|
|
|
@property
|
|
def fe_identifier(self):
|
|
try:
|
|
return self.fe_identifiers[0]
|
|
except IndexError:
|
|
return None
|
|
|
|
@property
|
|
def fe_organization(self):
|
|
try:
|
|
return self.fe_organizations[0]
|
|
except IndexError:
|
|
return None
|
|
|
|
def tofacho(self):
|
|
tax_identifier = self.tax_identifer
|
|
if tax_identifer is None:
|
|
tax_identifer = ''
|
|
return form.Party(
|
|
name = self.name,
|
|
ident = tax_identifer,
|
|
responsability_code = self.fe_identifier.fe_code,
|
|
organization_code = self.fe_organization.fe_code,
|
|
)
|
|
|
|
|
|
class PartyFeOrganization(ModelSQL, ModelView):
|
|
'Party Facturacion Electronica Organizacion'
|
|
__name__ = 'account_invoice_facho.party.fe_organization'
|
|
|
|
name = fields.Char('Name', required=True)
|
|
party = fields.Many2One('party.party', 'Party', ondelete='CASCADE',
|
|
required=True, select=True)
|
|
fe_tipo = fields.Many2One('account_invoice_facho.fe_generic_code',
|
|
'Tipo Organizacion',
|
|
domain=[('resource', '=', 'tipo_organizacion')],
|
|
required=True)
|
|
|
|
@property
|
|
def fe_code(self):
|
|
return self.fe_tipo.code
|
|
|
|
|
|
class PartyFeIdentifier(ModelSQL, ModelView):
|
|
'Party Facturaction Electron Identificador'
|
|
__name__ = 'account_invoice_facho.party.fe_identifier'
|
|
_rec_name = 'code'
|
|
code = fields.Char('Code', required=True)
|
|
party = fields.Many2One('party.party', 'Party', ondelete='CASCADE',
|
|
required=True, select=True)
|
|
fe_tipo = fields.Many2One('account_invoice_facho.fe_generic_code',
|
|
'Tipo Responsabilidad',
|
|
domain=[('resource', '=', 'tipo_responsabilidad')],
|
|
required=True)
|
|
|
|
@property
|
|
def fe_code(self):
|
|
return self.fe_tipo.code
|
|
|
|
|
|
class Invoice(metaclass=PoolMeta):
|
|
__name__ = 'account.invoice'
|
|
|
|
_states = {'invisible': ~Eval('is_fe_colombia', False),
|
|
'readonly': Eval('state') != 'draft'}
|
|
_states_readonly = {'readonly': Eval('state') != 'draft'}
|
|
_depends = ['is_fe_colombia', 'state']
|
|
|
|
is_fe_colombia = fields.Boolean('FE Colombia?',
|
|
states={'invisible': False})
|
|
fe_habilitacion = fields.Boolean('FE Habilitacion?',
|
|
states=_states,
|
|
depends=_depends)
|
|
fe_delivery_state = fields.Selection([
|
|
('draft', 'draft'),
|
|
('queued', 'Queued'), # local encola
|
|
('delivered', 'Delivered'), # remoto encola
|
|
('exception', 'Exception'), # local exception
|
|
('failure', 'Failure'), # remoto fallo
|
|
('done', 'Done') # remoto ok
|
|
], 'Delivery State', states=_states_readonly, depends=_depends)
|
|
fe_delivery_trackid = fields.Char('Delivery TrackID',
|
|
states=_states_readonly,
|
|
depends=_depends)
|
|
fe_delivery_status = fields.Text('Delivery Status',
|
|
states=_states_readonly,
|
|
depends=_depends)
|
|
fe_delivery_checked_at = fields.DateTime('Delivery Checked At',
|
|
states=_states_readonly,
|
|
depends=_depends)
|
|
|
|
del _states, _states_readonly
|
|
del _depends
|
|
|
|
@staticmethod
|
|
def default_fe_delivery_state():
|
|
return 'draft'
|
|
|
|
@staticmethod
|
|
def default_is_fe_colombia():
|
|
return Transaction().context.get('is_fe_colombia', False)
|
|
|
|
@classmethod
|
|
def validate(cls, invoices):
|
|
super().validate(invoices)
|
|
|
|
for invoice in invoices:
|
|
for line in invoice.lines:
|
|
for taxes, unit_price, quantity in line.taxable_lines:
|
|
for tax in taxes:
|
|
if tax.type != 'percentage':
|
|
raise UserError('Solo se soporta impuesto tipo porcentaje para producto')
|
|
|
|
facho_invoice = invoice.tofacho()
|
|
validator = form.DianResolucion0001Validator()
|
|
validator.validate(facho_invoice)
|
|
for (model, field, error) in validator.errors:
|
|
raise UserError('model %s in field %s has %s' % (model, field, error))
|
|
|
|
def tofacho(self):
|
|
inv = form.Invoice()
|
|
inv.set_period(self.invoice_date, self.invoice_date)
|
|
inv.set_issue(self.invoice_date)
|
|
inv.set_ident(self.number)
|
|
inv.set_customer(self.party.tofacho())
|
|
inv.set_supplier(self.company.party.tofacho())
|
|
|
|
for line in self.lines:
|
|
inv.add_invoice_line(line.tofacho())
|
|
|
|
inv.calculate()
|
|
return inv
|
|
|
|
def do_dian_request(self, request, config=None):
|
|
if config is None:
|
|
config = Pool().get('account_invoice_facho.configuration')(1)
|
|
client = dian.DianSignatureClient(config.dian_fe_llave_privada,
|
|
config.dian_fe_llave_publica,
|
|
password=config.dian_fe_llave_frasepaso)
|
|
return client.request(request)
|
|
|
|
def fe_update_status(self):
|
|
req = dian.GetStatusZip
|
|
if self.fe_habilitacion:
|
|
req = dian.Habilitacion.GetStatusZip
|
|
resp = self.do_dian_request(req(trackId = self.fe_delivery_trackid))
|
|
|
|
params = {}
|
|
params['fe_delivery_status'] = resp.StatusDescription
|
|
if resp.IsValid:
|
|
params['fe_delivery_state'] = 'donde'
|
|
else:
|
|
params['fe_delivery_state'] = 'failure'
|
|
|
|
self._force_write(params)
|
|
|
|
def _force_write(self, params):
|
|
self.state = 'draft'
|
|
params['fe_delivery_checked_at'] = datetime.now()
|
|
self.write([self], params)
|
|
|
|
def fe_delivery_test(self):
|
|
config = Pool().get('account_invoice_facho.configuration')(1)
|
|
if self.fe_delivery_state not in ['queued', 'draft']:
|
|
return
|
|
|
|
|
|
facho_invoice = self.tofacho()
|
|
xml_invoice = form.DIANInvoiceXML(facho_invoice)
|
|
|
|
zipdata = io.BytesIO()
|
|
with fe.DianZIP(zipdata) as dianzip:
|
|
dianzip.add_invoice_xml(facho_invoice.invoice_ident, str(xml_invoice))
|
|
|
|
zipdata.seek(0)
|
|
filename = 'invoice_%s' % (facho_invoice.invoice_ident)
|
|
res = self.do_dian_request(dian.Habilitacion.SendTestSetAsync(
|
|
filename, zipdata.read(),
|
|
config.dian_fe_test_setid
|
|
))
|
|
if not res.ZipKey:
|
|
raise UserError(str(res))
|
|
self._force_write({'fe_delivery_state': 'delivered',
|
|
'fe_delivery_trackid': res.ZipKey})
|
|
|
|
def fe_process(self):
|
|
if self.fe_habilitacion:
|
|
# TODO forzar facturas contabilidadas
|
|
self.fe_delivery_test()
|
|
|
|
@classmethod
|
|
def fe_delivery(cls):
|
|
for inv in cls.search([('is_fe_colombia', '=', True),
|
|
('state', 'in', ['posted', 'paid'])]):
|
|
inv.fe_process()
|
|
if inv.fe_delivery_state in ['delivered', 'failure']:
|
|
inv.fe_update_status()
|
|
|
|
|
|
class Product(metaclass=PoolMeta):
|
|
__name__ = 'product.product'
|
|
|
|
def tofacho(self):
|
|
return form.StandardItem(
|
|
description = self.name,
|
|
id = self.code
|
|
)
|
|
|
|
|
|
class InvoiceLine(metaclass=PoolMeta):
|
|
__name__ = 'account.invoice.line'
|
|
|
|
def tofacho(self):
|
|
tax_subtotals = []
|
|
for taxes, unit_price, quantity in self.taxable_lines:
|
|
for tax in taxes:
|
|
tax_subtotals.append(form.TaxSubTotal(
|
|
percent = tax.rate
|
|
))
|
|
|
|
return form.InvoiceLine(
|
|
quantity = self.quantity,
|
|
description = self.description,
|
|
# TODO debe ser decimal
|
|
price_amount = float(self.unit_price),
|
|
item = self.product.tofacho(),
|
|
tax = form.TaxTotal(
|
|
subtotals = tax_subtotals
|
|
)
|
|
)
|