devplacepy/devplacepy/models.py

# retoor <retoor@molodetz.nl>

import re

from datetime import datetime
from typing import Literal, Optional
from urllib.parse import urlsplit
from pydantic import BaseModel, Field, field_validator, model_validator
from devplacepy.constants import TOPICS, REACTION_EMOJI
from devplacepy.config import DEFAULT_CORRECTION_PROMPT, DEFAULT_MODIFIER_PROMPT


def normalize_european_date(value):
    if not value:
        return ""
    text = str(value).strip()
    if not text:
        return ""
    for fmt in ("%d/%m/%Y", "%Y-%m-%d"):
        try:
            return datetime.strptime(text, fmt).strftime("%Y-%m-%d")
        except ValueError:
            continue
    raise ValueError("Date must be in DD/MM/YYYY format")


def normalize_poll_options(value):
    if value is None:
        return []
    if isinstance(value, str):
        value = [value]
    if not isinstance(value, list):
        return value
    if len(value) == 1 and isinstance(value[0], str):
        single = value[0]
        separator = "\n" if "\n" in single else ("," if "," in single else "")
        if separator:
            return [part.strip() for part in single.split(separator) if part.strip()]
    return value


class SignupForm(BaseModel):
    username: str = Field(min_length=3, max_length=32)
    email: str = Field(min_length=1, max_length=255)
    password: str = Field(min_length=6, max_length=128)
    confirm_password: str = Field(min_length=1, max_length=128)

    @field_validator("username")
    @classmethod
    def username_chars(cls, value):
        if not value.isascii() or not all(
            c.isalnum() or c in ("-", "_") for c in value
        ):
            raise ValueError(
                "Username can only contain letters, numbers, hyphens, and underscores"
            )
        return value

    @field_validator("email")
    @classmethod
    def email_has_at(cls, value):
        if "@" not in value:
            raise ValueError("Valid email is required")
        return value

    @model_validator(mode="after")
    def passwords_match(self):
        if self.password != self.confirm_password:
            raise ValueError("Passwords do not match")
        return self


class LoginForm(BaseModel):
    email: str = Field(min_length=1, max_length=255)
    password: str = Field(min_length=1, max_length=128)
    remember_me: str = ""
    next: str = ""


class ForgotPasswordForm(BaseModel):
    email: str = Field(min_length=1, max_length=255)

    @field_validator("email")
    @classmethod
    def email_has_at(cls, value):
        if "@" not in value:
            raise ValueError("Valid email is required")
        return value


class ResetPasswordForm(BaseModel):
    password: str = Field(min_length=6, max_length=128)
    confirm_password: str = Field(min_length=1, max_length=128)

    @model_validator(mode="after")
    def passwords_match(self):
        if self.password != self.confirm_password:
            raise ValueError("Passwords do not match")
        return self


class PostForm(BaseModel):
    content: str = Field(min_length=10, max_length=125000)
    title: str = Field(default="", max_length=500)
    topic: str = "random"
    project_uid: str = Field(default="", max_length=36)
    attachment_uids: list[str] = []
    poll_question: str = Field(default="", max_length=200)
    poll_options: list[str] = []

    @field_validator("poll_options")
    @classmethod
    def poll_options_max_length(cls, value):
        if value is None:
            return value
        for opt in value:
            if isinstance(opt, str) and len(opt) > 200:
                raise ValueError("Each poll option must be 200 characters or fewer")
        return value

    @field_validator("topic")
    @classmethod
    def valid_topic(cls, value):
        return value if value in TOPICS else "random"

    @field_validator("poll_options", mode="before")
    @classmethod
    def split_poll_options(cls, value):
        return normalize_poll_options(value)


