RRRosmontis
34089d0042
Include utility modules, shared channels, prompts, read-only skills, wardrobe memory and regression tests. Externalize legacy credentials and exclude private runtime state.
143 lines
4.6 KiB
Python
143 lines
4.6 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""通过站内 OpenAPI 获取单个 Bilibili 视频，并以文件形式发送。"""
|
|
|
|
import asyncio
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import tempfile
|
|
from urllib.parse import urlencode
|
|
|
|
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
|
|
CACHE_DIR = Path(BASE_DIR) / "bilibili" / "cache"
|
|
OPENAPI_URL = "https://openapi.rrrosmontis.icu/bilibili/get_video"
|
|
HTTP_TIMEOUT = (15, 600)
|
|
MAX_DOWNLOAD_BYTES = 512 * 1024 * 1024
|
|
|
|
HELP_TEXT = (
|
|
"Bilibili 视频\n"
|
|
"━━━━━━━━━━━━\n"
|
|
"• .bilibili <BV号> 获取视频并以文件形式发送\n"
|
|
"示例: .bilibili BV17XxP6CE7n\n"
|
|
"━━━━━━━━━━━━\n"
|
|
"默认请求最高画质；大会员或版权资源可能无法获取"
|
|
)
|
|
|
|
|
|
def sanitize_bv(text):
|
|
"""只接受完整 BV 号，避免把任意文本或 URL 拼进 OpenAPI 请求。"""
|
|
match = re.fullmatch(r"(?i:BV)[0-9A-Za-z]{10}", (text or "").strip())
|
|
return "BV" + match.group(0)[2:] if match else None
|
|
|
|
|
|
def openapi_url(bv):
|
|
return OPENAPI_URL + "?" + urlencode({"bv": bv})
|
|
|
|
|
|
def cleanup_file(local_path):
|
|
try:
|
|
if local_path:
|
|
os.remove(local_path)
|
|
except FileNotFoundError:
|
|
pass
|
|
except OSError as exc:
|
|
print(f"[bilibili] 临时文件清理失败: {exc}")
|
|
|
|
|
|
def download_video(bv):
|
|
"""流式下载到独立临时文件；任何失败都清理残缺文件。"""
|
|
import requests
|
|
|
|
CACHE_DIR.mkdir(parents=True, exist_ok=True)
|
|
local_path = None
|
|
response = requests.get(
|
|
OPENAPI_URL,
|
|
params={"bv": bv},
|
|
stream=True,
|
|
timeout=HTTP_TIMEOUT,
|
|
)
|
|
try:
|
|
if response.status_code != 200:
|
|
detail = ""
|
|
content_type = response.headers.get("Content-Type", "").lower()
|
|
if "json" in content_type or "text" in content_type:
|
|
detail = response.text[:240].strip()
|
|
raise RuntimeError(
|
|
f"OpenAPI 下载失败 HTTP {response.status_code}"
|
|
+ (f": {detail}" if detail else "")
|
|
)
|
|
|
|
content_type = response.headers.get("Content-Type", "").split(";", 1)[0].strip().lower()
|
|
if "json" in content_type or content_type.startswith("text/"):
|
|
raise RuntimeError(f"OpenAPI 未返回视频: {response.text[:240].strip()}")
|
|
suffix = {
|
|
"video/x-flv": ".flv",
|
|
"video/webm": ".webm",
|
|
"video/x-matroska": ".mkv",
|
|
}.get(content_type, ".mp4")
|
|
fd, local_path = tempfile.mkstemp(prefix=bv + "_", suffix=suffix, dir=CACHE_DIR)
|
|
total = 0
|
|
with os.fdopen(fd, "wb") as output:
|
|
for chunk in response.iter_content(1024 * 1024):
|
|
if not chunk:
|
|
continue
|
|
total += len(chunk)
|
|
if total > MAX_DOWNLOAD_BYTES:
|
|
raise RuntimeError("视频超过 512 MB，已停止下载")
|
|
output.write(chunk)
|
|
if total == 0:
|
|
raise RuntimeError("OpenAPI 返回了空视频")
|
|
return Path(local_path)
|
|
except Exception:
|
|
cleanup_file(local_path)
|
|
raise
|
|
finally:
|
|
response.close()
|
|
|
|
|
|
async def handle_bilibili(channel, message_type, target_id, user_id, raw_text):
|
|
from channel import send_message
|
|
|
|
body = re.sub(r"^\s*\.bilibili\b", "", raw_text, flags=re.IGNORECASE).strip()
|
|
if not body or body.lower() in ("help", "-h", "帮助"):
|
|
await send_message(channel, message_type, target_id, HELP_TEXT)
|
|
return
|
|
|
|
bv = sanitize_bv(body)
|
|
if not bv:
|
|
await send_message(
|
|
channel,
|
|
message_type,
|
|
target_id,
|
|
"格式: .bilibili <BV号>\n示例: .bilibili BV17XxP6CE7n",
|
|
)
|
|
return
|
|
|
|
await send_message(channel, message_type, target_id, f"正在获取 Bilibili 视频 ({bv})，请稍候...")
|
|
local_path = None
|
|
try:
|
|
local_path = await asyncio.to_thread(download_video, bv)
|
|
ok = await send_message(
|
|
channel,
|
|
message_type,
|
|
target_id,
|
|
[{
|
|
"type": "file",
|
|
"data": {
|
|
"file": local_path.resolve().as_uri(),
|
|
"manual": openapi_url(bv),
|
|
},
|
|
}],
|
|
)
|
|
print(f"[bilibili] {bv} 文件投递{'成功' if ok else '失败'}")
|
|
except Exception as exc:
|
|
print(f"[bilibili] {bv} 获取失败: {type(exc).__name__}: {exc}")
|
|
await send_message(
|
|
channel,
|
|
message_type,
|
|
target_id,
|
|
f"{bv} 视频获取失败：{str(exc)[:400]}",
|
|
)
|
|
finally:
|
|
await asyncio.to_thread(cleanup_file, local_path)
|