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.')