class PostEditForm(BaseModel):
    content: str = Field(min_length=10, max_length=125000)
    title: str = Field(default="", max_length=500)
    topic: str = "random"
    poll_question: str = Field(default="", max_length=200)
    poll_options: list[str] = []

    @field_validator("poll_options")
    @classmethod
    def poll_options_max_length(cls, value):
        if value is None:
            return value
        for opt in value:
            if isinstance(opt, str) and len(opt) > 200:
                raise ValueError("Each poll option must be 200 characters or fewer")
        return value

    @field_validator("topic")
    @classmethod
    def valid_topic(cls, value):
        return value if value in TOPICS else "random"

    @field_validator("poll_options", mode="before")
    @classmethod
    def split_poll_options(cls, value):
        return normalize_poll_options(value)


class CommentForm(BaseModel):
    content: str = Field(min_length=3, max_length=125000)
    target_uid: str = Field(default="", max_length=36)
    post_uid: str = Field(default="", max_length=36)
    target_type: Literal["post", "project", "news", "issue", "gist"] = "post"
    parent_uid: str = Field(default="", max_length=36)
    attachment_uids: list[str] = []

    @model_validator(mode="after")
    def require_target(self):
        if not (self.target_uid or self.post_uid):
            raise ValueError("A target is required")
        return self


class CommentEditForm(BaseModel):
    content: str = Field(min_length=3, max_length=125000)


class ProjectForm(BaseModel):
    title: str = Field(min_length=1, max_length=200)
    description: str = Field(min_length=1, max_length=5000)
    release_date: str = ""
    demo_date: str = ""
    project_type: Literal["game", "game_asset", "software", "mobile_app", "website"] = (
        "software"
    )
    platforms: str = Field(default="", max_length=500)
    status: str = Field(default="In Development", max_length=100)
    is_private: bool = False
    attachment_uids: list[str] = []

    @field_validator("release_date", "demo_date", mode="before")
    @classmethod
    def normalize_dates(cls, value):
        return normalize_european_date(value)


class ProjectEditForm(BaseModel):
    title: str = Field(min_length=1, max_length=200)
    description: str = Field(min_length=1, max_length=5000)
    release_date: str = ""
    demo_date: str = ""
    project_type: Literal["game", "game_asset", "software", "mobile_app", "website"] = (
        "software"
    )
    platforms: str = Field(default="", max_length=500)
    status: str = Field(default="In Development", max_length=100)

    @field_validator("release_date", "demo_date", mode="before")
    @classmethod
    def normalize_dates(cls, value):
        return normalize_european_date(value)


class BackupRunForm(BaseModel):
    target: Literal["database", "uploads", "keys", "full"] = "full"


class BackupScheduleForm(BaseModel):
    name: str = Field(min_length=1, max_length=120)
    target: Literal["database", "uploads", "keys", "full"] = "full"
    kind: Literal["interval", "cron"] = "interval"
    every_seconds: int = Field(default=86400, ge=60)
    cron: str = Field(default="", max_length=120)
    keep_last: int = Field(default=7, ge=0, le=1000)

    @model_validator(mode="after")
    def _validate_schedule(self) -> "BackupScheduleForm":
        if self.kind == "cron" and not self.cron.strip():
            raise ValueError("A cron schedule requires a cron expression")
        return self


class ProjectFlagForm(BaseModel):
    value: bool = False


class CustomizationToggleForm(BaseModel):
    value: bool = False


class AwardGiveForm(BaseModel):
    description: str = Field(min_length=1, max_length=125)


class NotificationPrefForm(BaseModel):
    notification_type: str = Field(min_length=1, max_length=40)
    channel: Literal["in_app", "push", "telegram"]
    value: bool = False


class NotificationDefaultForm(BaseModel):
    notification_type: str = Field(min_length=1, max_length=40)
    channel: Literal["in_app", "push", "telegram"]
    value: bool = False


class AiCorrectionForm(BaseModel):
    enabled: bool = False
    sync: bool = False
    prompt: str = Field(default=DEFAULT_CORRECTION_PROMPT, max_length=20000)


class AiModifierForm(BaseModel):
    enabled: bool = False
    sync: bool = False
    prompt: str = Field(default=DEFAULT_MODIFIER_PROMPT, max_length=20000)


