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)