openapi 源码 + token 模板 + nginx 统一入口配置
This commit is contained in:
RRRosmontis
@@ -0,0 +1,175 @@
|
||||
import asyncio
|
||||
import sys
|
||||
import base64
|
||||
import os
|
||||
import re
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
import httpx
|
||||
from aiohttp import web
|
||||
|
||||
COOKIE_FILE = str(Path(__file__).parent / "cookies.txt")
|
||||
UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36"
|
||||
|
||||
|
||||
def _prepare_cookie_file():
|
||||
"""Convert simple name=value cookies.txt to Netscape format for yt-dlp."""
|
||||
try:
|
||||
with open(COOKIE_FILE) as f:
|
||||
cookies = [line.strip() for line in f if line.strip() and not line.startswith("#")]
|
||||
except FileNotFoundError:
|
||||
return None
|
||||
if not cookies:
|
||||
return None
|
||||
tmp = tempfile.NamedTemporaryFile(
|
||||
mode="w", suffix=".txt", prefix="bilibili_cookie_", delete=False
|
||||
)
|
||||
tmp.write("# Netscape HTTP Cookie File\n")
|
||||
for c in cookies:
|
||||
if "=" not in c:
|
||||
continue
|
||||
name, value = c.split("=", 1)
|
||||
# Netscape format: domain flag path secure expires name value
|
||||
tmp.write(f".bilibili.com\tTRUE\t/\tFALSE\t0\t{name}\t{value}\n")
|
||||
tmp.close()
|
||||
return tmp.name
|
||||
|
||||
|
||||
async def resolve_bvid(link):
|
||||
# 直接是 BV 号
|
||||
m = re.search(r"(BV[a-zA-Z0-9]{10})", link)
|
||||
if m:
|
||||
return m.group(1)
|
||||
|
||||
# b23.tv 短链接，跟随跳转
|
||||
if "b23.tv" in link:
|
||||
async with httpx.AsyncClient(timeout=10) as client:
|
||||
resp = await client.get(
|
||||
link.strip(),
|
||||
follow_redirects=False,
|
||||
headers={"User-Agent": UA},
|
||||
)
|
||||
location = resp.headers.get("Location", "")
|
||||
if m := re.search(r"(BV[a-zA-Z0-9]{10})", location):
|
||||
return m.group(1)
|
||||
if m := re.search(r"(BV[a-zA-Z0-9]{10})", resp.text):
|
||||
return m.group(1)
|
||||
|
||||
# 普通 bilibili.com URL
|
||||
if m := re.search(r"(BV[a-zA-Z0-9]{10})", link):
|
||||
return m.group(1)
|
||||
|
||||
raise ValueError(f"unable to resolve BV from: {link}")
|
||||
|
||||
|
||||
def decode_url(url_b64):
|
||||
try:
|
||||
return base64.urlsafe_b64decode(url_b64.encode()).decode()
|
||||
except Exception:
|
||||
return base64.b64decode(url_b64.encode()).decode()
|
||||
|
||||
|
||||
async def get_video(url_b64=None, bv=None):
|
||||
if bv:
|
||||
bvid = bv
|
||||
else:
|
||||
try:
|
||||
url = decode_url(url_b64)
|
||||
except Exception:
|
||||
return web.json_response({"error": "invalid base64 encoding"}, status=400)
|
||||
|
||||
if not url:
|
||||
return web.json_response({"error": "empty url"}, status=400)
|
||||
|
||||
if "bilibili.com" not in url and "b23.tv" not in url and not url.startswith("BV"):
|
||||
return web.json_response({"error": "not a bilibili url"}, status=400)
|
||||
|
||||
try:
|
||||
bvid = await resolve_bvid(url)
|
||||
except ValueError as e:
|
||||
return web.json_response({"error": str(e)}, status=400)
|
||||
|
||||
cookie_tmp = _prepare_cookie_file()
|
||||
if not cookie_tmp:
|
||||
return web.json_response({"error": "missing bilibili cookie"}, status=500)
|
||||
|
||||
tmpdir = tempfile.mkdtemp(prefix="bilibili_")
|
||||
output_template = os.path.join(tmpdir, "video.%(ext)s")
|
||||
cmd = [
|
||||
os.path.join(os.path.dirname(sys.executable), "yt-dlp"),
|
||||
"--cookies", cookie_tmp,
|
||||
"--add-header", f"User-Agent:{UA}",
|
||||
"--add-header", "Referer:https://www.bilibili.com/",
|
||||
"-f", "bestvideo*+bestaudio/best",
|
||||
"--merge-output-format", "mp4",
|
||||
"-o", output_template,
|
||||
f"https://www.bilibili.com/video/{bvid}",
|
||||
]
|
||||
try:
|
||||
proc = await asyncio.create_subprocess_exec(
|
||||
*cmd,
|
||||
stdout=asyncio.subprocess.PIPE,
|
||||
stderr=asyncio.subprocess.PIPE,
|
||||
)
|
||||
try:
|
||||
stdout, stderr = await asyncio.wait_for(
|
||||
proc.communicate(), timeout=300
|
||||
)
|
||||
except asyncio.TimeoutError:
|
||||
proc.kill()
|
||||
await proc.wait()
|
||||
return web.json_response({"error": "download timeout"}, status=504)
|
||||
finally:
|
||||
if cookie_tmp and os.path.exists(cookie_tmp):
|
||||
os.unlink(cookie_tmp)
|
||||
|
||||
video_files = [p for p in Path(tmpdir).glob("*") if p.is_file()]
|
||||
if proc.returncode != 0 and not video_files:
|
||||
err_msg = stderr.decode()[-1000:] if stderr else "unknown error"
|
||||
return web.json_response(
|
||||
{"error": "download failed", "detail": err_msg},
|
||||
status=502,
|
||||
)
|
||||
|
||||
if not video_files:
|
||||
return web.json_response({"error": "no output file"}, status=502)
|
||||
|
||||
video_path = max(video_files, key=lambda p: p.stat().st_size)
|
||||
|
||||
ext = video_path.suffix.lower()
|
||||
content_types = {
|
||||
".mp4": "video/mp4",
|
||||
".flv": "video/x-flv",
|
||||
".mkv": "video/x-matroska",
|
||||
".webm": "video/webm",
|
||||
}
|
||||
content_type = content_types.get(ext, "application/octet-stream")
|
||||
|
||||
return web.FileResponse(
|
||||
path=video_path,
|
||||
headers={
|
||||
"Content-Type": content_type,
|
||||
"Content-Disposition": f'attachment; filename="{video_path.name}"',
|
||||
},
|
||||
)
|
||||
except Exception as e:
|
||||
print(f"[bilibili] Error: {e}")
|
||||
return web.json_response({"error": str(e)}, status=500)
|
||||
|
||||
|
||||
async def handle_bilibili(request):
|
||||
bv = request.query.get("bv", "")
|
||||
if bv:
|
||||
return await get_video(bv=bv)
|
||||
url_b64 = request.query.get("url", "")
|
||||
if not url_b64:
|
||||
return web.json_response({"error": "missing url or bv parameter"}, status=400)
|
||||
return await get_video(url_b64=url_b64)
|
||||
|
||||
|
||||
app = web.Application()
|
||||
app.router.add_get("/bilibili/get_video", handle_bilibili)
|
||||
|
||||
if __name__ == "__main__":
|
||||
web.run_app(app, host="127.0.0.1", port=19990)
|
||||
@@ -0,0 +1,58 @@
|
||||
import asyncio
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
from bilibili_api.login_v2 import QrCodeLogin, QrCodeLoginEvents
|
||||
|
||||
QR_PATH = Path("/home/bilibili/qrcode.png")
|
||||
COOKIE_FILE = Path("/home/bilibili/cookies.txt")
|
||||
CF_QR_PATH = "/mnt/E/temp/bilibili_qrcode.png"
|
||||
|
||||
async def main():
|
||||
login = QrCodeLogin()
|
||||
await login.generate_qrcode()
|
||||
pic = login.get_qrcode_picture()
|
||||
QR_PATH.write_bytes(pic.content)
|
||||
print(f"QR saved to {QR_PATH}")
|
||||
print("Please scan the QR code with Bilibili App")
|
||||
|
||||
# copy to cf
|
||||
import subprocess
|
||||
try:
|
||||
subprocess.run([
|
||||
"sshpass", "-p", "ftA406u@all#2309",
|
||||
"scp", "-P", "10526", str(QR_PATH), "root@rrrosmontis.icu:/mnt/E/temp/bilibili_qrcode.png"
|
||||
], check=True, timeout=30)
|
||||
print(f"QR copied to cf:{CF_QR_PATH}")
|
||||
except Exception as e:
|
||||
print(f"Failed to copy QR to cf: {e}")
|
||||
|
||||
# wait up to 5 minutes
|
||||
for i in range(60):
|
||||
await asyncio.sleep(5)
|
||||
state = await login.check_state()
|
||||
print(f"[{i*5}s] state: {state}")
|
||||
if state == QrCodeLoginEvents.DONE:
|
||||
cred = login.get_credential()
|
||||
lines = []
|
||||
if cred.sessdata:
|
||||
lines.append(f"SESSDATA={cred.sessdata}")
|
||||
if cred.bili_jct:
|
||||
lines.append(f"bili_jct={cred.bili_jct}")
|
||||
if cred.buvid3:
|
||||
lines.append(f"buvid3={cred.buvid3}")
|
||||
if cred.dedeuserid:
|
||||
lines.append(f"dedeuserid={cred.dedeuserid}")
|
||||
if cred.ac_time_value:
|
||||
lines.append(f"ac_time_value={cred.ac_time_value}")
|
||||
COOKIE_FILE.write_text("\n".join(lines) + "\n")
|
||||
print(f"Login success. Cookie saved to {COOKIE_FILE}")
|
||||
print(f"ac_time_value present: {bool(cred.ac_time_value)}")
|
||||
return
|
||||
elif state == QrCodeLoginEvents.TIMEOUT:
|
||||
print("QR code timeout")
|
||||
sys.exit(1)
|
||||
print("QR login timeout after 5 minutes")
|
||||
sys.exit(1)
|
||||
|
||||
asyncio.run(main())
|
||||
@@ -0,0 +1,52 @@
|
||||
import asyncio
|
||||
import sys
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
from bilibili_api.login_v2 import QrCodeLogin, QrCodeLoginEvents, QrCodeLoginChannel
|
||||
|
||||
QR_PATH = Path("/home/bilibili/qrcode.png")
|
||||
COOKIE_FILE = Path("/home/bilibili/cookies.txt")
|
||||
CF_QR_PATH = "root@rrrosmontis.icu:/mnt/E/temp/bilibili_qrcode.png"
|
||||
|
||||
async def main():
|
||||
# try TV channel which may have longer expiration
|
||||
login = QrCodeLogin(platform=QrCodeLoginChannel.TV)
|
||||
await login.generate_qrcode()
|
||||
pic = login.get_qrcode_picture()
|
||||
QR_PATH.write_bytes(pic.content)
|
||||
print(f"QR regenerated: {QR_PATH}")
|
||||
|
||||
# sync to cf immediately
|
||||
try:
|
||||
subprocess.run([
|
||||
"sshpass", "-p", "ftA406u@all#2309",
|
||||
"scp", "-P", "10526", str(QR_PATH), CF_QR_PATH
|
||||
], check=True, timeout=10)
|
||||
print("QR synced to cf")
|
||||
except Exception as e:
|
||||
print(f"Sync failed: {e}")
|
||||
|
||||
print("Scan now! Polling every 3s for 3 minutes...")
|
||||
for i in range(60):
|
||||
await asyncio.sleep(3)
|
||||
state = await login.check_state()
|
||||
print(f"[{i*3}s] {state}")
|
||||
if state == QrCodeLoginEvents.DONE:
|
||||
cred = login.get_credential()
|
||||
lines = []
|
||||
if cred.sessdata: lines.append(f"SESSDATA={cred.sessdata}")
|
||||
if cred.bili_jct: lines.append(f"bili_jct={cred.bili_jct}")
|
||||
if cred.buvid3: lines.append(f"buvid3={cred.buvid3}")
|
||||
if cred.dedeuserid: lines.append(f"dedeuserid={cred.dedeuserid}")
|
||||
if cred.ac_time_value: lines.append(f"ac_time_value={cred.ac_time_value}")
|
||||
COOKIE_FILE.write_text("\n".join(lines) + "\n")
|
||||
print("LOGIN SUCCESS")
|
||||
print(f"ac_time_value: {bool(cred.ac_time_value)}")
|
||||
return
|
||||
elif state == QrCodeLoginEvents.TIMEOUT:
|
||||
print("QR TIMEOUT")
|
||||
sys.exit(1)
|
||||
print("POLLING TIMEOUT")
|
||||
sys.exit(1)
|
||||
|
||||
asyncio.run(main())
|
||||
@@ -0,0 +1,113 @@
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
import random
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
|
||||
from bilibili_api import Credential, sync
|
||||
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format="%(asctime)s [%(levelname)s] %(message)s",
|
||||
datefmt="%Y-%m-%d %H:%M:%S",
|
||||
)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
COOKIE_FILE = Path("cookies.txt")
|
||||
|
||||
def load_credential() -> Credential:
|
||||
"""从 KEY=VALUE 格式的 txt 文件加载 Credential 对象"""
|
||||
if not COOKIE_FILE.exists():
|
||||
logger.error("Cookie file not found: %s", COOKIE_FILE)
|
||||
sys.exit(1)
|
||||
|
||||
try:
|
||||
data = {}
|
||||
with open(COOKIE_FILE, "r") as f:
|
||||
for line in f:
|
||||
line = line.strip()
|
||||
if not line:
|
||||
continue
|
||||
if "=" in line:
|
||||
key, _, value = line.partition("=")
|
||||
data[key.strip()] = value.strip()
|
||||
# 构建 Credential 对象（key 名大小写不敏感）
|
||||
cred = Credential(
|
||||
sessdata=data.get("SESSDATA") or data.get("sessdata"),
|
||||
bili_jct=data.get("bili_jct") or data.get("BILI_JCT"),
|
||||
buvid3=data.get("buvid3") or data.get("BUVID3"),
|
||||
dedeuserid=data.get("dedeuserid") or data.get("DEDEUSERID"),
|
||||
ac_time_value=data.get("ac_time_value") or data.get("AC_TIME_VALUE"),
|
||||
)
|
||||
logger.info("Credential loaded from %s", COOKIE_FILE)
|
||||
return cred
|
||||
except Exception as e:
|
||||
logger.error("Failed to load credential: %s", e)
|
||||
sys.exit(1)
|
||||
|
||||
async def check_and_refresh(cred: Credential) -> bool:
|
||||
"""检查有效性，如果有效则刷新 token。返回是否成功刷新（或仍有效）"""
|
||||
try:
|
||||
valid = await cred.check_valid()
|
||||
if not valid:
|
||||
logger.warning("Credential is invalid (maybe expired). Refresh may fail.")
|
||||
else:
|
||||
logger.info("Credential is valid, proceeding to refresh...")
|
||||
|
||||
# 无论 valid 状态如何，都尝试 refresh（refresh 接口本身会处理）
|
||||
await cred.refresh()
|
||||
logger.info("Token refresh successful. New SESSDATA: %s...", cred.sessdata[:20])
|
||||
# 保存更新后的凭证到文件
|
||||
save_credential(cred)
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.error("Refresh failed: %s", e)
|
||||
return False
|
||||
|
||||
def save_credential(cred: Credential):
|
||||
"""将 credential 的关键字段保存回 txt 文件（KEY=VALUE 格式）"""
|
||||
lines = []
|
||||
if cred.sessdata:
|
||||
lines.append(f"SESSDATA={cred.sessdata}")
|
||||
if cred.bili_jct:
|
||||
lines.append(f"bili_jct={cred.bili_jct}")
|
||||
if cred.buvid3:
|
||||
lines.append(f"buvid3={cred.buvid3}")
|
||||
if cred.dedeuserid:
|
||||
lines.append(f"dedeuserid={cred.dedeuserid}")
|
||||
if cred.ac_time_value:
|
||||
lines.append(f"ac_time_value={cred.ac_time_value}")
|
||||
|
||||
with open(COOKIE_FILE, "w") as f:
|
||||
f.write("\n".join(lines) + "\n")
|
||||
logger.info("Credential saved to %s", COOKIE_FILE)
|
||||
|
||||
async def main_loop():
|
||||
logger.info("Starting token refresh daemon...")
|
||||
cred = load_credential()
|
||||
|
||||
while True:
|
||||
# 随机间隔 50~70 分钟（单位：秒）
|
||||
interval = random.randint(50 * 60, 70 * 60)
|
||||
next_refresh = datetime.now().timestamp() + interval
|
||||
logger.info("Next refresh in %d minutes (at %s)", interval // 60, datetime.fromtimestamp(next_refresh).strftime("%Y-%m-%d %H:%M:%S"))
|
||||
|
||||
await asyncio.sleep(interval)
|
||||
|
||||
logger.info("Refreshing token...")
|
||||
success = await check_and_refresh(cred)
|
||||
if not success:
|
||||
logger.error("Refresh failed, will retry in next cycle.")
|
||||
else:
|
||||
logger.info("Refresh cycle completed.")
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
asyncio.run(main_loop())
|
||||
except KeyboardInterrupt:
|
||||
logger.info("Daemon stopped by user.")
|
||||
except Exception as e:
|
||||
logger.exception("Unexpected error: %s", e)
|
||||
sys.exit(1)
|
||||
@@ -0,0 +1,55 @@
|
||||
from aiohttp import web, ClientSession, ClientTimeout
|
||||
|
||||
|
||||
PIXIV_BACKEND = "http://127.0.0.1:9088"
|
||||
BILIBILI_BACKEND = "http://127.0.0.1:19990"
|
||||
|
||||
|
||||
async def handle_pixiv(request):
|
||||
pid = request.query.get("id", "")
|
||||
if not pid:
|
||||
return web.json_response({"error": "missing id parameter"}, status=400)
|
||||
target = f"{PIXIV_BACKEND}/{pid}"
|
||||
return await proxy(request, target)
|
||||
|
||||
|
||||
async def handle_bilibili(request):
|
||||
qs = request.query_string
|
||||
target = f"{BILIBILI_BACKEND}/bilibili/get_video"
|
||||
if qs:
|
||||
target += "?" + qs
|
||||
return await proxy(request, target)
|
||||
|
||||
|
||||
async def proxy(request, target):
|
||||
timeout = ClientTimeout(total=300)
|
||||
async with ClientSession(timeout=timeout) as session:
|
||||
try:
|
||||
async with session.get(target) as resp:
|
||||
body = await resp.read()
|
||||
headers = {}
|
||||
for key in ("Content-Type", "Content-Disposition", "Cache-Control"):
|
||||
if key in resp.headers:
|
||||
headers[key] = resp.headers[key]
|
||||
return web.Response(
|
||||
body=body,
|
||||
status=resp.status,
|
||||
headers=headers,
|
||||
)
|
||||
except Exception as e:
|
||||
return web.json_response(
|
||||
{"error": "backend unreachable", "detail": str(e)}, status=502
|
||||
)
|
||||
|
||||
|
||||
async def handle_root(request):
|
||||
raise web.HTTPFound("https://rrrosmontis.icu/articles/index.html?id=00003")
|
||||
|
||||
|
||||
app = web.Application()
|
||||
app.router.add_get("/", handle_root)
|
||||
app.router.add_get("/pixiv/get_image", handle_pixiv)
|
||||
app.router.add_get("/bilibili/get_video", handle_bilibili)
|
||||
|
||||
if __name__ == "__main__":
|
||||
web.run_app(app, host="127.0.0.1", port=9888)
|
||||
@@ -0,0 +1,199 @@
|
||||
from flask import Flask, jsonify, send_file
|
||||
import requests
|
||||
import os
|
||||
import json
|
||||
import urllib3
|
||||
from pathlib import Path
|
||||
|
||||
IMAGE_PROXY_DOMAIN = "openapi.rrrosmontis.icu"
|
||||
TOKEN_FILE = "tokens.json"
|
||||
API_HOST = "app-api.pixiv.net"
|
||||
USER_AGENT = "PixivIOSApp/7.13.0 (iOS 14.6; iPhone)"
|
||||
|
||||
urllib3.disable_warnings()
|
||||
|
||||
IMAGE_DIR = Path("/home/pixiv/images")
|
||||
IMAGE_DIR.mkdir(exist_ok=True)
|
||||
|
||||
app = Flask(__name__)
|
||||
|
||||
|
||||
def get_access_token():
|
||||
token_path = Path(__file__).parent / TOKEN_FILE
|
||||
if not token_path.exists():
|
||||
return None
|
||||
try:
|
||||
with open(token_path, 'r', encoding='utf-8') as f:
|
||||
data = json.load(f)
|
||||
return data.get("access_token")
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def fetch_illust_url(pid, page):
|
||||
access_token = get_access_token()
|
||||
if not access_token:
|
||||
return None
|
||||
try:
|
||||
resp = requests.get(
|
||||
f"https://{API_HOST}/v1/illust/detail",
|
||||
params={"illust_id": pid},
|
||||
headers={
|
||||
"Host": API_HOST,
|
||||
"User-Agent": USER_AGENT,
|
||||
"Authorization": f"Bearer {access_token}"
|
||||
},
|
||||
timeout=20,
|
||||
verify=False
|
||||
)
|
||||
if resp.status_code != 200:
|
||||
return None
|
||||
data = resp.json()
|
||||
if data.get("error"):
|
||||
return None
|
||||
illust = data.get("illust")
|
||||
if not illust:
|
||||
return None
|
||||
page_count = illust.get("page_count", 1)
|
||||
if page >= page_count:
|
||||
return None
|
||||
original_url = None
|
||||
if page_count > 1:
|
||||
meta_pages = illust.get("meta_pages")
|
||||
if meta_pages and page < len(meta_pages):
|
||||
original_url = meta_pages[page]["image_urls"]["original"]
|
||||
else:
|
||||
meta_single = illust.get("meta_single_page")
|
||||
if meta_single:
|
||||
original_url = meta_single.get("original_image_url")
|
||||
if not original_url:
|
||||
original_url = illust.get("image_urls", {}).get("large")
|
||||
if not original_url:
|
||||
return None
|
||||
return original_url.replace("i.pximg.net", IMAGE_PROXY_DOMAIN)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def make_re_url(pid, page):
|
||||
if page == 0:
|
||||
return f"https://pixiv.re/{pid}.jpg"
|
||||
else:
|
||||
return f"https://pixiv.re/{pid}_p{page}.jpg"
|
||||
|
||||
|
||||
def make_cat_url(pid, page):
|
||||
if page == 0:
|
||||
return f"https://pixiv.cat/{pid}.jpg"
|
||||
else:
|
||||
return f"https://pixiv.cat/{pid}-{page+1}.jpg"
|
||||
|
||||
|
||||
def check_cache(pid, page):
|
||||
pattern = f"{pid}.*" if page == 0 else f"{pid}-p{page+1}.*"
|
||||
for f in IMAGE_DIR.glob(pattern):
|
||||
ext = f.suffix.lstrip('.')
|
||||
mime = 'image/jpeg'
|
||||
if ext == 'png':
|
||||
mime = 'image/png'
|
||||
elif ext == 'gif':
|
||||
mime = 'image/gif'
|
||||
return f, mime
|
||||
return None, None
|
||||
|
||||
|
||||
def download_image(url, pid, page):
|
||||
cached_path, mime = check_cache(pid, page)
|
||||
if cached_path:
|
||||
return cached_path, mime
|
||||
try:
|
||||
resp = requests.get(
|
||||
url,
|
||||
headers={"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"},
|
||||
timeout=30,
|
||||
verify=False
|
||||
)
|
||||
if resp.status_code != 200:
|
||||
return None, None
|
||||
ct = resp.headers.get('Content-Type', 'image/jpeg')
|
||||
ext = 'jpg'
|
||||
if 'png' in ct:
|
||||
ext = 'png'
|
||||
elif 'gif' in ct:
|
||||
ext = 'gif'
|
||||
filename = f"{pid}.{ext}" if page == 0 else f"{pid}-p{page+1}.{ext}"
|
||||
tmp = IMAGE_DIR / (filename + '.tmp')
|
||||
final = IMAGE_DIR / filename
|
||||
with open(tmp, 'wb') as f:
|
||||
f.write(resp.content)
|
||||
os.replace(tmp, final)
|
||||
return final, ct
|
||||
except Exception:
|
||||
return None, None
|
||||
|
||||
|
||||
@app.route('/<path:path>')
|
||||
def serve_image(path):
|
||||
import re
|
||||
m = re.match(r'^(\d+)(?:-(\d+))?(?:\.\w+)?$', path)
|
||||
if not m:
|
||||
return jsonify(error="Invalid path format, expected /{pid} or /{pid}-{page}"), 400
|
||||
pid = int(m.group(1))
|
||||
page = int(m.group(2)) - 1 if m.group(2) else 0
|
||||
|
||||
cached_file, mime = check_cache(pid, page)
|
||||
if cached_file:
|
||||
return send_file(cached_file, mimetype=mime, max_age=3600)
|
||||
|
||||
urls_to_try = []
|
||||
official_url = fetch_illust_url(pid, page)
|
||||
if official_url:
|
||||
urls_to_try.append(official_url)
|
||||
urls_to_try.append(make_re_url(pid, page))
|
||||
urls_to_try.append(make_cat_url(pid, page))
|
||||
|
||||
for url in urls_to_try:
|
||||
img_path, mime = download_image(url, pid, page)
|
||||
if img_path:
|
||||
return send_file(img_path, mimetype=mime, max_age=3600)
|
||||
|
||||
return jsonify(error="Failed to fetch image from all sources"), 500
|
||||
|
||||
|
||||
@app.route('/health')
|
||||
def health():
|
||||
access_token = get_access_token()
|
||||
if not access_token:
|
||||
return jsonify(status="error", detail="tokens.json missing or invalid")
|
||||
try:
|
||||
resp = requests.get(
|
||||
f"https://{API_HOST}/v1/illust/detail",
|
||||
params={"illust_id": 105383871},
|
||||
headers={
|
||||
"Host": API_HOST,
|
||||
"User-Agent": USER_AGENT,
|
||||
"Authorization": f"Bearer {access_token}"
|
||||
},
|
||||
timeout=15,
|
||||
verify=False
|
||||
)
|
||||
if resp.status_code == 200:
|
||||
return jsonify(status="ok", token="valid")
|
||||
else:
|
||||
return jsonify(status="error", detail=f"API returned {resp.status_code}")
|
||||
except Exception as e:
|
||||
return jsonify(status="error", detail=str(e))
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
token_path = Path(__file__).parent / TOKEN_FILE
|
||||
if not token_path.exists():
|
||||
print(f"Warning: {TOKEN_FILE} not found, access_token may be missing")
|
||||
else:
|
||||
token = get_access_token()
|
||||
if token:
|
||||
print("Access token loaded")
|
||||
else:
|
||||
print("Warning: access_token missing in tokens.json")
|
||||
print(f"Cache directory: {IMAGE_DIR}")
|
||||
app.run(host='0.0.0.0', port=9088, debug=False)
|
||||
@@ -0,0 +1,4 @@
|
||||
{
|
||||
"access_token": "<PIXIV_ACCESS_TOKEN>",
|
||||
"refresh_token": "<PIXIV_REFRESH_TOKEN>"
|
||||
}
|
||||
Reference in New Issue
Block a user