希望添加一个命令行的备份恢复json的功能

[zhlhlf]

这样就可以在git的action下自动恢复然后挂载webdav上传文件

[pongfcnkl]

自己写一个不就行了

import argparse
import asyncio
import datetime
import json
import os
from dataclasses import dataclass
from typing import TypeVar, Generic, Type, Dict, Any, Union, Literal
from urllib.parse import urljoin
import httpx

# Configuration defined somewhere, e.g., settings.py or environment variables
@dataclass
class Config:
    alist_host: str

T = TypeVar("T")

@dataclass
class AListAPIResponse(Generic[T]):
    def __init__(self, code: int, message: str, data: T | list[T] | dict | None = None):
        self.code = code
        self.message = message
        self.data = data
        self.raw_data: dict | list | None = None

    @classmethod
    def from_dict(cls, result: dict, data_class: Type[T] | None) -> "AListAPIResponse[T]":
        response = cls(code=result.get("code"), message=result.get("message"))
        response._check_code()
        data = result.get("data")
        response.raw_data = data
        response.data = data_class.from_dict(data) if data_class else data
        return response

    def _check_code(self):
        if self.code == 401 and self.message == "that's not even a token":
            raise AListTokenError()

    def __repr__(self):
        return f"APIResponse(code={self.code}, message={self.message}, data={self.data})"

class AListError(Exception):
    def __init__(self, message: str):
        self.message = message

    def __str__(self):
        return self.message

class AListTokenError(AListError):
    def __init__(self):
        super().__init__("AList Token错误")

class AListAPI:
    def __init__(self, host):
        self.host = host
        self.headers = {
            "UserAgent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/112.0.0.0 Safari/537.36",
            "Content-Type": "application/json",
            "Authorization": "", 
        }

    async def _request(self, method: Literal["GET", "POST", "PUT"], url, *,
                       data_class: Type[T] = None, headers: Dict[str, str] = None,
                       json: Dict[str, Any] = None, params: Dict[str, Any] = None,
                       data: Any = None, timeout: int = 10) -> AListAPIResponse[T]:
        url = urljoin(self.host, url)
        headers = self.headers if headers is None else headers

        async with httpx.AsyncClient() as client:
            if method == "GET":
                response = await client.get(url, headers=headers, params=params, timeout=timeout)
            elif method == "POST":
                response = await client.post(url, headers=headers, json=json, timeout=timeout)
            elif method == "PUT":
                response = await client.put(url, headers=headers, data=data, timeout=timeout)
        
        response.raise_for_status()
        result = response.json()
        return AListAPIResponse.from_dict(result, data_class)

    async def login(self, username: str, password: str) -> str:
        payload = {"username": username, "password": password}
        response = await self._request("POST", "/api/auth/login", json=payload)
        if response.code == 200 and response.data:
            self.headers["Authorization"] = response.data.get("token", "")
            return response.data.get("token", "")
        raise AListError("Failed to login and obtain token")

    async def setting_list(self):
        return await self._request("GET", "/api/admin/setting/list")

    async def user_list(self):
        return await self._request("GET", "/api/admin/user/list")

    async def storage_list(self):
        return await self._request("GET", "/api/admin/storage/list")

    async def meta_list(self):
        return await self._request("GET", "/api/admin/meta/list")

async def backup_config(alist: AListAPI, output_directory: str):
    bc_dic = {
        "encrypted": "",
        "settings": await alist.setting_list(),
        "users": await alist.user_list(),
        "storages": await alist.storage_list(),
        "metas": await alist.meta_list(),
    }
    for k, v in bc_dic.items():
        if isinstance(v, AListAPIResponse):
            bc_dic[k] = (
                v.raw_data.get("content")
                if isinstance(v.raw_data, dict)
                else v.raw_data
            )

    data = json.dumps(bc_dic, indent=4, ensure_ascii=False)
    now = datetime.datetime.now()
    current_time = now.strftime("%Y_%m_%d_%H_%M_%S")
    bc_file_name = os.path.join(os.getcwd(), output_directory, f"alist_backup_{current_time}.json")

    try:
        with open(bc_file_name, "w", encoding="utf-8") as b:
            json.dump(bc_dic, b, indent=4, ensure_ascii=False)
        print(f"JSON backup saved to: {bc_file_name}")
    except Exception as e:
        print(f"Error saving JSON backup: {e}")

async def main(username: str, password: str, alist_host: str, output_directory: str):
    alist = AListAPI(alist_host)
    try:
        token = await alist.login(username, password)
        print(f"Token: {token}")
        await backup_config(alist, output_directory)
    except AListError as e:
        print(f"Error: {e}")

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="AList Backup Tool")
    parser.add_argument("--username", required=True, help="Username for AList")
    parser.add_argument("--password", required=True, help="Password for AList")
    parser.add_argument("--alist-host", required=True, help="Host URL for AList API")
    parser.add_argument("--output-directory", default="", help="Output directory for the backup")

    args = parser.parse_args()

    asyncio.run(main(args.username, args.password, args.alist_host, args.output_directory))`

[zhlhlf]

大佬 牛逼

[zhlhlf]

不行 这个是备份的吧 我要一个恢复的

[pongfcnkl]

不行 这个是备份的吧 我要一个恢复的

恢复我一般直接去网页上了，你可以参考一下自己写

[zhlhlf]

没事 AI搞定了