maintenance-tools/maintenance_service_http_monitoring/models/maintenance_equipment.py

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(
        default=False,
        tracking=True,
    )
    maintenance_mode_start = fields.Datetime(
        readonly=True,
    )
    maintenance_mode_end = fields.Datetime(
        readonly=True,
        help="Computed from start + configured duration",
    )

    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_service):
        """
        Create or return the open maintenance.request for a single KO service.

        Deduplication: if ko_service already has an open (non-done) request,
        return it without creating a new one.
        """
        self.ensure_one()
        existing = ko_service.http_maintenance_request
        if existing and not existing.stage_id.done:
            return existing
        status_code = ko_service.last_http_status_code
        if status_code == -1:
            error_detail = "Erreur réseau (timeout / DNS / SSL)"
        else:
            error_detail = f"HTTP {status_code}"
        name = f"[HTTP KO] {ko_service.service_url}"
        description = f"Service KO: {ko_service.service_url}\n{error_detail}"
        vals = {
            "name": name,
            "equipment_id": self.id,
            "priority": "2",
            "maintenance_type": "corrective",
            "description": description,
        }
        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)
        ko_service.http_maintenance_request = request.id
        self._notify_webhook(request, ko_service)
        return request

    def _notify_webhook(self, request, ko_service):
        """
        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}&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,
            )