feat(integrations): add SaleOrder service for Odoo sales

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

services/integrations/app/schemas/__init__.py

@@ -0,0 +1,13 @@
+from app.schemas.sale import (
+    SaleOrderLine,
+    SaleOrderResponse,
+    SaleOrderSearchResult,
+    QuotationCreate,
+)
+
+__all__ = [
+    "SaleOrderLine",
+    "SaleOrderResponse",
+    "SaleOrderSearchResult",
+    "QuotationCreate",
+]

services/integrations/app/schemas/crm.py

@@ -0,0 +1,38 @@
+from pydantic import BaseModel
+from typing import Optional
+
+
+class LeadCreate(BaseModel):
+    name: str
+    partner_id: Optional[int] = None
+    contact_name: Optional[str] = None
+    phone: Optional[str] = None
+    mobile: Optional[str] = None
+    email_from: Optional[str] = None
+    description: Optional[str] = None
+    expected_revenue: Optional[float] = None
+    source: Optional[str] = "WhatsApp"
+
+
+class LeadResponse(BaseModel):
+    id: int
+    name: str
+    stage_id: int
+    stage_name: str
+    partner_id: Optional[int] = None
+    partner_name: Optional[str] = None
+    contact_name: Optional[str] = None
+    phone: Optional[str] = None
+    email_from: Optional[str] = None
+    expected_revenue: float
+    probability: float
+    user_id: Optional[int] = None
+    user_name: Optional[str] = None
+
+
+class LeadSearchResult(BaseModel):
+    id: int
+    name: str
+    stage_name: str
+    expected_revenue: float
+    probability: float

services/integrations/app/schemas/partner.py

@@ -0,0 +1,43 @@
+from pydantic import BaseModel
+from typing import Optional
+
+
+class PartnerBase(BaseModel):
+    name: str
+    phone: Optional[str] = None
+    mobile: Optional[str] = None
+    email: Optional[str] = None
+    street: Optional[str] = None
+    city: Optional[str] = None
+    country_id: Optional[int] = None
+    comment: Optional[str] = None
+
+
+class PartnerCreate(PartnerBase):
+    pass
+
+
+class PartnerUpdate(BaseModel):
+    name: Optional[str] = None
+    phone: Optional[str] = None
+    mobile: Optional[str] = None
+    email: Optional[str] = None
+    street: Optional[str] = None
+    city: Optional[str] = None
+    comment: Optional[str] = None
+
+
+class PartnerResponse(PartnerBase):
+    id: int
+    display_name: Optional[str] = None
+    credit: Optional[float] = None
+    debit: Optional[float] = None
+    credit_limit: Optional[float] = None
+
+
+class PartnerSearchResult(BaseModel):
+    id: int
+    name: str
+    phone: Optional[str] = None
+    mobile: Optional[str] = None
+    email: Optional[str] = None

services/integrations/app/schemas/product.py

@@ -0,0 +1,31 @@
+from pydantic import BaseModel
+from typing import Optional
+
+
+class ProductResponse(BaseModel):
+    id: int
+    name: str
+    default_code: Optional[str] = None
+    list_price: float
+    qty_available: float
+    virtual_available: float
+    description: Optional[str] = None
+    categ_name: Optional[str] = None
+
+
+class ProductSearchResult(BaseModel):
+    id: int
+    name: str
+    default_code: Optional[str] = None
+    list_price: float
+    qty_available: float
+
+
+class StockInfo(BaseModel):
+    product_id: int
+    product_name: str
+    qty_available: float
+    qty_reserved: float
+    qty_incoming: float
+    qty_outgoing: float
+    virtual_available: float

services/integrations/app/schemas/sale.py

