157 lines
5.1 KiB
Python
157 lines
5.1 KiB
Python
"""Blocklist source and import log Pydantic models.
|
|
|
|
Data shapes for blocklist source management, import operations, scheduling,
|
|
and import log retrieval.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from enum import StrEnum
|
|
|
|
from pydantic import AnyHttpUrl, Field
|
|
|
|
from app.models.response import BanGuiBaseModel
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Blocklist source
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class BlocklistSource(BanGuiBaseModel):
|
|
"""Domain model for a blocklist source definition."""
|
|
|
|
id: int
|
|
name: str
|
|
url: str
|
|
enabled: bool
|
|
created_at: str
|
|
updated_at: str
|
|
|
|
class BlocklistSourceCreate(BanGuiBaseModel):
|
|
"""Payload for ``POST /api/blocklists``.
|
|
|
|
URL must use http/https scheme. The hostname must resolve to a public IP
|
|
(not private, loopback, link-local, or reserved). Validation happens
|
|
asynchronously in the service layer.
|
|
"""
|
|
|
|
name: str = Field(..., min_length=1, max_length=100, description="Human-readable source name.")
|
|
url: AnyHttpUrl = Field(..., description="URL of the blocklist file (http/https only).")
|
|
enabled: bool = Field(default=True)
|
|
|
|
class BlocklistSourceUpdate(BanGuiBaseModel):
|
|
"""Payload for ``PUT /api/blocklists/{id}``. All fields are optional.
|
|
|
|
If URL is provided, it must use http/https scheme.
|
|
"""
|
|
|
|
name: str | None = Field(default=None, min_length=1, max_length=100)
|
|
url: AnyHttpUrl | None = Field(default=None)
|
|
enabled: bool | None = Field(default=None)
|
|
|
|
class BlocklistListResponse(BanGuiBaseModel):
|
|
"""Response for ``GET /api/blocklists``."""
|
|
|
|
sources: list[BlocklistSource] = Field(default_factory=list)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Import log
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class ImportLogEntry(BanGuiBaseModel):
|
|
"""A single blocklist import run record."""
|
|
|
|
id: int
|
|
source_id: int | None
|
|
source_url: str
|
|
timestamp: str
|
|
ips_imported: int
|
|
ips_skipped: int
|
|
errors: str | None
|
|
|
|
class ImportLogListResponse(BanGuiBaseModel):
|
|
"""Response for ``GET /api/blocklists/log``."""
|
|
|
|
items: list[ImportLogEntry] = Field(default_factory=list)
|
|
total: int = Field(..., ge=0)
|
|
page: int = Field(default=1, ge=1)
|
|
page_size: int = Field(default=50, ge=1)
|
|
total_pages: int = Field(default=1, ge=1)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Schedule
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class ScheduleFrequency(StrEnum):
|
|
"""Available import schedule frequency presets."""
|
|
|
|
hourly = "hourly"
|
|
daily = "daily"
|
|
weekly = "weekly"
|
|
|
|
class ScheduleConfig(BanGuiBaseModel):
|
|
"""Import schedule configuration.
|
|
|
|
The interpretation of fields depends on *frequency*:
|
|
|
|
- ``hourly``: ``interval_hours`` controls how often (every N hours).
|
|
- ``daily``: ``hour`` and ``minute`` specify the daily run time (UTC).
|
|
- ``weekly``: additionally uses ``day_of_week`` (0=Monday … 6=Sunday).
|
|
"""
|
|
|
|
# No strict=True here: FastAPI and json.loads() both supply enum values as
|
|
# plain strings; strict mode would reject string→enum coercion.
|
|
|
|
frequency: ScheduleFrequency = ScheduleFrequency.daily
|
|
interval_hours: int = Field(default=24, ge=1, le=168, description="Used when frequency=hourly")
|
|
hour: int = Field(default=3, ge=0, le=23, description="UTC hour for daily/weekly runs")
|
|
minute: int = Field(default=0, ge=0, le=59, description="Minute for daily/weekly runs")
|
|
day_of_week: int = Field(
|
|
default=0,
|
|
ge=0,
|
|
le=6,
|
|
description="Day of week for weekly runs (0=Monday … 6=Sunday)",
|
|
)
|
|
|
|
class ScheduleInfo(BanGuiBaseModel):
|
|
"""Current schedule configuration together with runtime metadata."""
|
|
|
|
config: ScheduleConfig
|
|
next_run_at: str | None
|
|
last_run_at: str | None
|
|
last_run_errors: bool | None = None
|
|
"""``True`` if the most recent import had errors, ``False`` if clean, ``None`` if never run."""
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Import results
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class ImportSourceResult(BanGuiBaseModel):
|
|
"""Result of importing a single blocklist source."""
|
|
|
|
source_id: int | None
|
|
source_url: str
|
|
ips_imported: int
|
|
ips_skipped: int
|
|
error: str | None
|
|
|
|
class ImportRunResult(BanGuiBaseModel):
|
|
"""Aggregated result from a full import run across all enabled sources."""
|
|
|
|
results: list[ImportSourceResult] = Field(default_factory=list)
|
|
total_imported: int
|
|
total_skipped: int
|
|
errors_count: int
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Preview
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class PreviewResponse(BanGuiBaseModel):
|
|
"""Response for ``GET /api/blocklists/{id}/preview``."""
|
|
|
|
entries: list[str] = Field(default_factory=list, description="Sample of valid IP entries")
|
|
total_lines: int
|
|
valid_count: int
|
|
skipped_count: int
|