Coverage for ingadhoc-odoo-saas-adhoc / saas_provider_upgrade / models / saas_provider_upgrade_util.py: 54%
72 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-03-09 19:24 +0000
1##############################################################################
2# For copyright and license notices, see __manifest__.py file in module root
3# directory
4##############################################################################
5import difflib
6import json
7import re
8from typing import Any, Literal
10import requests
11from markupsafe import Markup
12from odoo import api, models
13from odoo.models import BaseModel
14from odoo.tools import safe_eval
16from ..constants import UL_TYPES_FREEZE_IGNORES
19class SaasProviderUpgradeUtil(models.AbstractModel):
20 _name = "saas.provider.upgrade.util"
21 _description = "saas.provider.upgrade.util"
23 @api.model
24 def build_display_notification_action(
25 self, title: str, message: str, type: Literal["success", "warning", "danger", "info"] = "success"
26 ) -> dict[str, Any]:
27 """
28 Used to build "display_notification" action. Default type is "success"
30 :param title: Notification title
31 :param message: Notification message
32 :param type: Notification type (success, warning, danger, info)
33 :return: A dictionary representing the notification action
34 """
35 return {
36 "type": "ir.actions.client",
37 "tag": "display_notification",
38 "params": {
39 "title": title,
40 "type": type,
41 "message": message,
42 "next": {"type": "ir.actions.act_window_close"},
43 },
44 }
46 @api.model
47 def create_message(self, title: str, body: str) -> Markup:
48 """
49 Creates a message with a title and a body. It uses 'type' to determine the message type.
50 By default is an error message.
52 :param title: The title of the message
53 :param body: The body of the message (can contain HTML)
54 """
55 message = (Markup("<p><b>") + "ERROR: " + Markup("</b>") + "%s") % (title,)
56 message += Markup("<blockquote>%s</blockquote>") % (Markup(body))
57 return message
59 @api.model
60 def urgent_communication(self, msg: str):
61 """
62 Method used to send a message to the Telegram channel configured in the saas_provider_upgrade module.
63 Is used in the after_done UL scripts, and in the after_error scripts in the Upgrade Types.
65 :param msg: The message to send to the Telegram channel
66 """
67 chat_id = self.env["ir.config_parameter"].sudo().get_param("saas_provider_upgrade.telegram_chat_id", False)
68 bot_token = self.env["ir.config_parameter"].sudo().get_param("saas_provider_upgrade.telegram_bot_token", False)
69 channel = self._get_channel_upgrade()
70 url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
71 if bot_token and chat_id:
72 response = requests.post(
73 url,
74 json={"chat_id": chat_id, "text": msg},
75 headers={"Content-Type": "application/json"},
76 timeout=30,
77 )
78 if response.status_code != 200:
79 msg += f" (Fail posting on telegram with: error {response.status_code}"
80 channel.message_post(
81 body=msg,
82 subtype_id=self.env.ref("mail.mt_comment").id,
83 message_type="comment",
84 body_is_html=True,
85 )
87 if not (bot_token and chat_id):
88 channel.message_post(
89 body=msg,
90 subtype_id=self.env.ref("mail.mt_comment").id,
91 message_type="comment",
92 body_is_html=True,
93 )
95 @api.model
96 def similar_content(self, content_1: str, content_2: str, threshold: float = 0.85):
97 """
98 Check if two contents have similar content based on a threshold.
100 :param content_1: First content
101 :param content_2: Second content
102 :param threshold: Similarity threshold (default is 0.85)
103 :return: True if contents are similar, False otherwise
104 """
105 if content_1 and content_2:
106 ratio = difflib.SequenceMatcher(None, content_1, content_2).ratio()
107 return ratio >= threshold
109 return False
111 @api.model
112 def is_serializable(self, value: Any) -> bool:
113 """
114 Method to check if a value is JSON serializable.
116 :param value: Value to check
117 :return: True if the value is serializable, False otherwise
118 """
119 try:
120 json.dumps(value)
121 return True
122 except Exception:
123 return False
125 def action_reload(self) -> dict:
126 """
127 Reload the client.
129 :return: Action to reload the client
130 """
131 return {"type": "ir.actions.client", "tag": "soft_reload"}
133 @api.model
134 def _production_freeze_ignore(self, ul_type: str) -> bool:
135 """
136 Check if the upgrade line type should be ignored during production freeze.
138 :param ul_type: Upgrade line type
139 :return: True if the upgrade line type should be ignored, False otherwise
140 """
141 return ul_type in UL_TYPES_FREEZE_IGNORES
143 @api.model
144 def get_portal_requests_domain(self, ticket_id: int) -> list:
145 return [
146 ("ticket_id.id", "=", ticket_id),
147 ("hide_from_portal", "=", False),
148 "&",
149 ("aim", "!=", "production"),
150 "|",
151 "&",
152 ("original_database_id", "=", False),
153 ("upgraded_database_id", "=", False),
154 ("state", "!=", "cancel"),
155 "|",
156 ("original_database_state", "!=", "deleted"),
157 ("upgraded_database_state", "!=", "deleted"),
158 ("next_request_id", "=", False),
159 ]
161 @api.model
162 def _get_filters_eval_context(self) -> dict:
163 return {
164 "datetime": safe_eval.datetime,
165 "dateutil": safe_eval.dateutil,
166 "today": safe_eval.datetime.datetime.today(),
167 "time": safe_eval.time,
168 "user": self.env.user,
169 }
171 @api.model
172 def extract_qweb_variables(self, message: str) -> set[str]:
173 """
174 Extracts variable names used in a QWeb message template.
176 Matches attributes like `t-out="var"`, `t-esc='var'`, `t-if="cond"`,
177 `t-elif`, `t-foreach`, `t-set`, etc., returning the variable names found.
179 :param message: The QWeb template message to analyze
180 :return: Set of variable names used in the template
181 """
182 # Pattern to match t-* attributes with simple variable values
183 pattern = r't-(?:out|esc|if|elif|foreach|set)\s*=\s*["\']([a-zA-Z_][a-zA-Z0-9_]*)["\']'
184 return set(re.findall(pattern, message))
186 @api.model
187 def json_to_display(self, val: dict | list) -> str:
188 try:
189 return json.dumps(val, ensure_ascii=False, separators=(",", ": "))
190 except Exception:
191 return str(val)
193 @api.model
194 def _get_channel_upgrade(self) -> BaseModel:
195 """Retrieve the id of the upgrade discuss channel
197 Returns:
198 Recordset: Record of the upgrade discuss channel
199 """
200 channel_id = self.env["ir.config_parameter"].sudo().get_param("saas_provider_upgrade.discuss_channel_id", "0")
201 return self.env["discuss.channel"].sudo().browse(int(channel_id))