92 lines
2.9 KiB
Python
92 lines
2.9 KiB
Python
|
# This file is part of Tryton. The COPYRIGHT file at the top level of
|
||
|
|
# this repository contains the full copyright notices and license terms.
|
||
|
|
import logging
|
||
|
|
import os
|
||
|
|
|
||
|
|
from trytond.pool import PoolMeta
|
||
|
|
|
||
|
|
logger = logging.getLogger(__name__)
|
||
|
|
|
||
|
|
|
||
|
|
def _build_error_message(inv, error):
|
||
|
|
return (
|
||
|
|
f"ID: {inv.id}"
|
||
|
|
f" | Numero: {getattr(inv, 'number', 'N/D')}"
|
||
|
|
f" | NCF: {getattr(inv, 'ncf_number', 'N/D')}"
|
||
|
|
f" | Fecha: {getattr(inv, 'invoice_date', 'N/D')}"
|
||
|
|
f" | Descripcion: {getattr(inv, 'description', 'N/D')}"
|
||
|
|
f" | Error: {error}"
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def _write_error_log(path, header, error_lines):
|
||
|
|
os.makedirs(os.path.dirname(path), exist_ok=True)
|
||
|
|
with open(path, 'a', encoding='utf-8') as f:
|
||
|
|
f.write(f"===== {header} =====\n")
|
||
|
|
for line in error_lines:
|
||
|
|
f.write(line + '\n')
|
||
|
|
|
||
|
|
|
||
|
|
class Invoice(metaclass=PoolMeta):
|
||
|
|
__name__ = 'account.invoice'
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def _cron_process_invoices(cls, invoice_type, error_log_filename):
|
||
|
|
invoices = cls.search([
|
||
|
|
('type', '=', invoice_type),
|
||
|
|
('state', '=', 'posted'),
|
||
|
|
])
|
||
|
|
logger.info(
|
||
|
|
"cron_process_posted_%s_invoices: found %d invoices",
|
||
|
|
'customer' if invoice_type == 'out' else 'supplier',
|
||
|
|
len(invoices),
|
||
|
|
)
|
||
|
|
|
||
|
|
errors = []
|
||
|
|
total = len(invoices)
|
||
|
|
for idx, inv in enumerate(invoices, start=1):
|
||
|
|
try:
|
||
|
|
logger.info(
|
||
|
|
"Processing invoice %d/%d id=%s number=%s",
|
||
|
|
idx, total, inv.id, getattr(inv, 'number', 'N/D'),
|
||
|
|
)
|
||
|
|
cls.process([inv])
|
||
|
|
except Exception as e:
|
||
|
|
msg = _build_error_message(inv, e)
|
||
|
|
logger.error(msg, exc_info=True)
|
||
|
|
errors.append(msg)
|
||
|
|
|
||
|
|
if errors:
|
||
|
|
log_dir = os.path.expanduser('~/legna/tryton')
|
||
|
|
_write_error_log(
|
||
|
|
os.path.join(log_dir, error_log_filename),
|
||
|
|
f"ERRORES CRON {'CUSTOMER' if invoice_type == 'out' else 'SUPPLIER'} INVOICES",
|
||
|
|
errors,
|
||
|
|
)
|
||
|
|
logger.warning(
|
||
|
|
"cron finished with %d error(s)", len(errors))
|
||
|
|
else:
|
||
|
|
logger.info("cron finished successfully")
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def cron_process_posted_customer_invoices(cls):
|
||
|
|
cls._cron_process_invoices('out', 'invoice_out_cron_errors.log')
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def cron_process_posted_supplier_invoices(cls):
|
||
|
|
cls._cron_process_invoices('in', 'invoice_in_cron_errors.log')
|
||
|
|
|
||
|
|
|
||
|
|
class Cron(metaclass=PoolMeta):
|
||
|
|
__name__ = 'ir.cron'
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def __setup__(cls):
|
||
|
|
super().__setup__()
|
||
|
|
cls.method.selection.extend([
|
||
|
|
('account.invoice|cron_process_posted_customer_invoices',
|
||
|
|
"Process Posted Customer Invoices"),
|
||
|
|
('account.invoice|cron_process_posted_supplier_invoices',
|
||
|
|
"Process Posted Supplier Invoices"),
|
||
|
|
])
|