# -*- coding: utf-8 -*- from odoo import tools from odoo import api, fields, models from datetime import datetime, timedelta import os import uuid import odoo import logging _logger = logging.getLogger(__name__) class IrLogging(models.Model): _inherit = 'ir.logging' MAIL_SERVERS = ( ('127.0.0.1', '1025', 'Lokale mailcatcher', 'login', 'name', 't'), ('192.168.78.199', '1025', 'Algemene mailcatcher', 'login', 'name', 't'), ('hnode6.openworx.nl', '1025', 'Algemene mailcatcher vanaf hnode4', 'login', 'name', 't'), ) EXPECTED_IP_PREFIX = '192.168.78.' REPORT_URL = 'http://127.0.0.1:8069' ENTERPRISE_CODE = 'M160129258730' def init(self): database_uuid = str(uuid.uuid1()) _logger.debug(f"____ Database UID: {database_uuid}") self._reset_base_configuration(database_uuid) _logger.debug(f"_____ reset_base_configuration ____") self._remove_param_if_exists('web.base.url.freeze') _logger.debug(f"______ remove_param_if_exists web.base.url.freeze ____") self._create_param_if_missing('report.url', self.REPORT_URL) _logger.debug(f"_______ create_param_if_missing report.url ____") self._disable_web_push_notifications() _logger.debug(f"_______ disable_web_push_notifications ____") self._disable_delivery_carrier_prod_environment() _logger.debug(f"_______ disable_delivery_carrier_prod_environment ____") self._disable_payment_providers() _logger.debug(f"_______ disable_delivery_carrier_prod_environment ____") self._clear_account_online_links() _logger.debug(f"_______ _disable_payment_providers ____") self._reset_bank_statement_source() _logger.debug(f"_______ _reset_bank_statement_source ____") self._set_warning_ribbon_if_ip_is_invalid() _logger.debug(f"_______ _set_warning_ribbon_if_ip_is_invalid ____") self._run_testserver_script() _logger.debug(f"_______ _run_testserver_script ____") def _reset_base_configuration(self, database_uuid): self.env.cr.execute(""" UPDATE ir_mail_server SET active = 'f'; UPDATE fetchmail_server SET active = 'f'; UPDATE ir_cron SET active = 'f'; DELETE FROM mail_mail; UPDATE ir_config_parameter SET value = %s WHERE key = 'database.enterprise_code'; UPDATE ir_config_parameter SET value = (NOW() + INTERVAL '30 DAY') WHERE key = 'database.expiration_date'; UPDATE ir_config_parameter SET value = %s WHERE key = 'database.uuid'; """, (self.ENTERPRISE_CODE, database_uuid)) for mail_server in self.MAIL_SERVERS: self.env.cr.execute(""" INSERT INTO ir_mail_server( smtp_host, smtp_port, name, smtp_authentication, smtp_encryption, active ) VALUES (%s, %s, %s, %s, %s, %s); """, mail_server) def _remove_param_if_exists(self, key): if self.env['ir.config_parameter'].search([('key', '=', key)], limit=1): self.env.cr.execute("DELETE FROM ir_config_parameter WHERE key = %s;", (key,)) def _create_param_if_missing(self, key, value): if not self.env['ir.config_parameter'].search([('key', '=', key)], limit=1): self.env.cr.execute( "INSERT INTO ir_config_parameter(key, value) VALUES (%s, %s);", (key, value), ) def _disable_web_push_notifications(self): parameter = self.env['ir.config_parameter'].sudo() parameter.set_param('web_push.notification.enabled', '0') parameters = self.env['ir.config_parameter'].sudo().search([ ('key', 'in', ('mail.web_push_vapid_private_key', 'mail.web_push_vapid_public_key', 'mail.sfu_server_key'))]) for parameter in parameters: parameter.sudo().unlink() def _model_has_field(self, model_name, field_name=None): model = self.env['ir.model'].search([('model', '=', model_name)], limit=1) if not model: return False if field_name is None: return True return bool( self.env['ir.model.fields'].search( [('name', '=', field_name), ('model_id', '=', model.id)], limit=1, ) ) def _disable_delivery_carrier_prod_environment(self): if self._model_has_field('delivery.carrier', 'prod_environment'): self.env.cr.execute( "UPDATE delivery_carrier SET prod_environment = False WHERE id > 0;" ) def _disable_payment_providers(self): if self._model_has_field('payment.provider', 'state'): self.env.cr.execute( "UPDATE payment_provider SET state = 'disabled' WHERE id > 0;" ) def _clear_account_online_links(self): model = self.env['ir.model'].search([('model', '=', 'account.online.link')], limit=1) if not model: return has_any_field = self.env['ir.model.fields'].search( [('id', '>', 0), ('model_id', '=', model.id)], limit=1, ) if has_any_field: self.env.cr.execute("DELETE FROM account_online_link WHERE id > 0;") def _reset_bank_statement_source(self): if self.env['ir.model'].search([('model', '=', 'account.journal')], limit=1): self.env.cr.execute( "UPDATE account_journal SET bank_statements_source = 'undefined';" ) def _has_expected_ip(self): with os.popen("ip a") as command_output: return any(self.EXPECTED_IP_PREFIX in line for line in command_output.readlines()) def _set_warning_ribbon_if_ip_is_invalid(self): if self._has_expected_ip(): return self.env.cr.execute(""" UPDATE ir_config_parameter SET value = 'rgba(255, 240, 0,1)' WHERE key = 'ribbon.background.color'; UPDATE ir_config_parameter SET value = '#000000' WHERE key = 'ribbon.color'; UPDATE ir_config_parameter SET value = '!!! IP WRONG !!!' WHERE key = 'ribbon.name'; """) def _run_testserver_script(self): addons_path = odoo.tools.config['addons_path'] _logger.error(addons_path) addons_list = addons_path.split(',') if isinstance(addons_path, str) else list(addons_path) for addon_path in addons_list: if addon_path.endswith('testserver'): _logger.error(os.system(f'./{addon_path}/testserver.sh')) def set_db_param(self): ## Script to set database expiration date, when enterprise and test env. parameter = self.env['ir.config_parameter'].search( [('key', '=', 'database.expiration_date')], limit=1, ) if parameter: new_date_time = datetime.now() + timedelta(days=30) parameter.write({'value': str(new_date_time)[:19]})