164 lines
5.4 KiB
Python
164 lines
5.4 KiB
Python
import logging
|
||
from datetime import timedelta
|
||
|
||
from odoo import api, fields, models
|
||
|
||
|
||
try:
|
||
import requests as http_requests
|
||
except ImportError:
|
||
http_requests = None
|
||
|
||
_logger = logging.getLogger(__name__)
|
||
|
||
WEBHOOK_TIMEOUT = 10 # seconds
|
||
|
||
class MaintenanceEquipment(models.Model):
|
||
_inherit = "maintenance.equipment"
|
||
|
||
maintenance_mode = fields.Boolean(
|
||
string="Maintenance Mode",
|
||
default=False,
|
||
tracking=True,
|
||
)
|
||
maintenance_mode_start = fields.Datetime(
|
||
string="Maintenance Mode Start",
|
||
readonly=True,
|
||
)
|
||
maintenance_mode_end = fields.Datetime(
|
||
string="Maintenance Mode End",
|
||
readonly=True,
|
||
help="Computed from start + configured duration",
|
||
)
|
||
http_maintenance_request = fields.Many2one(
|
||
"maintenance.request",
|
||
string="HTTP Maintenance Request",
|
||
readonly=True,
|
||
)
|
||
|
||
def action_activate_maintenance_mode(self):
|
||
for rec in self:
|
||
duration = int(
|
||
self.env["ir.config_parameter"]
|
||
.sudo()
|
||
.get_param(
|
||
"maintenance_service_http_monitoring.maintenance_mode_duration", 4
|
||
)
|
||
)
|
||
now = fields.Datetime.now()
|
||
rec.write(
|
||
{
|
||
"maintenance_mode": True,
|
||
"maintenance_mode_start": now,
|
||
"maintenance_mode_end": now + timedelta(hours=duration),
|
||
}
|
||
)
|
||
|
||
def action_deactivate_maintenance_mode(self):
|
||
for rec in self:
|
||
rec.write(
|
||
{
|
||
"maintenance_mode": False,
|
||
"maintenance_mode_start": False,
|
||
"maintenance_mode_end": False,
|
||
}
|
||
)
|
||
|
||
@api.model
|
||
def cron_deactivate_expired_maintenance_mode(self):
|
||
now = fields.Datetime.now()
|
||
expired = self.search(
|
||
[
|
||
("maintenance_mode", "=", True),
|
||
("maintenance_mode_end", "<=", now),
|
||
]
|
||
)
|
||
expired.action_deactivate_maintenance_mode()
|
||
|
||
def create_http_maintenance_request(self, ko_services):
|
||
self.ensure_one()
|
||
today = fields.Date.context_today(self)
|
||
name = f"[HTTP KO] {self.name}"
|
||
domain = [
|
||
("name", "=", name),
|
||
("equipment_id", "=", self.id),
|
||
("maintenance_type", "=", "corrective"),
|
||
("create_date", ">=", f"{today} 00:00:00"),
|
||
("create_date", "<=", f"{today} 23:59:59"),
|
||
]
|
||
existing = self.env["maintenance.request"].search(domain, limit=1)
|
||
# Check if a task with same name already exist for the day, if it’s the case : skip
|
||
if existing:
|
||
self.http_maintenance_request = existing.id
|
||
return existing
|
||
request = self.http_maintenance_request
|
||
if request and not request.stage_id.done:
|
||
return request
|
||
vals = {
|
||
"name": name,
|
||
"equipment_id": self.id,
|
||
"priority": "2",
|
||
"maintenance_type": "corrective",
|
||
"description": self._build_ko_services_description(ko_services),
|
||
}
|
||
if self.employee_id:
|
||
vals["employee_id"] = self.employee_id.id
|
||
if self.technician_user_id:
|
||
vals["user_id"] = self.technician_user_id.id
|
||
if self.maintenance_team_id:
|
||
vals["maintenance_team_id"] = self.maintenance_team_id.id
|
||
else:
|
||
team = self.env["maintenance.team"].search([], limit=1)
|
||
if team:
|
||
vals["maintenance_team_id"] = team.id
|
||
request = self.env["maintenance.request"].create(vals)
|
||
self.http_maintenance_request = request.id
|
||
self._notify_webhook(request, ko_services)
|
||
return request
|
||
|
||
def _notify_webhook(self, request, ko_services):
|
||
"""Send a webhook notification when a new maintenance request is created."""
|
||
ICP = self.env['ir.config_parameter'].sudo()
|
||
webhook_url = ICP.get_param(
|
||
'maintenance_service_http_monitoring.webhook_url', ''
|
||
)
|
||
if not webhook_url:
|
||
return
|
||
webhook_user = ICP.get_param(
|
||
'maintenance_service_http_monitoring.webhook_user', ''
|
||
)
|
||
webhook_password = ICP.get_param(
|
||
'maintenance_service_http_monitoring.webhook_password', ''
|
||
)
|
||
base_url = ICP.get_param('web.base.url', '')
|
||
link = (
|
||
f"{base_url}/web#id={request.id}"
|
||
f"&model=maintenance.request&view_type=form"
|
||
)
|
||
payload = {
|
||
'id': request.id,
|
||
'name': request.name,
|
||
'description': request.description or '',
|
||
'equipment': self.name,
|
||
'link': link,
|
||
}
|
||
auth = None
|
||
if webhook_user and webhook_password:
|
||
auth = (webhook_user, webhook_password)
|
||
try:
|
||
http_requests.post(
|
||
webhook_url,
|
||
json=payload,
|
||
auth=auth,
|
||
timeout=WEBHOOK_TIMEOUT,
|
||
)
|
||
except Exception as e:
|
||
_logger.warning(
|
||
"Webhook notification failed for maintenance request %s: %s",
|
||
request.id, e,
|
||
)
|
||
|
||
def _build_ko_services_description(self, ko_services):
|
||
lines = [f"Service KO: {s.service_url or s.name}" for s in ko_services]
|
||
return "\n".join(lines)
|