@@ -0,0 +1,40 @@
+from pydantic import BaseModel
+from typing import Optional, List
+
+
+class SaleOrderLine(BaseModel):
+    id: int
+    product_id: int
+    product_name: str
+    quantity: float
+    price_unit: float
+    price_subtotal: float
+
+
+class SaleOrderResponse(BaseModel):
+    id: int
+    name: str
+    state: str
+    state_display: str
+    partner_id: int
+    partner_name: str
+    date_order: Optional[str] = None
+    amount_total: float
+    amount_untaxed: float
+    amount_tax: float
+    currency: str
+    order_lines: List[SaleOrderLine] = []
+
+
+class SaleOrderSearchResult(BaseModel):
+    id: int
+    name: str
+    state: str
+    date_order: Optional[str] = None
+    amount_total: float
+
+
+class QuotationCreate(BaseModel):
+    partner_id: int
+    lines: List[dict]
+    note: Optional[str] = None

services/integrations/app/services/__init__.py

@@ -0,0 +1,5 @@
+from app.services.sale import SaleOrderService
+
+__all__ = [
+    "SaleOrderService",
+]

services/integrations/app/services/crm.py

@@ -0,0 +1,108 @@
+from typing import List, Optional
+from app.odoo import get_odoo_client, OdooNotFoundError
+from app.schemas.crm import LeadCreate, LeadResponse, LeadSearchResult
+
+
+class CRMService:
+    """Service for Odoo CRM operations"""
+
+    MODEL = "crm.lead"
+
+    def __init__(self):
+        self.client = get_odoo_client()
+
+    def create_lead(self, data: LeadCreate) -> int:
+        """Create a new lead/opportunity"""
+        values = data.model_dump(exclude_none=True)
+
+        if data.source:
+            source_ids = self.client.search(
+                "utm.source",
+                [("name", "=", data.source)],
+                limit=1,
+            )
+            if source_ids:
+                values["source_id"] = source_ids[0]
+            if "source" in values:
+                del values["source"]
+
+        return self.client.create(self.MODEL, values)
+
+    def get_by_id(self, lead_id: int) -> LeadResponse:
+        """Get lead details"""
+        results = self.client.read(
+            self.MODEL,
+            [lead_id],
+            [
+                "id", "name", "stage_id", "partner_id", "contact_name",
+                "phone", "email_from", "expected_revenue", "probability",
+                "user_id",
+            ],
+        )
+
+        if not results:
+            raise OdooNotFoundError(f"Lead {lead_id} not found")
+
+        lead = results[0]
+        return LeadResponse(
+            id=lead["id"],
+            name=lead["name"],
+            stage_id=lead["stage_id"][0] if lead.get("stage_id") else 0,
+            stage_name=lead["stage_id"][1] if lead.get("stage_id") else "",
+            partner_id=lead["partner_id"][0] if lead.get("partner_id") else None,
+            partner_name=lead["partner_id"][1] if lead.get("partner_id") else None,
+            contact_name=lead.get("contact_name"),
+            phone=lead.get("phone"),
+            email_from=lead.get("email_from"),
+            expected_revenue=lead.get("expected_revenue", 0),
+            probability=lead.get("probability", 0),
+            user_id=lead["user_id"][0] if lead.get("user_id") else None,
+            user_name=lead["user_id"][1] if lead.get("user_id") else None,
+        )
+
+    def search_by_partner(
+        self,
+        partner_id: int,
+        limit: int = 10,
+    ) -> List[LeadSearchResult]:
+        """Search leads by partner"""
+        results = self.client.search_read(
+            self.MODEL,
+            [("partner_id", "=", partner_id)],
+            fields=["id", "name", "stage_id", "expected_revenue", "probability"],
+            limit=limit,
+            order="create_date desc",
+        )
+
+        return [
+            LeadSearchResult(
+                id=r["id"],
+                name=r["name"],
+                stage_name=r["stage_id"][1] if r.get("stage_id") else "",
+                expected_revenue=r.get("expected_revenue", 0),
+                probability=r.get("probability", 0),
+            )
+            for r in results
+        ]
+
+    def update_stage(self, lead_id: int, stage_id: int) -> bool:
+        """Move lead to different stage"""
+        return self.client.write(self.MODEL, [lead_id], {"stage_id": stage_id})
+
+    def add_note(self, lead_id: int, note: str) -> int:
+        """Add internal note to lead"""
+        return self.client.create("mail.message", {
+            "model": self.MODEL,
+            "res_id": lead_id,
+            "body": note,
+            "message_type": "comment",
+        })
+
+    def get_stages(self) -> List[dict]:
+        """Get all CRM stages"""
+        return self.client.search_read(
+            "crm.stage",
+            [],
+            fields=["id", "name", "sequence"],
+            order="sequence",
+        )

