# -*- coding: utf-8 -*- """ Network Log Model - Stores HTTP request logs in the database. """ import logging from datetime import timedelta from urllib.parse import urlparse from odoo import api, fields, models _logger = logging.getLogger(__name__) class NetworkLog(models.Model): _name = 'network.log' _description = 'Network Traffic Log' _order = 'timestamp desc' timestamp = fields.Datetime( string='Timestamp', default=fields.Datetime.now, required=True, index=True, ) method = fields.Selection([ ('GET', 'GET'), ('POST', 'POST'), ('PUT', 'PUT'), ('DELETE', 'DELETE'), ('PATCH', 'PATCH'), ('HEAD', 'HEAD'), ('OPTIONS', 'OPTIONS'), ], string='Method', required=True, index=True) url = fields.Char(string='URL', required=True, index=True) domain = fields.Char(string='Domain', compute='_compute_domain', store=True, index=True) is_odoo_call = fields.Boolean( string='Odoo External Call', default=False, index=True, help='True if this request was made to an Odoo server', ) status_code = fields.Integer(string='Status Code') status_type = fields.Selection([ ('success', 'Success'), ('redirect', 'Redirect'), ('client_error', 'Client Error'), ('server_error', 'Server Error'), ('blocked', 'Blocked'), ('error', 'Error'), ('pending', 'Pending'), ], string='Status Type', compute='_compute_status_type', store=True) response_time = fields.Float(string='Response Time (s)', digits=(10, 4)) error_message = fields.Text(string='Error Message') request_headers = fields.Text(string='Request Headers') request_body = fields.Text(string='Request Body') @api.depends('url') def _compute_domain(self): for record in self: try: parsed = urlparse(record.url or '') record.domain = parsed.netloc or 'unknown' except Exception: record.domain = 'unknown' @api.depends('status_code', 'error_message') def _compute_status_type(self): for record in self: if record.error_message: if 'blocked' in (record.error_message or '').lower(): record.status_type = 'blocked' else: record.status_type = 'error' elif not record.status_code: record.status_type = 'pending' elif 200 <= record.status_code < 300: record.status_type = 'success' elif 300 <= record.status_code < 400: record.status_type = 'redirect' elif 400 <= record.status_code < 500: record.status_type = 'client_error' elif record.status_code >= 500: record.status_type = 'server_error' else: record.status_type = 'pending' @api.model def log_request(self, method, url, is_odoo_call=False, status_code=None, response_time=None, error_message=None): """Create a new network log entry.""" try: vals = { 'method': method.upper(), 'url': url[:2048] if url else '', 'is_odoo_call': is_odoo_call, 'status_code': status_code, 'response_time': response_time, 'error_message': error_message[:4096] if error_message else None, } return self.sudo().create(vals) except Exception as e: _logger.warning("Failed to log network request: %s", e) return self.browse() @api.model def clear_old_logs(self, days=7): """Delete logs older than specified days.""" cutoff = fields.Datetime.now() - timedelta(days=days) old_logs = self.search([('timestamp', '<', cutoff)]) count = len(old_logs) old_logs.unlink() _logger.info("Cleared %d network logs older than %d days", count, days) return count @api.model def clear_all_logs(self): """Delete all network logs.""" count = self.search_count([]) self.search([]).unlink() _logger.info("Cleared all %d network logs", count) return count