|
|
# -*- encoding: utf-8 -*- ############################################################################## # # mis_builder module for Odoo, Management Information System Builder # Copyright (C) 2014-2015 ACSONE SA/NV (<http://acsone.eu>) # # This file is a part of mis_builder # # mis_builder is free software: you can redistribute it and/or modify # it under the terms of the GNU Affero General Public License v3 or later # as published by the Free Software Foundation, either version 3 of the # License, or (at your option) any later version. # # mis_builder is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License v3 or later for more details. # # You should have received a copy of the GNU Affero General Public License # v3 or later along with this program. # If not, see <http://www.gnu.org/licenses/>. # ##############################################################################
from datetime import datetime, timedelta import re import traceback
import pytz
from openerp import api, fields, models, _ from openerp.tools.safe_eval import safe_eval
from .aep import AccountingExpressionProcessor as AEP
class AutoStruct(object):
def __init__(self, **kwargs): for k, v in kwargs.items(): setattr(self, k, v)
def _get_selection_label(selection, value): for v, l in selection: if v == value: return l return ''
def _utc_midnight(d, tz_name, add_day=0): d = fields.Date.from_string(d) if add_day: d = d + timedelta(days=add_day) utc_tz = pytz.timezone('UTC') context_tz = pytz.timezone(tz_name) local_timestamp = context_tz.localize(d, is_dst=False) return fields.Datetime.to_string(local_timestamp.astimezone(utc_tz))
def _python_var(var_str): return re.sub(r'\W|^(?=\d)', '_', var_str).lower()
def _is_valid_python_var(name): return re.match("[_A-Za-z][_a-zA-Z0-9]*$", name)
class MisReportKpi(models.Model): """ A KPI is an element (ie a line) of a MIS report.
In addition to a name and description, it has an expression to compute it based on queries defined in the MIS report. It also has various informations defining how to render it (numeric or percentage or a string, a suffix, divider) and how to render comparison of two values of the KPI. KPI's have a sequence and are ordered inside the MIS report. """
_name = 'mis.report.kpi'
name = fields.Char(size=32, required=True, string='Name') description = fields.Char(required=True, string='Description', translate=True) expression = fields.Char(required=True, string='Expression') default_css_style = fields.Char(string='Default CSS style') css_style = fields.Char(string='CSS style expression') type = fields.Selection([('num', _('Numeric')), ('pct', _('Percentage')), ('str', _('String'))], required=True, string='Type', default='num') divider = fields.Selection([('1e-6', _('µ')), ('1e-3', _('m')), ('1', _('1')), ('1e3', _('k')), ('1e6', _('M'))], string='Factor', default='1') dp = fields.Integer(string='Rounding', default=0) suffix = fields.Char(size=16, string='Suffix') compare_method = fields.Selection([('diff', _('Difference')), ('pct', _('Percentage')), ('none', _('None'))], required=True, string='Comparison Method', default='pct') sequence = fields.Integer(string='Sequence', default=100) report_id = fields.Many2one('mis.report', string='Report', ondelete='cascade')
_order = 'sequence, id'
@api.one @api.constrains('name') def _check_name(self): return _is_valid_python_var(self.name)
@api.onchange('name') def _onchange_name(self): if self.name and not _is_valid_python_var(self.name): return { 'warning': { 'title': 'Invalid name %s' % self.name, 'message': 'The name must be a valid python identifier' } }
@api.onchange('description') def _onchange_description(self): """ construct name from description """ if self.description and not self.name: self.name = _python_var(self.description)
@api.onchange('type') def _onchange_type(self): # TODO: change compare_method, divider and dp for all 3 types if self.type == 'pct': self.compare_method = 'diff' elif self.type == 'str': self.compare_method = 'none' self.divider = '' self.dp = 0
def render(self, lang_id, value): """ render a KPI value as a unicode string, ready for display """ assert len(self) == 1 if value is None: return '#N/A' elif self.type == 'num': return self._render_num(lang_id, value, self.divider, self.dp, self.suffix) elif self.type == 'pct': return self._render_num(lang_id, value, 0.01, self.dp, '%') else: return unicode(value)
def render_comparison(self, lang_id, value, base_value, average_value, average_base_value): """ render the comparison of two KPI values, ready for display """ assert len(self) == 1 if value is None or base_value is None: return '' if self.type == 'pct': return self._render_num( lang_id, value - base_value, 0.01, self.dp, _('pp'), sign='+') elif self.type == 'num': if average_value: value = value / float(average_value) if average_base_value: base_value = base_value / float(average_base_value) if self.compare_method == 'diff': return self._render_num( lang_id, value - base_value, self.divider, self.dp, self.suffix, sign='+') elif self.compare_method == 'pct': if round(base_value, self.dp) != 0: return self._render_num( lang_id, (value - base_value) / abs(base_value), 0.01, self.dp, '%', sign='+') return ''
def _render_num(self, lang_id, value, divider, dp, suffix, sign='-'): divider_label = _get_selection_label( self._columns['divider'].selection, divider) if divider_label == '1': divider_label = '' # format number following user language value = round(value / float(divider or 1), dp) or 0 value = self.env.registry['res.lang'].format( self.env.cr, self.env.uid, [lang_id], '%%%s.%df' % (sign, dp), value, grouping=True, context=self.env.context) value = u'%s\N{NO-BREAK SPACE}%s%s' % \ (value, divider_label, suffix or '') value = value.replace('-', u'\N{NON-BREAKING HYPHEN}') return value
class MisReportQuery(models.Model): """ A query to fetch arbitrary data for a MIS report.
A query works on a model and has a domain and list of fields to fetch. At runtime, the domain is expanded with a "and" on the date/datetime field. """
_name = 'mis.report.query'
@api.one @api.depends('field_ids') def _compute_field_names(self): field_names = [field.name for field in self.field_ids] self.field_names = ', '.join(field_names)
name = fields.Char(size=32, required=True, string='Name') model_id = fields.Many2one('ir.model', required=True, string='Model') field_ids = fields.Many2many('ir.model.fields', required=True, string='Fields to fetch') field_names = fields.Char(compute='_compute_field_names', string='Fetched fields name') aggregate = fields.Selection([('sum', _('Sum')), ('avg', _('Average')), ('min', _('Min')), ('max', _('Max'))], string='Aggregate') date_field = fields.Many2one('ir.model.fields', required=True, string='Date field', domain=[('ttype', 'in', ('date', 'datetime'))]) domain = fields.Char(string='Domain') report_id = fields.Many2one('mis.report', string='Report', ondelete='cascade')
_order = 'name'
@api.one @api.constrains('name') def _check_name(self): return _is_valid_python_var(self.name)
class MisReport(models.Model): """ A MIS report template (without period information)
The MIS report holds: * a list of explicit queries; the result of each query is stored in a variable with same name as a query, containing as list of data structures populated with attributes for each fields to fetch; when queries have an aggregate method and no fields to group, it returns a data structure with the aggregated fields * a list of KPI to be evaluated based on the variables resulting from the accounting data and queries (KPI expressions can references queries and accounting expression - see AccoutingExpressionProcessor) """
_name = 'mis.report'
name = fields.Char(required=True, string='Name', translate=True) description = fields.Char(required=False, string='Description', translate=True) query_ids = fields.One2many('mis.report.query', 'report_id', string='Queries') kpi_ids = fields.One2many('mis.report.kpi', 'report_id', string='KPI\'s')
# TODO: kpi name cannot be start with query name
class MisReportInstancePeriod(models.Model): """ A MIS report instance has the logic to compute
a report template for a given date period.
Periods have a duration (day, week, fiscal period) and are defined as an offset relative to a pivot date. """
@api.one @api.depends('report_instance_id.pivot_date', 'type', 'offset', 'duration') def _compute_dates(self): self.date_from = False self.date_to = False self.period_from = False self.period_to = False self.valid = False d = fields.Date.from_string(self.report_instance_id.pivot_date) if self.type == 'd': date_from = d + timedelta(days=self.offset) date_to = date_from + timedelta(days=self.duration - 1) self.date_from = fields.Date.to_string(date_from) self.date_to = fields.Date.to_string(date_to) self.valid = True elif self.type == 'w': date_from = d - timedelta(d.weekday()) date_from = date_from + timedelta(days=self.offset * 7) date_to = date_from + timedelta(days=(7 * self.duration) - 1) self.date_from = fields.Date.to_string(date_from) self.date_to = fields.Date.to_string(date_to) self.valid = True elif self.type == 'fp': current_periods = self.env['account.period'].search( [('special', '=', False), ('date_start', '<=', d), ('date_stop', '>=', d), ('company_id', '=', self.report_instance_id.company_id.id)]) if current_periods: all_periods = self.env['account.period'].search( [('special', '=', False), ('company_id', '=', self.report_instance_id.company_id.id)], order='date_start') all_period_ids = [p.id for p in all_periods] p = all_period_ids.index(current_periods[0].id) + self.offset if p >= 0 and p + self.duration <= len(all_period_ids): periods = all_periods[p:p + self.duration] self.date_from = periods[0].date_start self.date_to = periods[-1].date_stop self.period_from = periods[0] self.period_to = periods[-1] self.valid = True
_name = 'mis.report.instance.period'
name = fields.Char(size=32, required=True, string='Description', translate=True) type = fields.Selection([('d', _('Day')), ('w', _('Week')), ('fp', _('Fiscal Period')), # ('fy', _('Fiscal Year')) ], required=True, string='Period type') offset = fields.Integer(string='Offset', help='Offset from current period', default=-1) duration = fields.Integer(string='Duration', help='Number of periods', default=1) date_from = fields.Date(compute='_compute_dates', string="From") date_to = fields.Date(compute='_compute_dates', string="To") period_from = fields.Many2one(compute='_compute_dates', comodel_name='account.period', string="From period") period_to = fields.Many2one(compute='_compute_dates', comodel_name='account.period', string="To period") valid = fields.Boolean(compute='_compute_dates', type='boolean', string='Valid') sequence = fields.Integer(string='Sequence', default=100) report_instance_id = fields.Many2one('mis.report.instance', string='Report Instance', ondelete='cascade') comparison_column_ids = fields.Many2many( comodel_name='mis.report.instance.period', relation='mis_report_instance_period_rel', column1='period_id', column2='compare_period_id', string='Compare with') normalize_factor = fields.Integer( string='Factor', help='Factor to use to normalize the period (used in comparison', default=1)
_order = 'sequence, id'
_sql_constraints = [ ('duration', 'CHECK (duration>0)', 'Wrong duration, it must be positive!'), ('normalize_factor', 'CHECK (normalize_factor>0)', 'Wrong normalize factor, it must be positive!'), ('name_unique', 'unique(name, report_instance_id)', 'Period name should be unique by report'), ]
@api.multi def drilldown(self, expr): assert len(self) == 1 if AEP.has_account_var(expr): aep = AEP(self.env) aep.parse_expr(expr) aep.done_parsing(self.report_instance_id.root_account) domain = aep.get_aml_domain_for_expr( expr, self.date_from, self.date_to, self.period_from, self.period_to, self.report_instance_id.target_move) return { 'name': expr + ' - ' + self.name, 'domain': domain, 'type': 'ir.actions.act_window', 'res_model': 'account.move.line', 'views': [[False, 'list'], [False, 'form']], 'view_type': 'list', 'view_mode': 'list', 'target': 'current', } else: return False
def _fetch_queries(self): assert len(self) == 1 res = {} for query in self.report_instance_id.report_id.query_ids: model = self.env[query.model_id.model] domain = query.domain and safe_eval(query.domain) or [] if query.date_field.ttype == 'date': domain.extend([(query.date_field.name, '>=', self.date_from), (query.date_field.name, '<=', self.date_to)]) else: datetime_from = _utc_midnight( self.date_from, self._context.get('tz', 'UTC')) datetime_to = _utc_midnight( self.date_to, self._context.get('tz', 'UTC'), add_day=1) domain.extend([(query.date_field.name, '>=', datetime_from), (query.date_field.name, '<', datetime_to)]) # TODO: we probably don't need company_id here if model._columns.get('company_id'): domain.extend(['|', ('company_id', '=', False), ('company_id', '=', self.report_instance_id.company_id.id)]) field_names = [f.name for f in query.field_ids] if not query.aggregate: data = model.search_read(domain, field_names) res[query.name] = [AutoStruct(**d) for d in data] elif query.aggregate == 'sum': data = model.read_group( domain, field_names, []) s = AutoStruct(count=data[0]['__count']) for field_name in field_names: v = data[0][field_name] setattr(s, field_name, v) res[query.name] = s else: data = model.search_read(domain, field_names) s = AutoStruct(count=len(data)) if query.aggregate == 'min': agg = min elif query.aggregate == 'max': agg = max elif query.aggregate == 'avg': agg = lambda l: sum(l) / float(len(l)) for field_name in field_names: setattr(s, field_name, agg([d[field_name] for d in data])) res[query.name] = s return res
def _compute(self, lang_id, aep): res = {}
localdict = { 'registry': self.pool, 'sum': sum, 'min': min, 'max': max, 'len': len, 'avg': lambda l: sum(l) / float(len(l)), }
localdict.update(self._fetch_queries())
aep.do_queries(self.date_from, self.date_to, self.period_from, self.period_to, self.report_instance_id.target_move)
compute_queue = self.report_instance_id.report_id.kpi_ids recompute_queue = [] while True: for kpi in compute_queue: try: kpi_val_comment = kpi.name + " = " + kpi.expression kpi_eval_expression = aep.replace_expr(kpi.expression) kpi_val = safe_eval(kpi_eval_expression, localdict) except ZeroDivisionError: kpi_val = None kpi_val_rendered = '#DIV/0' kpi_val_comment += '\n\n%s' % (traceback.format_exc(),) except (NameError, ValueError): recompute_queue.append(kpi) kpi_val = None kpi_val_rendered = '#ERR' kpi_val_comment += '\n\n%s' % (traceback.format_exc(),) except: kpi_val = None kpi_val_rendered = '#ERR' kpi_val_comment += '\n\n%s' % (traceback.format_exc(),) else: kpi_val_rendered = kpi.render(lang_id, kpi_val)
localdict[kpi.name] = kpi_val try: kpi_style = None if kpi.css_style: kpi_style = safe_eval(kpi.css_style, localdict) except: # TODO: log warning kpi_style = None
drilldown = (kpi_val is not None and AEP.has_account_var(kpi.expression))
res[kpi.name] = { 'val': kpi_val, 'val_r': kpi_val_rendered, 'val_c': kpi_val_comment, 'style': kpi_style, 'suffix': kpi.suffix, 'dp': kpi.dp, 'is_percentage': kpi.type == 'pct', 'period_id': self.id, 'expr': kpi.expression, 'drilldown': drilldown, }
if len(recompute_queue) == 0: # nothing to recompute, we are done break if len(recompute_queue) == len(compute_queue): # could not compute anything in this iteration # (ie real Value errors or cyclic dependency) # so we stop trying break # try again compute_queue = recompute_queue recompute_queue = []
return res
class MisReportInstance(models.Model): """The MIS report instance combines everything to compute
a MIS report template for a set of periods."""
@api.one @api.depends('date') def _compute_pivot_date(self): if self.date: self.pivot_date = self.date else: self.pivot_date = fields.Date.context_today(self)
@api.one @api.depends('company_id') def _compute_root_account(self): self.root_account = False accounts = self.env['account.account'].search( [('parent_id', '=', False), ('company_id', '=', self.company_id.id)]) if len(accounts) == 1: self.root_account = accounts[0]
_name = 'mis.report.instance'
name = fields.Char(required=True, string='Name', translate=True) description = fields.Char(required=False, string='Description', translate=True) date = fields.Date(string='Base date', help='Report base date ' '(leave empty to use current date)') pivot_date = fields.Date(compute='_compute_pivot_date', string="Pivot date") report_id = fields.Many2one('mis.report', required=True, string='Report') period_ids = fields.One2many('mis.report.instance.period', 'report_instance_id', required=True, string='Periods') target_move = fields.Selection([('posted', 'All Posted Entries'), ('all', 'All Entries')], string='Target Moves', required=True, default='posted') company_id = fields.Many2one('res.company', 'Company', required=True, default=lambda self: self.env['res.company']. _company_default_get('mis.report.instance')) root_account = fields.Many2one(compute='_compute_root_account', comodel_name='account.account', string="Account chart") landscape_pdf = fields.Boolean(string='Landscape PDF')
def _format_date(self, lang_id, date): # format date following user language date_format = self.env['res.lang'].browse(lang_id).date_format return datetime.strftime(fields.Date.from_string(date), date_format)
@api.multi def preview(self): assert len(self) == 1 view_id = self.env.ref('mis_builder.' 'mis_report_instance_result_view_form') return { 'type': 'ir.actions.act_window', 'res_model': 'mis.report.instance', 'res_id': self.id, 'view_mode': 'form', 'view_type': 'form', 'view_id': view_id.id, 'target': 'new', }
@api.multi def compute(self): assert len(self) == 1
# prepare AccountingExpressionProcessor aep = AEP(self.env) for kpi in self.report_id.kpi_ids: aep.parse_expr(kpi.expression) aep.done_parsing(self.root_account)
# fetch user language only once # TODO: is this necessary? lang = self.env.user.lang if not lang: lang = 'en_US' lang_id = self.env['res.lang'].search([('code', '=', lang)]).id
# compute kpi values for each period kpi_values_by_period_ids = {} for period in self.period_ids: if not period.valid: continue kpi_values = period._compute(lang_id, aep) kpi_values_by_period_ids[period.id] = kpi_values
# prepare header and content header = [] header.append({ 'kpi_name': '', 'cols': [] }) content = [] rows_by_kpi_name = {} for kpi in self.report_id.kpi_ids: rows_by_kpi_name[kpi.name] = { 'kpi_name': kpi.description, 'cols': [], 'default_style': kpi.default_css_style } content.append(rows_by_kpi_name[kpi.name])
# populate header and content for period in self.period_ids: if not period.valid: continue # add the column header # TODO: format period.date_from header[0]['cols'].append(dict( name=period.name, date=(period.duration > 1 or period.type == 'w') and _('from %s to %s' % (period.period_from and period.period_from.name or self._format_date(lang_id, period.date_from), period.period_to and period.period_to.name or self._format_date(lang_id, period.date_to))) or period.period_from and period.period_from.name or period.date_from)) # add kpi values kpi_values = kpi_values_by_period_ids[period.id] for kpi_name in kpi_values: rows_by_kpi_name[kpi_name]['cols'].append(kpi_values[kpi_name])
# add comparison columns for compare_col in period.comparison_column_ids: compare_kpi_values = \ kpi_values_by_period_ids.get(compare_col.id) if compare_kpi_values: # add the comparison column header # TODO: make 'vs' translatable header[0]['cols'].append( dict(name='%s vs %s' % (period.name, compare_col.name), date='')) # add comparison values for kpi in self.report_id.kpi_ids: rows_by_kpi_name[kpi.name]['cols'].append({ 'val_r': kpi.render_comparison( lang_id, kpi_values[kpi.name]['val'], compare_kpi_values[kpi.name]['val'], period.normalize_factor, compare_col.normalize_factor) })
return {'header': header, 'content': content}
|