services/integrations/app/services/partner.py

@@ -0,0 +1,97 @@
+from typing import Optional
+from app.odoo import get_odoo_client, OdooNotFoundError
+from app.schemas.partner import (
+    PartnerCreate,
+    PartnerUpdate,
+    PartnerResponse,
+    PartnerSearchResult,
+)
+
+
+class PartnerService:
+    """Service for Odoo res.partner operations"""
+
+    MODEL = "res.partner"
+    FIELDS = [
+        "id", "name", "display_name", "phone", "mobile", "email",
+        "street", "city", "country_id", "comment",
+        "credit", "debit", "credit_limit",
+    ]
+
+    def __init__(self):
+        self.client = get_odoo_client()
+
+    def search_by_phone(self, phone: str) -> Optional[PartnerSearchResult]:
+        """Search partner by phone number"""
+        normalized = phone.replace(" ", "").replace("-", "").replace("+", "")
+
+        domain = [
+            "|",
+            ("phone", "ilike", normalized[-10:]),
+            ("mobile", "ilike", normalized[-10:]),
+        ]
+
+        results = self.client.search_read(
+            self.MODEL,
+            domain,
+            fields=["id", "name", "phone", "mobile", "email"],
+            limit=1,
+        )
+
+        if results:
+            return PartnerSearchResult(**results[0])
+        return None
+
+    def search_by_email(self, email: str) -> Optional[PartnerSearchResult]:
+        """Search partner by email"""
+        results = self.client.search_read(
+            self.MODEL,
+            [("email", "=ilike", email)],
+            fields=["id", "name", "phone", "mobile", "email"],
+            limit=1,
+        )
+
+        if results:
+            return PartnerSearchResult(**results[0])
+        return None
+
+    def get_by_id(self, partner_id: int) -> PartnerResponse:
+        """Get partner by ID"""
+        results = self.client.read(self.MODEL, [partner_id], self.FIELDS)
+        if not results:
+            raise OdooNotFoundError(f"Partner {partner_id} not found")
+
+        data = results[0]
+        if data.get("country_id") and isinstance(data["country_id"], (list, tuple)):
+            data["country_id"] = data["country_id"][0]
+        return PartnerResponse(**data)
+
+    def create(self, data: PartnerCreate) -> int:
+        """Create a new partner"""
+        values = data.model_dump(exclude_none=True)
+        return self.client.create(self.MODEL, values)
+
+    def update(self, partner_id: int, data: PartnerUpdate) -> bool:
+        """Update a partner"""
+        values = data.model_dump(exclude_none=True)
+        if not values:
+            return True
+        return self.client.write(self.MODEL, [partner_id], values)
+
+    def get_balance(self, partner_id: int) -> dict:
+        """Get partner balance (credit/debit)"""
+        results = self.client.read(
+            self.MODEL,
+            [partner_id],
+            ["credit", "debit", "credit_limit"],
+        )
+        if not results:
+            raise OdooNotFoundError(f"Partner {partner_id} not found")
+
+        data = results[0]
+        return {
+            "credit": data.get("credit", 0),
+            "debit": data.get("debit", 0),
+            "balance": data.get("debit", 0) - data.get("credit", 0),
+            "credit_limit": data.get("credit_limit", 0),
+        }

services/integrations/app/services/product.py

