feat: init project

src/core/__init__.py

@@ -0,0 +1,55 @@
+"""Core module for aria2bot - constants, config, exceptions, and system utilities."""
+from src.core.constants import (
+    HOME,
+    ARIA2_BIN,
+    ARIA2_CONFIG_DIR,
+    ARIA2_CONF,
+    ARIA2_SESSION,
+    ARIA2_LOG,
+    DOWNLOAD_DIR,
+    SYSTEMD_USER_DIR,
+    ARIA2_SERVICE,
+)
+from src.core.exceptions import (
+    Aria2Error,
+    UnsupportedOSError,
+    UnsupportedArchError,
+    DownloadError,
+    ConfigError,
+    ServiceError,
+    NotInstalledError,
+)
+from src.core.config import Aria2Config, BotConfig
+from src.core.system import (
+    detect_os,
+    detect_arch,
+    generate_rpc_secret,
+    is_aria2_installed,
+    get_aria2_version,
+)
+
+__all__ = [
+    "HOME",
+    "ARIA2_BIN",
+    "ARIA2_CONFIG_DIR",
+    "ARIA2_CONF",
+    "ARIA2_SESSION",
+    "ARIA2_LOG",
+    "DOWNLOAD_DIR",
+    "SYSTEMD_USER_DIR",
+    "ARIA2_SERVICE",
+    "Aria2Error",
+    "UnsupportedOSError",
+    "UnsupportedArchError",
+    "DownloadError",
+    "ConfigError",
+    "ServiceError",
+    "NotInstalledError",
+    "Aria2Config",
+    "BotConfig",
+    "detect_os",
+    "detect_arch",
+    "generate_rpc_secret",
+    "is_aria2_installed",
+    "get_aria2_version",
+]

src/core/config.py

@@ -0,0 +1,42 @@
+"""Configuration dataclass for aria2bot."""
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass, field
+from pathlib import Path
+
+from src.core.constants import DOWNLOAD_DIR
+
+
+@dataclass
+class Aria2Config:
+    rpc_port: int = 6800
+    rpc_secret: str = ""
+    download_dir: Path = DOWNLOAD_DIR
+    max_concurrent_downloads: int = 5
+    max_connection_per_server: int = 16
+    bt_tracker_update: bool = True
+
+
+@dataclass
+class BotConfig:
+    token: str = ""
+    api_base_url: str = ""
+    aria2: Aria2Config = field(default_factory=Aria2Config)
+
+    @classmethod
+    def from_env(cls) -> "BotConfig":
+        """从环境变量加载配置"""
+        from dotenv import load_dotenv
+        load_dotenv()
+
+        token = os.environ.get("TELEGRAM_BOT_TOKEN", "")
+        aria2 = Aria2Config(
+            rpc_port=int(os.environ.get("ARIA2_RPC_PORT", "6800")),
+            rpc_secret=os.environ.get("ARIA2_RPC_SECRET", ""),
+        )
+        return cls(
+            token=token,
+            api_base_url=os.environ.get("TELEGRAM_API_BASE_URL", ""),
+            aria2=aria2,
+        )

src/core/constants.py

@@ -0,0 +1,12 @@
+"""Path constants for aria2bot."""
+from pathlib import Path
+
+HOME = Path.home()
+ARIA2_BIN = HOME / ".local" / "bin" / "aria2c"
+ARIA2_CONFIG_DIR = HOME / ".config" / "aria2"
+ARIA2_CONF = ARIA2_CONFIG_DIR / "aria2.conf"
+ARIA2_SESSION = ARIA2_CONFIG_DIR / "aria2.session"
+ARIA2_LOG = ARIA2_CONFIG_DIR / "aria2.log"
+DOWNLOAD_DIR = HOME / "downloads"
+SYSTEMD_USER_DIR = HOME / ".config" / "systemd" / "user"
+ARIA2_SERVICE = SYSTEMD_USER_DIR / "aria2.service"

src/core/exceptions.py

