[NEW] maintenance_service_http_monitoring : check HTTP 200 for services

maintenance_service_http_monitoring/models/__init__.py

@@ -0,0 +1,3 @@
+from . import service_instance
+from . import maintenance_equipment
+from . import res_config_settings

maintenance_service_http_monitoring/models/maintenance_equipment.py

@@ -0,0 +1,97 @@
+from datetime import timedelta
+from odoo import models, fields, api
+
+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
+        return request
+
+    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)

maintenance_service_http_monitoring/models/service_instance.py

@@ -0,0 +1,69 @@
+import logging
+from datetime import datetime
+from odoo import models, fields, api
+
+try:
+    import requests
+except ImportError:
+    requests = None
+
+_logger = logging.getLogger(__name__)
+
+HTTP_CHECK_TIMEOUT = 10  # seconds
+
+class ServiceInstance(models.Model):
+    _inherit = 'service.instance'
+
+    last_http_status_code = fields.Integer(
+        string="Last HTTP Status Code",
+        readonly=True,
+        default=0,
+    )
+    last_http_check_date = fields.Datetime(
+        string="Last HTTP Check Date",
+        readonly=True,
+    )
+    http_status_ok = fields.Boolean(
+        string="HTTP Status OK",
+        readonly=True,
+        default=True,
+    )
+
+    def check_http_status(self):
+        for rec in self:
+            if not rec.service_url or not rec.equipment_id:
+                continue
+            equipment = rec.equipment_id
+            if getattr(equipment, 'maintenance_mode', False):
+                continue
+            status_ok = False
+            status_code = -1
+            now = fields.Datetime.now()
+            url = rec.service_url
+            if not url.lower().startswith("https://"):
+                url = "https://" + url.lstrip("http://")
+            try:
+                response = requests.get(url, timeout=HTTP_CHECK_TIMEOUT)
+                status_code = response.status_code
+                status_ok = (status_code == 200)
+            except requests.exceptions.RequestException as e:
+                _logger.warning(f"HTTP check failed for %s: %s", rec.service_url, e)
+            rec.write({
+                'last_http_status_code': status_code,
+                'last_http_check_date': now,
+                'http_status_ok': status_ok,
+            })
+            if not status_ok:
+                # Delegate maintenance.request creation to equipment
+                if hasattr(equipment, 'create_http_maintenance_request'):
+                    equipment.create_http_maintenance_request([rec])
+
+    @api.model
+    def cron_check_http_services(self):
+        domain = [('service_url', '!=', False), ('equipment_id', '!=', False)]
+        services = self.search(domain)
+        for service in services:
+            equipment = service.equipment_id
+            if getattr(equipment, 'maintenance_mode', False):
+                continue
+            service.check_http_status()