# -*- coding: utf-8 -*- # pixiv.py — Pixiv 单图下载模块 # 流程: 下载(镜像/代理) -> 本地缓存 -> QQ 直发图片 -> 留存(oss/上传cf) -> sleep -> 清缓存 # SEND_IMAGE=0 时仅返回镜像 URL, 不下载不上传 import asyncio import base64 import hashlib import json import os from pathlib import Path BASE_DIR = os.path.dirname(os.path.abspath(__file__)) import re import shutil import threading import time from pathlib import Path # ---------- 配置 ---------- SEND_IMAGE = 1 # 1=启用: 下载->缓存->上传cf->删缓存->直发图片; 0=禁用: 仅返回镜像URL IMAGE_FETCH_MODE = "mirror" # 图片获取模式 (SEND_IMAGE=1 时生效): # "mirror" = 从 openapi.rrrosmontis.icu 镜像站获取 # "pximg" = 走本地代理获取 pixiv.re 源站原图 (获取失败时回退镜像站) # 2026-09-12: pixiv.re 全站 500 (hk 干净 IP 亦 500), 已停止尝试 pixiv.re, # 固定走镜像站; 镜像下载仍绕行本地代理 (bj->hk 直连实测仅 ~8KB/s) # 图片留存模式 (SEND_IMAGE=1 时生效): # "oss" = 保存到本机 C:/oss/pixiv, 不上传 cf # "upload" = 上传 cf /mnt/A/pixiv 存档 RETAIN_MODE = "oss" OSS_SAVE_DIR = os.environ.get("OSS_SAVE_DIR", "/mnt/oss/pixiv/images") # 运行 napcat 的机器上是否有可用的同一 oss 挂载 (仅 RETAIN_MODE="oss" 时读取): # True = NapCat 与 oss 同机/同一挂载 (如 cf 上 ossfs2 挂载 /mnt/A), 发图直接引用 # NAPCAT_SERVER_OSS_IMAGES_PATH 下 {pid}{后缀} 的本地路径, 不走 base64 内联 # False = 跨机器, 使用 base64 内联发图 IS_NAPCAT_SERVER_OSS_AVAILABLE = True NAPCAT_SERVER_OSS_IMAGES_PATH = os.environ.get("NAPCAT_OSS_IMAGES_PATH", "/mnt/oss/pixiv/images") PROXY_CONFIG_PATH = os.environ.get("BOT_PROXY_CONFIG", os.path.join(BASE_DIR, "proxies.json")) PIXIV_CACHE_DIR = os.path.join(BASE_DIR, "pixiv", "cache") CF_BASE = "https://pan.rrrosmontis.icu" # cf 文件服务器 CF_PASSWORD = os.environ.get('PIXIV_CF_PASSWORD', '') # fileserver 登录密码, 提交前做 SHA-256 CF_UPLOAD_DIR = "/mnt/A/pixiv" # cf 目标目录 IMAGE_MIRROR = "https://openapi.rrrosmontis.icu/pixiv/get_image?id={pid}" # 公开镜像 (经 hk pixiv 代理) SOURCE_IMAGE = "https://pixiv.re/{pid}.jpg" # 已弃用: 2026-09-12 起 pixiv.re 全站 500, 默认模式已固定 mirror SOURCE_PAGE = "https://www.pixiv.net/artworks/{pid}" HTTP_TIMEOUT = 60 SEND_AS_FILE_THRESHOLD = 20 * 1024 * 1024 # QQ 侧: 超过该字节数改发镜像链接 TG_DOC_LIMIT = 50 * 1024 * 1024 # TG 侧: Bot API send_document 上限 50MB HELP_TEXT = ( "Pixiv 单图\n" "━━━━━━━━━━━━\n" "• .pixiv 获取指定 PID 作品并发送链接/图片\n" "示例: .pixiv 12345678\n" "━━━━━━━━━━━━\n" "尽可能直接发图，否则返回镜像链接" ) # ---------- 工具 ---------- def _load_proxy_ports(): """读取 proxies.json 中的本地代理端口 (与 jm/setu 共用 xray 本地 http 端口)""" try: with open(PROXY_CONFIG_PATH, encoding="utf-8") as f: data = json.load(f) return [f"http://127.0.0.1:{p['localPort']}" for p in data.get("proxies", []) if p.get("localPort")] except Exception as e: print(f"[pixiv] 代理列表加载失败: {e}") return [] def _sanitize_pid(text): m = re.fullmatch(r"(\d{1,12})", text.strip()) return m.group(1) if m else None # 按 PID 的并发锁: setu/pixiv 共用本缓存目录, 同一 PID 并发任务须串行 # (同名缓存文件互写损坏 / 一方 cleanup 删掉另一方仍在使用的文件) _pid_locks = {} # pid -> asyncio.Lock _pid_locks_guard = threading.Lock() def get_pid_lock(pid): """取指定 PID 的 asyncio.Lock (下载->发送->清理 全段持锁, 同 PID 串行, 不同 PID 并行)""" key = str(pid) with _pid_locks_guard: lock = _pid_locks.get(key) if lock is None: lock = asyncio.Lock() _pid_locks[key] = lock return lock # ---------- .pixiv 帮助后的 PID 监听 ---------- # .pixiv / /pixiv（无参数）弹出帮助后, 监听该用户在本会话的下一条消息: # 纯数字 → 当作 PID 直接取图; 其它任何消息（含其它指令）→ 取消监听, 正常处理。 # 一次性监听, 超时 PENDING_PID_TTL 秒自动失效。 PENDING_PID_TTL = 120 _pending_pid = {} # (channel_name, msg_type, target_id, user_id) -> expires_ts _pending_pid_guard = threading.Lock() def arm_pid_listener(channel, msg_type, target_id, user_id): """登记一次性 PID 监听（同一用户/会话重复登记覆盖旧值）""" key = (getattr(channel, "name", ""), msg_type, int(target_id), int(user_id)) now = time.time() with _pending_pid_guard: # 顺手清理过期项 for k, v in list(_pending_pid.items()): if v <= now: del _pending_pid[k] _pending_pid[key] = now + PENDING_PID_TTL def try_consume_pid(channel, msg_type, target_id, user_id, text): """dispatch 在每条消息进指令分派前调用一次。 返回 True 表示该消息已被当作 PID 取图处理（调用方应 return）。""" key = (getattr(channel, "name", ""), msg_type, int(target_id), int(user_id)) with _pending_pid_guard: expires = _pending_pid.pop(key, None) # 一次性: 任何消息都结束监听 if expires is None: return False if time.time() > expires: return False text = (text or "").strip() if not re.fullmatch(r"\d{1,12}", text): return False asyncio.create_task(handle_pixiv(channel, msg_type, target_id, user_id, f".pixiv {text}")) return True def _image_url(pid): return IMAGE_MIRROR.format(pid=pid) def _download_failure_text(pid, error): if isinstance(error, PixivRestrictedError): return (f"PID {pid} 是 MyPixiv 限定作品，OpenAPI 没有返回原图。\n" f"请登录 Pixiv 后查看: {SOURCE_PAGE.format(pid=pid)}") return (f"PID {pid} 作品不存在或已删除\n" f"可访问源站确认: {SOURCE_PAGE.format(pid=pid)}") def image_base64(local_path): """读取本地图片, 返回 base64:// 数据 (供 NapCat 跨机器发图, 不依赖本地路径)""" with open(local_path, "rb") as f: data = base64.b64encode(f.read()).decode("ascii") return "base64://" + data # ---------- 缓存/上传 (setu 模块复用) ---------- class PixivNotFoundError(Exception): """作品不存在或已删除（镜像后端返回 HTTP 404 not_found）。 收到此错误应立即终止所有代理节点轮询——重试无意义。""" class PixivRestrictedError(Exception): """OpenAPI 返回 MyPixiv 限定占位图，而非作品原图。""" RESTRICTED_IMAGE_SHA256 = frozenset({ "caa0bd60e0a0c28a4bd80b6b7024fb4956ffff7c536f4fe9d2d2588b660524c7", }) def download_image(pid, image_url=None, proxy=None): """下载图片到本地缓存, 返回本地路径。失败抛异常 image_url: 指定下载源 (默认 openapi.rrrosmontis.icu 镜像); proxy: http 代理地址, 如 http://127.0.0.1:10810""" import requests url = image_url or _image_url(pid) kwargs = {"timeout": HTTP_TIMEOUT, "stream": True} if proxy: kwargs["proxies"] = {"http": proxy, "https": proxy} # pximg 源站要求 pixiv Referer, 否则 403; # requests 默认 UA (python-requests/...) 会被 i.pximg.net WAF 403, 必须带浏览器 UA (2026-09-02) kwargs["headers"] = { "Referer": "https://www.pixiv.net/", "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", } resp = requests.get(url, **kwargs) try: if resp.status_code == 404: # 镜像后端语义化错误码：作品不存在/已删除（不存在的作品换节点重试无意义） raise PixivNotFoundError(f"作品不存在或已删除 (PID {pid})") if resp.status_code != 200: raise RuntimeError(f"镜像下载失败 HTTP {resp.status_code}") ctype = resp.headers.get("Content-Type", "").split(";")[0].strip().lower() ext = {"image/jpeg": ".jpg", "image/png": ".png", "image/gif": ".gif", "image/webp": ".webp"}.get(ctype, ".jpg") cache_dir = Path(PIXIV_CACHE_DIR) cache_dir.mkdir(parents=True, exist_ok=True) save_path = cache_dir / f"{pid}{ext}" try: digest = hashlib.sha256() with open(save_path, "wb") as f: for chunk in resp.iter_content(65536): if chunk: f.write(chunk) digest.update(chunk) if digest.hexdigest() in RESTRICTED_IMAGE_SHA256: cleanup_cache(save_path) raise PixivRestrictedError(f"PID {pid} 为 MyPixiv 限定作品，OpenAPI 未提供原图") return save_path except BaseException: # Do not leave a truncated download for a later proxy attempt. cleanup_cache(save_path) raise finally: resp.close() # stream=True 的连接必须显式关闭, 否则失败路径泄漏连接 def upload_to_cf(local_path): """登录 fileserver 并上传图片到 /mnt/A/pixiv, 返回 (ok, msg)""" if not CF_PASSWORD: return False, "未配置 PIXIV_CF_PASSWORD，无法上传" import requests sess = requests.Session() try: pw_hash = hashlib.sha256(CF_PASSWORD.encode("utf-8")).hexdigest() r = sess.post(f"{CF_BASE}/api/login", json={"password": pw_hash}, timeout=HTTP_TIMEOUT) try: j = r.json() except Exception: return False, f"登录响应异常: HTTP {r.status_code}" if not j.get("success"): return False, f"登录失败: {j.get('error', '未知')}" b64 = lambda s: base64.b64encode(s.encode("utf-8")).decode("ascii") with open(local_path, "rb") as f: r = sess.post( f"{CF_BASE}/api/upload", data={"currentPath": b64(CF_UPLOAD_DIR), "fileName": b64(local_path.name)}, files={"file": (local_path.name, f)}, timeout=300, ) try: j = r.json() except Exception: return False, f"上传响应异常: HTTP {r.status_code}" if j.get("success"): return True, "ok" return False, str(j.get("error", "未知错误")) finally: sess.close() def cleanup_cache(local_path): """删除单次缓存文件""" try: if local_path and os.path.exists(local_path): os.remove(local_path) except Exception: pass def clear_cache(): """清空本地缓存目录, 返回提示文本""" d = Path(PIXIV_CACHE_DIR) if not d.exists(): return "本地缓存目录不存在, 无需清理" total = 0 count = 0 for p in d.rglob("*"): if p.is_file(): count += 1 try: total += p.stat().st_size except OSError: pass import shutil shutil.rmtree(d, ignore_errors=True) d.mkdir(parents=True, exist_ok=True) return f"已清理 pixiv 缓存: {count} 个, 释放 {total / 1024 / 1024:.1f} MB" # ---------- 指令入口 ---------- async def handle_pixiv(websocket, message_type, target_id, user_id, raw_text): from channel import send_message body = re.sub(r"^\s*\.pixiv\b", "", raw_text, flags=re.IGNORECASE).strip() if not body or body.lower() in ("help", "-h"): await send_message(websocket, message_type, target_id, HELP_TEXT + "\n━━━━━━━━━━━━\n也可以在激活此帮助后跟进发送 PID 以获取图片") arm_pid_listener(websocket, message_type, target_id, user_id) return if body.lower() in ("del-cache", "delcache", "clearcache", "clear"): if not websocket.is_admin(user_id): await send_message(websocket, message_type, target_id, "该指令仅管理员可用") return await send_message(websocket, message_type, target_id, await asyncio.to_thread(clear_cache)) return pid = _sanitize_pid(body) if not pid: await send_message(websocket, message_type, target_id, "格式: .pixiv \n示例: .pixiv 12345678") return if not SEND_IMAGE: await send_message(websocket, message_type, target_id, f"Pixiv PID {pid}: \n{_image_url(pid)}\n\n" f"源站: {SOURCE_PAGE.format(pid=pid)}") return await send_message(websocket, message_type, target_id, f"正在获取图片 (PID {pid}), 请稍候...") # 同一 PID 并发任务串行: 下载->发送->清理 全段持锁 (与 setu 共用缓存目录, 防同名互写/互删) pid_lock = get_pid_lock(pid) await pid_lock.acquire() try: # 图片获取模式: "mirror"=openapi.rrrosmontis.icu 镜像; "pximg"=走本地代理取 pixiv.re 源站原图 (失败回退镜像) download_url = None # 代理列表恒定加载 (2026-09-12 修正): 原先只在 pximg 模式加载, mirror 模式下 proxy_list 为空, # 会跳过"镜像绕行本地代理"直接落到 bj->hk 直连兜底 (~8KB/s), 2MB 原图必定 Read timed out。 proxy_list = _load_proxy_ports() if IMAGE_FETCH_MODE == "pximg": download_url = SOURCE_IMAGE.format(pid=pid) if download_url: print(f"[pixiv] 模式 pximg: 源站 {download_url}, 代理节点 {len(proxy_list)} 个") else: print(f"[pixiv] 模式 mirror: 从 openapi.rrrosmontis.icu 镜像获取, 代理节点 {len(proxy_list)} 个") local_path = None last_err = None not_found = False if download_url and proxy_list: for proxy_url in proxy_list: try: local_path = await asyncio.to_thread( download_image, pid, download_url, proxy_url) break except (PixivNotFoundError, PixivRestrictedError) as e: # 作品不存在是确定性结果：终止全部轮询，直接告知用户 not_found = True last_err = e print(f"[pixiv] 作品不存在 (PID {pid})，终止代理轮询") break except Exception as e: print(f"[pixiv] pximg 代理下载失败 ({proxy_url}): {e}") last_err = e if not_found: await send_message(websocket, message_type, target_id, _download_failure_text(pid, last_err)) return if local_path is None: if download_url: print(f"[pixiv] pximg 源站获取失败, 回退镜像站: {last_err}") # 镜像站优先绕行本地代理下载 (2026-09-02): 实测 bj->hk 直连跨境仅 ~8KB/s, # 大图 (如 11MB 原图) 必触发 IncompleteRead 断流; 经机场节点中转 ~200KB/s 稳定。 # 全部代理失败后再试直连兜底。 mirror_url = _image_url(pid) if proxy_list: for proxy_url in proxy_list: try: local_path = await asyncio.to_thread( download_image, pid, mirror_url, proxy_url) break except (PixivNotFoundError, PixivRestrictedError) as e: not_found = True last_err = e print(f"[pixiv] 作品不存在 (PID {pid})，终止镜像轮询") break except Exception as e: print(f"[pixiv] 镜像代理下载失败 ({proxy_url}): {e}") last_err = e if not_found: await send_message(websocket, message_type, target_id, _download_failure_text(pid, last_err)) return if local_path is None: try: local_path = await asyncio.to_thread(download_image, pid) except (PixivNotFoundError, PixivRestrictedError) as e: await send_message(websocket, message_type, target_id, _download_failure_text(pid, e)) return except Exception as e: print(f"[pixiv] 下载失败 {pid}: {e}") await send_message(websocket, message_type, target_id, f"PID {pid} 下载失败: {str(e)[:100]}\n" f"可尝试自行访问: {SOURCE_PAGE.format(pid=pid)}") return # 发图: oss 模式且 NapCat 与 oss 同机挂载时, 先留存到 oss 再引用本地路径发图 (不走 base64 内联); # 超过阈值则直接返回镜像站链接 (不内嵌 base64, 避免超限) use_oss_local_path = (RETAIN_MODE == "oss" and IS_NAPCAT_SERVER_OSS_AVAILABLE) oss_retained = False if use_oss_local_path: # 先留存: bj 侧 C:/oss (rclone 挂载) 与 cf 侧 /mnt/A (ossfs2 挂载) 为同一 OSS, # 复制完成后 NapCat 即可从 NAPCAT_SERVER_OSS_IMAGES_PATH 读到该文件 try: os.makedirs(OSS_SAVE_DIR, exist_ok=True) shutil.copyfile(str(local_path), os.path.join(OSS_SAVE_DIR, os.path.basename(str(local_path)))) print(f"[pixiv] 已留存 OSS: {os.path.basename(str(local_path))}") oss_retained = True except Exception as e: print(f"[pixiv] OSS 留存失败 pid={pid}: {e}") # TG 上限 50MB (send_document): 20~50MB 走文件通道; QQ 侧维持 20MB 改发链接 _is_tg = getattr(websocket, "name", "") == "tg" _size_limit = TG_DOC_LIMIT if _is_tg else SEND_AS_FILE_THRESHOLD if os.path.getsize(local_path) > _size_limit: await send_message(websocket, message_type, target_id, f"Pixiv PID {pid}: \n\n{_image_url(pid)}\n\n(图片过大请手动下载)") elif use_oss_local_path and oss_retained: # 注意: NapCat 只认 file: 前缀的本地路径 (checkUriType 源码), 裸绝对路径会报"识别URL失败"; # 用正斜杠拼接 (Windows os.path.join 会产出反斜杠, NapCat 在 Linux 上无法读取) oss_path = (Path(NAPCAT_SERVER_OSS_IMAGES_PATH) / os.path.basename(str(local_path))).as_uri() await asyncio.sleep(10) # 等 rclone 写缓存同步到 OSS (实测 4MB 约 6.3s), 确保 NapCat 能读到新文件 await send_message(websocket, message_type, target_id, [ {"type": "text", "data": {"text": f"Pixiv PID {pid}: "}}, {"type": "image", "data": {"file": oss_path}}, ]) else: img_b64 = await asyncio.to_thread(image_base64, local_path) await send_message(websocket, message_type, target_id, [ {"type": "text", "data": {"text": f"Pixiv PID {pid}: "}}, {"type": "image", "data": {"file": img_b64}}, ]) # 留存 (oss 同机已留存则跳过; 其余走 oss 本机目录 / upload=cf), 再 sleep 等待 NapCat 读图, 最后清缓存 if not oss_retained: if RETAIN_MODE == "oss": try: os.makedirs(OSS_SAVE_DIR, exist_ok=True) shutil.copyfile(str(local_path), os.path.join(OSS_SAVE_DIR, os.path.basename(str(local_path)))) print(f"[pixiv] 已留存 OSS: {os.path.basename(str(local_path))}") except Exception as e: print(f"[pixiv] OSS 留存失败 pid={pid}: {e}") await send_message(websocket, message_type, target_id, f"图片已发送, 但 OSS 留存失败: {str(e)[:80]}") else: try: up_ok, up_msg = await asyncio.to_thread(upload_to_cf, local_path) except Exception as e: up_ok, up_msg = False, str(e) if not up_ok: print(f"[pixiv] 上传失败 pid={pid}: {up_msg}") await send_message(websocket, message_type, target_id, f"图片已发送, 但存档上传失败: {up_msg[:80]}") await asyncio.sleep(30) # 等待 NapCat 读取本地缓存文件 (asyncio.sleep 让出事件循环, 不阻塞主线程) finally: try: if local_path is not None: # Sending, retention and cancellation must not strand cache files. await asyncio.to_thread(cleanup_cache, local_path) finally: pid_lock.release()