02.302后台服务¶
我已经写了几个可开箱即用的 302 模块
1. 一个最简 302 后台服务¶
首先需要安装依赖模块
pip install -U blacksheep p115client uvicorn
下面的代码分享了一个最简单的 302 服务，实现了 id 和 pickcode 查询，并对取得的下载链接进行 302 重定向
#!/usr/bin/env python3
# encoding: utf-8
__author__ = "ChenyangGao <https://github.com/ChenyangGao>"
__version__ = (0, 0, 1)
from pathlib import Path
from p115client import P115Client
client = P115Client(Path("~/115-cookies.txt").expanduser())
from blacksheep import json, redirect, Application, Request
app = Application(show_error_details=__debug__)
@app.router.route("/", methods=["GET", "HEAD"])
async def index(
request: Request,
id: int = 0,
pickcode: str = "",
):
if id > 0 and not pickcode:
pickcode = client.to_pickcode(id)
if not pickcode:
return json({"error": "请指定查询参数 id 或 pickcode，并确保有效"}, 500)
user_agent = request.get_first_header(b"user-agent") or b""
try:
url = await client.download_url(
pickcode,
headers={"user-agent": user_agent.decode("latin-1")},
app="android",
async_=True,
)
except (FileNotFoundError, IsADirectoryError):
return json({"pickcode": pickcode, "error": "not found"}, 404)
return redirect(url)
if __name__ == "__main__":
from uvicorn import run
run(app, host="0.0.0.0", port=8115)
2. 扩展：支持查询 sha1 和 name¶
首先需要安装依赖模块
pip install -U blacksheep p115client uvicorn
#!/usr/bin/env python3
# encoding: utf-8
__author__ = "ChenyangGao <https://github.com/ChenyangGao>"
__version__ = (0, 0, 2)
from pathlib import Path
from p115client import P115Client
client = P115Client(Path("~/115-cookies.txt").expanduser())
from blacksheep import json, redirect, Application, Request
app = Application(show_error_details=__debug__)
@app.router.route("/", methods=["GET", "HEAD"])
async def index(
request: Request,
id: int = 0,
pickcode: str = "",
sha1: str = "",
name: str = "",
):
if not pickcode:
if id > 0:
pickcode = client.to_pickcode(id)
elif sha1:
resp = await client.fs_shasearch(sha1, async_=True)
if not resp["state"]:
return json({"sha1": sha1, "resp": resp}, 404)
pickcode = resp["data"]["pick_code"]
elif name:
payload = {"fc": 2, "limit": 16, "search_value": name, "type": 99}
suffix = name.rpartition(".")[-1]
if len(suffix) < 5 and suffix.isalnum() and suffix[0].isalpha():
payload["suffix"] = suffix
resp = await client.fs_search(payload, async_=True)
if not resp["state"]:
return json({"name": name, "resp": resp}, 404)
for info in resp["data"]:
if info["n"] == name and info.get("sha"):
pickcode = info["pc"]
break
if not pickcode:
return json({"name": name, "error": "not found"}, 500)
user_agent = request.get_first_header(b"user-agent") or b""
try:
url = await client.download_url(
pickcode,
headers={"user-agent": user_agent.decode("latin-1")},
app="android",
async_=True,
)
except (FileNotFoundError, IsADirectoryError):
return json({"pickcode": pickcode, "error": "not found"}, 404)
return redirect(url)
if __name__ == "__main__":
from uvicorn import run
run(app, host="0.0.0.0", port=8115)
3. 扩展：支持对下载链接缓存一定时间¶
首先需要安装依赖模块
pip install -U blacksheep cachedict p115client uvicorn
#!/usr/bin/env python3
# encoding: utf-8
__author__ = "ChenyangGao <https://github.com/ChenyangGao>"
__version__ = (0, 0, 3)
from pathlib import Path
from p115client import P115Client
client = P115Client(Path("~/115-cookies.txt").expanduser())
from binascii import crc32
from urllib.parse import parse_qsl, urlsplit
from blacksheep import json, redirect, Application, Request
from cachedict import TLRUDict
app = Application(show_error_details=__debug__)
cached_urls = TLRUDict(1024)
@app.router.route("/", methods=["GET", "HEAD"])
async def index(
request: Request,
id: int = 0,
pickcode: str = "",
refresh: bool = False,
):
user_agent = request.get_first_header(b"user-agent") or b""
if id <= 0:
id = client.to_id(pickcode)
if id <= 0:
return json({"error": "请指定查询参数 id 或 pickcode，并确保有效"}, 500)
#key = (id, user_agent)
key = (len(user_agent) << id.bit_length()) + id << 32 | crc32(user_agent, id)
if not refresh and (pair := cached_urls.get(key)):
return redirect(pair[1])
if not pickcode:
pickcode = client.to_pickcode(id)
try:
url = await client.download_url(
pickcode,
headers={"user-agent": user_agent.decode("latin-1")},
app="android",
async_=True,
)
except (FileNotFoundError, IsADirectoryError):
return json({"pickcode": pickcode, "error": "not found"}, 404)
# NOTE: 缓存的过期时间为链接过期时间提前 5 分钟
expire_ts = int(next(v for k, v in parse_qsl(urlsplit(url).query) if k == "t")) - 60 * 5
cached_urls[key] = (expire_ts, url)
return redirect(url)
if __name__ == "__main__":
from uvicorn import run
run(app, host="0.0.0.0", port=8115)