@@ -0,0 +1,117 @@
+from typing import List, Optional
+from app.odoo import get_odoo_client, OdooNotFoundError
+from app.schemas.product import ProductResponse, ProductSearchResult, StockInfo
+
+
+class ProductService:
+    """Service for Odoo product operations"""
+
+    MODEL = "product.product"
+
+    def __init__(self):
+        self.client = get_odoo_client()
+
+    def search(
+        self,
+        query: str = None,
+        category_id: int = None,
+        limit: int = 20,
+    ) -> List[ProductSearchResult]:
+        """Search products"""
+        domain = [("sale_ok", "=", True)]
+
+        if query:
+            domain.append("|")
+            domain.append(("name", "ilike", query))
+            domain.append(("default_code", "ilike", query))
+
+        if category_id:
+            domain.append(("categ_id", "=", category_id))
+
+        results = self.client.search_read(
+            self.MODEL,
+            domain,
+            fields=["id", "name", "default_code", "list_price", "qty_available"],
+            limit=limit,
+        )
+
+        return [ProductSearchResult(**r) for r in results]
+
+    def get_by_id(self, product_id: int) -> ProductResponse:
+        """Get product details"""
+        results = self.client.read(
+            self.MODEL,
+            [product_id],
+            [
+                "id", "name", "default_code", "list_price",
+                "qty_available", "virtual_available",
+                "description_sale", "categ_id",
+            ],
+        )
+
+        if not results:
+            raise OdooNotFoundError(f"Product {product_id} not found")
+
+        p = results[0]
+        return ProductResponse(
+            id=p["id"],
+            name=p["name"],
+            default_code=p.get("default_code"),
+            list_price=p.get("list_price", 0),
+            qty_available=p.get("qty_available", 0),
+            virtual_available=p.get("virtual_available", 0),
+            description=p.get("description_sale"),
+            categ_name=p["categ_id"][1] if p.get("categ_id") else None,
+        )
+
+    def get_by_sku(self, sku: str) -> Optional[ProductResponse]:
+        """Get product by SKU (default_code)"""
+        ids = self.client.search(
+            self.MODEL,
+            [("default_code", "=", sku)],
+            limit=1,
+        )
+        if not ids:
+            return None
+        return self.get_by_id(ids[0])
+
+    def check_stock(self, product_id: int) -> StockInfo:
+        """Get stock info for a product"""
+        results = self.client.read(
+            self.MODEL,
+            [product_id],
+            [
+                "id", "name", "qty_available", "virtual_available",
+                "incoming_qty", "outgoing_qty",
+            ],
+        )
+
+        if not results:
+            raise OdooNotFoundError(f"Product {product_id} not found")
+
+        p = results[0]
+        qty_available = p.get("qty_available", 0)
+        virtual = p.get("virtual_available", 0)
+
+        return StockInfo(
+            product_id=p["id"],
+            product_name=p["name"],
+            qty_available=qty_available,
+            qty_reserved=max(0, qty_available - virtual),
+            qty_incoming=p.get("incoming_qty", 0),
+            qty_outgoing=p.get("outgoing_qty", 0),
+            virtual_available=virtual,
+        )
+
+    def check_availability(self, product_id: int, quantity: float) -> dict:
+        """Check if quantity is available"""
+        stock = self.check_stock(product_id)
+        available = stock.virtual_available >= quantity
+
+        return {
+            "available": available,
+            "requested": quantity,
+            "in_stock": stock.qty_available,
+            "virtual_available": stock.virtual_available,
+            "shortage": max(0, quantity - stock.virtual_available),
+        }

services/integrations/app/services/sale.py

