# -*- coding: utf-8 -*- import base64 import hashlib import json import re from odoo import api, fields, models from odoo.exceptions import UserError from odoo.addons.nacef_smdf_client.api import ( AgentRestNacefClient, MockNacefClient, NacefError, SchemaError, SMDFSignedTicket, SMDFStatus, SMDFTicketInfo, SICLogEntry, validate_ticket, ) # S-MDF status -> message shown to the taxpayer when a sale is blocked # (PROCTEST steps 4/7/8/15/19/21). _BLOCK_MESSAGES = { SMDFStatus.FACTORY.value: "Le contribuable doit obtenir un certificat électronique avant de " "réaliser des ventes.", SMDFStatus.CERT_REQUESTED.value: "La caisse doit être synchronisée avec le système NACEF avant de " "réaliser des ventes.", SMDFStatus.NOT_SYNCHRONIZED.value: "Il faut synchroniser la caisse avec le système NACEF avant de " "réaliser des ventes.", SMDFStatus.SUSPENDED.value: "La caisse a été suspendue. Vente impossible.", SMDFStatus.MAINTENANCE.value: "La caisse est en maintenance. Une synchronisation est nécessaire.", SMDFStatus.SYNCHRONIZATION_IN_PROGRESS.value: "Synchronisation en cours. Veuillez patienter.", } # A3 ticket schema (developers.nacef.tn, nacef-smdf-api-1.2.0, schema "Ticket"). # ``data_type``/``version`` are fixed consts required by the schema. _A3_DATA_TYPE = "ncf.cashier.operation" _A3_VERSION = "1.1.4" _SW_VERSION = "17.0.1.0.0" # A5 tax codification (Annexe A5): VAT 7% -> 10, VAT 19% -> 11, stamp -> 20. _A5_TAX_CODE = {7: "10", 19: "11"} # pos.payment.method -> A3 payment enum (collection_details.method). _A3_PAYMENT_ENUM = { "cash", "check", "bank_card", "restaurant_ticket", "mobile_payment", "contre_bon", "wire_transfer", } # Encaissement fields sealed by the inalterability chain (E1101/E1102): once an # order is hashed, changing any of these is forbidden by ``write``. _NACEF_PROTECTED_FIELDS = { "amount_total", "amount_tax", "amount_paid", "amount_return", "date_order", "pos_reference", "lines", "nacef_transaction_type", "nacef_operation_type", "nacef_origin_reference", "nacef_fiscal_advantage_ref", "nacef_fiscal_reference", "nacef_ticket_identifier", "nacef_imdf", "nacef_hash", "nacef_prev_hash", "nacef_secure_sequence", } class PosOrder(models.Model): _inherit = "pos.order" # transaction / ticket typing (E0403 / E0501) nacef_transaction_type = fields.Selection( [("SALE", "Vente"), ("TRAINING", "Formation")], default="SALE", copy=False) nacef_operation_type = fields.Selection( [("TICKET", "Normal"), ("PROFORMA", "Pro-forma"), ("REFUND", "Remboursement"), ("DUPLICATE", "Copie")], default="TICKET", copy=False) # fiscal protection results nacef_imdf = fields.Char(string="IMDF", copy=False, readonly=True) nacef_fiscal_reference = fields.Char( string="Fiscal reference", copy=False, readonly=True, help="Gapless chronological reference assigned by the fiscal " "sequence (E0402).") nacef_ticket_identifier = fields.Char( string="S-MDF ticket id", copy=False, readonly=True, help="Unique fiscal identifier returned by the S-MDF signature.") nacef_qr_code = fields.Char( string="QR code (base64)", copy=False, readonly=True) nacef_signed = fields.Boolean(copy=False, readonly=True) nacef_awaiting_signature = fields.Boolean( string="Awaiting browser signature", copy=False, readonly=True, help="Browser client mode only. The fiscal reference is allocated and " "the guards have passed, but the S-MDF signature has not come " "back from the till yet. Such an order is NOT a completed fiscal " "sale (E0302) and must be reconciled or voided.") nacef_online = fields.Boolean( string="Signed online", copy=False, readonly=True) nacef_origin_reference = fields.Char( string="Origin reference", copy=False, help="Reference of the original ticket for a DUPLICATE (E0505).") # E0506(l): fiscal advantage indicator. The attestation reference (achat en # suspension de taxes) is set when the customer benefits (AA); empty = SA. nacef_fiscal_advantage_ref = fields.Char( string="Fiscal-advantage attestation", copy=False, help="Référence de l'attestation d'achat en suspension de taxes. " "Renseignée = AA (avec avantage) ; vide = SA (sans avantage).") # inalterability chain (E1101/E1102) nacef_secure_sequence = fields.Integer(copy=False, readonly=True) nacef_prev_hash = fields.Char(copy=False, readonly=True) nacef_hash = fields.Char(copy=False, readonly=True, index=True) # E0404/E0505: once the original receipt is printed, further prints must go # through the "Ticket copie" duplicate flow. Persisted so a page refresh / # reopening the order can't unlock a free reprint. nacef_printed = fields.Boolean( string="Receipt printed", copy=False, readonly=True) # ------------------------------------------------------------------ # # inalterability enforcement (E1101/E1102) # ------------------------------------------------------------------ # def write(self, vals): sealed_fields = _NACEF_PROTECTED_FIELDS & set(vals) if sealed_fields: for order in self: if order.nacef_hash: raise UserError( "Cette transaction est scellée (inaltérabilité E1101) ; " "les champs %s ne peuvent plus être modifiés." % ", ".join(sorted(sealed_fields))) return super().write(vals) def unlink(self): # Sealed orders are immutable (E1101); the only sanctioned deletion is a # fiscal purge, which always archives first (E1401) and runs under the # ``nacef_purge`` context. if not self.env.context.get("nacef_purge"): if any(order.nacef_hash for order in self): raise UserError( "Une transaction scellée ne peut pas être supprimée " "(E1101).") return super().unlink() def _nacef_compute_hash(self, fiscal_ref, ticket_id, prev_hash, secure_seq): self.ensure_one() payload = { "imdf": self.nacef_imdf, "secure_sequence": secure_seq, "prev_hash": prev_hash, "fiscal_reference": fiscal_ref, "ticket_identifier": ticket_id, "amount_total": round(self.amount_total, 3), "amount_tax": round(self.amount_tax, 3), "date_order": str(self.date_order), "pos_reference": self.pos_reference, "operation_type": self.nacef_operation_type, "transaction_type": self.nacef_transaction_type, } blob = json.dumps(payload, sort_keys=True) return hashlib.sha256(blob.encode("utf-8")).hexdigest() @api.model def nacef_verify_inalterability(self, imdf, company=None): """Recompute the per-IMDF order chain; return (ok, first_broken_id).""" company = company or self.env.company orders = self.search([ ("company_id", "=", company.id), ("nacef_imdf", "=", imdf), ("nacef_hash", "!=", False), ], order="nacef_secure_sequence asc") prev = "GENESIS" for order in orders: expected = order._nacef_compute_hash( order.nacef_fiscal_reference, order.nacef_ticket_identifier, prev, order.nacef_secure_sequence) if order.nacef_prev_hash != prev or order.nacef_hash != expected: return False, order.id prev = order.nacef_hash return True, None # ------------------------------------------------------------------ # # order processing hook (server side) — sign or block # ------------------------------------------------------------------ # @api.model def create_from_ui(self, orders, draft=False): # The base sync response only carries id/pos_reference/account_move. # Augment it with the fiscal fields so the POS receipt can print the # fiscal id + QR (E0506 v) without an extra round trip. res = super().create_from_ui(orders, draft=draft) by_id = {o.id: o for o in self.browse( [r["id"] for r in res if r.get("id")])} for row in res: order = by_id.get(row.get("id")) if order: row.update({ "nacef_id": order.id, "nacef_signed": order.nacef_signed, "nacef_fiscal_reference": order.nacef_fiscal_reference, "nacef_ticket_identifier": order.nacef_ticket_identifier, "nacef_qr_code": order.nacef_qr_code, "nacef_imdf": order.nacef_imdf, "nacef_online": order.nacef_online, "nacef_operation_type": order.nacef_operation_type, "nacef_transaction_type": order.nacef_transaction_type, "nacef_printed": order.nacef_printed, "nacef_fiscal_advantage": order._nacef_fiscal_advantage(), "nacef_mention": order._nacef_mention(), }) return res def _process_order(self, order, draft, existing_order): order_id = super()._process_order(order, draft, existing_order) rec = self.browse(order_id) if isinstance(order, dict): data = order.get("data", {}) # E0506(l): fiscal-advantage attestation reference captured at POS. ref = data.get("nacef_fiscal_advantage_ref") if ref: rec.nacef_fiscal_advantage_ref = ref # E0501/E0502: Pro-forma ticket (a quote — signed as PROFORMA, # excluded from fiscal turnover). if data.get("nacef_proforma") and \ rec.nacef_operation_type == "TICKET": rec.nacef_operation_type = "PROFORMA" # Training mode (E0403/E1003): the POS flags the order as training, but # it may be activated ONLY by an administrator profile. A cashier- # supplied flag is refused (the sale is then a real, signed transaction) # and the attempt is traced. if isinstance(order, dict) and \ order.get("data", {}).get("nacef_training"): if self.env.user.has_group("point_of_sale.group_pos_manager"): rec.nacef_transaction_type = "TRAINING" elif rec._nacef_get_imdf(): rec._nacef_audit( "TRAINING", {"denied": "requires_admin_profile (E1003)", "user": self.env.user.login, "ref": rec.pos_reference}, imdf=rec._nacef_get_imdf(), ptype="ERREUR") rec._nacef_process_fiscal() return order_id def _nacef_mention(self): """Mandatory ticket mention (E0503/E0504/E0505).""" self.ensure_one() if self.nacef_transaction_type == "TRAINING": return "TICKET FORMATION" if self.nacef_operation_type == "DUPLICATE": return "TICKET COPIE" if self.nacef_operation_type == "REFUND": return "TICKET REMBOURSEMENT" if self.nacef_operation_type == "PROFORMA": return "PRO-FORMA" return "" def _nacef_process_fiscal(self): for order in self: if order.nacef_signed: continue imdf = order._nacef_get_imdf() if not imdf: # No IMDF configured -> the POS is not under fiscal enforcement. continue order.nacef_imdf = imdf # Auto-detect a refund (negative total) -> REFUND ticket (E0504). if order.amount_total < 0 and \ order.nacef_operation_type == "TICKET": order.nacef_operation_type = "REFUND" # E1202: no transaction may be recorded in a closed fiscal period. if self.env["nacef.fiscal.closing"].sudo().is_date_in_closed_period( imdf, order.date_order, order.company_id): order._nacef_audit( "CASHING", {"blocked": "closed_period", "ref": order.pos_reference}, imdf=imdf, ptype="ERREUR") raise UserError( "Impossible d'enregistrer une transaction sur une période " "clôturée (E1202).") if order.nacef_transaction_type == "TRAINING": # Formation: recorded and flagged, never signed (E1003). order._nacef_audit("CASHING", {"training": True, "ref": order.pos_reference}, imdf=imdf) continue if order._nacef_mode() == "browser": # Cloud POS: the Odoo server cannot reach the till's S-MDF, so # the signature is fetched by the cashier's browser and posted # back to /nacef/pos/sign/confirm. Run every guard now and # reserve the reference, but leave the order unsigned — it is # explicitly NOT a completed fiscal sale until confirmed. order._nacef_reserve_signature(imdf) continue order._nacef_sign(imdf) # ------------------------------------------------------------------ # # signing # ------------------------------------------------------------------ # def _nacef_signing_state(self, imdf): """Resolve the S-MDF state and enforce the E0302 sign-or-block guard. Shared by every client mode: whether the signature is fetched by this process or by the cashier's browser, a sale is refused unless the equipment is SYNCHRONIZED.""" self.ensure_one() # fiscal bookkeeping (state, sequence) runs elevated so a cashier's sale # can be signed without granting cashiers write access to fiscal models. state = self.env["nacef.smdf.state"].sudo().get_or_create( imdf, self.company_id) if not state.can_sign: self._nacef_audit( "CASHING", {"blocked": True, "status": state.status, "ref": self.pos_reference}, imdf=imdf, ptype="ERREUR") raise UserError(_BLOCK_MESSAGES.get( state.status, "La caisse ne peut pas signer les tickets (état %s)." % state.status)) return state def _nacef_sign(self, imdf, payment_source=None): self.ensure_one() state = self._nacef_signing_state(imdf) # Assign the gapless fiscal reference first so it is the ticket's # transaction id (A3 transaction.id / E0402). If signing fails the whole # savepoint rolls back, so the sequence is not consumed (no gap). fiscal_ref = self.env["nacef.fiscal.sequence"].sudo().next_value( imdf, self.company_id) client = self._nacef_client(imdf, state) info = self._nacef_ticket_info(fiscal_ref, payment_source=payment_source) try: signed = client.sign_ticket(info) except NacefError as err: self._nacef_audit( "SIGN_REQUEST", {"error": err.name, "code": err.code, "message": err.message, "ref": self.pos_reference}, imdf=imdf, ptype="ERREUR", module="lc.agent") raise UserError( "Signature du ticket impossible : %s" % err.message) from err return self._nacef_apply_signed(imdf, state, fiscal_ref, signed, client=client) def _nacef_apply_signed(self, imdf, state, fiscal_ref, signed, client=None): """Seal an order against a signature that has already been obtained. Split out of :meth:`_nacef_sign` so the browser client mode can reuse it verbatim: the S-MDF call is the only part that moves off the server, and everything that carries fiscal weight — the inalterability chain, the QR, the audit trail — still happens here, elevated, in one transaction.""" self.ensure_one() # inalterability chain (E1101): seal this order to the previous secured # order for the same company + IMDF. prev = self.search([ ("company_id", "=", self.company_id.id), ("nacef_imdf", "=", imdf), ("nacef_hash", "!=", False), ], order="nacef_secure_sequence desc", limit=1) prev_hash = prev.nacef_hash or "GENESIS" secure_seq = (prev.nacef_secure_sequence or 0) + 1 vals = { "nacef_fiscal_reference": fiscal_ref, "nacef_ticket_identifier": signed.ticket_identifier, "nacef_qr_code": self._nacef_qr_png(signed) or False, "nacef_signed": True, "nacef_awaiting_signature": False, "nacef_online": state.connection_state == "Online", "nacef_secure_sequence": secure_seq, "nacef_prev_hash": prev_hash, } vals["nacef_hash"] = self._nacef_compute_hash( fiscal_ref, signed.ticket_identifier, prev_hash, secure_seq) self.write(vals) # Persist the offline-ticket budget the S-MDF consumed for this # signature. Without this the budget was decremented on a throwaway # client instance and reset each sale, so the forced-resync-on- # exhaustion rule (NOT_SYNCHRONIZED) never fired in real usage. remaining = getattr(getattr(client, "manifest", None), "available_offline_tickets", None) if client else None if remaining is not None and remaining != state.available_offline_tickets: state.sudo().available_offline_tickets = remaining # traceability: audit trail (E0901) + mirror to S-MDF /log/ (SMDF-05) self._nacef_audit( "CASHING", {"ref": self.pos_reference, "fiscal_ref": fiscal_ref, "ticket_id": signed.ticket_identifier}, imdf=imdf) # In browser mode there is no server-side client to mirror through; the # till's own connector already carries the /log/ call. The audit trail # above is the authoritative record either way (E0901). if client is not None: try: client.log(SICLogEntry( module="lc.agent", operation="SIGN_REQUEST", level="INFO", message=signed.ticket_identifier)) except NacefError: pass # /log/ never blocks a sale return signed # ------------------------------------------------------------------ # # reprint lockdown -> "Ticket copie" (E0505 / E0501 DUPLICATE) # ------------------------------------------------------------------ # def action_nacef_create_duplicate(self): """A copy of a ticket is not a free reprint: it is a distinct recorded DUPLICATE transaction, stamped "Ticket copie", with a new fiscal number referencing the original (E0505). Reprinting the raw receipt must be disabled in the POS UI; this method is the only sanctioned copy path.""" self.ensure_one() if not self.nacef_signed: raise UserError("Seul un ticket signé peut être copié.") if self.nacef_operation_type == "DUPLICATE": raise UserError("Une copie ne peut pas être elle-même copiée.") imdf = self._nacef_get_imdf() duplicate = self.copy({ # core's pos_reference is copy=False; without it, the frontend # Orders screen crashes (_export_for_ui assumes a non-null string). # It must still differ from the original (E0501.iv: "un numéro # différent doit être affecté au ticket") — the -COPIE suffix # keeps the digit/dash run core's _export_for_ui parses intact. "pos_reference": (self.pos_reference or "") + "-COPIE", # core's name is copy=False (default '/'); its own create()/write() # only backfills it via ir.sequence when state == 'paid' (not # 'done'/'invoiced'), so a completed order's copy needs it set # explicitly or it prints as a bare "/". "name": self._compute_order_name(), "nacef_operation_type": "DUPLICATE", "nacef_transaction_type": self.nacef_transaction_type, "nacef_origin_reference": self.nacef_fiscal_reference, "nacef_imdf": imdf, "nacef_signed": False, "nacef_fiscal_reference": False, "nacef_ticket_identifier": False, "nacef_qr_code": False, "nacef_online": False, "nacef_secure_sequence": False, "nacef_prev_hash": False, "nacef_hash": False, # This is the same completed sale, re-issued — not a new draft # order — so it must keep the original's paid/done/invoiced state # (E1204 excludes it from turnover by operation_type, not state). "state": self.state, }) # payment_ids is copy=False (core): the duplicate carries no payment # of its own, so the mandatory "mode de règlement" (E0506.s) on its # ticket must be read from the original transaction it reproduces. duplicate._nacef_sign(imdf, payment_source=self) # config=False: a fiscal ticket is a thermal receipt, not a branded # A4 document — never gate it behind the company letterhead wizard. return self.env.ref( "nacef_pos.action_report_ticket_copy" ).report_action(duplicate, config=False) def _nacef_get_origin_order(self): """The sale a DUPLICATE re-issues (E0506.b), looked up by fiscal reference. Used at print time, in a fresh request where the Python ``self`` used at signature time (see action_nacef_create_duplicate) is no longer available, to display the original's payment method.""" self.ensure_one() if not self.nacef_origin_reference: return self.browse() return self.search([ ("company_id", "=", self.company_id.id), ("nacef_fiscal_reference", "=", self.nacef_origin_reference), ], limit=1) def _nacef_display_payments(self): """Payment lines to print on the ticket (E0506.s): a DUPLICATE has none of its own (payment_ids is copy=False) — fall back to the original sale it reproduces.""" self.ensure_one() if self.payment_ids: return self.payment_ids return self._nacef_get_origin_order().payment_ids def _nacef_cashier_name(self): """E0506(e): the operator "who took charge of the transaction data" — i.e. the actual cashier (pos_hr's employee_id, clocked in on the till), not user_id ("Responsible"), which is the account the POS session was opened under and stays the same across every cashier who used that session/till.""" self.ensure_one() employee = getattr(self, "employee_id", False) return (employee and employee.name) or self.user_id.name or "" # ------------------------------------------------------------------ # # helpers # ------------------------------------------------------------------ # def _nacef_get_imdf(self): self.ensure_one() config = self.config_id if config and config.nacef_effective_imdf: return config.nacef_effective_imdf return self.company_id.nacef_imdf or False # ------------------------------------------------------------------ # # browser client mode: signature fetched by the cashier's machine # ------------------------------------------------------------------ # def _nacef_mode(self): self.ensure_one() return (self.company_id.nacef_client_mode or self.env["ir.config_parameter"].sudo().get_param( "nacef.client_mode", "mock")) def _nacef_reserve_signature(self, imdf): """Run the guards and allocate the fiscal reference, without signing. The reference is consumed inside the same transaction that stores the order, so a browser that never comes back leaves a visible, auditable order awaiting signature rather than a hole in the sequence (E0402).""" self.ensure_one() self._nacef_signing_state(imdf) fiscal_ref = self.env["nacef.fiscal.sequence"].sudo().next_value( imdf, self.company_id) self.write({"nacef_fiscal_reference": fiscal_ref, "nacef_awaiting_signature": True}) self._nacef_audit( "SIGN_REQUEST", {"reserved": fiscal_ref, "ref": self.pos_reference}, imdf=imdf, module="lc.agent") return fiscal_ref def _nacef_signature_payload(self, payment_source=None): """The A3 ticket blob the browser must hand to the local S-MDF. Built server-side on purpose: schema validation (E0803) and the ticket contents must not be something the client can forge.""" self.ensure_one() if not self.nacef_awaiting_signature: raise UserError("Ce ticket n'attend pas de signature.") info = self._nacef_ticket_info(self.nacef_fiscal_reference, payment_source=payment_source) return info.to_dict() def _nacef_confirm_signature(self, response): """Seal the order with the signature the browser obtained. ``response`` is the S-MDF Agent's raw JSON reply, parsed by the same model class the server-side path uses, so a malformed or forged payload fails here rather than reaching the fiscal chain.""" self.ensure_one() if self.nacef_signed: return True # idempotent: a retried confirm must not double-sign if not self.nacef_awaiting_signature: raise UserError("Ce ticket n'attend pas de signature.") imdf = self.nacef_imdf or self._nacef_get_imdf() state = self._nacef_signing_state(imdf) # The SIC wraps replies in {object, errorCode, message}; the browser # relays that verbatim, so open the envelope the same way the # server-side client does before parsing the signature. try: payload = AgentRestNacefClient._unwrap(200, response or {}) except NacefError as err: self._nacef_audit( "SIGN_REQUEST", {"error": err.name, "code": err.code, "message": err.message, "ref": self.pos_reference}, imdf=imdf, ptype="ERREUR", module="lc.agent") raise UserError( "Signature refusée par le S-MDF : %s" % (err.message or err.name or "?")) from err try: signed = SMDFSignedTicket.from_dict(payload) except Exception as err: # noqa: BLE001 - any parse failure is fatal self._nacef_audit( "SIGN_REQUEST", {"error": "malformed_agent_response", "ref": self.pos_reference}, imdf=imdf, ptype="ERREUR", module="lc.agent") raise UserError( "Réponse du S-MDF illisible : %s" % err) from err if not signed.ticket_identifier: raise UserError("Le S-MDF n'a pas renvoyé d'identifiant de ticket.") self._nacef_apply_signed(imdf, state, self.nacef_fiscal_reference, signed, client=None) return True def _nacef_client(self, imdf, state): """Resolve a NacefClient. ``mock`` (default) is seeded from the persisted S-MDF state so signing behaves like the live equipment.""" mode = (self.company_id.nacef_client_mode or self.env["ir.config_parameter"].sudo().get_param( "nacef.client_mode", "mock")) if mode == "agent": url = (self.config_id.nacef_smdf_agent_url or self.company_id.nacef_smdf_agent_default_url or None) # url=None -> AgentRestNacefClient defaults to http://localhost:10006 return AgentRestNacefClient(url) # mock seeded from the persisted state client = MockNacefClient( imdf=imdf, offline_capacity=max(state.available_offline_tickets, 1)) m = client.manifest m.status = state.status or SMDFStatus.SYNCHRONIZED.value m.state = state.connection_state or "Online" m.available_offline_tickets = state.available_offline_tickets or 0 ci = m.certificate_info ci.cert_request_status = state.cert_request_status or \ "CERTIFICATE_GENERATED" ci.revoked = state.cert_revoked ci.expired = state.cert_expired return client # ------------------------------------------------------------------ # # A3 ticket JSON (developers.nacef.tn "Ticket" schema, v1.2.0) # ------------------------------------------------------------------ # @staticmethod def _nacef_millimes(amount): """TND amount -> integer millimes (3 decimals), as the A3 schema wants.""" return int(round((amount or 0.0) * 1000)) @staticmethod def _nacef_family_code(product): """A3 product.family_code (Annexe A4): 2..8 chars. Uses the category's configured A4 code (pos.category.nacef_family_code) when set; otherwise derives a schema-valid fallback from the category name until the real A4 codes are entered (E0802).""" cat = product.pos_categ_ids[:1] code = (cat.nacef_family_code or "").strip() if cat else "" if not 2 <= len(code) <= 8: base = "".join(c for c in (cat.name if cat else "") if c.isalnum()) code = base[:8].upper() if len(base) >= 2 else "01" return code @staticmethod def _nacef_tax_split(tax): """account.tax -> (type, tax_code, value) per A5 codification. An unmapped rate used to fall back to "00", which is not a valid A5 code. Our own schema accepted it (any non-empty string) but the S-MDF rejects the WHOLE ticket with 506 TICKET.SCHEMA_INVALID — and the cashier sees a refused sale with no clue which tax caused it. Fail here instead, naming the tax, so it is a configuration error rather than a mystery at the till. """ if tax.amount_type == "percent": code = (tax.nacef_a5_code or "").strip() \ or _A5_TAX_CODE.get(int(round(tax.amount))) if not code: raise UserError( "Taxe « %s » (%.2f %%) : aucun code NACEF (Annexe A5).\n\n" "Les taux reconnus automatiquement sont 7 %% et 19 %%. " "Pour tout autre taux, renseignez « Code taxe NACEF (A5) » " "sur la taxe (Comptabilité → Configuration → Taxes).\n\n" "Sans code A5 valide, le S-MDF refuse le ticket entier " "(506 TICKET.SCHEMA_INVALID)." % (tax.name, tax.amount)) return "percent", code, tax.amount return "fixed", (tax.nacef_a5_code or "20"), tax.amount # droit de timbre @staticmethod def _nacef_payment_method(pm): name = (pm.name or "").lower() if pm.is_cash_count: return "cash" for key, val in (("carte", "bank_card"), ("card", "bank_card"), ("bank", "bank_card"), ("chèq", "check"), ("cheq", "check"), ("check", "check"), ("restaurant", "restaurant_ticket"), ("mobile", "mobile_payment"), ("virement", "wire_transfer"), ("transfer", "wire_transfer"), ("contre", "contre_bon")): if key in name: return val return "cash" def _nacef_qr_png(self, signed): """Base64 PNG for the receipt QR (ISO/IEC 18004). In agent mode the S-MDF returns the real QR image; in mock mode we render a real, scannable QR from the QR content so the receipt always shows a valid code.""" if not signed.qrcode_image: return False raw = base64.b64decode(signed.qrcode_image) if raw[:4] == b"\x89PNG": return signed.qrcode_image # real S-MDF QR image content = raw.decode("utf-8", "replace") or signed.ticket_identifier png = self.env["ir.actions.report"].sudo().barcode( "QR", content, width=220, height=220) return base64.b64encode(png).decode("ascii") def _nacef_agent_mac(self): agent = self.env["nacef.smdf.agent"].sudo().search( [("config_id", "=", self.config_id.id)], limit=1) if not agent: agent = self.env["nacef.smdf.agent"].sudo().search( [("company_id", "=", self.company_id.id), ("imdf", "=", self.nacef_imdf)], limit=1) # null MAC fallback keeps the ticket schema-valid (minLength 3) when no # agent is registered; a real till always has a declared MAC. return agent.mac_address or "00:00:00:00:00:00" def _nacef_fiscal_advantage(self): """E0506(l) indicator: AA (avec avantage) or SA (sans avantage).""" self.ensure_one() ref = (self.nacef_fiscal_advantage_ref or "").strip() return "AA" if 3 <= len(ref) <= 48 else "SA" def _nacef_ticket_info(self, fiscal_ref, payment_source=None): self.ensure_one() ticket = self._nacef_build_ticket_dict( fiscal_ref, payment_source=payment_source) # E0803: the JSON object must be valid against the NACEF schema before # it is transmitted for signing. Block (and trace) an invalid ticket. try: validate_ticket(ticket) except SchemaError as err: self._nacef_audit( "SIGN_REQUEST", {"schema_invalid": err.errors[:8], "ref": self.pos_reference}, imdf=self.nacef_imdf, ptype="ERREUR", module="lc.agent") raise UserError( "Ticket fiscal non conforme au schéma NACEF (E0803) : %s" % err) from err blob = base64.b64encode( json.dumps(ticket, ensure_ascii=True, sort_keys=True) .encode("utf-8")).decode("ascii") summary = ticket["sale_summary"] return SMDFTicketInfo( base64_ticket=blob, total_ht=summary["total_excl_tax"], total_tax=summary["total_tax"], operation_type=self.nacef_operation_type or "TICKET", transaction_type=self.nacef_transaction_type or "SALE") def _nacef_build_ticket_dict(self, fiscal_ref, payment_source=None): """Build the A3 fiscal ticket JSON (schema "Ticket", nacef-smdf-api 1.2.0). Monetary values are integer millimes (TND, 3 decimals). ``payment_source``: order to read payment_ids/amount_return from (E0506.s). A DUPLICATE has no payments of its own (E0505: it is a re-issue of an already-settled sale, not a new one) — its "mode de règlement" must show the original's payment method.""" self.ensure_one() payment_source = payment_source or self company = self.company_id config = self.config_id M = self._nacef_millimes sale_details = [] tax_totals = {} # tax_code -> millimes for line in self.lines: qty = line.qty or 0.0 unit_ht = (line.price_subtotal / qty) if qty else line.price_subtotal taxation = [] if line.tax_ids and qty: computed = line.tax_ids.compute_all( line.price_unit, self.currency_id, qty, product=line.product_id) for tinfo in computed.get("taxes", []): tax = self.env["account.tax"].browse(tinfo["id"]) ttype, code, value = self._nacef_tax_split(tax) taxation.append( {"type": ttype, "value": value, "tax_code": code}) tax_totals[code] = tax_totals.get(code, 0) + M(tinfo["amount"]) discount = line.discount or 0.0 sale_details.append({ "product": { "family_code": self._nacef_family_code(line.product_id), "name": line.product_id.display_name or "", "price_pre_tax": M(unit_ht), }, "taxation": taxation, "quantity": max(1, int(round(qty))), # schema: integer >= 1 "discount_per_unit": { "percent": discount, "value": M(unit_ht * discount / 100.0), }, }) tax_summary = [{"tax_code": c, "total_amount": a} for c, a in sorted(tax_totals.items())] # Odoo core adds a synthetic negative payment line (amount = # -amount_return) to book the change given back to the customer # (point_of_sale/models/pos_order.py). That is not an encaissement — # the schema requires collection_details amounts >= 1 — and is # already represented below via ``returned_change``. collection = [{"method": self._nacef_payment_method(p.payment_method_id), "amount": M(p.amount)} for p in payment_source.payment_ids if p.amount > 0] returned_change = ([{"method": "cash", "amount": M(payment_source.amount_return)}] if (payment_source.amount_return or 0) > 0 else []) total_incl = M(self.amount_total) total_tax = M(self.amount_tax) # A "patentée" customer (PP) needs a valid Tunisian matricule # (^\d{7}[A-Z]$); otherwise the customer is non-patentée (NP) and the # optional id is omitted. vat = ((self.partner_id.vat or "").strip() if self.partner_id else "") customer = {} if re.match(r"^\d{7}[A-Z]$", vat): customer["id_type"] = "PP" customer["id"] = vat else: customer["id_type"] = "NP" # E0506(l): AA (avec avantage) -> attestation reference (schema # fiscal_advantages, 3..48 chars); SA (sans avantage) -> omitted. ref = (self.nacef_fiscal_advantage_ref or "").strip() if 3 <= len(ref) <= 48: customer["fiscal_advantages"] = ref # establishment reference must be 3 digits (^\d{3}$) store_ref = config.nacef_store_id or company.nacef_store_id or "" if not (store_ref.isdigit() and len(store_ref) == 3): store_ref = "000" return { "data_type": _A3_DATA_TYPE, "version": _A3_VERSION, "transaction": { "id": fiscal_ref, "timestamp": (self.date_order or fields.Datetime.now() ).isoformat(), "operation": { "op_type": self.nacef_operation_type or "TICKET", "context": self.nacef_transaction_type or "SALE", # schema: a DUPLICATE must reference the original transaction. **({"duplicated_transaction_identifier": self.nacef_origin_reference} if self.nacef_operation_type == "DUPLICATE" and self.nacef_origin_reference else {}), }, "originator": { "agent_identifier": self._nacef_agent_mac(), "imdf": self.nacef_imdf or "", "cash_register_serialnumber": config.nacef_ce_serial or "", "cash_register_software": _SW_VERSION, "accreditation_reference": company.nacef_accreditation_reference or "", }, }, "merchant_identity": { "id": company.nacef_matricule_fiscal or company.vat or "", "id_type": "MF", "taxpayer_establishment": { "commercial_name": company.nacef_nom_commercial or company.name or "", "reference": store_ref, "address": company.street or "", "city": company.city or "", }, }, "customer_identity": customer, "sale_details": sale_details, "tax_summary": tax_summary, "general_discount": {"percent": 0.0, "value": 0}, "additional_tax": {"type": "fixed", "value": 0, "tax_code": "20"}, "payment_details": { "collection_details": collection, "returned_change": returned_change, }, "sale_summary": { "total_excl_tax": total_incl - total_tax, "total_incl_tax": total_incl, "total_tax": total_tax, }, "delivery_details": {"type": "SELF_PICKUP"}, } def _nacef_audit(self, operation, message, imdf=None, ptype="INFO", module="lc"): self.env["nacef.audit.log"].log( module_code=module, operation_code=operation, message=message, ptype=ptype, imdf=imdf or self.nacef_imdf, company=self.company_id) # ------------------------------------------------------------------ # # expose fiscal fields to the POS frontend (receipt QR) # ------------------------------------------------------------------ # @api.model def nacef_mark_printed(self, order_ids): """Persist that the original receipt was printed (E0404/E0505), so a page refresh or reopening the order cannot unlock a free reprint. Called by the POS right after the first successful print.""" marked = 0 for o in self.browse(order_ids).exists().filtered( lambda r: r.nacef_signed and not r.nacef_printed): o.sudo().nacef_printed = True o._nacef_audit("PRINTER", {"printed": o.pos_reference, "fiscal_ref": o.nacef_fiscal_reference}) marked += 1 return marked def _nacef_ui_payload(self): """Fiscal data the POS frontend needs for one order. Single source of truth, because it is consumed by two different paths: ``_export_for_ui`` (order list, reprints) and the /nacef/pos/fiscal/state route the frontend calls after a sync. That route exists because ``create_from_ui`` does NOT return ``_export_for_ui``: in Odoo 17 it ends with search_read(fields=['id', 'pos_reference', 'account_move']) so every nacef_* key added to _export_for_ui is absent from the sync response. Reading fiscal state off that response silently yields ``undefined`` — an unsigned ticket with no error anywhere. """ self.ensure_one() return { "id": self.id, "pos_reference": self.pos_reference, "nacef_fiscal_reference": self.nacef_fiscal_reference, "nacef_ticket_identifier": self.nacef_ticket_identifier, "nacef_qr_code": self.nacef_qr_code, "nacef_imdf": self.nacef_imdf, "nacef_signed": self.nacef_signed, # Tells the frontend it must go and fetch the signature from the # till's connector before the receipt may be printed. "nacef_awaiting_signature": self.nacef_awaiting_signature, "nacef_agent_url": (self.config_id.nacef_smdf_agent_url or self.company_id.nacef_smdf_agent_default_url or "http://127.0.0.1:10016"), # A signature on an online S-MDF is not a local computation: the # equipment may reach the NACEF back end before answering. 15s was # far too tight and aborted valid signatures. Tunable per tenant # via the nacef.sign_timeout_ms system parameter. "nacef_sign_timeout_ms": int( self.env["ir.config_parameter"].sudo().get_param( "nacef.sign_timeout_ms", 90000)), "nacef_online": self.nacef_online, "nacef_operation_type": self.nacef_operation_type, "nacef_transaction_type": self.nacef_transaction_type, "nacef_printed": self.nacef_printed, "nacef_fiscal_advantage": self._nacef_fiscal_advantage(), "nacef_mention": self._nacef_mention(), } def _export_for_ui(self, order): result = super()._export_for_ui(order) result.update(order._nacef_ui_payload()) return result class PosCategory(models.Model): _inherit = "pos.category" nacef_family_code = fields.Char( string="Code famille (A4)", size=8, help="Code famille produit selon l'Annexe A4 (2 à 8 caractères), " "imprimé dans le ticket fiscal (family_code, E0802). Si vide, un " "code est dérivé du nom de la catégorie.") @api.constrains("nacef_family_code") def _check_nacef_family_code(self): for cat in self: code = (cat.nacef_family_code or "").strip() if code and not 2 <= len(code) <= 8: raise UserError( "Le code famille A4 doit comporter entre 2 et 8 " "caractères (catégorie « %s »)." % cat.name)