from __future__ import annotations import json import uuid from collections import defaultdict from datetime import datetime, timezone from decimal import Decimal, InvalidOperation from typing import Any, Dict, List, Optional, Tuple from sqlalchemy import text from app.config import settings from app.db import engine from app.odoo_client import OdooClient _SCHEMA_READY = False def _json(value: Any) -> str: return json.dumps(value or {}, ensure_ascii=False, default=str) def _money(value: Any, default: str = "0") -> str: try: if value is None or str(value).strip() == "": return f"{Decimal(default):.2f}" return f"{Decimal(str(value).replace(',', '.')).quantize(Decimal('0.01'))}" except (InvalidOperation, ValueError): return f"{Decimal(default):.2f}" def _m2o_id(value: Any) -> Optional[int]: if isinstance(value, (list, tuple)) and value: try: return int(value[0]) except Exception: return None if isinstance(value, int): return int(value) return None def _m2o_name(value: Any) -> str: if isinstance(value, (list, tuple)) and len(value) > 1: return str(value[1] or "") return "" def ensure_odoo_schema() -> None: global _SCHEMA_READY if _SCHEMA_READY: return with engine.begin() as conn: conn.execute(text(""" CREATE TABLE IF NOT EXISTS odoo_sync_runs ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), sync_type TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'running', started_at TIMESTAMPTZ NOT NULL DEFAULT now(), finished_at TIMESTAMPTZ, total_seen INTEGER NOT NULL DEFAULT 0, total_changed INTEGER NOT NULL DEFAULT 0, total_errors INTEGER NOT NULL DEFAULT 0, message TEXT, payload JSONB NOT NULL DEFAULT '{}'::jsonb ) """)) conn.execute(text("CREATE INDEX IF NOT EXISTS idx_odoo_sync_runs_type_started ON odoo_sync_runs(sync_type, started_at DESC)")) conn.execute(text("CREATE INDEX IF NOT EXISTS idx_products_metadata_odoo_id ON products ((metadata->'odoo'->>'product_id'))")) conn.execute(text("CREATE INDEX IF NOT EXISTS idx_products_metadata_odoo_sync ON products ((metadata->'odoo'->>'last_synced_at'))")) _SCHEMA_READY = True def test_odoo_connection() -> Dict[str, Any]: client = OdooClient() version = client.version() uid = client.authenticate() counts = {} for model in ["product.product", "stock.quant", "mrp.bom", "mrp.production", "sale.order"]: try: counts[model] = client.count(model, []) except Exception as exc: counts[model] = f"erro: {exc}" return { "ok": True, "base_url": settings.odoo_base_url, "db": settings.odoo_db, "username": settings.odoo_username, "uid": uid, "version": version, "counts": counts, } def _start_sync(sync_type: str, payload: Optional[Dict[str, Any]] = None) -> str: ensure_odoo_schema() sync_id = str(uuid.uuid4()) with engine.begin() as conn: conn.execute(text(""" INSERT INTO odoo_sync_runs (id, sync_type, status, payload) VALUES (CAST(:id AS UUID), :sync_type, 'running', CAST(:payload AS JSONB)) """), {"id": sync_id, "sync_type": sync_type, "payload": _json(payload or {})}) return sync_id def _finish_sync(sync_id: str, *, status: str, total_seen: int = 0, total_changed: int = 0, total_errors: int = 0, message: str = "", payload: Optional[Dict[str, Any]] = None) -> None: with engine.begin() as conn: conn.execute(text(""" UPDATE odoo_sync_runs SET status = :status, finished_at = now(), total_seen = :total_seen, total_changed = :total_changed, total_errors = :total_errors, message = :message, payload = payload || CAST(:payload AS JSONB) WHERE id = CAST(:id AS UUID) """), { "id": sync_id, "status": status, "total_seen": int(total_seen), "total_changed": int(total_changed), "total_errors": int(total_errors), "message": message, "payload": _json(payload or {}), }) def list_odoo_sync_runs(limit: int = 20) -> List[Dict[str, Any]]: ensure_odoo_schema() with engine.begin() as conn: rows = conn.execute(text(""" SELECT id::text, sync_type, status, started_at, finished_at, total_seen, total_changed, total_errors, message, payload FROM odoo_sync_runs ORDER BY started_at DESC LIMIT :limit """), {"limit": int(limit)}).mappings().all() return [dict(row) for row in rows] def _fetch_stock_by_product(client: OdooClient, *, limit: int = 10000) -> Dict[int, Dict[str, float]]: fields = ["product_id", "location_id", "quantity", "reserved_quantity"] try: quants = client.search_read( "stock.quant", [["location_id.usage", "=", "internal"]], fields, limit=limit, context={"active_test": False}, ) except Exception: fields = ["product_id", "location_id", "quantity"] quants = client.search_read( "stock.quant", [["location_id.usage", "=", "internal"]], fields, limit=limit, context={"active_test": False}, ) by_product: Dict[int, Dict[str, float]] = defaultdict(lambda: {"quantity": 0.0, "reserved": 0.0}) for q in quants: pid = _m2o_id(q.get("product_id")) if not pid: continue by_product[pid]["quantity"] += float(q.get("quantity") or 0) by_product[pid]["reserved"] += float(q.get("reserved_quantity") or 0) for vals in by_product.values(): vals["available"] = vals["quantity"] - vals["reserved"] return by_product def _fetch_bom_index(client: OdooClient, *, limit: int = 5000) -> Tuple[Dict[int, int], Dict[int, int]]: boms = client.search_read( "mrp.bom", [], ["id", "product_id", "product_tmpl_id", "type", "active"], limit=limit, context={"active_test": False}, ) by_product: Dict[int, int] = defaultdict(int) by_template: Dict[int, int] = defaultdict(int) for bom in boms: product_id = _m2o_id(bom.get("product_id")) template_id = _m2o_id(bom.get("product_tmpl_id")) if product_id: by_product[product_id] += 1 if template_id: by_template[template_id] += 1 return dict(by_product), dict(by_template) def _upsert_product(row: Dict[str, Any], metadata: Dict[str, Any]) -> bool: odoo_product_id = int(row["id"]) sku = str(row.get("default_code") or "").strip() if not sku: sku = f"ODOO-{odoo_product_id}" name = str(row.get("display_name") or row.get("name") or sku).strip() category = _m2o_name(row.get("categ_id")) or "Odoo" price = _money(row.get("lst_price") if "lst_price" in row else row.get("list_price")) with engine.begin() as conn: result = conn.execute(text(""" INSERT INTO products ( id, sku, name, category, description, default_unit_price, vat_rate, active, metadata, created_at, updated_at ) VALUES ( CAST(:id AS UUID), :sku, :name, :category, :description, :default_unit_price, 23, :active, CAST(:metadata AS JSONB), now(), now() ) ON CONFLICT (sku) DO UPDATE SET name = EXCLUDED.name, category = EXCLUDED.category, default_unit_price = EXCLUDED.default_unit_price, active = EXCLUDED.active, metadata = products.metadata || EXCLUDED.metadata, updated_at = now() RETURNING id """), { "id": str(uuid.uuid4()), "sku": sku, "name": name, "category": category, "description": str(row.get("description_sale") or ""), "default_unit_price": price, "active": bool(row.get("active", True)), "metadata": _json({"odoo": metadata}), }) return result.rowcount > 0 def sync_odoo_products(*, limit: int = 500, include_inactive: bool = True) -> Dict[str, Any]: ensure_odoo_schema() sync_id = _start_sync("products", {"limit": limit, "include_inactive": include_inactive}) total_seen = 0 total_changed = 0 total_errors = 0 errors: List[str] = [] try: client = OdooClient() client.authenticate() stock_by_product = _fetch_stock_by_product(client) bom_by_product, bom_by_template = _fetch_bom_index(client) fields = [ "id", "display_name", "default_code", "active", "lst_price", "standard_price", "categ_id", "product_tmpl_id", "sale_ok", "purchase_ok", "type", "description_sale", ] try: products = client.search_read( "product.product", [["sale_ok", "=", True]], fields, limit=int(limit), context={"active_test": not include_inactive}, ) except Exception: fallback_fields = ["id", "display_name", "default_code", "active", "lst_price", "categ_id", "product_tmpl_id", "sale_ok", "purchase_ok"] products = client.search_read( "product.product", [["sale_ok", "=", True]], fallback_fields, limit=int(limit), context={"active_test": not include_inactive}, ) synced_at = datetime.now(timezone.utc).isoformat() for product in products: total_seen += 1 try: product_id = int(product["id"]) template_id = _m2o_id(product.get("product_tmpl_id")) stock = stock_by_product.get(product_id, {"quantity": 0.0, "reserved": 0.0, "available": 0.0}) bom_count = int(bom_by_product.get(product_id, 0) + (bom_by_template.get(template_id, 0) if template_id else 0)) metadata = { "source": "odoo", "product_id": product_id, "template_id": template_id, "last_synced_at": synced_at, "sale_ok": bool(product.get("sale_ok")), "purchase_ok": bool(product.get("purchase_ok")), "type": product.get("type"), "cost": product.get("standard_price"), "stock": { "quantity_on_hand": stock["quantity"], "reserved": stock["reserved"], "available": stock["available"], }, "has_bom": bom_count > 0, "bom_count": bom_count, } if _upsert_product(product, metadata): total_changed += 1 except Exception as exc: total_errors += 1 errors.append(f"Produto {product.get('id')}: {exc}") status = "success" if total_errors == 0 else "partial" _finish_sync(sync_id, status=status, total_seen=total_seen, total_changed=total_changed, total_errors=total_errors, message="sync produtos concluído", payload={"errors": errors[:20]}) return { "ok": total_errors == 0, "sync_id": sync_id, "status": status, "total_seen": total_seen, "total_changed": total_changed, "total_errors": total_errors, "errors": errors[:20], } except Exception as exc: _finish_sync(sync_id, status="failed", total_seen=total_seen, total_changed=total_changed, total_errors=total_errors + 1, message=str(exc), payload={"errors": errors[:20]}) raise def get_odoo_product_snapshot(limit: int = 20) -> Dict[str, Any]: ensure_odoo_schema() with engine.begin() as conn: stats = conn.execute(text(""" SELECT count(*) FILTER (WHERE metadata ? 'odoo') AS synced_products, count(*) FILTER (WHERE COALESCE((metadata->'odoo'->>'has_bom')::boolean, false)) AS products_with_bom, count(*) FILTER (WHERE COALESCE((metadata->'odoo'->'stock'->>'available')::numeric, 0) > 0) AS products_with_available_stock, max(metadata->'odoo'->>'last_synced_at') AS last_synced_at FROM products """)).mappings().first() rows = conn.execute(text(""" SELECT id::text, sku, name, category, default_unit_price, active, metadata->'odoo' AS odoo FROM products WHERE metadata ? 'odoo' ORDER BY name LIMIT :limit """), {"limit": int(limit)}).mappings().all() return { "stats": dict(stats or {}), "products": [dict(row) for row in rows], "sync_runs": list_odoo_sync_runs(limit=10), } # === ClientFlow Odoo physical status integration === def _compact_m2o(value): if isinstance(value, (list, tuple)) and len(value) >= 2: return {"id": value[0], "name": value[1]} if isinstance(value, int): return {"id": value, "name": ""} return {"id": None, "name": ""} def _find_odoo_sale_order_link(opportunity_id: str) -> dict: with engine.begin() as conn: row = conn.execute(text(""" SELECT opportunity_id::text, external_id, external_name, external_url, status, payload FROM operation_links WHERE opportunity_id = CAST(:opportunity_id AS UUID) AND system = 'odoo' AND external_type = 'sale_order' LIMIT 1 """), {"opportunity_id": opportunity_id}).mappings().first() return dict(row or {}) def _find_sale_order(client: OdooClient, external_id: str = "", external_name: str = "") -> dict: fields = ["id", "name", "state", "partner_id", "amount_total", "date_order"] if external_id and str(external_id).isdigit(): rows = client.search_read("sale.order", [["id", "=", int(external_id)]], fields, limit=1) if rows: return dict(rows[0]) for ref in [external_name, external_id]: ref = str(ref or "").strip() if not ref: continue rows = client.search_read("sale.order", [["name", "=", ref]], fields, limit=1) if rows: return dict(rows[0]) return {} def _search_odoo_pickings(client: OdooClient, sale_name: str) -> list: if not sale_name: return [] fields = ["id", "name", "state", "origin", "picking_type_id", "scheduled_date", "date_done"] try: rows = client.search_read("stock.picking", [["origin", "ilike", sale_name]], fields, limit=100, order="id desc") except Exception: rows = [] return [dict(r) for r in rows] def _search_odoo_productions(client: OdooClient, sale_name: str) -> list: if not sale_name: return [] fields = ["id", "name", "state", "origin", "product_id", "product_qty", "date_start", "date_finished"] try: rows = client.search_read("mrp.production", [["origin", "ilike", sale_name]], fields, limit=100, order="id desc") except Exception: rows = [] return [dict(r) for r in rows] def _is_outgoing_picking(picking: dict) -> bool: name = str(picking.get("name") or "").upper() picking_type = _m2o_name(picking.get("picking_type_id")).lower() if "/OUT/" in name or name.startswith("WH/OUT"): return True keywords = ["delivery", "outgoing", "entrega", "expedição", "expedicao", "saída", "saida"] return any(k in picking_type for k in keywords) def _derive_physical_status(sale_order: dict, pickings: list, productions: list) -> dict: sale_state = str(sale_order.get("state") or "") outgoing = [p for p in pickings if _is_outgoing_picking(p)] or pickings picking_states = {str(p.get("state") or "") for p in outgoing} production_states = {str(mo.get("state") or "") for mo in productions} if sale_state in {"cancel", "cancelled"}: return { "physical_status": "cancelled", "label": "Cancelada", "reason": "A venda no Odoo está cancelada.", "ready_to_ship": False, "next_action": "Rever oportunidade no ClientFlow.", "stage": None, } if outgoing and all(str(p.get("state") or "") == "done" for p in outgoing): return { "physical_status": "shipped", "label": "Expedida no Odoo", "reason": "A entrega/picking no Odoo está concluída.", "ready_to_ship": False, "next_action": "Processo pronto para conclusão no ClientFlow quando fatura enviada e pagamento confirmado.", "stage": "SHIPPED", } if any(state == "assigned" for state in picking_states): return { "physical_status": "picking_assigned", "label": "Picking reservado — validação física pendente", "reason": "Odoo assigned indica stock reservado/disponível, mas não confirma que a encomenda foi fisicamente validada.", "ready_to_ship": False, "picking_reserved": True, "next_action": "Confirmar a preparação física da encomenda antes de criar envio/tracking.", "stage": "ORDER_PREPARATION", } if any(state in {"progress", "to_close", "confirmed"} for state in production_states): return { "physical_status": "in_production", "label": "A aguardar WH/OUT", "reason": "Ainda não existe entrega/WH-OUT pronta ou concluída; WH/MO é detalhe técnico.", "ready_to_ship": False, "next_action": "Aguardar encomenda/WH-OUT no Odoo.", "stage": "ODOO_ORDER_CREATED", } if any(state in {"waiting", "confirmed"} for state in picking_states): return { "physical_status": "waiting_stock", "label": "A aguardar stock/preparação", "reason": "A entrega ainda não está disponível para despacho.", "ready_to_ship": False, "next_action": "Aguardar stock, compra ou produção no Odoo.", "stage": None, } if sale_state in {"draft", "sent"}: return { "physical_status": "quote_only", "label": "Cotação no Odoo", "reason": "A venda ainda não está confirmada no Odoo.", "ready_to_ship": False, "next_action": "Confirmar venda/pagamento antes de preparar.", "stage": None, } if sale_state in {"sale", "done"}: return { "physical_status": "order_created", "label": "Venda criada", "reason": "Venda confirmada, mas sem picking pronto identificado.", "ready_to_ship": False, "next_action": "Verificar preparação física no Odoo.", "stage": "ODOO_ORDER_CREATED", } return { "physical_status": "unknown", "label": "Estado desconhecido", "reason": "Não foi possível interpretar o estado físico a partir do Odoo.", "ready_to_ship": False, "next_action": "Rever venda diretamente no Odoo.", "stage": None, } def _safe_apply_odoo_derived_stage(opportunity_id: str, stage: str, *, reason: str = "") -> dict: """Apply an Odoo-derived stage, including one guarded physical rewind. ``picking assigned`` is stronger evidence than a stale ClientFlow stage. It may safely move READY_TO_SHIP/SHIPMENT_CREATED back to ORDER_PREPARATION only when no physical validation, shipment or tracking exists. No commercial/financial stage is rewound. """ stage = str(stage or "").strip().upper() if not stage: return {"changed": False, "reason": "no_stage"} if stage != "ORDER_PREPARATION": if stage in {"IN_PRODUCTION", "READY_TO_SHIP", "SHIPMENT_CREATED", "SHIPPED"}: try: from app.opportunity_service import set_opportunity_stage changed = set_opportunity_stage( opportunity_id, stage, note=reason or "Estado físico atualizado pelo Odoo.", created_by="odoo_sync" ) return {"changed": bool(changed), "stage": stage, "mode": "normal"} except Exception as exc: return {"changed": False, "stage": stage, "reason": str(exc)} return {"changed": False, "stage": stage, "reason": "stage_not_managed"} with engine.begin() as conn: current = conn.execute(text(""" SELECT id::text, stage, status, COALESCE(metadata, '{}'::jsonb) AS metadata FROM opportunities WHERE id = CAST(:opportunity_id AS UUID) FOR UPDATE """), {"opportunity_id": opportunity_id}).mappings().first() if not current: return {"changed": False, "reason": "opportunity_not_found"} guards = conn.execute(text(""" SELECT EXISTS ( SELECT 1 FROM operation_links WHERE opportunity_id = CAST(:opportunity_id AS UUID) AND system = 'odoo' AND external_type = 'physical_validation' AND status IN ('validated','ready_to_ship') ) AS physical_validated, EXISTS ( SELECT 1 FROM shipments WHERE opportunity_id = CAST(:opportunity_id AS UUID) AND (COALESCE(external_reference,'') <> '' OR COALESCE(tracking_code,'') <> '' OR status NOT IN ('draft','cancelled','failed')) ) AS shipment_exists, EXISTS ( SELECT 1 FROM operation_links WHERE opportunity_id = CAST(:opportunity_id AS UUID) AND ((system = 'packlink' AND external_type = 'shipment') OR external_type IN ('shipment','tracking')) AND status NOT IN ('not_created','cancelled','failed') ) AS shipment_link_exists """), {"opportunity_id": opportunity_id}).mappings().first() or {} if guards.get("physical_validated") or guards.get("shipment_exists") or guards.get("shipment_link_exists"): return { "changed": False, "stage": current.get("stage"), "reason": "physical_validation_or_shipment_exists", "guards": dict(guards), } old_stage = str(current.get("stage") or "") allowed_current = { "ODOO_ORDER_CREATED", "IN_PRODUCTION", "ORDER_PREPARATION", "READY_TO_SHIP", "SHIPMENT_CREATED", } if old_stage not in allowed_current: return {"changed": False, "stage": old_stage, "reason": "commercial_stage_not_rewindable"} conn.execute(text(""" UPDATE opportunities SET stage = 'ORDER_PREPARATION', status = 'open', last_action_code = 'VALIDATE_PHYSICAL_ORDER', metadata = COALESCE(metadata, '{}'::jsonb) || CAST(:metadata AS JSONB), updated_at = now() WHERE id = CAST(:opportunity_id AS UUID) """), { "opportunity_id": opportunity_id, "metadata": _json({ "odoo_physical_stage_corrected": True, "odoo_physical_stage_from": old_stage, "odoo_physical_stage_to": "ORDER_PREPARATION", "odoo_physical_stage_reason": reason or "picking_assigned_without_physical_validation", "odoo_physical_stage_version": "v4928.1.5.132", }), }) superseded = conn.execute(text(""" UPDATE tasks SET status = 'ignored', done_at = COALESCE(done_at, now()), done_by = COALESCE(done_by, 'odoo_sync'), metadata = COALESCE(metadata, '{}'::jsonb) || CAST(:metadata AS JSONB), note = COALESCE(note, '') || E'\n\nIgnorada: picking Odoo assigned ainda requer validação física.', updated_at = now() WHERE opportunity_id = CAST(:opportunity_id AS UUID) AND status = 'pending' AND action_code = 'CREATE_SHIPMENT' RETURNING id::text """), { "opportunity_id": opportunity_id, "metadata": _json({ "superseded_by": "VALIDATE_PHYSICAL_ORDER", "superseded_reason": "picking_assigned_without_physical_validation", "superseded_version": "v4928.1.5.132", }), }).mappings().all() conn.execute(text(""" INSERT INTO opportunity_events ( id, opportunity_id, event_type, action_code, from_stage, to_stage, note, payload, created_by ) VALUES ( CAST(:id AS UUID), CAST(:opportunity_id AS UUID), 'odoo_physical_stage_corrected', 'VALIDATE_PHYSICAL_ORDER', CAST(:from_stage AS TEXT), 'ORDER_PREPARATION', CAST(:note AS TEXT), CAST(:payload AS JSONB), 'odoo_sync' ) """), { "id": str(uuid.uuid4()), "opportunity_id": opportunity_id, "from_stage": old_stage, "note": reason or "Picking reservado; validação física ainda pendente.", "payload": _json({"ignored_create_shipment_task_ids": [row["id"] for row in superseded]}), }) try: from app.opportunity_next_action_service import get_opportunity_next_action from app.opportunity_action_task_materializer import ensure_pending_task_for_next_action materialization = ensure_pending_task_for_next_action( opportunity_id, get_opportunity_next_action(opportunity_id), source="odoo_physical_stage_correction", actor="odoo_sync", ) except Exception as exc: materialization = {"created": False, "reason": str(exc)} return { "changed": old_stage != "ORDER_PREPARATION", "from_stage": old_stage, "stage": "ORDER_PREPARATION", "mode": "guarded_physical_rewind", "materialization": materialization, } def sync_opportunity_odoo_status(opportunity_id: str) -> dict: """Consulta Odoo e guarda um resumo físico simples na operation_links. Não altera o Odoo. Apenas lê sale.order, stock.picking e mrp.production, e guarda em operation_links external_type='physical_status'. """ link = _find_odoo_sale_order_link(opportunity_id) if not link: payload = { "physical_status": "no_order", "label": "Sem venda Odoo", "reason": "A oportunidade ainda não tem venda Odoo ligada.", "ready_to_ship": False, "next_action": "Criar ou associar venda Odoo.", } _upsert_odoo_physical_status_link(opportunity_id, "", "Sem venda Odoo", "", "no_order", payload) return payload client = OdooClient() sale = _find_sale_order(client, str(link.get("external_id") or ""), str(link.get("external_name") or "")) if not sale: payload = { "physical_status": "not_found", "label": "Venda não encontrada", "reason": "A referência guardada no ClientFlow não foi encontrada no Odoo.", "ready_to_ship": False, "next_action": "Confirmar o número/id da venda Odoo.", "linked_sale_order": link, } _upsert_odoo_physical_status_link(opportunity_id, str(link.get("external_id") or ""), str(link.get("external_name") or ""), str(link.get("external_url") or ""), "not_found", payload) return payload sale_name = str(sale.get("name") or "") pickings = _search_odoo_pickings(client, sale_name) productions = _search_odoo_productions(client, sale_name) derived = _derive_physical_status(sale, pickings, productions) payload = { **derived, "sale_order": { "id": sale.get("id"), "name": sale.get("name"), "state": sale.get("state"), "partner": _compact_m2o(sale.get("partner_id")), "amount_total": sale.get("amount_total"), "date_order": sale.get("date_order"), }, "pickings": [ { "id": p.get("id"), "name": p.get("name"), "state": p.get("state"), "type": _compact_m2o(p.get("picking_type_id")), "scheduled_date": p.get("scheduled_date"), "date_done": p.get("date_done"), } for p in pickings ], "productions": [ { "id": mo.get("id"), "name": mo.get("name"), "state": mo.get("state"), "product": _compact_m2o(mo.get("product_id")), "qty": mo.get("product_qty"), "date_start": mo.get("date_start"), "date_finished": mo.get("date_finished"), } for mo in productions ], } _upsert_odoo_physical_status_link( opportunity_id, str(sale.get("id") or link.get("external_id") or ""), sale_name, str(link.get("external_url") or ""), derived["physical_status"], payload, ) stage = derived.get("stage") payload["stage_application"] = _safe_apply_odoo_derived_stage( opportunity_id, str(stage or ""), reason=str(derived.get("reason") or ""), ) if str(derived.get("physical_status") or "").lower() == "shipped": from app.odoo_delivery_task_reconciliation import reconcile_odoo_delivery_done with engine.begin() as conn: payload["task_reconciliation"] = reconcile_odoo_delivery_done( conn, opportunity_id, evidence=payload, actor="odoo_sync", upsert_validation=True, ) return payload def _upsert_odoo_physical_status_link(opportunity_id: str, external_id: str, external_name: str, external_url: str, status: str, payload: dict) -> None: with engine.begin() as conn: conn.execute(text(""" INSERT INTO operation_links ( opportunity_id, system, external_type, external_id, external_name, external_url, status, payload, last_synced_at, updated_at ) VALUES ( CAST(:opportunity_id AS UUID), 'odoo', 'physical_status', :external_id, :external_name, :external_url, :status, CAST(:payload AS JSONB), now(), now() ) ON CONFLICT (opportunity_id, system, external_type) DO UPDATE SET external_id = EXCLUDED.external_id, external_name = EXCLUDED.external_name, external_url = EXCLUDED.external_url, status = EXCLUDED.status, payload = EXCLUDED.payload, last_synced_at = now(), updated_at = now() """), { "opportunity_id": opportunity_id, "external_id": external_id, "external_name": external_name, "external_url": external_url, "status": status, "payload": _json(payload), })