Coverage for ingadhoc-odoo-saas-adhoc / saas_provider_upgrade / models / helpdesk_ticket_customer_note.py: 75%
165 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-03-09 18:05 +0000
1##############################################################################
2# For copyright and license notices, see __manifest__.py file in module root
3# directory
4##############################################################################
6import hashlib
7import logging
8import re
9from datetime import timedelta
10from typing import TYPE_CHECKING
12from lxml import etree
13from markupsafe import Markup
14from odoo import SUPERUSER_ID, api, fields, models
15from odoo.addons.html_editor.tools import handle_history_divergence
16from odoo.exceptions import UserError
17from odoo.fields import Domain
18from odoo.models import BaseModel
19from odoo.tools import is_html_empty, safe_eval
21if TYPE_CHECKING:
22 from ..models.helpdesk_ticket_upgrade_request import HelpdeskTicketUpgradeRequest as UpgradeRequest
23 from ..models.saas_upgrade_line import SaasUpgradeLine as SaasUpgradeLine
25_logger = logging.getLogger(__name__)
28class HelpdeskTicketCustomerNote(models.Model):
29 _name = "helpdesk.ticket.customer_note"
30 _description = "Helpdesk Ticket Customer Notes"
31 _rec_name = "title"
32 _order = "priority desc, sequence, adhoc_product_id"
33 _inherit = [
34 "mail.thread",
35 "mail.activity.mixin",
36 "portal.mixin",
37 "saas.provider.upgrade.util",
38 "html.field.history.mixin",
39 ]
41 def _get_versioned_fields(self):
42 return [HelpdeskTicketCustomerNote.description.name]
44 sequence = fields.Integer()
45 adhoc_product_id = fields.Many2one(
46 "adhoc.product",
47 domain=[
48 ("parent_id", "!=", False),
49 ("type", "=", "horizontal"),
50 ],
51 )
52 title = fields.Char(
53 required=True,
54 tracking=True,
55 )
56 description = fields.Html(
57 copy=False,
58 )
59 html_field_history = fields.Json(copy=False)
60 notes = fields.Text()
61 status = fields.Selection(
62 [
63 ("ongoing", "In Progress"),
64 ("done", "Completed"),
65 ],
66 default="ongoing",
67 required=True,
68 tracking=True,
69 help="* Ongoing: represents notes that are being worked on.\n"
70 "* Done: represents notes that have already been analyzed and require no further action.",
71 )
72 responsible = fields.Char(tracking=True)
73 ticket_id = fields.Many2one(
74 "helpdesk.ticket",
75 required=True,
76 ondelete="cascade",
77 )
78 upgrade_type_id = fields.Many2one(
79 "saas.upgrade.type",
80 readonly=True,
81 help="Type to know to which upgrade jump this customer note belongs.",
82 )
83 upgrade_line_id = fields.Many2one(
84 "saas.upgrade.line",
85 ondelete="cascade",
86 readonly=True,
87 )
88 post_production = fields.Boolean()
89 code_validation = fields.Boolean(
90 related="upgrade_line_id.customer_note_validation",
91 depends=["upgrade_line_id.customer_note_validation"],
92 store=True,
93 )
94 upload_changes = fields.Boolean(
95 related="upgrade_line_id.customer_note_upload_changes",
96 store=True,
97 )
98 parent_adhoc_product_id = fields.Many2one(
99 related="adhoc_product_id.parent_id",
100 store=True,
101 )
102 active = fields.Boolean(
103 default=True,
104 tracking=True,
105 )
106 priority = fields.Selection(
107 related="upgrade_line_id.priority",
108 depends=["upgrade_line_id.priority"],
109 store=True,
110 )
111 used_placeholders_hash = fields.Char()
112 client_config_line_ids = fields.One2many(
113 "saas.upgrade.client.config.line",
114 "customer_note_id",
115 string="Client Configurations",
116 )
118 def action_open_client_config_lines(self) -> dict:
119 self.ensure_one()
120 action = self.env["ir.actions.act_window"]._for_xml_id(
121 "saas_provider_upgrade.action_saas_upgrade_client_config_line"
122 )
123 action["domain"] = [["customer_note_id", "=", self.id]]
124 action["context"] = {"default_customer_note_id": self.id}
125 return action
127 def action_create_ticket(self):
128 self.ensure_one()
129 description = self.env._("""
130 <p>Ticket generated from Customer Note: %s</p>
131 <h2>Description</h2>
132 %s
133 <h2>Notes</h2>
134 %s
135 """)
136 new_rec = self.env["helpdesk.ticket"].create(
137 {
138 "project_id": self.ticket_id.project_id.id,
139 "name": self.title,
140 "ticket_description": description % ("%s (#%s)" % (self.title, self.id), self.description, self.notes),
141 "adhoc_product_id": self.adhoc_product_id.id,
142 "parent_id": self.ticket_id.id,
143 }
144 )
145 msg = self.env._(
146 "Follow-up in ticket: <a target='_blank' href='%s'>%s</a>"
147 % (
148 new_rec.access_url,
149 new_rec.display_name,
150 )
151 )
152 self.message_post(body=Markup(msg))
153 self.notes = (self.notes + "<br/>" if self.notes else "") + msg
155 def action_create_task(self):
156 self.ensure_one()
157 description = self.env._("""
158 <p>Task generated from Customer Note: %s</p>
159 <h2>Description</h2>
160 %s
161 <h2>Notes</h2>
162 %s
163 """)
164 new_rec = self.env["project.task"].create(
165 {
166 "project_id": self.ticket_id.project_id.id,
167 "name": self.title,
168 "description": description % ("%s (#%s)" % (self.title, self.id), self.description, self.notes),
169 "adhoc_product_id": self.adhoc_product_id.id,
170 "ticket_ids": [(4, self.ticket_id.id, 0)],
171 }
172 )
173 msg = self.env._(
174 "Follow-up in ticket: <a target='_blank' href='%s'>%s</a>"
175 % (
176 new_rec.access_url,
177 new_rec.display_name,
178 )
179 )
180 self.message_post(body=Markup(msg))
181 self.notes = (self.notes + "<br/>" if self.notes else "") + msg
183 def action_toggle_status(self):
184 """
185 Function to change the status of the customer note.
186 Called from 'customer_note.js'.
188 :param self: Customer Note recordset
189 """
190 next_status = {
191 "ongoing": "done",
192 "done": "ongoing",
193 }
194 for rec in self:
195 if rec.code_validation and rec.status == "ongoing": 195 ↛ 199line 195 didn't jump to line 199 because the condition on line 195 was always true
196 action = rec._run_customer_note_validation()
197 if action and action["params"]["type"] in ["warning", "danger"]:
198 return action
199 rec.status = next_status[rec.status]
201 def write(self, vals):
202 if "description" in vals:
203 if len(self) == 1: 203 ↛ 205line 203 didn't jump to line 205 because the condition on line 203 was always true
204 handle_history_divergence(self, "description", vals)
205 return super().write(vals)
207 @api.model
208 def generate(self, upgrade_line: "SaasUpgradeLine", request: "UpgradeRequest", placeholders: dict) -> None:
209 """
210 Generates a Customer Note using data from 'upgrade_line_id' and 'placeholders'.
212 :param upgrade_line_id: The object containing the data to generate the Customer Note
213 :param request: The Upgrade Request object related to the Customer Note
214 :param placeholders: Placeholders to replace in the 'message' defined in the Upgrade Line.
215 The rendered message becomes the final Customer Note content.
216 """
217 ticket = request.ticket_id
218 domain = Domain("ticket_id", "=", ticket.id) & Domain("upgrade_line_id", "=", upgrade_line.id)
219 existing_customer_note = (
220 self.env["helpdesk.ticket.customer_note"].with_context(active_test=False).search(domain, limit=1)
221 )
222 layout_message = upgrade_line.message
223 # Sometimes the PO creates the message in the dev message field, so we don't have the 'productive'
224 # message yet. In that case, we don't generate a Customer Note.
225 if not layout_message: 225 ↛ 226line 225 didn't jump to line 226 because the condition on line 225 was never true
226 return
228 # Try to render the message with placeholders
229 try:
230 rendered_message, used_placeholders = self.render(layout_message, placeholders)
231 except Exception as e:
232 upgrade_line.run_info_id.set_error(e)
233 raise UserError(
234 self.env._("Error rendering Customer Note message for UL %s: %s") % (upgrade_line.name, str(e))
235 )
237 # Check if the rendered message is empty
238 if is_html_empty(rendered_message):
239 if existing_customer_note and existing_customer_note.active: 239 ↛ 240line 239 didn't jump to line 240 because the condition on line 239 was never true
240 existing_customer_note.active = False
241 existing_customer_note.message_post(
242 body=self.env._("This Customer Note has been deactivated because the message is empty.")
243 )
244 return
246 values = {
247 "title": upgrade_line.title,
248 "description": rendered_message,
249 "sequence": upgrade_line.customer_note_sequence,
250 "adhoc_product_id": upgrade_line.adhoc_product_id.id,
251 }
252 customer_note = existing_customer_note
253 new_used_placeholders_hash = hashlib.md5(str(used_placeholders).encode()).hexdigest()
254 create = not customer_note or (
255 customer_note.used_placeholders_hash != new_used_placeholders_hash
256 and customer_note.upgrade_type_id != request.upgrade_type_id
257 )
258 if create:
259 values.update(
260 {
261 "status": "ongoing",
262 "upgrade_line_id": upgrade_line.id,
263 "ticket_id": ticket.id,
264 "upgrade_type_id": request.upgrade_type_id.id,
265 }
266 )
267 customer_note = self.create(values)
268 else:
269 customer_note.upgrade_type_id = request.upgrade_type_id
270 if customer_note.used_placeholders_hash != new_used_placeholders_hash: 270 ↛ 279line 270 didn't jump to line 279 because the condition on line 270 was always true
271 status = customer_note.status
272 values.update(
273 {
274 "used_placeholders_hash": new_used_placeholders_hash,
275 "status": "ongoing" if upgrade_line.force_reopen_customer_note else status,
276 }
277 )
278 customer_note.write(values)
279 customer_note.action_unarchive()
281 customer_note._link_client_config_lines(upgrade_line, ticket)
283 @api.model
284 def render(self, html: str, placeholders: dict) -> tuple[str | bool, dict]:
285 """
286 Renders the 'html' with the variables in the 'placeholders' dict.
288 :param html: An HTML structure to render, defined in the UL's 'message' variable
289 :param placeholders: Dictionary with the placeholder values
290 :return: A tuple (rendered_message, used_placeholders)
291 """
292 uses_qweb = bool(self.extract_qweb_variables(html))
293 if uses_qweb and not placeholders:
294 # Return empty message if the Customer Note uses qweb placeholders but no placeholders are provided.
295 return "", {}
297 html = self._sanitize_html(html)
298 used_placeholders = self._sanitize_placeholders(placeholders)
299 return self.env["ir.qweb"]._render(etree.XML(html), used_placeholders), used_placeholders
301 @api.model
302 def _sanitize_html(self, html: str) -> str:
303 """
304 Sanitizes the HTML structure to be rendered in the Customer Note.
306 :param html: The HTML structure to sanitize
307 :return: Sanitized HTML structure
308 """
309 to_replace = [
310 (r" ", r" "),
311 (r"<br>", r"<br/>"),
312 (r"<p><br></p>", r"<br/>"),
313 (r"<hr>", r"<hr/>"),
314 (r"(<img[^>]*)(?<!/)>", r"\1 />"),
315 ]
316 for old, new in to_replace:
317 html = re.sub(old, new, html)
318 return f"<html>{html}</html>"
320 @api.model
321 def _sanitize_placeholders(self, placeholders: dict) -> dict:
322 """
323 Sanitizes placeholders before rendering.
324 Currently wraps HTML values with Markup().
326 :param placeholders: Dictionary with the placeholder values
327 :return: Sanitized placeholders dictionary
328 """
329 for key, value in placeholders.items():
330 if type(value) is str and bool(re.search(r"<.*?>", value)): 330 ↛ 331line 330 didn't jump to line 331 because the condition on line 330 was never true
331 placeholders[key] = Markup(value)
332 return placeholders
334 def _link_client_config_lines(self, upgrade_line: "SaasUpgradeLine", ticket: BaseModel) -> None:
335 """
336 Links client configuration lines to the Customer Note.
338 :param upgrade_line: Upgrade Line record
339 :param ticket: Helpdesk Ticket record
340 """
341 self.ensure_one()
342 config_lines = self.env["saas.upgrade.client.config.line"].search(
343 [
344 ("upgrade_line_id", "=", upgrade_line.id),
345 ("ticket_id", "=", ticket.id),
346 ]
347 )
348 config_lines.write({"customer_note_id": self.id})
350 def _run_customer_note_validation(self) -> dict | None:
351 """
352 Runs the validation script for the Customer Note.
353 Returns a display notification if validation fails.
355 :return: Action dictionary for display notification or None
356 """
357 self.ensure_one()
358 request = self.ticket_id.last_test_request_line_id
359 if not request: 359 ↛ 360line 359 didn't jump to line 360 because the condition on line 359 was never true
360 if self.ticket_id.all_request_line_ids.filtered(lambda x: x.aim == "test" and x.automatic):
361 message = self.env._(
362 "There is already a test request being generated for this Customer Note. "
363 "Please wait until it finishes before closing this Customer Note."
364 )
365 else:
366 self.env["helpdesk.ticket.upgrade.request"].with_user(SUPERUSER_ID)._create_request(
367 self.ticket_id.id, "test", automatic=True, bypass=True
368 )
369 message = self.env._(
370 "A test request is required to close this Customer Note. "
371 "One has been generated and will be available in a few hours."
372 )
373 return self.build_display_notification_action(
374 self.env._("Cannot close Customer Note"),
375 message,
376 "danger",
377 )
379 upgrade_line_sudo = self.sudo().upgrade_line_id
380 validation_script = upgrade_line_sudo.customer_note_validation_script
381 no_new_client = "new_client" not in validation_script
382 no_database = "old_client" not in validation_script and no_new_client
383 eval_context: dict = {}
384 with (
385 request.sudo()
386 .with_context(no_new_client=no_new_client, no_database=no_database)
387 ._get_eval_context(eval_context)
388 ):
389 # Add customer_note to the evaluation context for validation scripts
390 eval_context["customer_note"] = self
391 if validation_script: 391 ↛ 400line 391 didn't jump to line 400
392 try:
393 safe_eval.safe_eval(validation_script, eval_context, mode="exec")
394 except Exception as e:
395 body = self.env._(
396 f"Error running validation script (CN: {self.id}): <blockquote>{str(e)}</blockquote>"
397 )
398 request.message_post(body=Markup(body))
399 eval_context["result"] = self.env._("Validation failed. Please contact the upgrade team.")
400 result = eval_context.pop("result", False)
401 if result:
402 return self.build_display_notification_action(self.env._("Cannot close Customer Note"), result, "warning")
403 return None
405 @api.autovacuum
406 def _gc_archived_customer_notes(self):
407 cutoff_date = fields.Datetime.now() - timedelta(days=1)
408 archived_customer_notes = self.search(
409 [
410 ("create_date", "<", cutoff_date),
411 ("active", "=", False),
412 ("ticket_id.ticket_upgrade_status", "=", "no_upgraded"),
413 ]
414 )
415 if archived_customer_notes:
416 _logger.info("Autovacuum: Deleting %d archived customer notes", len(archived_customer_notes))
417 archived_customer_notes.unlink()
418 else:
419 _logger.info("Autovacuum: No archived customer notes to delete")