class InteractionsForm(BaseModel):
    enabled: bool = True
    reset: bool = False


class TelegramPairForm(BaseModel):
    action: str = Field(default="request", max_length=16)


class ForkForm(BaseModel):
    title: str = Field(min_length=1, max_length=200)


class PlanningForm(BaseModel):
    numbers: str = Field(default="", max_length=4096)


class UploadUrlForm(BaseModel):
    url: str = Field(min_length=1, max_length=2048)
    filename: Optional[str] = Field(default=None, max_length=255)


class ProjectFileWriteForm(BaseModel):
    path: str = Field(min_length=1, max_length=1024)
    content: str = Field(default="", max_length=400000)


class ProjectFileMkdirForm(BaseModel):
    path: str = Field(min_length=1, max_length=1024)


class ProjectFileMoveForm(BaseModel):
    from_path: str = Field(min_length=1, max_length=1024)
    to_path: str = Field(min_length=1, max_length=1024)


class ProjectFileDeleteForm(BaseModel):
    path: str = Field(min_length=1, max_length=1024)


class ProjectFileReplaceLinesForm(BaseModel):
    path: str = Field(min_length=1, max_length=1024)
    start: int = Field(ge=1)
    end: int = Field(ge=0)
    content: str = Field(default="", max_length=400000)


class ProjectFileInsertLinesForm(BaseModel):
    path: str = Field(min_length=1, max_length=1024)
    at: int = Field(ge=1)
    content: str = Field(default="", max_length=400000)


class ProjectFileDeleteLinesForm(BaseModel):
    path: str = Field(min_length=1, max_length=1024)
    start: int = Field(ge=1)
    end: int = Field(ge=1)


class ProjectFileAppendForm(BaseModel):
    path: str = Field(min_length=1, max_length=1024)
    content: str = Field(default="", max_length=400000)


class ContainerInstanceForm(BaseModel):
    name: str = Field(min_length=1, max_length=64)
    boot_command: str = Field(default="", max_length=500)
    boot_language: str = Field(default="none", max_length=10)
    boot_script: str = Field(default="", max_length=100000)
    run_as_uid: str = Field(default="", max_length=36)
    start_on_boot: bool = False
    env: str = Field(default="", max_length=10000)
    cpu_limit: str = Field(default="", max_length=16)
    mem_limit: str = Field(default="", max_length=16)
    ports: str = Field(default="", max_length=500)
    volumes: str = Field(default="", max_length=2000)
    restart_policy: str = Field(default="never", max_length=20)
    autostart: bool = True
    ingress_slug: str = Field(default="", max_length=64)
    ingress_port: Optional[int] = Field(default=None, ge=1, le=65535)

    @field_validator("ingress_port", mode="before")
    @classmethod
    def _blank_ingress_port(cls, value):
        if value is None or (isinstance(value, str) and not value.strip()):
            return None
        return value


class ContainerAdminCreateForm(ContainerInstanceForm):
    project_slug: str = Field(min_length=1, max_length=128)


class ContainerEditForm(BaseModel):
    run_as_uid: str = Field(default="", max_length=36)
    boot_language: str = Field(default="none", max_length=10)
    boot_script: str = Field(default="", max_length=100000)
    boot_command: str = Field(default="", max_length=500)
    restart_policy: str = Field(default="never", max_length=20)
    start_on_boot: bool = False
    cpu_limit: str = Field(default="", max_length=16)
    mem_limit: str = Field(default="", max_length=16)


class ContainerExecForm(BaseModel):
    command: str = Field(min_length=1, max_length=2000)


class ContainerScheduleForm(BaseModel):
    action: str = Field(max_length=10)
    kind: str = Field(max_length=10)
    cron: str = Field(default="", max_length=120)
    run_at: str = Field(default="", max_length=40)
    delay_seconds: Optional[int] = Field(default=None, ge=1)
    every_seconds: Optional[int] = Field(default=None, ge=1)
    max_runs: Optional[int] = Field(default=None, ge=1)