@@ -0,0 +1,29 @@
+"""Exception classes for aria2bot."""
+
+
+class Aria2Error(Exception):
+    """Base exception"""
+
+
+class UnsupportedOSError(Aria2Error):
+    """不支持的操作系统"""
+
+
+class UnsupportedArchError(Aria2Error):
+    """不支持的 CPU 架构"""
+
+
+class DownloadError(Aria2Error):
+    """下载失败"""
+
+
+class ConfigError(Aria2Error):
+    """配置错误"""
+
+
+class ServiceError(Aria2Error):
+    """服务操作失败"""
+
+
+class NotInstalledError(Aria2Error):
+    """aria2 未安装"""

src/core/system.py

@@ -0,0 +1,91 @@
+"""System detection utilities for aria2bot."""
+from __future__ import annotations
+
+import platform
+import secrets
+import shutil
+import string
+import subprocess
+from pathlib import Path
+
+from src.core.constants import ARIA2_BIN
+from src.core.exceptions import UnsupportedOSError, UnsupportedArchError
+
+
+def detect_os() -> str:
+    """检测操作系统，返回 'centos', 'debian', 'ubuntu' 或抛出 UnsupportedOSError"""
+    os_release_path = Path("/etc/os-release")
+    if os_release_path.exists():
+        info: dict[str, str] = {}
+        for line in os_release_path.read_text(encoding="utf-8", errors="ignore").splitlines():
+            if "=" not in line:
+                continue
+            key, value = line.split("=", 1)
+            info[key.strip()] = value.strip().strip('"').lower()
+        os_id = info.get("ID")
+        if os_id in {"ubuntu", "debian"}:
+            return os_id
+        if os_id in {"centos", "rhel", "rocky", "almalinux"}:
+            return "centos"
+
+    redhat_release = Path("/etc/redhat-release")
+    if redhat_release.exists():
+        content = redhat_release.read_text(encoding="utf-8", errors="ignore").lower()
+        if any(name in content for name in ("centos", "red hat", "rocky", "alma")):
+            return "centos"
+
+    raise UnsupportedOSError("Unsupported operating system")
+
+
+def detect_arch() -> str:
+    """检测 CPU 架构，返回 'amd64', 'arm64', 'armhf', 'i386' 或抛出 UnsupportedArchError"""
+    machine = platform.machine().lower()
+    if machine in {"x86_64", "amd64"}:
+        return "amd64"
+    if machine in {"aarch64", "arm64", "armv8"}:
+        return "arm64"
+    if machine.startswith("armv7") or machine.startswith("armv6"):
+        return "armhf"
+    if machine in {"i386", "i686", "x86"}:
+        return "i386"
+    raise UnsupportedArchError(f"Unsupported CPU architecture: {machine}")
+
+
+def generate_rpc_secret() -> str:
+    """生成 20 位随机 RPC 密钥"""
+    alphabet = string.ascii_letters + string.digits
+    return "".join(secrets.choice(alphabet) for _ in range(20))
+
+
+def is_aria2_installed() -> bool:
+    """检查 aria2c 是否已安装"""
+    if ARIA2_BIN.exists():
+        return True
+    return shutil.which("aria2c") is not None
+
+
+def get_aria2_version() -> str | None:
+    """获取已安装的 aria2 版本"""
+    candidates = [ARIA2_BIN] if ARIA2_BIN.exists() else []
+    path_cmd = shutil.which("aria2c")
+    if path_cmd:
+        candidates.append(Path(path_cmd))
+
+    if not candidates:
+        return None
+
+    for cmd in candidates:
+        result = subprocess.run(
+            [str(cmd), "-v"], capture_output=True, text=True, check=False
+        )
+        if result.returncode != 0:
+            continue
+        for line in result.stdout.splitlines():
+            lowered = line.lower()
+            if "aria2 version" in lowered:
+                parts = line.split()
+                return parts[-1] if parts else line.strip()
+        if result.stdout.strip():
+            return result.stdout.splitlines()[0].strip()
+
+    return None