from __future__ import annotations import logging from pathlib import Path from app.models import Mirror from app.services.host import get_ubuntu_codename, host_path, run_on_host, validate_codename logger = logging.getLogger(__name__) SOURCES_PATH = "etc/apt/sources.list" SOURCES_D_PATH = "etc/apt/sources.list.d" MIRROR_MARKER = "mirror-manager" def read_current_apt() -> dict: sources = host_path(SOURCES_PATH) return { "sources_list": sources.read_text(encoding="utf-8") if sources.exists() else None, "path": str(sources), } def _build_sources_content(base_url: str, codename: str) -> str: url = base_url.rstrip("/") return f"""# Managed by Mirror Manager - {MIRROR_MARKER} deb {url} {codename} main restricted universe multiverse deb {url} {codename}-updates main restricted universe multiverse deb {url} {codename}-backports main restricted universe multiverse deb {url} {codename}-security main restricted universe multiverse """ def apply_apt(mirror: Mirror) -> dict: if not mirror.url: return {"success": False, "error": "URL میرور APT تعریف نشده"} codename = validate_codename(get_ubuntu_codename()) sources_path = host_path(SOURCES_PATH) sources_path.parent.mkdir(parents=True, exist_ok=True) content = _build_sources_content(mirror.url, codename) sources_path.write_text(content, encoding="utf-8") sources_d = host_path(SOURCES_D_PATH) if sources_d.exists(): for f in sources_d.glob("*.list"): if MIRROR_MARKER not in f.read_text(encoding="utf-8", errors="ignore"): backup_name = f.with_suffix(".list.disabled") f.rename(backup_name) result = run_on_host(["apt-get", "update", "-qq"], timeout=180) if result.returncode != 0: return { "success": False, "error": (result.stderr or result.stdout or "apt-get update ناموفق")[:500], "codename": codename, } return { "success": True, "message": f"APT mirror اعمال شد: {mirror.name}", "codename": codename, "url": mirror.url, }