class MessageForm(BaseModel):
    content: str = Field(min_length=1, max_length=2000)
    receiver_uid: str = Field(min_length=1, max_length=36)
    attachment_uids: list[str] = []


class ProfileForm(BaseModel):
    bio: str = Field(default="", max_length=500)
    location: str = Field(default="", max_length=200)
    git_link: str = Field(default="", max_length=500)
    website: str = Field(default="", max_length=500)


class GistForm(BaseModel):
    title: str = Field(min_length=1, max_length=200)
    description: str = Field(default="", max_length=5000)
    source_code: str = Field(min_length=1, max_length=400000)
    language: str = Field(default="plaintext", max_length=50)
    attachment_uids: list[str] = []


class GistEditForm(BaseModel):
    title: str = Field(min_length=1, max_length=200)
    description: str = Field(default="", max_length=5000)
    source_code: str = Field(min_length=1, max_length=400000)
    language: str = Field(default="plaintext", max_length=50)


class IssueForm(BaseModel):
    title: str = Field(min_length=1, max_length=200)
    description: str = Field(min_length=1, max_length=5000)
    attachment_uids: list[str] = []


class IssueCommentForm(BaseModel):
    body: str = Field(min_length=1, max_length=5000)
    attachment_uids: list[str] = []


class IssueAttachmentForm(BaseModel):
    attachment_uids: list[str] = []


class IssueStatusForm(BaseModel):
    status: Literal["open", "closed"]


class VoteForm(BaseModel):
    value: int

    @field_validator("value")
    @classmethod
    def valid_value(cls, value):
        if value not in (1, -1):
            raise ValueError("value must be 1 or -1")
        return value


class ReactionForm(BaseModel):
    emoji: str = Field(min_length=1, max_length=16)

    @field_validator("emoji")
    @classmethod
    def valid_emoji(cls, value):
        if value not in REACTION_EMOJI:
            raise ValueError("Invalid reaction")
        return value


class PollVoteForm(BaseModel):
    option_uid: str = Field(min_length=1, max_length=36)


class SeoRunForm(BaseModel):
    url: str = Field(min_length=3, max_length=2000)
    mode: Literal["url", "sitemap"] = "url"
    max_pages: int = Field(default=10, ge=1, le=50)

    @field_validator("url")
    @classmethod
    def url_scheme(cls, value):
        text = value.strip()
        if not text:
            raise ValueError("A URL is required")
        return text


ISSLOP_URL_PATTERN = re.compile(r"^(https?://|git://|ssh://|git@)[\w./:@~^-]+$", re.IGNORECASE)
ISSLOP_SINGLE_SLASH_PATTERN = re.compile(r"^(https?|git|ssh):/(?!/)", re.IGNORECASE)
ISSLOP_SCHEME_PATTERN = re.compile(r"^[a-z][a-z0-9+.-]*://", re.IGNORECASE)


class IsslopRunForm(BaseModel):
    url: str = Field(min_length=4, max_length=2048)

    @field_validator("url")
    @classmethod
    def url_scheme(cls, value):
        text = value.strip()
        text = ISSLOP_SINGLE_SLASH_PATTERN.sub(lambda match: f"{match.group(1)}://", text)
        if not ISSLOP_SCHEME_PATTERN.match(text) and not text.startswith("git@"):
            text = f"https://{text}"
        if not ISSLOP_URL_PATTERN.match(text):
            raise ValueError("URL must be an http(s), git or ssh source location")
        return text


DEEPSEARCH_MIN_DEPTH = 1
DEEPSEARCH_MAX_DEPTH = 4
DEEPSEARCH_DEFAULT_DEPTH = 2
DEEPSEARCH_MIN_PAGES = 1
DEEPSEARCH_MAX_PAGES = 30
DEEPSEARCH_DEFAULT_PAGES = 12
DEEPSEARCH_MIN_QUERY = 3
DEEPSEARCH_MAX_QUERY = 500
DEEPSEARCH_MAX_MESSAGE = 2000


