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(
        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)