243 lines
8.5 KiB
Python
243 lines
8.5 KiB
Python
import logging
|
|
|
|
from odoo import models, api
|
|
|
|
_logger = logging.getLogger(__name__)
|
|
|
|
# Default business models, grouped by category.
|
|
# Only included if the corresponding module is installed.
|
|
_DEFAULT_BUSINESS_MODELS = {
|
|
'sale': ['sale.order', 'sale.order.line'],
|
|
'account': ['account.move', 'account.move.line', 'account.payment'],
|
|
'product': ['product.template', 'product.product', 'product.category'],
|
|
'base': ['res.partner'],
|
|
'stock': ['stock.picking', 'stock.move', 'stock.quant'],
|
|
'purchase': ['purchase.order', 'purchase.order.line'],
|
|
'crm': ['crm.lead'],
|
|
'project': ['project.project', 'project.task'],
|
|
'helpdesk': ['helpdesk.ticket'],
|
|
'hr': ['hr.employee', 'hr.leave'],
|
|
}
|
|
|
|
# Fields to always skip (technical/internal)
|
|
_SKIP_FIELDS = {
|
|
'__last_update', 'write_date', 'write_uid', 'create_date', 'create_uid',
|
|
'display_name', 'message_ids', 'message_follower_ids', 'message_partner_ids',
|
|
'message_channel_ids', 'message_attachment_count', 'message_has_error',
|
|
'message_has_error_counter', 'message_has_sms_error', 'message_is_follower',
|
|
'message_main_attachment_id', 'message_needaction', 'message_needaction_counter',
|
|
'message_unread', 'message_unread_counter', 'website_message_ids',
|
|
'activity_ids', 'activity_state', 'activity_user_id', 'activity_type_id',
|
|
'activity_date_deadline', 'activity_summary', 'activity_exception_decoration',
|
|
'activity_exception_icon', 'activity_type_icon', 'activity_calendar_event_id',
|
|
'rating_ids', 'rating_last_value', 'rating_last_feedback', 'rating_last_image',
|
|
'rating_count', 'rating_avg', 'rating_percentage_satisfaction',
|
|
'access_url', 'access_token', 'access_warning',
|
|
}
|
|
|
|
# Type abbreviations for compact format
|
|
_TYPE_MAP = {
|
|
'char': 'char',
|
|
'text': 'text',
|
|
'html': 'html',
|
|
'integer': 'int',
|
|
'float': 'float',
|
|
'monetary': 'mon',
|
|
'boolean': 'bool',
|
|
'date': 'date',
|
|
'datetime': 'dt',
|
|
'binary': 'bin',
|
|
'selection': 'sel',
|
|
'many2one': 'm2o',
|
|
'one2many': 'o2m',
|
|
'many2many': 'm2m',
|
|
'reference': 'ref',
|
|
}
|
|
|
|
|
|
class SchemaBuilder(models.AbstractModel):
|
|
_name = 'otk.schema.builder'
|
|
_description = "O'Toolkit Schema Builder"
|
|
|
|
@api.model
|
|
def get_business_models(self):
|
|
"""Return list of business-relevant model names.
|
|
|
|
Only includes models whose parent module is installed.
|
|
Can be overridden via ir.config_parameter 'otoolkit.business_models'.
|
|
"""
|
|
ICP = self.env['ir.config_parameter'].sudo()
|
|
custom = ICP.get_param('otoolkit.business_models', '')
|
|
if custom:
|
|
return [m.strip() for m in custom.split(',') if m.strip()]
|
|
|
|
result = []
|
|
IrModule = self.env['ir.module.module'].sudo()
|
|
for module_name, model_names in _DEFAULT_BUSINESS_MODELS.items():
|
|
if module_name == 'base':
|
|
result.extend(model_names)
|
|
continue
|
|
installed = IrModule.search([
|
|
('name', '=', module_name),
|
|
('state', '=', 'installed'),
|
|
], limit=1)
|
|
if installed:
|
|
# Verify model actually exists in registry
|
|
for model_name in model_names:
|
|
if model_name in self.env:
|
|
result.append(model_name)
|
|
return result
|
|
|
|
@api.model
|
|
def check_model_access(self, model_name):
|
|
"""Check if current user has read access to the given model."""
|
|
try:
|
|
self.env[model_name].check_access('read')
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
@api.model
|
|
def build_model_schema(self, model_name, depth=1):
|
|
"""Build a schema dict for a model with field metadata.
|
|
|
|
Args:
|
|
model_name: Technical model name (e.g. 'sale.order')
|
|
depth: How deep to follow relational fields (0 = no relations)
|
|
|
|
Returns:
|
|
dict with model info, fields list, and nested relation schemas
|
|
"""
|
|
if model_name not in self.env:
|
|
return None
|
|
|
|
Model = self.env[model_name]
|
|
model_desc = Model._description or model_name
|
|
|
|
fields_info = Model.fields_get(attributes=[
|
|
'string', 'type', 'relation', 'selection', 'required',
|
|
'readonly', 'store',
|
|
])
|
|
|
|
schema_fields = []
|
|
related_schemas = {}
|
|
|
|
for fname, finfo in sorted(fields_info.items()):
|
|
if fname.startswith('_') or fname in _SKIP_FIELDS:
|
|
continue
|
|
if not finfo.get('store', True):
|
|
continue
|
|
|
|
field_data = {
|
|
'name': fname,
|
|
'type': finfo['type'],
|
|
'label': finfo.get('string', fname),
|
|
}
|
|
|
|
if finfo['type'] == 'selection' and finfo.get('selection'):
|
|
field_data['selection'] = finfo['selection']
|
|
|
|
if finfo['type'] in ('many2one', 'one2many', 'many2many') and finfo.get('relation'):
|
|
field_data['relation'] = finfo['relation']
|
|
if depth > 0 and finfo['relation'] not in related_schemas:
|
|
related_schemas[finfo['relation']] = self.build_model_schema(
|
|
finfo['relation'], depth=depth - 1
|
|
)
|
|
|
|
if finfo.get('required'):
|
|
field_data['required'] = True
|
|
|
|
schema_fields.append(field_data)
|
|
|
|
return {
|
|
'model': model_name,
|
|
'description': model_desc,
|
|
'fields': schema_fields,
|
|
'related': related_schemas,
|
|
}
|
|
|
|
@api.model
|
|
def compress_schema(self, schema):
|
|
"""Compress a schema dict into compact text format for API.
|
|
|
|
Saves ~75% tokens vs JSON.
|
|
|
|
Format:
|
|
=== sale.order (Sales Order) ===
|
|
id:int name:char state:sel[draft,sent,sale] date_order:dt
|
|
partner_id->res.partner amount_total:mon
|
|
"""
|
|
if not schema:
|
|
return ''
|
|
|
|
lines = []
|
|
lines.append(f"=== {schema['model']} ({schema['description']}) ===")
|
|
|
|
field_parts = []
|
|
for f in schema['fields']:
|
|
ftype = _TYPE_MAP.get(f['type'], f['type'])
|
|
|
|
if f['type'] == 'selection' and f.get('selection'):
|
|
# Include labels for AI to understand meaning (key=Label)
|
|
entries = [f"{s[0]}={s[1]}" for s in f['selection']]
|
|
part = f"{f['name']}:sel[{','.join(entries)}]"
|
|
elif f['type'] in ('many2one', 'one2many', 'many2many') and f.get('relation'):
|
|
arrow = '->' if f['type'] == 'many2one' else ':o2m->' if f['type'] == 'one2many' else ':m2m->'
|
|
if f['type'] == 'many2one':
|
|
part = f"{f['name']}->{f['relation']}"
|
|
else:
|
|
part = f"{f['name']}:{ftype}->{f['relation']}"
|
|
else:
|
|
part = f"{f['name']}:{ftype}"
|
|
|
|
field_parts.append(part)
|
|
|
|
# Group field parts into lines of ~120 chars
|
|
current_line = []
|
|
current_len = 0
|
|
for part in field_parts:
|
|
if current_len + len(part) + 1 > 120 and current_line:
|
|
lines.append(' '.join(current_line))
|
|
current_line = []
|
|
current_len = 0
|
|
current_line.append(part)
|
|
current_len += len(part) + 1
|
|
if current_line:
|
|
lines.append(' '.join(current_line))
|
|
|
|
# Add related model schemas (depth-1 only, no nested relations)
|
|
for rel_name, rel_schema in (schema.get('related') or {}).items():
|
|
if rel_schema:
|
|
lines.append('')
|
|
lines.append(self.compress_schema(rel_schema))
|
|
|
|
return '\n'.join(lines)
|
|
|
|
@api.model
|
|
def build_context_for_models(self, model_names=None, depth=1):
|
|
"""Build compressed schema context for a list of models.
|
|
|
|
Args:
|
|
model_names: List of model names. If None, uses get_business_models().
|
|
depth: Relation traversal depth.
|
|
|
|
Returns:
|
|
Compressed schema string ready to send to API.
|
|
"""
|
|
if model_names is None:
|
|
model_names = self.get_business_models()
|
|
|
|
parts = []
|
|
seen = set()
|
|
for model_name in model_names:
|
|
if model_name in seen:
|
|
continue
|
|
seen.add(model_name)
|
|
if not self.check_model_access(model_name):
|
|
continue
|
|
schema = self.build_model_schema(model_name, depth=depth)
|
|
if schema:
|
|
parts.append(self.compress_schema(schema))
|
|
|
|
return '\n\n'.join(parts)
|