1#!/usr/bin/env python
 2# -*- coding: utf-8 -*-
 3import base64
 4import io
 5import shutil
 6from pathlib import Path
 7
 8import anyio
 9from glom import glom
10from loguru import logger
11
12from config import DB, DOWNLOAD_DIR
13from networking import hx_req
14
15
16async def list_alist() -> list[dict]:
17    """List from Alist."""
18    if not DB.ALIST_ENABLED:
19        return [{}]
20    api = DB.ALIST_SERVER.removesuffix("/") + "/api/fs/list"
21    logger.info(f"List Alist from: {api}")
22    res = await hx_req(api, method="POST", json_data={"path": "/"}, check_kv={"code": 200})
23    return glom(res, "data.content", default=[]) or []
24
25
26async def download_alist(fname: str, save_path: str | Path | None = None) -> str:
27    """Download file from Alist."""
28    if not DB.ALIST_ENABLED:
29        return ""
30    save_path = Path(DOWNLOAD_DIR) / fname if save_path is None else Path(save_path)
31    ext = Path(fname).suffix
32    # Headers DO NOT support Unicode characters
33    if any(ord(c) > 127 for c in fname):  # has Non-ASCII
34        b64str = base64.urlsafe_b64encode(fname.encode("utf-8")).decode("ascii").rstrip("=")
35        fname = b64str + ext
36    url = DB.ALIST_SERVER.removesuffix("/") + "/d/" + DB.ALIST_BASR_PATH.strip("/") + f"/{fname.lstrip('/')}"
37    res = await hx_req(url, rformat="content", check_keys=["content"])
38    logger.info(f"Download {fname} to {save_path.name} from: {url}")
39    async with await anyio.open_file(save_path, "wb") as f:
40        await f.write(res["content"])
41    return save_path.as_posix()
42
43
44async def upload_alist(path: str | Path) -> str:
45    """Upload to Alist."""
46    if not DB.ALIST_ENABLED:
47        return ""
48    path = Path(path).expanduser().resolve()
49    if not path.is_file():
50        return ""
51    api = DB.ALIST_SERVER.removesuffix("/") + "/api/fs/form"
52    # Headers DO NOT support Unicode characters
53    new_path = Path("/non-exist")
54    if any(ord(c) > 127 for c in path.name):  # has Non-ASCII
55        new_name = base64.urlsafe_b64encode(path.name.encode("utf-8")).decode("ascii").rstrip("=")
56        new_path = path.with_stem(new_name)
57        shutil.copy(path, new_path)
58        path = new_path
59    headers = {"File-Path": f"/{path.name}"}
60    logger.info(f"Upload {path.name} to: {api}")
61    async with await anyio.open_file(path, "rb") as f:
62        content = await f.read()
63        await hx_req(api, method="PUT", headers=headers, files={"file": (path.name, io.BytesIO(content))}, check_kv={"code": 200})
64    if "new_path" in locals():
65        new_path.unlink(missing_ok=True)
66    return DB.ALIST_SERVER.removesuffix("/") + "/d/" + DB.ALIST_BASR_PATH.strip("/") + f"/{path.name}"
67
68
69async def delete_alist(fname: str, *, ensure_ascii: bool = True) -> None:
70    """Delete from Alist."""
71    if not DB.ALIST_ENABLED:
72        return
73    # Get JWT Token
74    payload = {"username": DB.ALIST_USERNAME, "password": DB.ALIST_PASSWORD}
75    auth_api = DB.ALIST_SERVER.removesuffix("/") + "/api/auth/login"
76    res = await hx_req(auth_api, method="POST", json_data=payload, check_keys=["data.token"])
77    token = res["data"]["token"]
78
79    # Delete
80    api = DB.ALIST_SERVER.removesuffix("/") + "/api/fs/remove"
81    headers = {"Content-Type": "application/json", "Authorization": token}
82    if ensure_ascii and any(ord(c) > 127 for c in fname):  # has Non-ASCII
83        b64str = base64.urlsafe_b64encode(fname.encode("utf-8")).decode("ascii").rstrip("=")
84        payload = {"names": [b64str + Path(fname).suffix]}
85    else:
86        payload = {"names": [fname]}
87    logger.info(f"Delete {fname} from: {api}")
88    res = await hx_req(api, method="POST", headers=headers, json_data=payload, check_kv={"code": 200})