@@ -0,0 +1,128 @@
+from typing import List, Optional
+from app.odoo import get_odoo_client, OdooNotFoundError
+from app.schemas.sale import (
+    SaleOrderResponse,
+    SaleOrderSearchResult,
+    SaleOrderLine,
+    QuotationCreate,
+)
+
+
+STATE_DISPLAY = {
+    "draft": "Presupuesto",
+    "sent": "Presupuesto Enviado",
+    "sale": "Pedido de Venta",
+    "done": "Bloqueado",
+    "cancel": "Cancelado",
+}
+
+
+class SaleOrderService:
+    """Service for Odoo sale.order operations"""
+
+    MODEL = "sale.order"
+    LINE_MODEL = "sale.order.line"
+
+    def __init__(self):
+        self.client = get_odoo_client()
+
+    def search_by_partner(
+        self,
+        partner_id: int,
+        state: str = None,
+        limit: int = 10,
+    ) -> List[SaleOrderSearchResult]:
+        """Search orders by partner"""
+        domain = [("partner_id", "=", partner_id)]
+        if state:
+            domain.append(("state", "=", state))
+
+        results = self.client.search_read(
+            self.MODEL,
+            domain,
+            fields=["id", "name", "state", "date_order", "amount_total"],
+            limit=limit,
+            order="date_order desc",
+        )
+
+        return [SaleOrderSearchResult(**r) for r in results]
+
+    def get_by_id(self, order_id: int) -> SaleOrderResponse:
+        """Get order details"""
+        results = self.client.read(
+            self.MODEL,
+            [order_id],
+            [
+                "id", "name", "state", "partner_id", "date_order",
+                "amount_total", "amount_untaxed", "amount_tax",
+                "currency_id", "order_line",
+            ],
+        )
+
+        if not results:
+            raise OdooNotFoundError(f"Sale order {order_id} not found")
+
+        order = results[0]
+
+        lines = []
+        if order.get("order_line"):
+            line_data = self.client.read(
+                self.LINE_MODEL,
+                order["order_line"],
+                ["id", "product_id", "product_uom_qty", "price_unit", "price_subtotal"],
+            )
+            for line in line_data:
+                lines.append(SaleOrderLine(
+                    id=line["id"],
+                    product_id=line["product_id"][0] if line.get("product_id") else 0,
+                    product_name=line["product_id"][1] if line.get("product_id") else "",
+                    quantity=line.get("product_uom_qty", 0),
+                    price_unit=line.get("price_unit", 0),
+                    price_subtotal=line.get("price_subtotal", 0),
+                ))
+
+        return SaleOrderResponse(
+            id=order["id"],
+            name=order["name"],
+            state=order["state"],
+            state_display=STATE_DISPLAY.get(order["state"], order["state"]),
+            partner_id=order["partner_id"][0] if order.get("partner_id") else 0,
+            partner_name=order["partner_id"][1] if order.get("partner_id") else "",
+            date_order=order.get("date_order"),
+            amount_total=order.get("amount_total", 0),
+            amount_untaxed=order.get("amount_untaxed", 0),
+            amount_tax=order.get("amount_tax", 0),
+            currency=order["currency_id"][1] if order.get("currency_id") else "USD",
+            order_lines=lines,
+        )
+
+    def get_by_name(self, name: str) -> Optional[SaleOrderResponse]:
+        """Get order by name (SO001)"""
+        ids = self.client.search(self.MODEL, [("name", "=", name)], limit=1)
+        if not ids:
+            return None
+        return self.get_by_id(ids[0])
+
+    def create_quotation(self, data: QuotationCreate) -> int:
+        """Create a quotation"""
+        order_id = self.client.create(self.MODEL, {
+            "partner_id": data.partner_id,
+            "note": data.note,
+        })
+
+        for line in data.lines:
+            self.client.create(self.LINE_MODEL, {
+                "order_id": order_id,
+                "product_id": line["product_id"],
+                "product_uom_qty": line.get("quantity", 1),
+            })
+
+        return order_id
+
+    def confirm_order(self, order_id: int) -> bool:
+        """Confirm quotation to sale order"""
+        return self.client.execute(self.MODEL, "action_confirm", [order_id])
+
+    def get_pdf_url(self, order_id: int) -> str:
+        """Get URL to download order PDF"""
+        return f"{self.client.url}/report/pdf/sale.report_saleorder/{order_id}"