mirror/app/models.py

from __future__ import annotations

import json
from datetime import datetime, timezone

from app.extensions import db


def utcnow() -> datetime:
    return datetime.now(timezone.utc)


class Mirror(db.Model):
    __tablename__ = "mirrors"

    id = db.Column(db.Integer, primary_key=True)
    category = db.Column(db.String(32), nullable=False, index=True)
    name = db.Column(db.String(128), nullable=False)
    url = db.Column(db.String(512), nullable=True)
    ips = db.Column(db.Text, nullable=True)
    priority = db.Column(db.Integer, default=100, nullable=False)
    enabled = db.Column(db.Boolean, default=True, nullable=False)
    notes = db.Column(db.Text, nullable=True)
    meta_json = db.Column(db.Text, nullable=True)
    created_at = db.Column(db.DateTime, default=utcnow, nullable=False)
    updated_at = db.Column(db.DateTime, default=utcnow, onupdate=utcnow, nullable=False)

    test_results = db.relationship("TestResult", back_populates="mirror", cascade="all, delete-orphan")

    CATEGORIES = ("dns", "apt", "docker", "github", "pip", "npm")

    CATEGORY_LABELS = {
        "dns": "DNS",
        "apt": "مخزن APT",
        "docker": "رجیstry داکر",
        "github": "گیت‌هاب",
        "pip": "pip",
        "npm": "npm",
    }

    def get_ips(self) -> list[str]:
        if not self.ips:
            return []
        return [ip.strip() for ip in self.ips.split(",") if ip.strip()]

    def get_meta(self) -> dict:
        if not self.meta_json:
            return {}
        try:
            return json.loads(self.meta_json)
        except json.JSONDecodeError:
            return {}

    def set_meta(self, data: dict) -> None:
        self.meta_json = json.dumps(data, ensure_ascii=False)

    def __repr__(self) -> str:
        return f"<Mirror {self.category}:{self.name}>"


class Profile(db.Model):
    __tablename__ = "profiles"

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(128), nullable=False, unique=True)
    description = db.Column(db.Text, nullable=True)
    is_default = db.Column(db.Boolean, default=False, nullable=False)
    created_at = db.Column(db.DateTime, default=utcnow, nullable=False)

    items = db.relationship("ProfileItem", back_populates="profile", cascade="all, delete-orphan")
    apply_logs = db.relationship("ApplyLog", back_populates="profile")

    def mirrors_by_category(self) -> dict[str, Mirror]:
        result: dict[str, Mirror] = {}
        for item in sorted(self.items, key=lambda x: x.order):
            if item.mirror and item.mirror.enabled:
                result[item.category] = item.mirror
        return result


class ProfileItem(db.Model):
    __tablename__ = "profile_items"

    id = db.Column(db.Integer, primary_key=True)
    profile_id = db.Column(db.Integer, db.ForeignKey("profiles.id"), nullable=False)
    mirror_id = db.Column(db.Integer, db.ForeignKey("mirrors.id"), nullable=False)
    category = db.Column(db.String(32), nullable=False)
    order = db.Column(db.Integer, default=0, nullable=False)

    profile = db.relationship("Profile", back_populates="items")
    mirror = db.relationship("Mirror")


class Setting(db.Model):
    __tablename__ = "settings"

    key = db.Column(db.String(64), primary_key=True)
    value = db.Column(db.Text, nullable=False)

    @staticmethod
    def get(key: str, default: str | None = None) -> str | None:
        row = Setting.query.get(key)
        return row.value if row else default

    @staticmethod
    def set(key: str, value: str) -> None:
        row = Setting.query.get(key)
        if row:
            row.value = value
        else:
            db.session.add(Setting(key=key, value=value))
        db.session.commit()


class ApplyLog(db.Model):
    __tablename__ = "apply_logs"

    id = db.Column(db.Integer, primary_key=True)
    profile_id = db.Column(db.Integer, db.ForeignKey("profiles.id"), nullable=True)
    status = db.Column(db.String(32), nullable=False, default="pending")
    backup_path = db.Column(db.String(512), nullable=True)
    previous_backup_path = db.Column(db.String(512), nullable=True)
    details_json = db.Column(db.Text, nullable=True)
    rollback_at = db.Column(db.DateTime, nullable=True)
    confirmed = db.Column(db.Boolean, default=False, nullable=False)
    started_at = db.Column(db.DateTime, default=utcnow, nullable=False)
    finished_at = db.Column(db.DateTime, nullable=True)

    profile = db.relationship("Profile", back_populates="apply_logs")

    def get_details(self) -> dict:
        if not self.details_json:
            return {}
        try:
            return json.loads(self.details_json)
        except json.JSONDecodeError:
            return {}

    def set_details(self, data: dict) -> None:
        self.details_json = json.dumps(data, ensure_ascii=False)


class TestResult(db.Model):
    __tablename__ = "test_results"

    id = db.Column(db.Integer, primary_key=True)
    mirror_id = db.Column(db.Integer, db.ForeignKey("mirrors.id"), nullable=False)
    test_type = db.Column(db.String(64), nullable=False)
    success = db.Column(db.Boolean, nullable=False)
    latency_ms = db.Column(db.Float, nullable=True)
    error = db.Column(db.Text, nullable=True)
    tested_at = db.Column(db.DateTime, default=utcnow, nullable=False)

    mirror = db.relationship("Mirror", back_populates="test_results")


class SystemState(db.Model):
    __tablename__ = "system_state"

    id = db.Column(db.Integer, primary_key=True)
    initial_backup_path = db.Column(db.String(512), nullable=True)
    initial_backup_done = db.Column(db.Boolean, default=False, nullable=False)
    current_profile_id = db.Column(db.Integer, db.ForeignKey("profiles.id"), nullable=True)
    pending_apply_log_id = db.Column(db.Integer, db.ForeignKey("apply_logs.id"), nullable=True)
    updated_at = db.Column(db.DateTime, default=utcnow, onupdate=utcnow, nullable=False)

    current_profile = db.relationship("Profile", foreign_keys=[current_profile_id])
    pending_apply = db.relationship("ApplyLog", foreign_keys=[pending_apply_log_id])


class AdminUser(db.Model):
    __tablename__ = "admin_users"

    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(64), unique=True, nullable=False)
    password_hash = db.Column(db.String(256), nullable=False)

    def check_password(self, password: str) -> bool:
        from werkzeug.security import check_password_hash

        return check_password_hash(self.password_hash, password)

    def set_password(self, password: str) -> None:
        from werkzeug.security import generate_password_hash

        self.password_hash = generate_password_hash(password)

    @property
    def is_authenticated(self) -> bool:
        return True

    @property
    def is_active(self) -> bool:
        return True

    @property
    def is_anonymous(self) -> bool:
        return False

    def get_id(self) -> str:
        return str(self.id)