You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
239 lines
8.0 KiB
239 lines
8.0 KiB
# Copyright 2015-2019 Onestein (<https://www.onestein.eu>)
|
|
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl).
|
|
|
|
from collections import defaultdict
|
|
|
|
from odoo import api, models, registry
|
|
|
|
NO_BI_MODELS = [
|
|
'fetchmail.server'
|
|
]
|
|
|
|
NO_BI_TTYPES = [
|
|
'many2many',
|
|
'one2many',
|
|
'html',
|
|
'binary',
|
|
'reference'
|
|
]
|
|
|
|
|
|
def dict_for_field(field):
|
|
return {
|
|
'id': field.id,
|
|
'name': field.name,
|
|
'description': field.field_description,
|
|
'type': field.ttype,
|
|
'relation': field.relation,
|
|
'custom': False,
|
|
'model_id': field.model_id.id,
|
|
'model': field.model_id.model,
|
|
'model_name': field.model_id.name
|
|
}
|
|
|
|
|
|
def dict_for_model(model):
|
|
return {
|
|
'id': model.id,
|
|
'name': model.name,
|
|
'model': model.model
|
|
}
|
|
|
|
|
|
class IrModel(models.Model):
|
|
_inherit = 'ir.model'
|
|
|
|
@api.model
|
|
def _filter_bi_models(self, model):
|
|
|
|
def _check_name(model_model):
|
|
if model_model in NO_BI_MODELS:
|
|
return 1
|
|
return 0
|
|
|
|
def _check_startswith(model_model):
|
|
if model_model.startswith('workflow') or \
|
|
model_model.startswith('ir.') or \
|
|
model_model.startswith('base_'):
|
|
return 1
|
|
return 0
|
|
|
|
def _check_contains(model_model):
|
|
if 'mail' in model_model or \
|
|
'report' in model_model or \
|
|
'edi.' in model_model:
|
|
return 1
|
|
return 0
|
|
|
|
def _check_unknown(model_name):
|
|
if model_name == 'Unknown' or '.' in model_name:
|
|
return 1
|
|
return 0
|
|
|
|
model_model = model['model']
|
|
model_name = model['name']
|
|
count_check = 0
|
|
count_check += _check_name(model_model)
|
|
count_check += _check_startswith(model_model)
|
|
count_check += _check_contains(model_model)
|
|
count_check += _check_unknown(model_name)
|
|
if not count_check:
|
|
return self.env['ir.model.access'].check(
|
|
model['model'], 'read', False)
|
|
return False
|
|
|
|
def get_model_list(self, model_table_map):
|
|
if not model_table_map:
|
|
return []
|
|
domain = [('model_id', 'in', list(model_table_map.keys())),
|
|
('store', '=', True),
|
|
('ttype', '=', 'many2one')]
|
|
fields = self.env['ir.model.fields'].sudo().search(domain)
|
|
model_list = []
|
|
for field in fields:
|
|
for table_alias in model_table_map[field.model_id.id]:
|
|
model_list.append(dict(
|
|
dict_for_field(field),
|
|
table_alias=table_alias,
|
|
join_node=-1,
|
|
))
|
|
return model_list
|
|
|
|
def get_relation_list(self, model_table_map):
|
|
if not model_table_map:
|
|
return []
|
|
model_names = {}
|
|
for model in self.sudo().browse(model_table_map.keys()):
|
|
model_names.update({model.model: model.id})
|
|
|
|
domain = [('relation', 'in', list(model_names.keys())),
|
|
('store', '=', True),
|
|
('ttype', '=', 'many2one')]
|
|
fields = self.env['ir.model.fields'].sudo().search(domain)
|
|
relation_list = []
|
|
for field in fields:
|
|
model_id = model_names[field.relation]
|
|
for join_node in model_table_map[model_id]:
|
|
relation_list.append(dict(
|
|
dict_for_field(field),
|
|
join_node=join_node,
|
|
table_alias=-1
|
|
))
|
|
return relation_list
|
|
|
|
@api.model
|
|
def _get_related_models_domain(self, model_table_map):
|
|
domain = [('transient', '=', False)]
|
|
if model_table_map:
|
|
model_list = self.get_model_list(model_table_map)
|
|
relation_list = self.get_relation_list(model_table_map)
|
|
model_ids = [f['model_id'] for f in relation_list + model_list]
|
|
model_ids += list(model_table_map.keys())
|
|
relations = [f['relation'] for f in model_list]
|
|
domain += [
|
|
'|', ('id', 'in', model_ids), ('model', 'in', relations)]
|
|
return domain
|
|
|
|
@api.model
|
|
def get_related_models(self, model_table_map):
|
|
""" Return list of model dicts for all models that can be
|
|
joined with the already selected models.
|
|
"""
|
|
domain = self._get_related_models_domain(model_table_map)
|
|
return self.sudo().search(domain, order='name asc')
|
|
|
|
@api.model
|
|
def get_models(self, table_model_map=None):
|
|
""" Return list of model dicts for all available models.
|
|
"""
|
|
self = self.with_context(lang=self.env.user.lang)
|
|
model_table_map = defaultdict(list)
|
|
for k, v in (table_model_map or {}).items():
|
|
model_table_map[v].append(k)
|
|
|
|
models = self.get_related_models(model_table_map)
|
|
|
|
# filter out abstract models (they do not have DB tables)
|
|
non_abstract_models = self.env.registry.models.keys()
|
|
models = models.filtered(lambda m: m.model in non_abstract_models)
|
|
|
|
return list(map(dict_for_model, models))
|
|
|
|
@api.model
|
|
def get_join_nodes(self, field_data, new_field):
|
|
""" Return list of field dicts of join nodes
|
|
|
|
Return all possible join nodes to add new_field to the query
|
|
containing model_ids.
|
|
"""
|
|
def remove_duplicate_nodes(join_nodes):
|
|
seen = set()
|
|
nodes_list = []
|
|
for node in join_nodes:
|
|
node_tuple = tuple(node.items())
|
|
if node_tuple not in seen:
|
|
seen.add(node_tuple)
|
|
nodes_list.append(node)
|
|
return nodes_list
|
|
|
|
self = self.with_context(lang=self.env.user.lang)
|
|
|
|
keys = []
|
|
model_table_map = defaultdict(list)
|
|
for field in field_data:
|
|
model_table_map[field['model_id']].append(field['table_alias'])
|
|
if field.get('join_node', -1) != -1:
|
|
keys.append((field['table_alias'], field['id']))
|
|
|
|
# nodes in current model
|
|
existing_aliases = model_table_map[new_field['model_id']]
|
|
join_nodes = [{'table_alias': alias} for alias in existing_aliases]
|
|
|
|
# nodes in past selected models
|
|
for field in self.get_model_list(model_table_map):
|
|
if new_field['model'] == field['relation']:
|
|
if (field['table_alias'], field['id']) not in keys:
|
|
join_nodes.append(field)
|
|
|
|
# nodes in new model
|
|
for field in self.get_relation_list(model_table_map):
|
|
if new_field['model_id'] == field['model_id']:
|
|
if (field['table_alias'], field['id']) not in keys:
|
|
join_nodes.append(field)
|
|
|
|
return remove_duplicate_nodes(join_nodes)
|
|
|
|
@api.model
|
|
def get_fields(self, model_id):
|
|
self = self.with_context(lang=self.env.user.lang)
|
|
|
|
fields = self.env['ir.model.fields'].sudo().search([
|
|
('model_id', '=', model_id),
|
|
('store', '=', True),
|
|
('name', 'not in', models.MAGIC_COLUMNS),
|
|
('ttype', 'not in', NO_BI_TTYPES)
|
|
], order='field_description desc')
|
|
fields_dict = list(map(dict_for_field, fields))
|
|
return fields_dict
|
|
|
|
@api.model
|
|
def create(self, vals):
|
|
if self.env.context and self.env.context.get('bve'):
|
|
vals['state'] = 'base'
|
|
res = super().create(vals)
|
|
|
|
# this sql update is necessary since a write method here would
|
|
# be not working (an orm constraint is restricting the modification
|
|
# of the state field while updating ir.model)
|
|
q = "UPDATE ir_model SET state = 'manual' WHERE id = %s"
|
|
self.env.cr.execute(q, (res.id, ))
|
|
|
|
# # update registry
|
|
if self.env.context.get('bve'):
|
|
# setup models; this reloads custom models in registry
|
|
self.pool.setup_models(self._cr)
|
|
|
|
# signal that registry has changed
|
|
registry(self.env.cr.dbname).signal_changes()
|
|
|
|
return res
|