reporting-engine/kpi_dashboard/models/kpi_kpi.py

# Copyright 2020 Creu Blanca
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).

from odoo import api, fields, models, _
from odoo.exceptions import ValidationError
import ast
from odoo.tools.safe_eval import safe_eval
from odoo.addons.base.models.ir_cron import _intervalTypes
from odoo.tools.float_utils import float_compare
import re
import json
import datetime


class KpiKpi(models.Model):
    _name = "kpi.kpi"
    _description = "Kpi Kpi"

    name = fields.Char(required=True)
    active = fields.Boolean(default=True)
    cron_id = fields.Many2one("ir.cron", readonly=True, copy=False)
    computation_method = fields.Selection(
        [("function", "Function"), ("code", "Code")], required=True
    )
    value = fields.Serialized()
    dashboard_item_ids = fields.One2many("kpi.dashboard.item", inverse_name="kpi_id")
    model_id = fields.Many2one("ir.model",)
    function = fields.Char()
    args = fields.Char()
    kwargs = fields.Char()
    widget = fields.Selection(
        [("number", "Number"), ("meter", "Meter"), ("graph", "Graph")],
        required=True,
        default="number",
    )
    value_last_update = fields.Datetime(readonly=True)
    prefix = fields.Char()
    suffix = fields.Char()
    action_ids = fields.One2many(
        "kpi.kpi.action",
        inverse_name='kpi_id',
        help="Actions that can be opened from the KPI"
    )
    code = fields.Text("Code")
    store_history = fields.Boolean()
    store_history_interval = fields.Selection(
        selection=lambda self:
        self.env['ir.cron']._fields['interval_type'].selection,
    )
    store_history_interval_number = fields.Integer()
    compute_on_fly = fields.Boolean()
    history_ids = fields.One2many("kpi.kpi.history", inverse_name="kpi_id")
    computed_value = fields.Serialized(compute='_compute_computed_value')
    computed_date = fields.Datetime(compute='_compute_computed_value')

    @api.depends('value', 'value_last_update', 'compute_on_fly')
    def _compute_computed_value(self):
        for record in self:
            if record.compute_on_fly:
                record.computed_value = record._compute_value()
                record.computed_date = fields.Datetime.now()
            else:
                record.computed_value = record.value
                record.computed_date = record.value_last_update

    def _cron_vals(self):
        return {
            "name": self.name,
            "model_id": self.env.ref("kpi_dashboard.model_kpi_kpi").id,
            "interval_number": 1,
            "interval_type": "hours",
            "state": "code",
            "code": "model.browse(%s).compute()" % self.id,
            "active": True,
        }

    def compute(self):
        for record in self:
            record._compute()
        return True

    def _generate_history_vals(self, value):
        return {
            "kpi_id": self.id,
            "value": value,
            "widget": self.widget,
        }

    def _compute_value(self):
        return getattr(self, "_compute_value_%s" % self.computation_method)()

    def _compute(self):
        value = self._compute_value()
        self.write({"value": value})
        if self.store_history:
            last = self.env['kpi.kpi.history'].search([
                ('kpi_id', '=', self.id)
            ], limit=1)
            if (
                not last or
                not self.store_history_interval or
                last.create_date + _intervalTypes[self.store_history_interval](
                    self.store_history_interval_number) < fields.Datetime.now()
            ):
                self.env["kpi.kpi.history"].create(
                    self._generate_history_vals(value)
                )
        notifications = []
        for dashboard_item in self.dashboard_item_ids:
            channel = "kpi_dashboard_%s" % dashboard_item.dashboard_id.id
            notifications.append([channel, dashboard_item._read_dashboard()])
        if notifications:
            self.env["bus.bus"].sendmany(notifications)

    def _compute_value_function(self):
        obj = self
        if self.model_id:
            obj = self.env[self.model_id.model]
        args = ast.literal_eval(self.args or "[]")
        kwargs = ast.literal_eval(self.kwargs or "{}")
        return getattr(obj, self.function)(*args, **kwargs)

    def generate_cron(self):
        self.ensure_one()
        self.cron_id = self.env["ir.cron"].create(self._cron_vals())

    @api.multi
    def write(self, vals):
        if "value" in vals:
            vals["value_last_update"] = fields.Datetime.now()
        return super().write(vals)

    def _get_code_input_dict(self):
        return {
            "self": self,
            "model": self.browse(),
            "datetime": datetime,
            "float_compare": float_compare,
        }

    def _forbidden_code(self):
        return ["commit", "rollback", "getattr", "execute"]

    def _compute_value_code(self):
        forbidden = self._forbidden_code()
        search_terms = "(" + ("|".join(forbidden)) + ")"
        if re.search(search_terms, (self.code or "").lower()):
            message = ", ".join(forbidden[:-1]) or ""
            if len(message) > 0:
                message += _(" or ")
            message += forbidden[-1]
            raise ValidationError(_(
                "The code cannot contain the following terms: %s."
            ) % message)
        results = self._get_code_input_dict()
        savepoint = "kpi_formula_%s" % self.id
        self.env.cr.execute("savepoint %s" % savepoint)
        safe_eval(self.code or "", results, mode="exec", nocopy=True)
        self.env.cr.execute("rollback to %s" % savepoint)
        return results.get("result", {})

    def show_value(self):
        self.ensure_one()
        action = self.env.ref('kpi_dashboard.kpi_kpi_act_window')
        result = action.read()[0]
        result.update({
            'res_id': self.id,
            'target': 'new',
            'view_mode': 'form',
            'views': [(self.env.ref(
                'kpi_dashboard.kpi_kpi_widget_form_view'
            ).id, 'form')],
        })
        return result


class KpiKpiAction(models.Model):
    _name = 'kpi.kpi.action'
    _description = 'KPI action'

    kpi_id = fields.Many2one('kpi.kpi', required=True, ondelete='cascade')
    action = fields.Reference(
        selection=[('ir.actions.report', 'ir.actions.report'),
                   ('ir.actions.act_window', 'ir.actions.act_window'),
                   ('ir.actions.act_url', 'ir.actions.act_url'),
                   ('ir.actions.server', 'ir.actions.server'),
                   ('ir.actions.client', 'ir.actions.client')],
        required=True,
    )
    context = fields.Char()

    def read_dashboard(self):
        result = {}
        for r in self:
            result[r.id] = {
                'id': r.action.id,
                'type': r.action._name,
                'name': r.action.name,
                'context': safe_eval(r.context or '{}')
            }
        return result


class KpiKpiHistory(models.Model):
    _name = 'kpi.kpi.history'
    _description = 'KPI history'
    _order = 'create_date DESC'

    kpi_id = fields.Many2one(
        'kpi.kpi', required=True, ondelete='cascade', readonly=True
    )
    value = fields.Serialized(readonly=True)
    raw_value = fields.Char(compute='_compute_raw_value')
    name = fields.Char(related='kpi_id.name')
    widget = fields.Selection(
        selection=lambda self:
        self.env['kpi.kpi']._fields['widget'].selection,
        required=True)

    @api.depends('value')
    def _compute_raw_value(self):
        for record in self:
            record.raw_value = json.dumps(record.value)

    def show_form(self):
        self.ensure_one()
        action = self.env.ref('kpi_dashboard.kpi_kpi_history_act_window')
        result = action.read()[0]
        result.update({
            'res_id': self.id,
            'target': 'new',
            'view_mode': 'form',
            'views': [(self.env.context.get('form_id'), 'form')],
        })
        return result