Tower: upload cetmix_tower_webhook 16.0.1.0.5 (via marketplace)
This commit is contained in:
195
addons/cetmix_tower_webhook/models/cx_tower_webhook_log.py
Normal file
195
addons/cetmix_tower_webhook/models/cx_tower_webhook_log.py
Normal file
@@ -0,0 +1,195 @@
|
||||
# Copyright (C) 2025 Cetmix OÜ
|
||||
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
|
||||
|
||||
from datetime import timedelta
|
||||
|
||||
from odoo import api, fields, models
|
||||
from odoo.http import request
|
||||
|
||||
|
||||
class CxTowerWebhookLog(models.Model):
|
||||
_name = "cx.tower.webhook.log"
|
||||
_description = "Webhook Call Log"
|
||||
_order = "create_date desc"
|
||||
_rec_name = "display_name"
|
||||
|
||||
webhook_id = fields.Many2one(
|
||||
comodel_name="cx.tower.webhook",
|
||||
ondelete="cascade",
|
||||
index=True,
|
||||
help="Webhook that received the call.",
|
||||
)
|
||||
endpoint = fields.Char(
|
||||
readonly=True,
|
||||
)
|
||||
authenticator_id = fields.Many2one(
|
||||
comodel_name="cx.tower.webhook.authenticator",
|
||||
readonly=True,
|
||||
)
|
||||
request_method = fields.Selection(
|
||||
[
|
||||
("post", "POST"),
|
||||
("get", "GET"),
|
||||
],
|
||||
default="post",
|
||||
required=True,
|
||||
help="Select the HTTP method for this webhook.",
|
||||
)
|
||||
request_headers = fields.Text(
|
||||
help="Headers of the received HTTP request (JSON-encoded).",
|
||||
)
|
||||
request_payload = fields.Text(
|
||||
help="Payload/body of the received HTTP request (JSON-encoded).",
|
||||
)
|
||||
authentication_status = fields.Selection(
|
||||
[
|
||||
("success", "Success"),
|
||||
("failed", "Failed"),
|
||||
("not_required", "Not Required"),
|
||||
],
|
||||
required=True,
|
||||
default="failed",
|
||||
help="Result of authentication for this webhook call.",
|
||||
)
|
||||
code_status = fields.Selection(
|
||||
[
|
||||
("success", "Success"),
|
||||
("failed", "Failed"),
|
||||
("skipped", "Skipped"),
|
||||
],
|
||||
string="Webhook Code Status",
|
||||
required=True,
|
||||
default="skipped",
|
||||
help="Result of webhook code execution.",
|
||||
)
|
||||
http_status = fields.Integer(
|
||||
string="HTTP Status",
|
||||
help="HTTP status code returned to the client.",
|
||||
)
|
||||
result_message = fields.Text(
|
||||
help="Message returned by the webhook code or authenticator (if any).",
|
||||
)
|
||||
error_message = fields.Text(
|
||||
help="Error message in case of authentication or code failure.",
|
||||
)
|
||||
user_id = fields.Many2one(
|
||||
comodel_name="res.users",
|
||||
string="Run as User",
|
||||
help="User as which the webhook code was executed (if set).",
|
||||
)
|
||||
ip_address = fields.Char(
|
||||
string="IP Address",
|
||||
help="IP address of the client that made the request.",
|
||||
)
|
||||
country_id = fields.Many2one(
|
||||
comodel_name="res.country",
|
||||
help="Country of the client that made the request.",
|
||||
)
|
||||
display_name = fields.Char(
|
||||
compute="_compute_display_name",
|
||||
store=True,
|
||||
readonly=True,
|
||||
)
|
||||
|
||||
@api.depends("webhook_id", "endpoint", "http_status")
|
||||
def _compute_display_name(self):
|
||||
"""Compute display name."""
|
||||
for rec in self:
|
||||
rec.display_name = (
|
||||
f"{rec.webhook_id.display_name or ''} ({rec.endpoint}) "
|
||||
f"[{rec.http_status or ''}]"
|
||||
)
|
||||
|
||||
@api.model
|
||||
def _get_country_id(self):
|
||||
"""
|
||||
Return the country ID of the client based on geoip information.
|
||||
|
||||
Returns:
|
||||
int | bool: Country ID if found, otherwise False.
|
||||
"""
|
||||
country_code = None
|
||||
if request and hasattr(request, "geoip") and request.geoip:
|
||||
country_code = request.geoip.get("country_code")
|
||||
if country_code:
|
||||
country = (
|
||||
self.env["res.country"]
|
||||
.sudo()
|
||||
.search([("code", "=", country_code)], limit=1)
|
||||
)
|
||||
if country:
|
||||
return country.id
|
||||
return False
|
||||
|
||||
@api.model
|
||||
def _get_ip_address(self):
|
||||
"""
|
||||
Return the IP address of the client making the request.
|
||||
|
||||
Returns:
|
||||
str | None: IP address string, or None if unavailable.
|
||||
"""
|
||||
if not request:
|
||||
return None
|
||||
# Check for forwarded IP (common proxy headers)
|
||||
forwarded_for = request.httprequest.headers.get("X-Forwarded-For")
|
||||
if forwarded_for:
|
||||
# Return the first IP in the chain
|
||||
return forwarded_for.split(",")[0].strip()
|
||||
return request.httprequest.remote_addr
|
||||
|
||||
@api.model
|
||||
def create_from_call(self, **kwargs):
|
||||
"""
|
||||
Create a log entry from webhook call parameters.
|
||||
|
||||
Args:
|
||||
**kwargs: Values passed to `_prepare_values`.
|
||||
|
||||
Returns:
|
||||
CxTowerWebhookLog: Newly created log record.
|
||||
"""
|
||||
values = self._prepare_values(**kwargs)
|
||||
return self.create(values)
|
||||
|
||||
@api.model
|
||||
def _prepare_values(self, webhook=None, **kwargs):
|
||||
"""
|
||||
Prepare values for creating a webhook log record.
|
||||
|
||||
Args:
|
||||
webhook (RecordSet, optional): Webhook record.
|
||||
**kwargs: Additional fields such as endpoint, request_method, etc.
|
||||
|
||||
Returns:
|
||||
dict: Prepared values for log creation.
|
||||
"""
|
||||
vals = {
|
||||
"webhook_id": webhook.id if webhook else None,
|
||||
"endpoint": webhook.endpoint if webhook else kwargs.get("endpoint"),
|
||||
"authenticator_id": webhook.authenticator_id.id if webhook else None,
|
||||
"request_method": webhook.method
|
||||
if webhook
|
||||
else kwargs.get("request_method"),
|
||||
"user_id": webhook.user_id.id if webhook else None,
|
||||
"ip_address": self._get_ip_address(),
|
||||
"country_id": self._get_country_id(),
|
||||
**kwargs,
|
||||
}
|
||||
return vals
|
||||
|
||||
@api.autovacuum
|
||||
def _gc_delete_old_logs(self):
|
||||
"""
|
||||
Remove old webhook log records beyond configured retention period.
|
||||
|
||||
This method is automatically triggered by Odoo's autovacuum.
|
||||
"""
|
||||
days = int(
|
||||
self.env["ir.config_parameter"]
|
||||
.sudo()
|
||||
.get_param("cetmix_tower_webhook.webhook_log_duration", 30)
|
||||
)
|
||||
cutoff = fields.Datetime.now() - timedelta(days=days)
|
||||
logs_to_delete = self.sudo().search([("create_date", "<", cutoff)])
|
||||
logs_to_delete.unlink()
|
||||
Reference in New Issue
Block a user