odoo-ai/agent_service/agents/sales_agent.py

from __future__ import annotations
import logging
from .base_agent import BaseAgent, AgentReport, AgentDirective, SweepReport
from ..tools.sales_tools import SalesTools

logger = logging.getLogger(__name__)

SALES_TOOLS = [
    {'name': 'get_sales_orders', 'description': 'Retrieve confirmed sales orders',
     'parameters': {'state': {'type': 'string', 'optional': True},
                    'partner_id': {'type': 'integer', 'optional': True},
                    'date_from': {'type': 'string', 'optional': True},
                    'date_to': {'type': 'string', 'optional': True},
                    'limit': {'type': 'integer', 'optional': True}}},
    {'name': 'get_quotations', 'description': 'Get open quotations',
     'parameters': {'partner_id': {'type': 'integer', 'optional': True},
                    'limit': {'type': 'integer', 'optional': True}}},
    {'name': 'get_sales_summary', 'description': 'Get sales summary and rep breakdown',
     'parameters': {'date_from': {'type': 'string', 'optional': True},
                    'date_to': {'type': 'string', 'optional': True}}},
    {'name': 'get_customer_orders', 'description': 'Get all orders for a specific customer',
     'parameters': {'partner_id': {'type': 'integer'},
                    'limit': {'type': 'integer', 'optional': True}}},
    {'name': 'confirm_quotation', 'description': 'Confirm a draft quotation to sales order',
     'parameters': {'order_id': {'type': 'integer'}}},
    {'name': 'update_order_note', 'description': 'Update the internal note on an order',
     'parameters': {'order_id': {'type': 'integer'}, 'note': {'type': 'string'}}},
    {'name': 'flag_for_review', 'description': 'Flag a sales order for review',
     'parameters': {'model': {'type': 'string'}, 'record_id': {'type': 'integer'},
                    'reason': {'type': 'string'},
                    'severity': {'type': 'string', 'optional': True}}},
    {'name': 'post_chatter_note', 'description': 'Post a note on a record',
     'parameters': {'model': {'type': 'string'}, 'record_id': {'type': 'integer'},
                    'note': {'type': 'string'}}},
]


class SalesAgent(BaseAgent):
    name = 'sales_agent'
    domain = 'sales'
    required_odoo_module = 'sale'
    system_prompt_file = 'sales_system.txt'
    tools = SALES_TOOLS

    def __init__(self, odoo, llm, peer_bus=None):
        super().__init__(odoo, llm, peer_bus)
        self._st = SalesTools(odoo)
        self._gathered_data = {}
        self._actions_taken = []
        self._escalations_list = []

    async def _plan(self, directive: AgentDirective) -> dict:
        intent = (directive.intent or '').lower()
        return {
            'fetch_summary': any(k in intent for k in ('summary', 'overview', 'report', 'revenue')),
            'fetch_quotations': any(k in intent for k in ('quotation', 'quote', 'draft')),
            'fetch_orders': any(k in intent for k in ('order', 'sale')),
            'partner_id': directive.context.get('partner_id'),
            'date_from': directive.context.get('date_from'),
            'date_to': directive.context.get('date_to'),
        }

    async def _gather(self, ctx: dict) -> dict:
        plan = ctx.get('plan', {})
        data: dict = {}
        if plan.get('fetch_summary') or not any([plan.get('fetch_quotations'), plan.get('fetch_orders')]):
            data['summary'] = await self._st.get_sales_summary(
                date_from=plan.get('date_from'), date_to=plan.get('date_to'),
            )
        if plan.get('fetch_quotations'):
            data['quotations'] = await self._st.get_quotations(
                partner_id=plan.get('partner_id'), limit=20,
            )
        if plan.get('fetch_orders'):
            data['orders'] = await self._st.get_sales_orders(
                partner_id=plan.get('partner_id'), limit=20,
            )
        self._gathered_data = data
        return data

    async def _reason(self, ctx: dict) -> dict:
        data = self._gathered_data
        analysis: dict = {'escalations': []}
        summary = data.get('summary', {})
        if summary and summary.get('total_revenue', 0) == 0:
            analysis['escalations'].append('No confirmed sales orders in the period.')
        self._escalations_list = analysis['escalations']
        return analysis

    async def _act(self, ctx: dict) -> list:
        return []

    async def _report(self, ctx: dict) -> AgentReport:
        data = self._gathered_data
        parts = []
        summary = data.get('summary', {})
        if summary:
            parts.append(
                f'Sales: {summary.get("order_count", 0)} orders, '
                f'total revenue {summary.get("total_revenue", 0):.2f}.'
            )
        if not parts:
            parts.append('Sales review complete.')
        return AgentReport(agent=self.name, summary=chr(10).join(parts),
                           data=data, escalations=self._escalations_list, actions_taken=[])

    async def _dispatch_tool(self, name: str, args: dict):
        dispatch = {
            'get_sales_orders': self._st.get_sales_orders,
            'get_quotations': self._st.get_quotations,
            'get_sales_summary': self._st.get_sales_summary,
            'get_customer_orders': self._st.get_customer_orders,
            'confirm_quotation': self._st.confirm_quotation,
            'update_order_note': self._st.update_order_note,
            'flag_for_review': self._st.flag_for_review,
            'post_chatter_note': self._st.post_chatter_note,
        }
        if name not in dispatch:
            raise ValueError(f'Unknown tool: {name}')
        return await dispatch[name](**args)

    async def handle_peer_request(self, request_type: str, params: dict, directive_id: str) -> dict:
        try:
            if request_type == 'sales_summary':
                return await self._st.get_sales_summary()
            if request_type == 'customer_orders':
                return {'orders': await self._st.get_customer_orders(partner_id=params['partner_id'])}
            return {'error': f'Unknown type: {request_type}'}
        except Exception as exc:
            return {'error': str(exc)}

    async def sweep(self) -> SweepReport:
        findings = []
        try:
            quotations = await self._st.get_quotations(limit=50)
            import datetime
            today = str(datetime.date.today())
            expired = [q for q in quotations if q.get('validity_date') and q['validity_date'] < today]
            for q in expired:
                findings.append({'type': 'expired_quotation', 'order_id': q.get('id'),
                                  'partner': q.get('partner_id', [0, 'Unknown'])[1] if isinstance(q.get('partner_id'), list) else '',
                                  'severity': 'low'})
        except Exception as exc:
            return SweepReport(agent=self.name, findings=[], actions=[], error=str(exc))
        return SweepReport(agent=self.name, findings=findings, actions=[],
                           summary=f'Sales sweep: {len(findings)} expired quotations found.')