1from __future__ import annotations
  2
  3import re
  4from typing import TYPE_CHECKING
  5
  6from plain import models
  7
  8if TYPE_CHECKING:
  9    from plain.http import Request
 10
 11
 12def _get_client_ip(request: Request) -> str | None:
 13    if x_forwarded_for := request.headers.get("X-Forwarded-For"):
 14        return x_forwarded_for.split(",")[0].strip()
 15    else:
 16        return request.meta.get("REMOTE_ADDR")
 17
 18
 19@models.register_model
 20class Redirect(models.Model):
 21    from_pattern = models.CharField(max_length=255)
 22    to_pattern = models.CharField(max_length=255)
 23    http_status = models.PositiveSmallIntegerField(
 24        default=301
 25    )  # Default to permanent - could be choices?
 26    created_at = models.DateTimeField(auto_now_add=True)
 27    updated_at = models.DateTimeField(auto_now=True)
 28    order = models.PositiveSmallIntegerField(default=0)
 29    enabled = models.BooleanField(default=True)
 30    is_regex = models.BooleanField(default=False)
 31
 32    # query params?
 33    # logged in or not? auth not required necessarily...
 34    # headers?
 35
 36    model_options = models.Options(
 37        ordering=["order", "-created_at"],
 38        indexes=[
 39            models.Index(fields=["order"]),
 40            models.Index(fields=["created_at"]),
 41        ],
 42        constraints=[
 43            models.UniqueConstraint(
 44                fields=["from_pattern"],
 45                name="plainredirects_redirect_unique_from_pattern",
 46            ),
 47        ],
 48    )
 49
 50    def __str__(self) -> str:
 51        return f"{self.from_pattern}"
 52
 53    def matches_request(self, request: Request) -> bool:
 54        """
 55        Decide whether a request matches this Redirect,
 56        automatically checking whether the pattern is path based or full URL based.
 57        """
 58
 59        if self.from_pattern.startswith("http"):
 60            # Full url with query params
 61            url = request.build_absolute_uri()
 62        else:
 63            # Doesn't include query params or host
 64            url = request.path
 65
 66        if self.is_regex:
 67            return bool(re.match(self.from_pattern, url))
 68        else:
 69            return url == self.from_pattern
 70
 71    def get_redirect_url(self, request: Request) -> str:
 72        if not self.is_regex:
 73            return self.to_pattern
 74
 75        # Replace any regex groups in the to_pattern
 76        if self.from_pattern.startswith("http"):
 77            url = request.build_absolute_uri()
 78        else:
 79            url = request.path
 80
 81        return re.sub(self.from_pattern, self.to_pattern, url)
 82
 83
 84@models.register_model
 85class RedirectLog(models.Model):
 86    redirect = models.ForeignKey(Redirect, on_delete=models.CASCADE)
 87
 88    # The actuals that were used to redirect
 89    from_url = models.URLField(max_length=512)
 90    to_url = models.URLField(max_length=512)
 91    http_status = models.PositiveSmallIntegerField(default=301)
 92
 93    # Request metadata
 94    ip_address = models.GenericIPAddressField()
 95    user_agent = models.CharField(required=False, max_length=512)
 96    referrer = models.CharField(required=False, max_length=512)
 97
 98    created_at = models.DateTimeField(auto_now_add=True)
 99
100    model_options = models.Options(
101        ordering=["-created_at"],
102        indexes=[
103            models.Index(fields=["created_at"]),
104        ],
105    )
106
107    @classmethod
108    def from_redirect(cls, redirect: Redirect, request: Request) -> RedirectLog:
109        from_url = request.build_absolute_uri()
110        to_url = redirect.get_redirect_url(request)
111
112        if not to_url.startswith("http"):
113            to_url = request.build_absolute_uri(to_url)
114
115        if from_url == to_url:
116            raise ValueError("Redirecting to the same URL")
117
118        return cls.query.create(
119            redirect=redirect,
120            from_url=from_url,
121            to_url=to_url,
122            http_status=redirect.http_status,
123            ip_address=_get_client_ip(request),
124            user_agent=request.headers.get("User-Agent", ""),
125            referrer=request.headers.get("Referer", ""),
126        )
127
128
129@models.register_model
130class NotFoundLog(models.Model):
131    url = models.URLField(max_length=512)
132
133    # Request metadata
134    ip_address = models.GenericIPAddressField()
135    user_agent = models.CharField(required=False, max_length=512)
136    referrer = models.CharField(required=False, max_length=512)
137
138    created_at = models.DateTimeField(auto_now_add=True)
139
140    model_options = models.Options(
141        ordering=["-created_at"],
142        indexes=[
143            models.Index(fields=["created_at"]),
144        ],
145    )
146
147    @classmethod
148    def from_request(cls, request: Request) -> NotFoundLog:
149        return cls.query.create(
150            url=request.build_absolute_uri(),
151            ip_address=_get_client_ip(request),
152            user_agent=request.headers.get("User-Agent", ""),
153            referrer=request.headers.get("Referer", ""),
154        )