🔍 Source Information from Helpdesk Ticket #{self.id}
This sale order was created from helpdesk ticket {self.name}
"""
# Add original description if available
if description:
message_body += f"""
Original Ticket Description:
{description}
"""
# Add chatter messages if available (limited to avoid huge messages)
if chatter_messages:
# Limit the size of chatter messages to avoid huge messages
max_chars = 2000
if len(chatter_messages) > max_chars:
chatter_messages = chatter_messages[:max_chars] + "... (truncated)"
message_body += f"""
Relevant Chatter Messages:
{chatter_messages}
"""
# Get the original attachments from the helpdesk ticket
attachments = self.env['ir.attachment'].search([
('res_id', '=', self.id),
('res_model', '=', self._name)
])
# Add attachment information to the message
if attachments:
message_body += f"""
Attachments: {len(attachments)} file(s) attached to this message
"""
# Close the main div
message_body += "
"
try:
_logger.debug(f"Attempting to post source information message for sale order {sale_order.id}")
# Prepare attachment IDs to forward with the message
attachment_ids = attachments.ids if attachments else []
_logger.debug(f"Forwarding {len(attachment_ids)} attachments from ticket {self.id} to sale order {sale_order.id}")
# Post the message with attachments
result = sale_order.message_post(
body=message_body,
subject="Source Information",
body_is_html=True,
attachment_ids=attachment_ids
)
_logger.debug(f"Message post result: {result}")
_logger.debug(f"Successfully posted source information message for sale order {sale_order.id}")
except Exception as e:
_logger.error(f"Error posting source information message: {e}")
def _get_sale_order_values(self) -> dict:
"""
Generate sales order values using AI to analyze ticket content, chatter messages, and attachments.
The AI will identify products from the content and match them to Odoo products.
Returns:
dict: Values for creating a sales order including order lines
"""
self.ensure_one()
_logger.debug(f"Generating AI sales order values for ticket {self.id}")
# Get the ticket data
ticket_data = self._prepare_ai_prompt_data()
description = ticket_data.get('ticket_description', '')
chatter_messages = ticket_data.get('ticket_messages', '')
attachments_info = ticket_data.get('attachments_info', '')
attachment_contents = ticket_data.get('attachment_contents', '')
# Log the content being analyzed
_logger.debug(f"AI Analysis - Content lengths: Description={len(description)}, Chatter={len(chatter_messages)}, Attachments={len(attachment_contents)}")
# Get the OpenWebUI provider from company settings
company = self.env.company
provider = company.openwebui_provider_id
if not provider:
_logger.error("No OpenWebUI provider configured for company")
return {"order_line": []}
# Get the OpenWebUI client from the provider
ai_client = provider.get_client()
# Register the product finding methods as tools
# STEP 1: Extract potential products from ticket content and attachments
_logger.debug("Step 1: Extracting potential products from ticket content and attachments")
potential_products = self._ai_extract_potential_products(ticket_data)
_logger.debug(f"Step 1 result - Potential products: {potential_products}")
# Extract delivery date if available (stored in field ai_extracted_delivery_date)
delivery_date = self.ai_extracted_delivery_date
_logger.debug(f"Step 1 result - Extracted delivery date: {delivery_date}")
# If no potential products found, return empty order
if not potential_products:
_logger.warning("No potential products found in ticket content")
return {
"order_line": [],
"note": "No products identified in ticket content"
}
# STEP 2: Match extracted products to database products using tools
_logger.debug("Step 2: Matching extracted products to database products")
matched_products = self._ai_match_products_to_database(potential_products)
_logger.debug(f"Step 2 result - Matched products: {matched_products}")
# If no matched products, return empty order
if not matched_products:
_logger.warning("No products could be matched to database")
return {
"order_line": [],
"note": "No products could be matched to database"
}
# STEP 3: Format the matched products into final JSON structure
_logger.debug("Step 3: Formatting matched products into final sales order structure")
final_order = self._ai_format_final_sale_order(matched_products, ticket_data, delivery_date)
_logger.debug(f"Step 3 result - Final order: {final_order}")
# Return the final formatted sales order along with potential and matched products for unmatched products tracking
final_order['_potential_products_for_tracking'] = potential_products
final_order['_matched_products_for_tracking'] = matched_products
# Return the final formatted sales order
return final_order
def _prepare_ai_prompt_data(self):
"""
Extract and prepare all relevant data from the helpdesk ticket for AI analysis.
This includes ticket description, chatter messages, and attachment contents.
Returns:
dict: Dictionary containing ticket data for AI analysis
"""
self.ensure_one()
_logger.debug(f"Preparing AI prompt data for ticket {self.id}")
result = {
'ticket_description': '',
'ticket_messages': '',
'attachments_info': '',
'attachment_contents': ''
}
# Get ticket description
if self.description:
result['ticket_description'] = self.description
# Get chatter messages
messages = []
if self.message_ids:
for message in self.message_ids:
if message.body and not message.is_internal:
# Skip system messages and focus on actual conversation
if not message.author_id or message.author_id.name != 'OdooBot':
# Format: [Author] on [Date]: [Message]
author = message.author_id.name if message.author_id else 'System'
date = message.date.strftime('%Y-%m-%d %H:%M') if message.date else ''
# Clean HTML from message body
body = re.sub(r'<[^>]+>', ' ', message.body)
messages.append(f"[{author}] on {date}: {body}")
result['ticket_messages'] = '\n\n'.join(messages)
# Get attachments
attachments = self.env['ir.attachment'].search([('res_id', '=', self.id), ('res_model', '=', self._name)])
attachment_infos = []
attachment_contents = []
for attachment in attachments:
# Add attachment metadata
attachment_infos.append(f"File: {attachment.name} ({attachment.mimetype}, {attachment.file_size} bytes)")
# Add PDF content reference
if attachment.mimetype == 'application/pdf':
attachment_contents.append(f"IMPORTANT PDF CONTENT: {attachment.name} - The AI must analyze this PDF for ALL product mentions and include ANY products found as either matched products or unmatched line notes")
# Add text file content reference
elif attachment.mimetype == 'text/plain':
attachment_contents.append(f"IMPORTANT TEXT CONTENT: {attachment.name} - The AI must analyze this text file for ALL product mentions and include ANY products found as either matched products or unmatched line notes")
result['attachments_info'] = '\n'.join(attachment_infos)
result['attachment_contents'] = '\n\n'.join(attachment_contents)
return result
def _prepare_order_line_values(self, product, quantity, description=""):
"""Prepare values for creating a sale order line"""
return {
'product_id': product.id,
'product_uom_qty': quantity,
'name': description or product.name,
}
def _ai_find_product_id_by_name(self, product_name: str) -> int | None:
"""Find a product by name
Args:
product_name: The name of the product to find
Returns:
The ID of the product if found, or None if not found
"""
return self.env['product.product'].search([
('name', 'ilike', product_name),
('sale_ok', '=', True)
], limit=1).id
def _ai_find_product_id_by_code(self, product_reference: str) -> int | None:
"""Find a product by code
Args:
product_reference: The code of the product to find
Returns:
The ID of the product if found, or None if not found
"""
return self.env['product.product'].search([
('default_code', 'ilike', product_reference),
('sale_ok', '=', True)
], limit=1).id
def _ai_extract_potential_products(self, ticket_data: dict) -> list:
"""
First AI call: Extract potential products from ticket content and attachments.
Only this step will have access to the files/attachments.
Args:
ticket_data (dict): Dictionary containing ticket description, messages, and attachment info
Returns:
list: List of potential products identified by the AI
"""
self.ensure_one()
_logger.debug("Starting Step 1: Extracting potential products")
description = ticket_data.get('ticket_description', '')
chatter_messages = ticket_data.get('ticket_messages', '')
attachments_info = ticket_data.get('attachments_info', '')
attachment_contents = ticket_data.get('attachment_contents', '')
# Log the content being analyzed
_logger.debug(f"AI Analysis - Content lengths: Description={len(description)}, Chatter={len(chatter_messages)}, Attachments={len(attachment_contents)}")
# Get the OpenWebUI provider from company settings
company = self.env.company
provider = company.openwebui_provider_id
if not provider:
_logger.error("No OpenWebUI provider configured for company")
return []
# Get the OpenWebUI client from the provider
ai_client = provider.get_client()
# Process PDF and text attachments for analysis
attachments_list = []
attachments = self.env['ir.attachment'].search([('res_id', '=', self.id), ('res_model', '=', self._name)])
# Upload PDF and text attachments to OpenWebUI
uploaded_files = []
for attachment in attachments:
if attachment.mimetype in ['application/pdf', 'text/plain']:
try:
temp_path = Path(f"/tmp/{attachment.name}")
shutil.copy(attachment._full_path(attachment.store_fname), str(temp_path))
# Upload the file to OpenWebUI and get a FileObject with id
file_object = ai_client.files.from_path(temp_path)
uploaded_files.append(file_object)
# Clean up temporary file
temp_path.unlink(missing_ok=True)
except Exception as e:
_logger.error(f"Error processing attachment {attachment.name}: {e}")
import traceback
_logger.error(f"Traceback: {traceback.format_exc()}")
# Create the prompt for the AI to extract potential products AND delivery date
prompt = f"""IMPORTANT: YOU ARE NOT A CONVERSATIONAL ASSISTANT. YOU ARE A DATA EXTRACTION SYSTEM.
Your ONLY function is to analyze the provided content and return BOTH a list of potential products AND any delivery date mentioned.
DO NOT introduce yourself, explain what you can or cannot do, or engage in conversation.
ONLY RETURN THE REQUESTED JSON DATA STRUCTURE.
TASK: Extract ALL potential products AND any delivery date mentioned in the following content:
Customer Request:
{description}
Chatter Messages (IMPORTANT - CAREFULLY ANALYZE THESE FOR PRODUCT INFORMATION):
{chatter_messages}
Attachments Information:
{attachments_info}
Attachment Contents (CRITICALLY IMPORTANT - ANALYZE PDF AND TEXT CONTENTS FOR ALL PRODUCT DETAILS):
{attachment_contents}
WORKFLOW - FOLLOW THESE STEPS EXACTLY:
1. Identify ALL potential products mentioned in the content (product names, codes, references) from BOTH chatter messages AND attachments
2. For EACH potential product identified:
a. Extract the product name
b. Extract the product's default code
c. Extract any quantity information
d. Extract any price information
e. Extract any additional description
3. Identify ANY date mentioned in the content - look for ALL of these:
a. Explicit delivery dates (phrases like 'delivery date', 'ship by', 'needed by', 'required by', etc.)
b. Quotation dates or dates labeled as 'Quotation Date'
d. ANY date in formats like MM/DD/YYYY, DD/MM/YYYY, YYYY-MM-DD, or similar formats
4. Convert any found dates to YYYY-MM-DD format (e.g., '2025-08-15')
5. Return a structured JSON object containing both the products list and delivery date
RESPONSE FORMAT:
Your response MUST ONLY be a valid JSON object with the following structure:
{{
"products": [
{{
"name": "Product name",
"default_code": "Product default code",
"quantity": "Quantity if mentioned (number or text)",
"price": "Price if mentioned"
}},
...
],
"delivery_date": "YYYY-MM-DD format if a delivery date is mentioned, null otherwise"
}}
CRITICAL RULES FOR IDENTIFYING PRODUCTS:
1. A potential product is ANYTHING that could be a product - be liberal in identification
2. If a product's default code is explicitly mentioned (often in brackets or as a standalone identifier), put it in the "default_code" field
3. If only a product name is mentioned, put it in the "name" field
4. DO NOT include descriptions in the "name" field - only use actual product names or references
5. DO NOT include a "description" field - it's not needed for product matching
6. DO NOT filter or validate products at this stage - that will happen later
7. Return ONLY valid JSON with no text before or after
8. DO NOT explain what you're doing or respond conversationally
CRITICAL RULES FOR IDENTIFYING DATES:
1. SCAN THE ENTIRE DOCUMENT FOR ANY DATES IN ANY FORMAT - especially MM/DD/YYYY format
2. Look for dates near labels like 'Quotation Date', 'Expiration', 'Delivery Date', etc.
3. If you find a date labeled as 'Quotation Date', use it as the delivery date if no explicit delivery date is found
4. If you find a date labeled as 'Expiration', use it as the delivery date if no explicit delivery date or quotation date is found
5. Convert all dates to YYYY-MM-DD format (e.g., if you see '07/02/2025', convert to '2025-07-02')
6. If multiple dates are found, prioritize in this order: explicit delivery date > quotation date > expiration date
7. DO NOT return null for delivery_date if ANY date is found in the document
IMPORTANT: Include EVERYTHING that might be a product, even if uncertain.
IMPORTANT: ALWAYS extract at least one date from the document if any date exists, regardless of its label.
"""
# Call the AI to extract potential products
try:
_logger.debug("Sending request to AI for product extraction")
response = ai_client.chat.completions.create(
model=provider.default_model_id.technical_name if provider.default_model_id else None,
messages=[
{
"role": "system",
"content": "You are a data extraction system. Your ONLY job is to extract ALL potential products mentioned in the provided content and return them in a structured JSON format. Be liberal in identifying potential products - include anything that might be a product."
},
{
"role": "user",
"content": prompt
}
],
stream=False,
files=uploaded_files
)
_logger.debug(f"Received AI response for product extraction")
except Exception as e:
_logger.error(f"Error in AI product extraction request: {e}")
import traceback
_logger.error(f"Traceback: {traceback.format_exc()}")
return []
# Process the AI response
_logger.debug(f"AI product extraction response type: {type(response)}")
# Add detailed logging of the response content
try:
response_content = response.choices[0].message.content if hasattr(response, 'choices') else str(response)
_logger.debug(f"AI product extraction response content: {response_content[:1000]}...")
except:
_logger.debug("Could not serialize AI product extraction response for logging")
# Extract the content from the response
response_content = response.choices[0].message.content if hasattr(response, 'choices') else str(response)
# Handle string responses (extract JSON if possible)
if isinstance(response_content, str):
_logger.debug("AI returned text response, attempting to extract JSON")
try:
# Look for JSON pattern in the text
json_pattern = r'```(?:json)?\s*(\[[\s\S]*?\]|{[\s\S]*?})\s*```'
json_matches = re.findall(json_pattern, response_content)
if json_matches:
response_data = json.loads(json_matches[0])
# Handle new structure with products and delivery_date
if isinstance(response_data, dict) and 'products' in response_data:
# Store delivery date as instance variable for later use
self.ai_extracted_delivery_date = response_data.get('delivery_date')
_logger.debug(f"Extracted delivery date: {self.ai_extracted_delivery_date}")
return response_data['products']
return response_data if isinstance(response_data, list) else [response_data]
elif response_content.strip().startswith('[') and response_content.strip().endswith(']'):
response_data = json.loads(response_content.strip())
# Handle old format (just products list)
self.ai_extracted_delivery_date = False
_logger.debug("Using old format - no delivery date extracted")
return response_data if isinstance(response_data, list) else [response_data]
elif response_content.strip().startswith('{') and response_content.strip().endswith('}'):
response_data = json.loads(response_content.strip())
# Handle new structure with products and delivery_date
if isinstance(response_data, dict) and 'products' in response_data:
# Store delivery date as instance variable for later use
self.ai_extracted_delivery_date = response_data.get('delivery_date')
_logger.debug(f"Extracted delivery date: {self.ai_extracted_delivery_date}")
return response_data['products']
return [response_data]
else:
_logger.error("Could not extract JSON from text response")
self.ai_extracted_delivery_date = False
return []
except Exception as parse_error:
_logger.error(f"Failed to extract JSON from text response: {parse_error}")
self.ai_extracted_delivery_date = False
return []
# If we get here, the response is in an unexpected format
_logger.error(f"Unexpected response format: {type(response_content)}")
self.ai_extracted_delivery_date = False
return []
def _ai_match_products_to_database(self, potential_products: list) -> list:
"""
Second AI call: Match extracted products to database products using tools.
This step will not have access to files/attachments.
Args:
potential_products (list): List of potential products from the first step
Returns:
list: List of matched products with database IDs or unmatched product notes
"""
self.ensure_one()
_logger.debug("Starting Step 2: Matching products to database")
_logger.debug(f"Input products to match: {potential_products}")
# We'll pass potential products to the unmatched products method instead of storing as instance attributes
# Get the OpenWebUI provider from company settings
company = self.env.company
provider = company.openwebui_provider_id
if not provider:
_logger.error("No OpenWebUI provider configured for company")
return []
# Get the OpenWebUI client from the provider
ai_client = provider.get_client()
# Register the product finding methods as tools
registry = ai_client.tool_registry
registry.register(
self._ai_find_product_id_by_name,
non_ai_params=["self"],
description="Find a product by its name and return its ID. Input: name (string) - The name of the product to find. Returns the product ID if found, or null if not found."
)
registry.register(
self._ai_find_product_id_by_code,
non_ai_params=["self"],
description="Find a product by its code/reference and return its ID. Input: code (string) - The code/reference of the product to find. Returns the product ID if found, or null if not found."
)
# Create the prompt for the AI to match products
products_json = json.dumps(potential_products, indent=2)
prompt = f"""IMPORTANT: YOU ARE NOT A CONVERSATIONAL ASSISTANT. YOU ARE A DATA PROCESSING SYSTEM.
Your ONLY function is to match the provided potential products to actual database products using the available tools.
DO NOT introduce yourself, explain what you can or cannot do, or engage in conversation.
ONLY RETURN THE REQUESTED JSON DATA STRUCTURE.
TASK: Match the following potential products to actual database products:
Potential Products:
{products_json}
WORKFLOW - FOLLOW THESE STEPS EXACTLY:
1. For EACH potential product:
a. If the product has a code/reference, call _ai_find_product_id_by_code with that code
b. If the product has a name, call _ai_find_product_id_by_name with that name
c. If the product is not found in the database, create a line note entry
2. Return a structured JSON list of matched products and line notes
RESPONSE FORMAT:
Your response MUST ONLY be a valid JSON array with the following structure:
[
{{
"product_id": NUMERIC_ID_FROM_TOOL_CALL, // Must be an actual ID returned from a tool call
"product_uom_qty": QUANTITY,
"price_unit": PRICE,
"name": "Product name (for reference)"
}},
...
]
CRITICAL RULES FOR MATCHING:
1. You MUST use the provided tools for EVERY potential product
2. product_id MUST be a numeric ID returned by a tool call, NEVER make up IDs
3. For products not found in the database, simply exclude them from the response (DO NOT include line notes)
4. Return ONLY valid JSON with no text before or after
5. DO NOT explain what you're doing or respond conversationally
IMPORTANT: You must invoke the tools directly using function calling, not just output text that looks like a tool call.
"""
# Call the AI with tools to match products
try:
_logger.debug("Sending request to AI for product matching")
response = ai_client.chat_with_tools(
messages=[
{
"role": "system",
"content": "You are a data processing system with access to tools for finding product IDs in an Odoo database. Your ONLY job is to match the provided potential products to actual database products using the available tools. For ANY product that cannot be found in the database, you should simply exclude it from your response."
},
{
"role": "user",
"content": prompt
}
],
tools=["_ai_find_product_id_by_name", "_ai_find_product_id_by_code"],
tool_params={},
max_tool_calls=25
)
_logger.debug(f"Received AI response for product matching")
except Exception as e:
_logger.error(f"Error in AI product matching request: {e}")
import traceback
_logger.error(f"Traceback: {traceback.format_exc()}")
return []
# Process the AI response
_logger.debug(f"AI product matching response type: {type(response)}")
# If it's a dictionary, use it directly
if isinstance(response, dict):
_logger.debug("AI returned dictionary response, using directly")
# Check if it's a single object that should be in a list
if 'product_id' in response or 'display_type' in response:
return [response]
# If it's a dictionary but not a single product object, return empty list
return []
# Handle string responses (extract JSON if possible)
if isinstance(response, str):
_logger.debug("AI returned text response, attempting to extract JSON")
try:
# Look for JSON pattern in the text
json_pattern = r'```(?:json)?\s*(\[[\s\S]*?\]|{{[\s\S]*?}})\s*```'
json_matches = re.findall(json_pattern, response)
if json_matches:
response_data = json.loads(json_matches[0])
return response_data if isinstance(response_data, list) else [response_data]
if response.strip().startswith('[') and response.strip().endswith(']'):
response_data = json.loads(response.strip())
return [response_data]
elif response.strip().startswith('{{') and response.strip().endswith('}}'):
response_data = json.loads(response.strip())
return [response_data]
else:
_logger.error("Could not extract JSON from text response")
return []
except Exception as parse_error:
_logger.error(f"Failed to extract JSON from text response: {parse_error}")
return []
# If it's already a list, return it
if isinstance(response, list):
# Store matched products for later use
self._matched_products_for_tracking = response
return response
# If we get here, the response is in an unexpected format
_logger.error(f"Unexpected response format: {type(response)}")
return []
def _post_unmatched_products_message(self, matched_products: list, potential_products: list = None):
"""
Post a temporary chatter message listing products that the AI couldn't find in the database.
This message will be deleted after the sale order is created.
Args:
matched_products (list): List of products that were successfully matched
potential_products (list): List of potential products identified by AI (optional)
"""
self.ensure_one()
# Use provided potential products or fall back to stored attribute
if potential_products is None:
if not hasattr(self, '_potential_products_for_tracking'):
return
potential_products = self._potential_products_for_tracking
# Create a set of matched product names/refs for quick lookup
matched_names = set()
matched_refs = set()
for product in matched_products:
if isinstance(product, dict):
# Add product name if available
if 'name' in product and product['name']:
matched_names.add(product['name'].lower().strip())
# Add product reference/code if available
if 'product_code' in product and product['product_code']:
matched_refs.add(product['product_code'].lower().strip())
# Identify unmatched products
unmatched_products = []
for product in potential_products:
if isinstance(product, dict):
# Check if product was matched by name or reference
name_matched = False
ref_matched = False
# Check by name
if 'name' in product and product['name']:
name_matched = product['name'].lower().strip() in matched_names
# Check by reference/code
if 'product_code' in product and product['product_code']:
ref_matched = product['product_code'].lower().strip() in matched_refs
# If neither name nor reference matched, add to unmatched list
if not name_matched and not ref_matched:
product_info = {}
if 'name' in product:
product_info['name'] = product['name']
if 'product_code' in product:
product_info['product_code'] = product['product_code']
if 'product_uom_qty' in product:
product_info['quantity'] = product['product_uom_qty']
unmatched_products.append(product_info)
# Remove duplicates while preserving order
seen = set()
unique_unmatched = []
for product in unmatched_products:
# Create a unique key for the product
key_parts = []
if 'name' in product:
key_parts.append(product['name'].lower().strip())
if 'product_code' in product:
key_parts.append(product['product_code'].lower().strip())
key = '|'.join(key_parts)
if key not in seen:
seen.add(key)
unique_unmatched.append(product)
# If we have unmatched products, post a temporary message
if unique_unmatched:
message_lines = ["The following products were mentioned in the ticket but could not be found in the product database:
")
message_lines.append("