Connecting Odoo

ocost_connect/
  __init__.py              from . import models
  __manifest__.py
  models/__init__.py       from . import account_move
  models/account_move.py
  data/cron.xml

{
    "name": "Ocost Connect",
    "version": "19.0.1.0.0",
    "depends": ["account"],
    "data": ["data/cron.xml"],
    "license": "LGPL-3",
}

import base64
import logging
from decimal import Decimal

import requests

from odoo import _, api, fields, models
from odoo.fields import Command

API = "https://api.ocost.io/public/v1"
log = logging.getLogger(__name__)


class AccountMove(models.Model):
    _inherit = "account.move"

    ocost_file_id = fields.Char(copy=False, readonly=True)
    ocost_state = fields.Selection(
        [("sent", "Sent"), ("read", "Read"), ("not_read", "Not read"), ("failed", "Failed")],
        copy=False,
        readonly=True,
    )

    def _ocost(self):
        key = self.env["ir.config_parameter"].sudo().get_param("ocost.api_key")
        session = requests.Session()
        session.headers["Authorization"] = f"Bearer {key}"
        return session

    def action_ocost_send(self):
        """Sends each draft vendor bill's PDF to Ocost (a button, a server
        action or the cron). The same PDF twice is the same file: safe."""
        session = self._ocost()
        bills = self.filtered(
            lambda m: m.move_type in ("in_invoice", "in_refund")
            and m.state == "draft"
            and not m.ocost_file_id
        )
        for move in bills:
            attachment = move.message_main_attachment_id or move.attachment_ids[:1]
            if not attachment:
                continue
            answer = session.post(
                f"{API}/files",
                files={
                    "file": (
                        attachment.name,
                        base64.b64decode(attachment.datas),
                        attachment.mimetype,
                    )
                },
                data={"external_ref": f"account.move,{move.id}"},
                timeout=30,
            )
            if answer.status_code == 429 or answer.status_code >= 500:
                continue  # the next run tries again
            if answer.status_code == 401:
                log.warning("Ocost refused the API key: replace it")
                return
            if answer.status_code in (413, 415):
                move.ocost_state = "failed"
                move.message_post(body=answer.json().get("detail"))
                continue
            answer.raise_for_status()
            file = answer.json()
            move.write({"ocost_file_id": file["id"], "ocost_state": "sent"})
            if file["status"] == "read":
                move._ocost_fill(session)

    @api.model
    def _cron_ocost(self):
        """Every few minutes: send what waits, then fill what Ocost read."""
        self.search(
            [("move_type", "in", ("in_invoice", "in_refund")), ("state", "=", "draft"),
             ("ocost_file_id", "=", False), ("message_main_attachment_id", "!=", False)],
            limit=20,
        ).action_ocost_send()
        session = self._ocost()
        for move in self.search([("ocost_state", "=", "sent")], limit=50):
            file = session.get(f"{API}/files/{move.ocost_file_id}", timeout=15).json()
            if file["status"] in ("received", "reading"):
                continue
            if file["status"] == "read":
                move._ocost_fill(session)
            else:
                move.ocost_state = file["status"]
                move.message_post(body=_("Ocost didn't read it: %s", file["reason"]))

    def _ocost_fill(self, session):
        """Fills the draft bill's empty fields from Ocost's reading."""
        self.ensure_one()
        doc = session.get(f"{API}/files/{self.ocost_file_id}/document", timeout=15).json()
        notes = []
        if doc["direction"] == "outgoing" or doc["kind"] not in ("invoice", "credit_note"):
            self.ocost_state = "read"
            self.message_post(body=_("Ocost read a %s, not a bill to fill.", doc["kind_detail"]))
            return
        vals = {}
        seller = doc["seller"] or {}
        contact = seller.get("contact") or {}
        vat_sure = doc["sources"].get("seller.vat_id", {}).get("confidence") != "doubt"
        if not self.partner_id and seller.get("name"):
            partner = self.env["res.partner"]._retrieve_partner(
                name=seller["name"],
                vat=seller.get("vat_id") if vat_sure else None,
                email=contact.get("email"),
                company=self.company_id,
            )
            if not partner:
                address = seller.get("address") or {}
                country = self.env["res.country"].search(
                    [("code", "=", address.get("country"))], limit=1
                )
                partner = self.env["res.partner"].create({
                    "name": seller["name"],
                    "is_company": True,
                    "vat": seller.get("vat_id") if vat_sure else False,
                    "street": address.get("street"),
                    "zip": address.get("postal_code"),
                    "city": address.get("city"),
                    "country_id": country.id or False,
                    "email": contact.get("email"),
                    "phone": contact.get("phone"),
                    "supplier_rank": 1,
                })
            vals["partner_id"] = partner.id
        if not self.ref and doc["number"]:
            vals["ref"] = doc["number"]
        if not self.invoice_date and doc["issue_date"]:
            vals["invoice_date"] = doc["issue_date"]
        if doc["due_date"]:
            vals["invoice_date_due"] = doc["due_date"]
        if doc["order_reference"] and not self.invoice_origin:
            vals["invoice_origin"] = doc["order_reference"]
        currency = self.env["res.currency"].search([("name", "=", doc["currency"])], limit=1)
        if currency:
            vals["currency_id"] = currency.id
        if doc["kind"] == "credit_note" and self.move_type == "in_invoice" and not self.invoice_line_ids:
            vals["move_type"] = "in_refund"
        if not self.invoice_line_ids:
            vals["invoice_line_ids"] = [
                Command.create(self._ocost_line(line, doc["totals"]["prices_include_vat"], notes))
                for line in doc["lines"]
            ] or [
                # No lines read: one line per VAT rate.
                Command.create(self._ocost_line(
                    {"name": doc["number"] or _("Invoice"), "quantity": "1",
                     "price": group["taxable_amount"], "vat_rate": group["rate"]},
                    False, notes,
                ))
                for group in doc["vat_breakdown"]
            ]
        self.write(vals)
        self.ocost_state = "read"
        if not all(c["passed"] for c in doc["checks"] if c["sums"]):
            notes.append(_("The sums don't add up as read: check the amounts."))
        doubts = [path for path, s in doc["sources"].items() if s["confidence"] == "doubt"]
        if doubts:
            notes.append(_("Ocost isn't sure of: %s", ", ".join(doubts)))
        if doc["possible_duplicate_of"]:
            notes.append(_("Ocost read a bill of this number from this seller before."))
        if doc["early_payment_discounts"]:
            skonto = doc["early_payment_discounts"][0]
            notes.append(_("Skonto: %s %% until %s", skonto["percent"], skonto["due_date"]))
        self.message_post(body="\n".join([_("Filled from Ocost's reading. Check before posting."), *notes]))

    def _ocost_line(self, line, included, notes):
        price = Decimal(line.get("price") or line.get("net_amount") or "0")
        per = Decimal(line.get("base_quantity") or "1")
        values = {
            "name": " ".join(filter(None, [
                line.get("seller_item_id") and f"[{line['seller_item_id']}]", line["name"],
            ])),
            "quantity": float(Decimal(line.get("quantity") or "1")),
            "price_unit": float(price / per),
            "discount": float(Decimal(line.get("discount_percent") or "0")),
        }
        if line.get("vat_rate") is not None:
            tax = self.env["account.tax"].search([
                ("company_id", "=", self.company_id.id),
                ("type_tax_use", "=", "purchase"),
                ("amount_type", "=", "percent"),
                ("amount", "=", float(line["vat_rate"])),
                ("price_include", "=", included),
            ], limit=1)
            if tax:
                values["tax_ids"] = [Command.set(tax.ids)]
            else:
                notes.append(_("No purchase tax of %s %%: set the line's tax.", line["vat_rate"]))
        return values

<odoo>
  <record id="cron_ocost" model="ir.cron">
    <field name="name">Ocost: send and fill</field>
    <field name="model_id" ref="account.model_account_move"/>
    <field name="state">code</field>
    <field name="code">model._cron_ocost()</field>
    <field name="interval_number">5</field>
    <field name="interval_type">minutes</field>
  </record>
</odoo>

    @api.model
    def ocost_search(self, query, kind=None):
        """Bills by what their PDFs say: a list of (bill, page, text)."""
        params = {"q": query, "limit": 20}
        if kind:
            params["kind"] = kind
        found = self._ocost().get(f"{API}/search", params=params, timeout=15).json()
        out = []
        for r in found["results"]:
            bill = self.search([("ocost_file_id", "=", r["file_id"])], limit=1)
            passage = r["passages"][0] if r["passages"] else {}
            out.append((bill, passage.get("page"), passage.get("text")))
        return out