355 lines
12 KiB
Python
355 lines
12 KiB
Python
# The COPYRIGHT file at the top level of this repository contains the full
|
|
# copyright notices and license terms.
|
|
from decimal import Decimal
|
|
from trytond.model import ModelView, fields, Workflow
|
|
from trytond.pool import Pool, PoolMeta
|
|
from trytond.pyson import Eval, Bool
|
|
from trytond.transaction import Transaction
|
|
|
|
from sql import Null
|
|
from sql.aggregate import Max
|
|
|
|
from .aeat import (
|
|
OPERATION_KEY, BOOK_KEY, SEND_SPECIAL_REGIME_KEY,
|
|
RECEIVE_SPECIAL_REGIME_KEY, AEAT_INVOICE_STATE, IVA_SUBJECTED,
|
|
EXCEMPTION_CAUSE, INTRACOMUNITARY_TYPE, COMMUNICATION_TYPE)
|
|
|
|
|
|
__all__ = ['Invoice', 'Sale', 'Purchase', 'Invoice2']
|
|
|
|
_SII_INVOICE_KEYS = ['sii_book_key', 'sii_issued_key', 'sii_received_key',
|
|
'sii_subjected_key', 'sii_excemption_key',
|
|
'sii_intracomunity_key']
|
|
|
|
|
|
class Invoice(metaclass=PoolMeta):
|
|
__name__ = 'account.invoice'
|
|
|
|
sii_book_key = fields.Selection(BOOK_KEY, 'SII Book Key',
|
|
states={
|
|
'required': Eval('state').in_(['posted', 'paid'])
|
|
}, depends=['state'])
|
|
sii_operation_key = fields.Selection(OPERATION_KEY, 'SII Operation Key',
|
|
states={
|
|
'required': Eval('state').in_(['posted', 'paid'])
|
|
}, depends=['state'])
|
|
sii_issued_key = fields.Selection(SEND_SPECIAL_REGIME_KEY,
|
|
'SII Issued Key',
|
|
states={
|
|
'invisible': ~Eval('sii_book_key').in_(['E']),
|
|
}, depends=['sii_book_key'])
|
|
sii_received_key = fields.Selection(RECEIVE_SPECIAL_REGIME_KEY,
|
|
'SII Recived Key',
|
|
states={
|
|
'invisible': ~Eval('sii_book_key').in_(['R']),
|
|
}, depends=['sii_book_key'])
|
|
sii_subjected_key = fields.Selection(IVA_SUBJECTED, 'Subjected')
|
|
sii_excemption_key = fields.Selection(EXCEMPTION_CAUSE,
|
|
'Excemption Cause')
|
|
sii_intracomunity_key = fields.Selection(INTRACOMUNITARY_TYPE,
|
|
'SII Intracommunity Key',
|
|
states={
|
|
'invisible': ~Eval('sii_book_key').in_(['U']),
|
|
}, depends=['sii_book_key'])
|
|
sii_records = fields.One2Many('aeat.sii.report.lines', 'invoice',
|
|
"SII Report Lines")
|
|
sii_state = fields.Function(fields.Selection(AEAT_INVOICE_STATE,
|
|
'SII State'), 'get_sii_state', searcher='search_sii_state')
|
|
sii_communication_type = fields.Function(fields.Selection(
|
|
COMMUNICATION_TYPE + [(None, '')], 'SII Communication Type'),
|
|
'get_sii_state')
|
|
|
|
@classmethod
|
|
def __setup__(cls):
|
|
super(Invoice, cls).__setup__()
|
|
sii_fields = ['sii_book_key', 'sii_operation_key',
|
|
'sii_received_key', 'sii_issued_key', 'sii_subjected_key',
|
|
'sii_excemption_key', 'sii_intracomunity_key']
|
|
cls._check_modify_exclude += sii_fields
|
|
cls._error_messages.update({
|
|
'invoices_sii': 'The next invoices are related with SII books:\n'
|
|
'%s.\n\nIf you edit them take care if you need to update '
|
|
'again to SII',
|
|
})
|
|
|
|
cls._buttons.update({
|
|
'reset_sii_keys': {
|
|
'invisible': Bool(Eval('sii_state', None)),
|
|
'icon': 'tryton-executable'}
|
|
})
|
|
if hasattr(cls, '_intercompany_excluded_fields'):
|
|
cls._intercompany_excluded_fields += sii_fields
|
|
cls._intercompany_excluded_fields += ['sii_records']
|
|
|
|
@staticmethod
|
|
def default_sii_operation_key():
|
|
type_ = Transaction().context.get('type', 'out_invoice')
|
|
if type_ in ('in_credit_note', 'out_credit_note'):
|
|
return 'R1'
|
|
return 'F1'
|
|
|
|
@classmethod
|
|
def search_sii_state(cls, name, clause):
|
|
pool = Pool()
|
|
SIILines = pool.get('aeat.sii.report.lines')
|
|
|
|
assert clause[1] in ('=', 'in')
|
|
|
|
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_)
|
|
|
|
values = clause[-1]
|
|
is_none = False
|
|
if isinstance(clause[-1], (list, tuple, set)):
|
|
if None in clause[-1]:
|
|
is_none = True
|
|
values.remove(None)
|
|
else:
|
|
is_none = bool(clause[-1] is None)
|
|
|
|
c0 = []
|
|
if is_none:
|
|
c0 = [('sii_records', '=', None)]
|
|
|
|
clause2 = [
|
|
('state', ) + tuple(clause[1:2]) + (values, ),
|
|
('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):
|
|
credit = super(Invoice, self)._credit()
|
|
for field in _SII_INVOICE_KEYS:
|
|
setattr(credit, field, getattr(self, field))
|
|
|
|
credit.sii_operation_key = 'R4'
|
|
return credit
|
|
|
|
def _set_sii_keys(self):
|
|
tax = None
|
|
for t in self.taxes:
|
|
if t.tax.sii_book_key:
|
|
tax = t.tax
|
|
break
|
|
if not tax:
|
|
return
|
|
for field in _SII_INVOICE_KEYS:
|
|
setattr(self, field, getattr(tax, field))
|
|
|
|
@fields.depends(*_SII_INVOICE_KEYS)
|
|
def _on_change_lines_taxes(self):
|
|
super(Invoice, self)._on_change_lines_taxes()
|
|
for field in _SII_INVOICE_KEYS:
|
|
if getattr(self, field):
|
|
return
|
|
self._set_sii_keys()
|
|
|
|
@classmethod
|
|
def update_taxes(cls, invoices, exception=False):
|
|
super().update_taxes(invoices, exception=exception)
|
|
to_save = []
|
|
for invoice in invoices:
|
|
set_keys = True
|
|
for field in _SII_INVOICE_KEYS:
|
|
if getattr(invoice, field):
|
|
set_keys = False
|
|
break
|
|
if set_keys:
|
|
invoice._set_sii_keys()
|
|
to_save.append(invoice)
|
|
if to_save:
|
|
cls.save(to_save)
|
|
|
|
@classmethod
|
|
def copy(cls, records, default=None):
|
|
if default is None:
|
|
default = {}
|
|
default = default.copy()
|
|
default['sii_records'] = None
|
|
return super(Invoice, cls).copy(records, default=default)
|
|
|
|
@classmethod
|
|
@ModelView.button
|
|
def reset_sii_keys(cls, records):
|
|
to_write = []
|
|
for record in records:
|
|
record._set_sii_keys()
|
|
record.sii_operation_key = ('R1'
|
|
if record.untaxed_amount < Decimal('0.0') else 'F1')
|
|
to_write.extend(([record], record._save_values))
|
|
|
|
if to_write:
|
|
cls.write(*to_write)
|
|
|
|
@classmethod
|
|
def process(cls, invoices):
|
|
super(Invoice, cls).process(invoices)
|
|
invoices_sii = ''
|
|
for invoice in invoices:
|
|
if invoice.state != 'draft':
|
|
continue
|
|
if invoice.sii_state:
|
|
invoices_sii += '\n%s: %s' % (invoice.number, invoice.sii_state)
|
|
if invoices_sii:
|
|
warning_name = 'invoices_sii_report_%s' % ",".join([str(x.id) for x in invoices])
|
|
cls.raise_user_warning(warning_name, 'invoices_sii', invoices_sii)
|
|
|
|
|
|
class Invoice2(metaclass=PoolMeta):
|
|
__name__ = 'account.invoice'
|
|
|
|
@classmethod
|
|
def __setup__(cls):
|
|
super().__setup__()
|
|
cls._sii_state_deny_draft = {
|
|
'Correcto', 'Correcta', 'AceptadoConErrores', 'AceptadaConErrores'
|
|
}
|
|
cls._deny_modify_sii_fields = {
|
|
'reference', 'party', 'invoice_date',
|
|
}
|
|
cls._error_messages.update({
|
|
'deny_draft_invalid_sii_state':
|
|
'You can not set to draft invoice "%(invoice)s" because the '
|
|
'SII state is "%(sii_state)s". Please first include the '
|
|
'invoice in a Delete Invoice SII Book and send to AEAT.',
|
|
'modify_invalid_sii_state':
|
|
'Invoice "%(invoice)s" was sent to SII and the state is '
|
|
'"%(sii_state)s". Please remember after posting to send it '
|
|
'again with a Modify SII Book.',
|
|
'deny_modify_sii_pk':
|
|
'Cannot modify "%(fields)s" on invoice "%(invoice)s" '
|
|
'because its SII state is "%(sii_state)s". Please first '
|
|
'include the invoice in a Delete Invoice SII Book and send '
|
|
'to AEAT.'
|
|
})
|
|
|
|
@classmethod
|
|
@ModelView.button
|
|
@Workflow.transition('draft')
|
|
def draft(cls, invoices):
|
|
for invoice in invoices:
|
|
invoice.check_sent_sii()
|
|
super().draft(invoices)
|
|
|
|
@classmethod
|
|
@ModelView.button
|
|
@Workflow.transition('posted')
|
|
def post(cls, invoices):
|
|
for invoice in invoices:
|
|
if invoice.state in ('draft', 'validated'):
|
|
invoice.check_sent_sii()
|
|
super().post(invoices)
|
|
|
|
@classmethod
|
|
def write(cls, *args):
|
|
actions = iter(args)
|
|
args = []
|
|
for records, values in zip(actions, actions):
|
|
sii_vals = set(values) & cls._deny_modify_sii_fields
|
|
if sii_vals:
|
|
for record in records:
|
|
if record.type == 'in':
|
|
record.check_sent_sii(list(sii_vals))
|
|
args.extend((records, values))
|
|
super().write(*args)
|
|
|
|
def check_sent_sii(self, fields=[]):
|
|
pool = Pool()
|
|
ModelData = pool.get('ir.model.data')
|
|
User = pool.get('res.user')
|
|
Group = pool.get('res.group')
|
|
|
|
if self.sii_state in self._sii_state_deny_draft and \
|
|
self.sii_communication_type != 'D0':
|
|
|
|
# check group
|
|
def in_group():
|
|
group = Group(ModelData.get_id(
|
|
'aeat_sii',
|
|
'group_invoice_sent_sii_posted2draft'))
|
|
transaction = Transaction()
|
|
user_id = transaction.user
|
|
if user_id == 0:
|
|
user_id = transaction.context.get('user', user_id)
|
|
if user_id == 0:
|
|
return True
|
|
user = User(user_id)
|
|
return group in user.groups
|
|
|
|
if fields:
|
|
self.raise_user_error('deny_modify_sii_pk', {
|
|
'fields': ', '.join(
|
|
[f['string'] for f in self.__class__.fields_get(
|
|
fields_names=fields).values()]),
|
|
'invoice': self.rec_name,
|
|
'sii_state': self.sii_state
|
|
})
|
|
elif not in_group():
|
|
self.raise_user_error('deny_draft_invalid_sii_state', {
|
|
'invoice': self.rec_name,
|
|
'sii_state': self.sii_state
|
|
})
|
|
else:
|
|
self.raise_user_warning(
|
|
'modify_invalid_sii_state_%s' % self.id,
|
|
'modify_invalid_sii_state', {
|
|
'invoice': self.rec_name,
|
|
'sii_state': self.sii_state
|
|
})
|