class DeepsearchRunForm(BaseModel):
    query: str = Field(min_length=DEEPSEARCH_MIN_QUERY, max_length=DEEPSEARCH_MAX_QUERY)
    depth: int = Field(
        default=DEEPSEARCH_DEFAULT_DEPTH,
        ge=DEEPSEARCH_MIN_DEPTH,
        le=DEEPSEARCH_MAX_DEPTH,
    )
    max_pages: int = Field(
        default=DEEPSEARCH_DEFAULT_PAGES,
        ge=DEEPSEARCH_MIN_PAGES,
        le=DEEPSEARCH_MAX_PAGES,
    )

    @field_validator("query")
    @classmethod
    def query_present(cls, value):
        text = value.strip()
        if not text:
            raise ValueError("A research question is required")
        return text


class DeepsearchChatForm(BaseModel):
    message: str = Field(min_length=1, max_length=DEEPSEARCH_MAX_MESSAGE)

    @field_validator("message")
    @classmethod
    def message_present(cls, value):
        text = value.strip()
        if not text:
            raise ValueError("A message is required")
        return text


class AdminRoleForm(BaseModel):
    role: Literal["member", "admin"]


class AdminPasswordForm(BaseModel):
    password: str = Field(min_length=6, max_length=128)


class AdminSettingsForm(BaseModel):
    site_name: str = Field(default="", max_length=200)
    site_description: str = Field(default="", max_length=500)
    site_tagline: str = Field(default="", max_length=500)
    site_url: str = Field(default="", max_length=300)
    max_upload_size_mb: str = Field(default="", max_length=10)
    allowed_file_types: str = Field(default="", max_length=1000)
    max_attachments_per_resource: str = Field(default="", max_length=10)
    rate_limit_per_minute: str = Field(default="", max_length=10)
    rate_limit_window_seconds: str = Field(default="", max_length=10)
    session_max_age_days: str = Field(default="", max_length=10)
    session_remember_days: str = Field(default="", max_length=10)
    registration_open: str = Field(default="", max_length=1)
    maintenance_mode: str = Field(default="", max_length=1)
    maintenance_message: str = Field(default="", max_length=300)
    docs_search_mode: str = Field(default="", max_length=20)
    outbound_proxy_url: str = Field(default="", max_length=500)
    extra_head: str = Field(default="", max_length=50000)

    @field_validator("outbound_proxy_url")
    @classmethod
    def validate_outbound_proxy_url(cls, value):
        text = value.strip()
        if not text:
            return text
        parsed = urlsplit(text)
        if parsed.scheme not in ("http", "https", "socks5", "socks5h") or not parsed.hostname:
            raise ValueError("Proxy URL must be http(s):// or socks5(h):// with a host, e.g. http://user:pass@host:port")
        return text


class GamePlantForm(BaseModel):
    slot: int = Field(ge=0, le=64)
    crop: str = Field(min_length=1, max_length=40)


class GameSlotForm(BaseModel):
    slot: int = Field(ge=0, le=64)


class GamePerkForm(BaseModel):
    perk: str = Field(min_length=1, max_length=40)


class GameQuestForm(BaseModel):
    quest: str = Field(min_length=1, max_length=40)
    scope: str = Field(default="daily", min_length=1, max_length=10)


class GameLegacyForm(BaseModel):
    key: str = Field(min_length=1, max_length=40)


class GameInfraForm(BaseModel):
    key: str = Field(min_length=1, max_length=40)


class GameCosmeticForm(BaseModel):
    key: str = Field(min_length=1, max_length=40)


class GameMasteryForm(BaseModel):
    key: str = Field(min_length=1, max_length=40)


class GameEraStartForm(BaseModel):
    name: str = Field(min_length=1, max_length=60)
    duration_days: int = Field(default=28, ge=1, le=180)