AniBT Wiki

公開 RSS 回傳含 .torrent 下載連結的 XML。番劇與字幕組篩選可組合；其他類資源使用獨立分類 RSS。

RSS URLs

https://anibt.net/rss/anime.xml?bgmId=543360&groupSlug=kirara-fantasia
https://anibt.net/rss/anime.xml?bgmId=543360&resolution=1080p&language=CHS
https://anibt.net/rss/group/kirara-fantasia.xml
https://anibt.net/rss/other/music.xml

以下腳本使用 Python 3 標準函式庫。先儲存正文，再儲存 ETag；304 只重用本機已有正文。

conditional_rss_poll.py

import os
from pathlib import Path
from urllib.error import HTTPError
from urllib.request import Request, urlopen

url = os.environ.get("FEED_URL", "https://anibt.net/rss/anime.xml?bgmId=543360")
feed_file = Path("feed.xml")
etag_file = Path("feed.etag")
headers = {"User-Agent": "AniBT-RSS-example/1.0"}
if feed_file.exists() and etag_file.exists():
    headers["If-None-Match"] = etag_file.read_text().strip()
try:
    with urlopen(Request(url, headers=headers), timeout=30) as response:
        body = response.read()
        etag = response.headers.get("ETag", "")
        status = response.status
except HTTPError as error:
    if error.code == 304:
        print("Unchanged; reuse feed.xml")
        raise SystemExit(0)
    raise SystemExit(f"HTTP {error.code}; Retry-After={error.headers.get('Retry-After', 'unknown')}")
if status != 200:
    raise SystemExit(f"Unexpected HTTP {status}")
# Commit the body before its validator; never mark an undownloaded body current.
feed_file.with_suffix(".tmp").write_bytes(body)
feed_file.with_suffix(".tmp").replace(feed_file)
etag_file.write_text(etag)
print("Saved feed.xml")

刪除示例從環境變數讀取 RELEASE_ID 與 ANIBT_API_KEY，最多輪詢 180 秒。只有明確的 completed 才算完成；HTTP 錯誤、failed 或無效正文都會停止。

delete_and_wait.py

import json
import os
import time
from urllib.error import HTTPError
from urllib.request import Request, urlopen

base = os.environ.get("ANIBT_BASE_URL", "https://anibt.net").rstrip("/")
release_id = os.environ["RELEASE_ID"]
key = os.environ["ANIBT_API_KEY"]
deadline = time.monotonic() + 180

def call(method, path):
    request = Request(base + path, method=method, headers={"Authorization": f"Bearer {key}"})
    try:
        response = urlopen(request, timeout=20)
    except HTTPError as error:
        response = error
    with response:
        raw = response.read()
        status = response.status
        retry_after = response.headers.get("Retry-After")
        request_id = response.headers.get("X-Request-ID", "unknown")
    if status == 429:
        wait = int(retry_after) if retry_after and retry_after.isdigit() else 5
        return None, max(1, wait)
    try:
        data = json.loads(raw)
    except ValueError:
        raise SystemExit(f"Non-JSON response: HTTP {status}, requestId={request_id}")
    if status not in (200, 202) or data.get("ok") is not True:
        code = data.get("error", {}).get("code", "unknown")
        raise SystemExit(f"HTTP {status}: {code}, requestId={request_id}")
    result = data.get("result")
    if not isinstance(result, dict) or result.get("state") not in ("pending", "completed", "failed"):
        raise SystemExit("Invalid deletion receipt")
    wait = int(retry_after) if retry_after and retry_after.isdigit() else 2
    return result, max(1, wait)

receipt, wait = call("DELETE", f"/api/releases/{release_id}")
if receipt is None:
    raise SystemExit(f"DELETE rate limited; retry after {wait} seconds")
while True:
    if receipt is not None and receipt["state"] == "completed":
        print("Deletion completed")
        break
    if receipt is not None and receipt["state"] == "failed":
        raise SystemExit(f"Deletion failed: {receipt.get('failure', {}).get('code', 'unknown')}")
    remaining = deadline - time.monotonic()
    if remaining <= wait:
        raise SystemExit("Polling timed out; keep the releaseId and check its status later")
    time.sleep(wait)
    receipt, wait = call("GET", f"/api/releases/{release_id}/deletion")

整合示例