Datalife
/
trytond-aeat_sii
mirror of https://gitlab.com/datalifeit/trytond-aeat_sii
1
0
Fork 0
Code Issues 1 Projects Releases Wiki Activity

Calculate SII by cron

This commit is contained in:
Raimon Esteve 2020-11-16 10:18:00 +01:00
parent a408dbf706
commit 2b7335c1ff
10 changed files with 350 additions and 310 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
__init__.py
View File

@ -1,6 +1,7 @@
# The COPYRIGHT file at the top level of this repository contains the full
# copyright notices and license terms.
from trytond.pool import Pool
from . import cron
from . import invoice
from . import aeat
from . import party
@ -14,8 +15,10 @@ from . import purchase
def register():
Pool.register(
account.Configuration,
account.TemplateTax,
account.Tax,
cron.Cron,
party.Party,
party.PartyIdentifier,
company.Company,

38
account.py
View File

@ -2,12 +2,48 @@
# copyright notices and license terms.
from trytond.model import fields
from trytond.pool import PoolMeta
from trytond.pyson import Eval
from .aeat import (BOOK_KEY, OPERATION_KEY, SEND_SPECIAL_REGIME_KEY,
RECEIVE_SPECIAL_REGIME_KEY, IVA_SUBJECTED, EXCEMPTION_CAUSE,
INTRACOMUNITARY_TYPE)
__all__ = ['TemplateTax', 'Tax']
__all__ = ['Configuration', 'TemplateTax', 'Tax']
class Configuration(metaclass=PoolMeta):
__name__ = 'account.configuration'
aeat_pending_sii = fields.Boolean('AEAT Pending SII',
help='Automatically generate AEAT Pending SII reports by cron')
aeat_pending_sii_send = fields.Boolean('AEAT Pending SII Send',
states={
'invisible': ~Eval('aeat_pending_sii', False),
}, depends=['aeat_pending_sii'],
help='Automatically send AEAT Pending SII reports by cron')
aeat_received_sii = fields.Boolean('AEAT Received SII',
help='Automatically generate AEAT Received SII reports by cron')
aeat_received_sii_send = fields.Boolean('AEAT Received SII Send',
states={
'invisible': ~Eval('aeat_received_sii', False),
}, depends=['aeat_received_sii'],
help='Automatically send AEAT Received SII reports by cron')
@staticmethod
def default_aeat_pending_sii():
return False
@staticmethod
def default_aeat_received_sii():
return False
@staticmethod
def default_aeat_pending_sii_send():
return False
@staticmethod
def default_aeat_received_sii_send():
return False
class TemplateTax(metaclass=PoolMeta):

8
account.xml
View File

@ -3,6 +3,14 @@
copyright notices and license terms. -->
<tryton>
<data>
<!-- account.configuration -->
<record model="ir.ui.view" id="configuration_view_form">
<field name="model">account.configuration</field>
<field name="inherit" ref="account.configuration_view_form"/>
<field name="name">configuration_form</field>
</record>
<!-- account.tax -->
<record model="ir.ui.view" id="aeat_sii_tax_form_view">
<field name="model">account.tax</field>
<field name="inherit" ref="account.tax_view_form"/>

273
aeat.py
View File

@ -8,6 +8,7 @@ from datetime import datetime
from zeep import helpers
import json
from collections import namedtuple
from ast import literal_eval
from pyAEATsii import service
from pyAEATsii import mapping
@ -20,6 +21,7 @@ from trytond.transaction import Transaction
from trytond.config import config
from trytond.i18n import gettext
from trytond.exceptions import UserError
from trytond.tools import grouped_slice
from . import tools
@ -38,7 +40,7 @@ _ZERO = Decimal('0.0')
# AEAT SII test
SII_TEST = config.getboolean('aeat', 'sii_test', default=True)
MAX_SII_LINES = config.getint('aeat', 'sii_lines', default=300)
def _decimal(x):
return Decimal(x) if x is not None else None
@ -940,6 +942,252 @@ class SIIReport(Workflow, ModelSQL, ModelView):
if pagination == 'S':
self.query_recieved_invoices(last_invoice=last_invoice)
@classmethod
def get_issued_sii_reports(cls):
pool = Pool()
Invoice = pool.get('account.invoice')
SIIReportLine = pool.get('aeat.sii.report.lines')
issued_invoices = {
'A0': {}, # 'A0', 'Registration of invoices/records'
'A1': {}, # 'A1', 'Amendment of invoices/records (registration errors)'
'D0': {}, # 'D0', 'Delete Invoices'
}
issued_invs = Invoice.search([
('sii_pending_sending', '=', True),
('sii_state', '=', 'Correcto'),
('sii_header', '!=', None),
('type', '=', 'out'),
])
# search issued invoices [delete]
delete_issued_invoices = []
# search issued invoices [modify]
modify_issued_invoices = []
for issued_inv in issued_invs:
if not issued_inv.sii_records:
continue
sii_record_id = max([s.id for s in issued_inv.sii_records])
sii_record = SIIReportLine(sii_record_id)
if issued_inv.sii_header:
if (literal_eval(issued_inv.sii_header) ==
literal_eval(sii_record.sii_header)):
modify_issued_invoices.append(issued_inv)
else:
delete_issued_invoices.append(issued_inv)
periods = {}
for invoice in delete_issued_invoices:
period = invoice.move.period
if period in periods:
periods[period].append(invoice,)
else:
periods[period] = [invoice]
issued_invoices['D0'] = periods
periods2 = {}
for invoice in modify_issued_invoices:
period = invoice.move.period
if period in periods2:
periods2[period].append(invoice,)
else:
periods2[period] = [invoice]
issued_invoices['A1'] = periods2
# search issued invoices [new]
new_issued_invoices = Invoice.search([
('sii_state', 'in', (None, 'Incorrecto')),
('sii_pending_sending', '=', True),
('type', '=', 'out'),
])
# search possible deleted invoices in SII and not uploaded again
new_issued_invoices += Invoice.search([
('sii_state', '=', 'Anulada'),
('sii_pending_sending', '=', True),
('type', '=', 'out'),
('state', 'in', ['paid', 'posted']),
])
new_issued_invoices += delete_issued_invoices
periods1 = {}
for invoice in new_issued_invoices:
period = invoice.move.period
if period in periods1:
periods1[period].append(invoice,)
else:
periods1[period] = [invoice]
issued_invoices['A0'] = periods1
book_type = 'E' # Issued
return cls.create_sii_book(issued_invoices, book_type)
@classmethod
def get_received_sii_reports(cls):
pool = Pool()
Invoice = pool.get('account.invoice')
SIIReportLine = pool.get('aeat.sii.report.lines')
received_invoices = {
'A0': {}, # 'A0', 'Registration of invoices/records'
'A1': {}, # 'A1', 'Amendment of invoices/records (registration errors)'
'D0': {}, # 'D0', 'Delete Invoices'
}
received_invs = Invoice.search([
('sii_pending_sending', '=', True),
('sii_state', '=', 'Correcto'),
('sii_header', '!=', None),
('type', '=', 'in'),
])
# search received invoices [delete]
delete_received_invoices = []
# search received invoices [modify]
modify_received_invoices = []
for received_inv in received_invs:
if not received_inv.sii_records:
continue
sii_record_id = max([s.id for s in received_inv.sii_records])
sii_record = SIIReportLine(sii_record_id)
if received_inv.sii_header:
if (literal_eval(received_inv.sii_header) ==
literal_eval(sii_record.sii_header)):
modify_received_invoices.append(received_inv)
else:
delete_received_invoices.append(received_inv)
periods2 = {}
for invoice in modify_received_invoices:
period = invoice.move.period
if period in periods2:
periods2[period].append(invoice,)
else:
periods2[period] = [invoice]
received_invoices['A1'] = periods2
periods = {}
for invoice in delete_received_invoices:
period = invoice.move.period
if period in periods:
periods[period].append(invoice,)
else:
periods[period] = [invoice]
received_invoices['D0'] = periods
# search received invoices [new]
new_received_invoices = Invoice.search([
('sii_state', 'in', (None, 'Incorrecto')),
('sii_pending_sending', '=', True),
('type', '=', 'in'),
])
# search possible deleted invoices in SII and not uploaded again
new_received_invoices += Invoice.search([
('sii_state', '=', 'Anulada'),
('sii_pending_sending', '=', True),
('type', '=', 'in'),
('state', 'in', ['paid', 'posted']),
])
new_received_invoices += delete_received_invoices
periods1 = {}
for invoice in new_received_invoices:
period = invoice.move.period
if period in periods1:
periods1[period].append(invoice,)
else:
periods1[period] = [invoice]
received_invoices['A0'] = periods1
book_type = 'R' # Received
return cls.create_sii_book(received_invoices, book_type)
@classmethod
def create_sii_book(cls, book_invoices, book):
pool = Pool()
SIIReport = pool.get('aeat.sii.report')
SIIReportLine = pool.get('aeat.sii.report.lines')
Company = Pool().get('company.company')
company = Transaction().context.get('company')
company = Company(company)
company_vat = company.party.sii_vat_code
cursor = Transaction().connection.cursor()
report_line_table = SIIReportLine.__table__()
reports = []
for operation in ['D0', 'A1', 'A0']:
values = book_invoices[operation]
delete = True if operation == 'D0' else False
for period, invoices in values.items():
for invs in grouped_slice(invoices, MAX_SII_LINES):
report = SIIReport()
report.company = company
report.company_vat = company_vat
report.fiscalyear = period.fiscalyear
report.period = period
report.operation_type = operation
report.book = book
report.save()
reports.append(report)
values = []
for inv in invs:
sii_header = str(inv.get_sii_header(inv, delete))
values.append([report.id, inv.id, sii_header, company.id])
cursor.execute(*report_line_table.insert(
columns=[report_line_table.report,
report_line_table.invoice,
report_line_table.sii_header,
report_line_table.company],
values=values
))
return reports
@classmethod
def find_reports(cls, book='E'):
return cls.search([
('state', 'in', ('draft', 'confirmed')),
('book', '=', book),
], count=True)
@classmethod
def calculate_sii(cls):
Configuration = Pool().get('account.configuration')
config = Configuration(1)
if not config.aeat_pending_sii and not config.aeat_received_sii:
return
if config.aeat_pending_sii:
reports = cls.find_reports(book='E')
if reports:
_logger.info('Not calculate pending SII report '
'because is other reports pending to sending')
return
issued_reports = cls.get_issued_sii_reports()
if config.aeat_pending_sii_send:
cls.confirm(issued_reports)
cls.send(issued_reports)
if config.aeat_received_sii:
reports = cls.find_reports(book='R')
if reports:
_logger.info('Not calculate received SII report '
'because is other reports pending to sending')
return
received_reports = cls.get_received_sii_reports()
if config.aeat_received_sii_send:
cls.confirm(received_reports)
cls.send(received_reports)
class SIIReportLine(ModelSQL, ModelView):
'''
@ -1161,17 +1409,12 @@ class CreateSiiIssuedPending(Wizard):
create_ = StateAction('aeat_sii.act_aeat_sii_issued_report')
def do_create_(self, action):
pool = Pool()
Invoice = pool.get('account.invoice')
Report = pool.get('aeat.sii.report')
Report = Pool().get('aeat.sii.report')
reports = Report.search([
('state', 'in', ('draft', 'confirmed')),
('book', '=', 'E'),
])
reports = Report.find_reports(book='E')
if reports:
raise UserError(gettext('aeat_sii.reports_exist'))
reports = Invoice.get_issued_sii_reports()
reports = Report.get_issued_sii_reports()
reports = [x.id for x in reports] if reports else []
action['pyson_domain'] = PYSONEncoder().encode([
('id', 'in', reports),
@ -1199,19 +1442,13 @@ class CreateSiiReceivedPending(Wizard):
])
create_ = StateAction('aeat_sii.act_aeat_sii_received_report')
def do_create_(self, action):
pool = Pool()
Invoice = pool.get('account.invoice')
Report = pool.get('aeat.sii.report')
Report = Pool().get('aeat.sii.report')
reports = Report.search([
('state', 'in', ('draft', 'confirmed')),
('book', '=', 'R'),
])
reports = Report.find_reports(book='R')
if reports:
raise UserError(gettext('aeat_sii.reports_exist'))
reports = Invoice.get_received_sii_reports()
reports = Report.get_received_sii_reports()
reports = [x.id for x in reports] if reports else []
action['pyson_domain'] = PYSONEncoder().encode([
('id', 'in', reports),

2
company.py
View File

@ -78,7 +78,7 @@ class Company(metaclass=PoolMeta):
@contextmanager
def tmp_ssl_credentials(self):
if not self.pem_certificate:
if not self.pem_certificate or not self.private_key:
raise UserError(gettext('aeat_sii.msg_missing_pem_cert'))
with NamedTemporaryFile(suffix='.crt') as crt:
with NamedTemporaryFile(suffix='.pem') as key:

15
cron.py Normal file
View File

@ -0,0 +1,15 @@
# The COPYRIGHT file at the top level of this repository contains the full
# copyright notices and license terms.
from trytond.pool import PoolMeta
__all__ = ['Cron']
class Cron(metaclass=PoolMeta):
__name__ = 'ir.cron'
@classmethod
def __setup__(cls):
super().__setup__()
cls.method.selection.extend([
('aeat.sii.report|calculate_sii', "AEAT SII"),
])

12
cron.xml Normal file
View File

@ -0,0 +1,12 @@
<?xml version="1.0"?>
<!-- The COPYRIGHT file at the top level of this repository contains the full
copyright notices and license terms. -->
<tryton>
<data>
<record model="ir.cron" id="cron_calculate_sii">
<field name="interval_number" eval="1"/>
<field name="interval_type">days</field>
<field name="method">aeat.sii.report|calculate_sii</field>
</record>
</data>
</tryton>

291
invoice.py
View File

@ -1,18 +1,14 @@
# The COPYRIGHT file at the top level of this repository contains the full
# copyright notices and license terms.
import hashlib
from ast import literal_eval
from decimal import Decimal
from trytond.model import ModelView, fields
from trytond.pool import Pool, PoolMeta
from trytond.pyson import Eval, Bool
from trytond.pyson import Eval
from trytond.transaction import Transaction
from trytond.i18n import gettext
from trytond.exceptions import UserError, UserWarning
from trytond.wizard import Wizard, StateView, StateTransition, Button
from sql import Null
from sql.aggregate import Max
from trytond.tools import grouped_slice
from .aeat import (
OPERATION_KEY, BOOK_KEY, SEND_SPECIAL_REGIME_KEY,
RECEIVE_SPECIAL_REGIME_KEY, AEAT_INVOICE_STATE, IVA_SUBJECTED,
@ -25,8 +21,6 @@ _SII_INVOICE_KEYS = ['sii_book_key', 'sii_operation_key', 'sii_issued_key',
'sii_received_key', 'sii_subjected_key', 'sii_excemption_key',
'sii_intracomunity_key']
MAX_SII_LINES = 300
class Invoice(metaclass=PoolMeta):
__name__ = 'account.invoice'
@ -77,289 +71,6 @@ class Invoice(metaclass=PoolMeta):
def default_sii_pending_sending():
return False
@classmethod
def get_issued_sii_reports(cls):
pool = Pool()
Invoice = pool.get('account.invoice')
SIIReportLine = pool.get('aeat.sii.report.lines')
issued_invoices = {
'A0': {}, # 'A0', 'Registration of invoices/records'
'A1': {}, # 'A1', 'Amendment of invoices/records (registration errors)'
'D0': {}, # 'D0', 'Delete Invoices'
}
issued_invs = Invoice.search([
('sii_pending_sending', '=', True),
('sii_state', '=', 'Correcto'),
('sii_header', '!=', None),
('type', 'in', ['out']),
])
# search issued invoices [delete]
delete_issued_invoices = []
# search issued invoices [modify]
modify_issued_invoices = []
for issued_inv in issued_invs:
if not issued_inv.sii_records:
continue
sii_record_id = max([s.id for s in issued_inv.sii_records])
sii_record = SIIReportLine(sii_record_id)
if issued_inv.sii_header:
if (literal_eval(issued_inv.sii_header) ==
literal_eval(sii_record.sii_header)):
modify_issued_invoices.append(issued_inv)
else:
delete_issued_invoices.append(issued_inv)
periods = {}
for invoice in delete_issued_invoices:
period = invoice.move.period
if period in periods:
periods[period].append(invoice,)
else:
periods[period] = [invoice]
issued_invoices['D0'] = periods
periods2 = {}
for invoice in modify_issued_invoices:
period = invoice.move.period
if period in periods2:
periods2[period].append(invoice,)
else:
periods2[period] = [invoice]
issued_invoices['A1'] = periods2
# search issued invoices [new]
new_issued_invoices = Invoice.search([
('sii_state', 'in', (None, 'Incorrecto')),
('sii_pending_sending', '=', True),
('type', '=', 'out'),
])
# search possible deleted invoices in SII and not uploaded again
new_issued_invoices += Invoice.search([
('sii_state', '=', 'Anulada'),
('sii_pending_sending', '=', True),
('type', '=', 'out'),
('state', 'in', ['paid', 'posted']),
])
new_issued_invoices += delete_issued_invoices
periods1 = {}
for invoice in new_issued_invoices:
period = invoice.move.period
if period in periods1:
periods1[period].append(invoice,)
else:
periods1[period] = [invoice]
issued_invoices['A0'] = periods1
book_type = 'E' # Issued
return cls.create_sii_book(issued_invoices, book_type)
@classmethod
def get_received_sii_reports(cls):
pool = Pool()
Invoice = pool.get('account.invoice')
SIIReportLine = pool.get('aeat.sii.report.lines')
received_invoices = {
'A0': {}, # 'A0', 'Registration of invoices/records'
'A1': {}, # 'A1', 'Amendment of invoices/records (registration errors)'
'D0': {}, # 'D0', 'Delete Invoices'
}
received_invs = Invoice.search([
('sii_pending_sending', '=', True),
('sii_state', '=', 'Correcto'),
('sii_header', '!=', None),
('type', '=', 'in'),
])
# search received invoices [delete]
delete_received_invoices = []
# search received invoices [modify]
modify_received_invoices = []
for received_inv in received_invs:
if not received_inv.sii_records:
continue
sii_record_id = max([s.id for s in received_inv.sii_records])
sii_record = SIIReportLine(sii_record_id)
if received_inv.sii_header:
if (literal_eval(received_inv.sii_header) ==
literal_eval(sii_record.sii_header)):
modify_received_invoices.append(received_inv)
else:
delete_received_invoices.append(received_inv)
periods2 = {}
for invoice in modify_received_invoices:
period = invoice.move.period
if period in periods2:
periods2[period].append(invoice,)
else:
periods2[period] = [invoice]
received_invoices['A1'] = periods2
periods = {}
for invoice in delete_received_invoices:
period = invoice.move.period
if period in periods:
periods[period].append(invoice,)
else:
periods[period] = [invoice]
received_invoices['D0'] = periods
# search received invoices [new]
new_received_invoices = Invoice.search([
('sii_state', 'in', (None, 'Incorrecto')),
('sii_pending_sending', '=', True),
('type', '=', 'in'),
])
# search possible deleted invoices in SII and not uploaded again
new_received_invoices += Invoice.search([
('sii_state', '=', 'Anulada'),
('sii_pending_sending', '=', True),
('type', '=', 'in'),
('state', 'in', ['paid', 'posted']),
])
new_received_invoices += delete_received_invoices
periods1 = {}
for invoice in new_received_invoices:
period = invoice.move.period
if period in periods1:
periods1[period].append(invoice,)
else:
periods1[period] = [invoice]
received_invoices['A0'] = periods1
book_type = 'R' # Received
return cls.create_sii_book(received_invoices, book_type)
@classmethod
def create_sii_book(cls, book_invoices, book):
pool = Pool()
SIIReport = pool.get('aeat.sii.report')
SIIReportLine = pool.get('aeat.sii.report.lines')
Company = Pool().get('company.company')
company = Transaction().context.get('company')
company = Company(company)
company_vat = company.party.sii_vat_code
cursor = Transaction().connection.cursor()
report_line_table = SIIReportLine.__table__()
reports = []
for operation in ['D0', 'A1', 'A0']:
values = book_invoices[operation]
delete = True if operation == 'D0' else False
for period, invoices in values.items():
for invs in grouped_slice(invoices, MAX_SII_LINES):
report = SIIReport()
report.company = company
report.company_vat = company_vat
report.fiscalyear = period.fiscalyear
report.period = period
report.operation_type = operation
report.book = book
report.save()
reports.append(report)
values = []
for inv in invs:
sii_header = str(inv.get_sii_header(inv, delete))
values.append([report.id, inv.id, sii_header, company.id])
cursor.execute(*report_line_table.insert(
columns=[report_line_table.report,
report_line_table.invoice,
report_line_table.sii_header,
report_line_table.company],
values=values
))
return reports
@classmethod
def search_sii_state(cls, name, clause):
pool = Pool()
SIILines = pool.get('aeat.sii.report.lines')
table = SIILines.__table__()
cursor = Transaction().connection.cursor()
cursor.execute(*table.select(Max(table.id), table.invoice,
group_by=table.invoice))
invoices = []
lines = []
for id_, invoice in cursor.fetchall():
invoices.append(invoice)
lines.append(id_)
is_none = False
c = clause[-1]
if isinstance(clause[-1], list):
if None in clause[-1]:
is_none = True
c.remove(None)
c0 = []
if clause[-1] == None or is_none:
c0 = [('id', 'not in', invoices)]
clause2 = [tuple(('state',)) + tuple(clause[1:])] + \
[('id', 'in', lines)]
res_lines = SIILines.search(clause2)
if is_none:
return ['OR', c0, [('id', 'in', [x.invoice.id for x in res_lines])]]
else:
return [('id', 'in', [x.invoice.id for x in res_lines])]
@classmethod
def get_sii_state(cls, invoices, names):
pool = Pool()
SIILines = pool.get('aeat.sii.report.lines')
SIIReport = pool.get('aeat.sii.report')
result = {}
for name in names:
result[name] = dict((i.id, None) for i in invoices)
table = SIILines.__table__()
report = SIIReport.__table__()
cursor = Transaction().connection.cursor()
join = table.join(report, condition=table.report == report.id)
cursor.execute(*table.select(Max(table.id), table.invoice,
where=(table.invoice.in_([x.id for x in invoices]) &
(table.state != Null)),
group_by=table.invoice))
lines = [a[0] for a in cursor.fetchall()]
if lines:
cursor.execute(*join.select(table.state, report.operation_type,
table.invoice,
where=((table.id.in_(lines)) & (table.state != Null) &
(table.company == report.company))))
for state, op, inv in cursor.fetchall():
if 'sii_state' in names:
result['sii_state'][inv] = state
if 'sii_communication_type' in names:
result['sii_communication_type'][inv] = op
return result
def _credit(self, **values):
credit = super(Invoice, self)._credit(**values)
for field in _SII_INVOICE_KEYS:

1
tryton.cfg
View File

@ -12,6 +12,7 @@ extras_depend:
xml:
account.xml
aeat.xml
cron.xml
invoice.xml
party.xml
company.xml

17
view/configuration_form.xml Normal file
View File

@ -0,0 +1,17 @@
<?xml version="1.0"?>
<!-- The COPYRIGHT file at the top level of this repository contains the full
copyright notices and license types. -->
<data>
<xpath expr="/form/separator[@id='invoice']" position="before">
<separator id="aeat_sii" string="AEAT SII" colspan="4"/>
<label name="aeat_pending_sii"/>
<field name="aeat_pending_sii"/>
<label name="aeat_pending_sii_send"/>
<field name="aeat_pending_sii_send"/>
<label name="aeat_received_sii"/>
<field name="aeat_received_sii"/>
<label name="aeat_received_sii_send"/>
<field name="aeat_received_sii_send"/>
<newline/>
</xpath>
</data>