[IMP] maintenance_project_task_domain, maintenance_server_data, maintenance_service_http_monitoring, maintenance_create_requests_from_project_task: pre-commit execution

This commit is contained in:
Stéphan Sainléger
2026-03-17 22:01:38 +01:00
parent 30a19649d2
commit a563a9f860
43 changed files with 786 additions and 1051 deletions

View File

@@ -1,6 +1,6 @@
import logging
from datetime import datetime
from odoo import models, fields, api
from odoo import api, fields, models
try:
import requests
@@ -11,8 +11,9 @@ _logger = logging.getLogger(__name__)
HTTP_CHECK_TIMEOUT = 10 # seconds
class ServiceInstance(models.Model):
_inherit = 'service.instance'
_inherit = "service.instance"
last_http_status_code = fields.Integer(
string="Last HTTP Status Code",
@@ -34,7 +35,7 @@ class ServiceInstance(models.Model):
if not rec.service_url or not rec.equipment_id:
continue
equipment = rec.equipment_id
if getattr(equipment, 'maintenance_mode', False):
if getattr(equipment, "maintenance_mode", False):
continue
status_ok = False
status_code = -1
@@ -45,25 +46,31 @@ class ServiceInstance(models.Model):
try:
response = requests.get(url, timeout=HTTP_CHECK_TIMEOUT)
status_code = response.status_code
status_ok = (status_code == 200)
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,
})
_logger.warning("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'):
if hasattr(equipment, "create_http_maintenance_request"):
equipment.create_http_maintenance_request([rec])
@api.model
def cron_check_http_services(self):
domain = [('active', '=', True), ('service_url', '!=', False), ('equipment_id', '!=', False)]
domain = [
("active", "=", True),
("service_url", "!=", False),
("equipment_id", "!=", False),
]
services = self.search(domain)
for service in services:
equipment = service.equipment_id
if getattr(equipment, 'maintenance_mode', False):
if getattr(equipment, "maintenance_mode", False):
continue
service.check_http_status()