409 lines
14 KiB
Python
409 lines
14 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 base64
|
|
from decimal import Decimal
|
|
from functools import wraps
|
|
|
|
import requests
|
|
|
|
from trytond.i18n import gettext
|
|
from trytond.model import fields
|
|
from trytond.pool import Pool, PoolMeta
|
|
from trytond.pyson import Eval
|
|
|
|
from .exceptions import TyplessCredentialWarning, TyplessError
|
|
|
|
MIME_TYPES = {
|
|
'application/pdf',
|
|
'image/jpeg',
|
|
'image/png',
|
|
'image/tiff',
|
|
}
|
|
EXTRACT_DATA = 'https://developers.typless.com/api/extract-data'
|
|
ADD_DOCUMENT_FEEDBACK = (
|
|
'https://developers.typless.com/api/add-document-feedback')
|
|
|
|
SUPPLIER_INVOICE_FIELDS = [
|
|
'company_name',
|
|
'company_tax_identifier',
|
|
'supplier_name',
|
|
'tax_identifier',
|
|
'currency',
|
|
'number',
|
|
'description',
|
|
'invoice_date',
|
|
'payment_term_date',
|
|
'payment_reference',
|
|
'total_amount',
|
|
'untaxed_amount',
|
|
'tax_amount',
|
|
'purchase_orders',
|
|
]
|
|
SUPPLIER_INVOICE_FIELDS = [(x, x) for x in SUPPLIER_INVOICE_FIELDS]
|
|
SUPPLIER_INVOICE_LINE_ITEM_FIELDS = [
|
|
'product_name',
|
|
'description',
|
|
'unit',
|
|
'quantity',
|
|
'unit_price',
|
|
'amount',
|
|
'purchase_order',
|
|
]
|
|
SUPPLIER_INVOICE_LINE_ITEM_FIELDS = [
|
|
(x, x) for x in SUPPLIER_INVOICE_LINE_ITEM_FIELDS]
|
|
|
|
|
|
def typless_api(func):
|
|
@wraps(func)
|
|
def wrapper(*args, **kwargs):
|
|
try:
|
|
return func(*args, **kwargs)
|
|
except requests.HTTPError as e:
|
|
error_message = e.args[0]
|
|
raise TyplessError(
|
|
gettext('document_incoming_ocr_typless'
|
|
'.msg_typless_webserver_error',
|
|
message=error_message))
|
|
return wrapper
|
|
|
|
|
|
def get_best_value(fields, name):
|
|
for field in fields:
|
|
if field.get('name') == name:
|
|
break
|
|
else:
|
|
return
|
|
values = field.get('values', [])
|
|
values = sorted(
|
|
values, key=lambda v: v.get('confidence_score'), reverse=True)
|
|
if values:
|
|
return values[0].get('value')
|
|
|
|
|
|
class IncomingOCRService(metaclass=PoolMeta):
|
|
__name__ = 'document.incoming.ocr.service'
|
|
|
|
_states = {
|
|
'required': Eval('type') == 'typless',
|
|
'invisible': Eval('type') != 'typless',
|
|
}
|
|
|
|
typless_api_key = fields.Char(
|
|
"API Key", states=_states,
|
|
help="The standard token from Typless settings page.")
|
|
typless_document_type = fields.Char(
|
|
"Document Type", states=_states,
|
|
help="The name of the document type on Typless.")
|
|
typless_fields = fields.MultiSelection(
|
|
'get_typless_fields', "Fields",
|
|
translate=False,
|
|
states={
|
|
'invisible': _states['invisible'],
|
|
},
|
|
help="The metadata fields setup for this document type.")
|
|
typless_line_item_fields = fields.MultiSelection(
|
|
'get_typless_line_item_fields', "Line Item Fields",
|
|
translate=False,
|
|
states={
|
|
'invisible': _states['invisible'],
|
|
},
|
|
help="The line item fields setup for this document type.")
|
|
typless_vat_rates = fields.Boolean(
|
|
"VAT Rates",
|
|
states={
|
|
'invisible': _states['invisible'],
|
|
},
|
|
help="Check if the vat rate net plugin is activated "
|
|
"for this document type.")
|
|
|
|
@classmethod
|
|
def __setup__(cls):
|
|
super().__setup__()
|
|
cls.type.selection.append(('typless', "Typless"))
|
|
|
|
def get_typless_fields(self):
|
|
return [('document_type', 'document_type')]
|
|
|
|
def get_typless_line_item_fields(self):
|
|
return []
|
|
|
|
def match_mime_type(self, mime_type):
|
|
match = super().match_mime_type(mime_type)
|
|
if self.type == 'typless':
|
|
match = mime_type in MIME_TYPES
|
|
return match
|
|
|
|
@typless_api
|
|
def _process_typless(self, document):
|
|
payload = self._typless_extract_payload(document)
|
|
headers = {
|
|
'Accept': 'application/json',
|
|
'Content-Type': 'application/json',
|
|
'Authorization': self.typless_api_key,
|
|
}
|
|
response = requests.post(EXTRACT_DATA, json=payload, headers=headers)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
def _typless_extract_payload(self, document):
|
|
return {
|
|
'file': base64.b64encode(document.data).decode('utf-8'),
|
|
'file_name': document.name,
|
|
'document_type_name': self.typless_document_type,
|
|
}
|
|
|
|
def _get_document_incoming_typless(self, document):
|
|
document_data = {}
|
|
if not document.parsed_data:
|
|
return document_data
|
|
fields = document.parsed_data.get('extracted_fields', [])
|
|
document_type = get_best_value(fields, 'document_type')
|
|
if document_type is not None:
|
|
document_data['document_type'] = document_type
|
|
return document_data
|
|
|
|
@typless_api
|
|
def _send_feedback_typless(self, document):
|
|
payload = self._typless_feedback_payload(document)
|
|
headers = {
|
|
'Accept': 'application/json',
|
|
'Content-Type': 'application/json',
|
|
'Authorization': self.typless_api_key,
|
|
}
|
|
response = requests.post(
|
|
ADD_DOCUMENT_FEEDBACK, json=payload, headers=headers)
|
|
response.raise_for_status()
|
|
|
|
def _typless_feedback_payload(self, document):
|
|
source = document.result
|
|
fields = document.ocr_service.typless_fields
|
|
learning_fields = []
|
|
getter = getattr(self, f'_typless_feedback_payload_{document.type}')
|
|
for name in fields:
|
|
value = getter(source, name)
|
|
learning_fields.append({
|
|
'name': name,
|
|
'value': value,
|
|
})
|
|
payload = {
|
|
'document_type_name': self.typless_document_type,
|
|
'document_object_id': document.parsed_data['object_id'],
|
|
'learning_fields': learning_fields,
|
|
}
|
|
line_items = document.ocr_service.typless_line_item_fields
|
|
if line_items:
|
|
lines = []
|
|
getter = getattr(
|
|
self, f'_typless_feedback_payload_line_items_{document.type}')
|
|
for line in self._typless_feedback_payload_line_items(source):
|
|
line_item = []
|
|
for name in line_items:
|
|
value = getter(line, name)
|
|
line_item.append({
|
|
'name': name,
|
|
'value': value,
|
|
})
|
|
lines.append(line_item)
|
|
payload['line_items'] = lines
|
|
if document.ocr_service.typless_vat_rates:
|
|
taxes = []
|
|
getter = getattr(
|
|
self, f'_typless_feedback_payload_vat_rates_{document.type}')
|
|
for tax_line in self._typless_feedback_payload_vat_rates(source):
|
|
percentage = getter(tax_line, 'vat_rate_percentage')
|
|
net = getter(tax_line, 'vat_rate_net')
|
|
if percentage and net:
|
|
taxes.append([{
|
|
'name': 'vat_rate_percentage',
|
|
'value': percentage,
|
|
}, {
|
|
'name': 'vat_rate_net',
|
|
'value': net,
|
|
}])
|
|
payload['vat_rates'] = taxes
|
|
return payload
|
|
|
|
def _typless_feedback_payload_document_incoming(self, document, name):
|
|
if name == 'document_type':
|
|
return document.type
|
|
|
|
def _typless_feedback_payload_line_items(self, record):
|
|
yield from []
|
|
|
|
def _typless_feedback_payload_line_items_document_incoming(
|
|
self, line, name):
|
|
pass
|
|
|
|
def _typless_feedback_payload_vat_rates(self, record):
|
|
yield from []
|
|
|
|
def _typless_feedback_payload_vat_rates_document_incoming(
|
|
self, tax_line, name):
|
|
pass
|
|
|
|
@classmethod
|
|
def check_modification(cls, mode, services, values=None, external=False):
|
|
pool = Pool()
|
|
Warning = pool.get('res.user.warning')
|
|
|
|
super().check_modification(
|
|
mode, services, values=values, external=external)
|
|
|
|
if mode == 'write' and external and 'typless_api_key' in values:
|
|
warning_name = Warning.format('typless_credential', services)
|
|
if Warning.check(warning_name):
|
|
raise TyplessCredentialWarning(
|
|
warning_name,
|
|
gettext('document_incoming_ocr_typless'
|
|
'.msg_typless_credential_modified'))
|
|
|
|
|
|
class IncomingOCRService_IncomingInvoice(metaclass=PoolMeta):
|
|
__name__ = 'document.incoming.ocr.service'
|
|
|
|
@fields.depends('document_type')
|
|
def get_typless_fields(self):
|
|
selection = super().get_typless_fields()
|
|
if self.document_type == 'supplier_invoice':
|
|
selection += SUPPLIER_INVOICE_FIELDS
|
|
return selection
|
|
|
|
@fields.depends('document_type')
|
|
def get_typless_line_item_fields(self):
|
|
selection = super().get_typless_line_item_fields()
|
|
if self.document_type == 'supplier_invoice':
|
|
selection += SUPPLIER_INVOICE_LINE_ITEM_FIELDS
|
|
return selection
|
|
|
|
def _get_supplier_invoice_typless(self, document):
|
|
invoice_data = {}
|
|
if not document.parsed_data:
|
|
return invoice_data
|
|
fields = document.parsed_data.get('extracted_fields', [])
|
|
for name in self.typless_fields:
|
|
value = get_best_value(fields, name)
|
|
if value is not None:
|
|
if name == 'total_amount':
|
|
value = Decimal(value)
|
|
invoice_data[name] = value
|
|
invoice_data['lines'] = lines = []
|
|
for parsed_line in document.parsed_data.get('line_items', []):
|
|
lines.append(self._get_supplier_invoice_typless_line(
|
|
parsed_line, invoice_data))
|
|
invoice_data['taxes'] = taxes = []
|
|
for parsed_tax in document.parsed_data.get('vat_rates', []):
|
|
taxes.append(self._get_supplier_invoice_typless_tax(
|
|
parsed_tax, invoice_data))
|
|
return invoice_data
|
|
|
|
def _get_supplier_invoice_typless_line(self, parsed_line, invoice_data):
|
|
line = {}
|
|
if 'purchase_orders' in invoice_data:
|
|
line['purchase_orders'] = invoice_data['purchase_orders']
|
|
for name in self.typless_line_item_fields:
|
|
value = get_best_value(parsed_line, name)
|
|
if value is not None:
|
|
if name in {'unit_price', 'amount'}:
|
|
value = Decimal(value)
|
|
elif name == 'quantity':
|
|
value = float(value)
|
|
line[name] = value
|
|
return line
|
|
|
|
def _get_supplier_invoice_typless_tax(sef, parsed_tax, invoice_data):
|
|
tax = {'type': 'percentage'}
|
|
percentage = get_best_value(parsed_tax, 'vat_rate_percentage')
|
|
if percentage is not None:
|
|
tax['rate'] = Decimal(percentage) / 100
|
|
net = get_best_value(parsed_tax, 'vat_rate_net')
|
|
if net is not None:
|
|
tax['base'] = Decimal(net)
|
|
return tax
|
|
|
|
def _typless_feedback_payload_supplier_invoice(self, invoice, name):
|
|
if name == 'company_name':
|
|
return invoice.company.party.name
|
|
elif name == 'company_tax_identifier':
|
|
if invoice.tax_identifier:
|
|
return invoice.tax_identifier.code
|
|
else:
|
|
return ''
|
|
elif name == 'supplier_name':
|
|
return invoice.party.name
|
|
elif name == 'tax_identifier':
|
|
if invoice.party_tax_identifier:
|
|
return invoice.party_tax_identifier.code
|
|
else:
|
|
return ''
|
|
elif name == 'currency':
|
|
return invoice.currency.code
|
|
elif name == 'number':
|
|
return invoice.reference
|
|
elif name == 'description':
|
|
return invoice.description
|
|
elif name == 'invoice_date':
|
|
return invoice.invoice_date.isoformat()
|
|
elif name == 'payment_term_date':
|
|
if invoice.payment_term_date:
|
|
return invoice.payment_term_date.isoformat()
|
|
else:
|
|
return ''
|
|
elif name == 'payment_reference':
|
|
return invoice.supplier_payment_reference
|
|
elif name == 'untaxed_amount':
|
|
return str(invoice.untaxed_amount)
|
|
elif name == 'tax_amount':
|
|
return str(invoice.tax_amount)
|
|
elif name == 'total_amount':
|
|
return str(invoice.total_amount)
|
|
elif name == 'purchase_orders':
|
|
return invoice.origins
|
|
|
|
def _typless_feedback_payload_line_items(self, record):
|
|
pool = Pool()
|
|
Invoice = pool.get('account.invoice')
|
|
yield from super()._typless_feedback_payload_line_items(record)
|
|
if isinstance(record, Invoice):
|
|
yield from record.line_lines
|
|
|
|
def _typless_feedback_payload_line_items_supplier_invoice(
|
|
self, line, name):
|
|
if name == 'product_name':
|
|
if line.product:
|
|
return line.product.name
|
|
else:
|
|
return ''
|
|
elif name == 'description':
|
|
return line.description
|
|
elif name == 'unit':
|
|
if line.unit:
|
|
return line.unit.name
|
|
else:
|
|
return ''
|
|
elif name == 'quantity':
|
|
return str(line.quantity)
|
|
elif name == 'unit_price':
|
|
return str(line.unit_price)
|
|
elif name == 'amount':
|
|
return str(line.amount)
|
|
elif name == 'purchase_order':
|
|
return line.origin_name
|
|
|
|
def _typless_feedback_payload_vat_rates(self, record):
|
|
pool = Pool()
|
|
Invoice = pool.get('account.invoice')
|
|
yield from super()._typless_feedback_payload_vat_rates(record)
|
|
if isinstance(record, Invoice):
|
|
yield from record.taxes
|
|
|
|
def _typless_feedback_payload_vat_rates_supplier_invoice(
|
|
self, tax_line, name):
|
|
if name == 'vat_rate_percentage':
|
|
if tax_line.tax and tax_line.tax.type == 'percentage':
|
|
return str(tax_line.tax.rate * 100)
|
|
else:
|
|
return ''
|
|
elif name == 'vat_rate_net':
|
|
return str(tax_line.base)
|