title / plot / number / genre / performer"] --> B["2. AI understands intent
intent · target · source · explain"] + B --> C["3. Reply 🔍 Searching: target"] + C --> D["4. Select the best crawler / API"] + D --> D1{intent} + D1 -->|number| E["Search by number
index sites"] + D1 -->|performer| F["Performer search"] + D1 -->|title / genre / plot| G["Keyword / BT search"] + E --> H["5. Reply ⏳ Searching ..."] + F --> H + G --> H + H --> I["6. Return magnet links
or BT alternatives"] + I --> J["7. Reply 📄 Search results"] + J --> K["8. Save best magnet to PikPak"] + K --> L["9. Reply ✅ Saved successfully"] +``` + +## Project structure + +``` +bot.py Main program: config, logging, message handlers, AI flow +ai.py AI intent client using OpenAI-compatible /chat/completions +pikpak.py PikPak official OpenAPI client for login, token refresh, and offline download +database.py Data layer: BotFileDb (JSON records) + BotCacheDb (SQLite cache) +requirements.txt Dependencies +docker-compose.yml One-click deployment +``` + +Runtime data is stored under `~/.tg_search_bot/`: `config.yaml`, `record.json`, `cache.db`, `pikpak_token.json`, and `log.txt`. + +## Usage + +### 1. Configure + +Edit `~/.tg_search_bot/config.yaml`: + +```yaml +# Required: Telegram chat ID +tg_chat_id: +# Required: Telegram bot token +tg_bot_token: +# Required: use global proxy, 1 = yes, 0 = no +use_proxy: +# Optional: proxy address, required when use_proxy is 1 +proxy_addr: +# Required: enable NSFW content, 1 = yes, 0 = no +enable_nsfw: 0 +# Optional: PikPak account for auto-saving magnets +pikpak_username: +pikpak_password: +# Optional: AI natural-language search via any OpenAI-compatible API +ai_base_url: +ai_api_key: +ai_model: +``` + +### 2. Run + +```sh +# Option 1: Docker deployment +docker-compose up -d + +# Option 2: run directly (Python 3.9+) +pip install -r requirements.txt +python3 bot.py +``` + +### 3. Commands + +| Command | Description | +| --- | --- | +| `/help` | Show help | +| `/stars` | View collected performers | +| `/ids` | View collected numbers | +| `/record` | Export the records file | + +Simply send a movie title, keyword, number, plot summary, performer name, or genre. The AI will understand and search automatically. + +### 4. Multi-turn follow-up + +BT search results are paginated in groups of 5. You can continue with: + +- `next batch` / `next page` / `more` — show the next page +- `previous page` — show the previous page +- `save item N` or `save N` — save the N-th magnet to PikPak + +## Development + +Python 3.9+ is recommended, preferably with a virtual environment: + +```sh +git clone https://github.com/akynazh/tg-search-bot.git +cd tg-search-bot +python3 -m venv .venv +source ./.venv/bin/activate +pip3 install -r requirements.txt +``` + +## Thanks + + +
+
+Thanks to JetBrains for supporting this project.
diff --git a/telegram/tg-search-bot/README.zh.md b/telegram/tg-search-bot/README.zh.md
new file mode 100644
index 0000000..5ecf640
--- /dev/null
+++ b/telegram/tg-search-bot/README.zh.md
@@ -0,0 +1,143 @@
+# tg-search-bot
+
+[English](README.md)｜[中文](README.zh.md)
+
+一个基于 Python3 的 Telegram 资源搜索机器人：搜索各类视频磁力链接，支持收藏、导出记录、自动保存到网盘，可配置 NSFW 开关与代理上网；同时集成 AI 自然语言理解能力，可直接根据用户自然语言意图自动识别搜索目标、类型与数据源。
+
+- 数据来源：TorrentKitty（中文）+ apibay（英文）+ 影视磁力索引 API（Jvav）
+- 网盘：Pikpak 官方 OpenAPI
+- AI：任意 OpenAI 兼容接口，支持意图识别与自动搜索
+
+## 功能
+
+- 按编号 / 演职人员 / 关键词搜索影片，返回封面、评分、日期、标签、演职人员及磁力链接（支持 HD / 字幕等过滤）
+- 获取预览视频、完整视频与影片截图
+- 收藏与导出记录（`record.json`）
+- 排行榜、随机高分 / 最新影片
+- 自动把最优磁力链接保存到 Pikpak（登录 token 直传离线任务）
+- AI 自然语言理解搜索：识别编号 / 演职人员 / 片名 / 剧情 / 类型并自动选择数据源
+- BT 种子搜索（中文 / 英文双引擎），结果分页
+- 多轮追问：`换一批` / `下一页` / `上一页` / `保存第N个`
+- 可配置 NSFW 开关与代理上网
+
+## 核心流程（AI 自然语言搜索）
+
+搜索流程：
+
+1. 用户输入想要的资源 / 资源相关信息（电影、电视剧名称、影片编号、剧情 / 对话、演职人员、影片类型）
+2. AI 根据输入理解用户目标（根据剧情推测片名、根据类型推荐等），输出 `intent / target / source / explain`
+3. 机器人回复：`🔍 搜索：{目标}`
+4. AI 选择最合适的爬虫 / API 进行搜索（编号 → 编号索引站，演职人员 → 演职人员搜索，其余 → 关键词或 BT）
+5. 机器人回复：`⏳ 搜索中 ...`
+6. 爬虫 / API 搜索得到磁力链接（没有磁力时返回 BT 种子等其他格式内容）
+7. 机器人回复：`📄 搜索结果`
+8. 程序将最优磁力链接保存到已配置的网盘（PikPak）
+9. 机器人回复：`✅ 保存结果：已提交离线任务到 PikPak`
+
+> 未配置 AI 接口时自动回退到原来的编号识别 + BT 关键词搜索逻辑。
+
+```mermaid
+flowchart TD
+ A["① 用户输入电影 / 电视剧 / 编号 / 剧情 / 演职人员 / 类型"] --> B["② AI 理解意图
intent · target · source · explain"] + B --> C["③ 回复 🔍 搜索：目标"] + C --> D["④ 选择最合适的爬虫 / API"] + D --> D1{意图 intent} + D1 -->|编号| E["按编号搜索
编号索引站"] + D1 -->|演职人员| F["演职人员搜索"] + D1 -->|关键词 / 类型 / 剧情| G["关键词 → BT 种子"] + E --> H["⑤ 回复 ⏳ 搜索中"] + F --> H + G --> H + H --> I["⑥ 搜索得到磁力链接
（无磁力则返回 BT 等其他格式）"] + I --> J["⑦ 回复 📄 搜索结果"] + J --> K["⑧ 保存最优磁力链接到 PikPak"] + K --> L["⑨ 回复 ✅ 保存结果"] +``` + +## 目录结构 + +``` +bot.py 主程序：配置、日志、消息 / 回调处理、AI 搜索流程 +ai.py AI 意图理解客户端（OpenAI 兼容 /chat/completions） +pikpak.py Pikpak 官方 OpenAPI 客户端（登录 / 刷新 token、离线下载） +database.py 数据层：BotFileDb（JSON 收藏）+ BotCacheDb（SQLite 缓存） +requirements.txt 依赖 +docker-compose.yml 一键部署 +``` + +配置与数据文件存放于 `~/.tg_search_bot/`：`config.yaml`（主配置）、`record.json`（收藏）、`cache.db`（缓存）、`pikpak_token.json`（Pikpak 登录态）、`log.txt`（日志）。 + +## 使用 + +### 1. 配置 + +编辑 `~/.tg_search_bot/config.yaml`： + +```yaml +# 必填，你的 Telegram chat id +tg_chat_id: +# 必填，你的 Telegram bot token +tg_bot_token: +# 必填，是否启用全局代理，1 是 / 0 否 +use_proxy: +# 可选，代理地址（use_proxy 为 1 时必填），如 http://127.0.0.1:7890 +proxy_addr: +# 必填，是否开启 NSFW 内容，1 是 / 0 否 +enable_nsfw: 0 +# 可选，Pikpak 账号（用于自动保存磁力到网盘，官方 API） +pikpak_username: +pikpak_password: +# 可选，AI 自然语言搜索（任意 OpenAI 兼容接口） +ai_base_url: +ai_api_key: +ai_model: +``` + +### 2. 运行 + +```sh +# 方式一：Docker 一键部署 +docker-compose up -d + +# 方式二：直接运行（Python >= 3.9） +pip install -r requirements.txt +python3 bot.py +``` + +### 3. 命令 + +| 命令 | 说明 | +| --- | --- | +| `/help` | 查看帮助 | +| `/stars` | 查看已收藏演职人员 | +| `/ids` | 查看已收藏编号 | +| `/record` | 导出收藏记录文件 | + +直接发送电影名 / 关键词 / 编号 / 剧情 / 演员名 / 类型即可，AI 会自动识别并搜索。 + +### 4. 多轮追问 + +BT 搜索结果每页 5 条，可继续回复： + +- `换一批` / `下一页` / `还有吗` — 下一页 +- `上一页` — 上一页 +- `保存第N个`（或 `save N`）— 把第 N 条磁力保存到 Pikpak + +## 开发 + +推荐使用 Python 3.9，并建议使用虚拟环境： + +```sh +git clone https://github.com/akynazh/tg-search-bot.git +cd tg-search-bot +python3 -m venv .venv +source ./.venv/bin/activate +pip3 install -r requirements.txt +``` + +## 致谢 + + +
+
+感谢 JetBrains 对这个项目的支持！
diff --git a/telegram/tg-search-bot/ai.py b/telegram/tg-search-bot/ai.py
new file mode 100644
index 0000000..e4aec46
--- /dev/null
+++ b/telegram/tg-search-bot/ai.py
@@ -0,0 +1,135 @@
+# -*- coding: UTF-8 -*-
+"""AI 意图理解客户端。
+
+对接任意 OpenAI 兼容接口（可自定义 base_url），将用户的自然语言输入解析为
+结构化的搜索意图，供机器人选择合适的爬虫 / API 进行资源搜索。
+"""
+
+import json
+import logging
+import re
+
+import requests
+
+LOG = logging.getLogger(__name__)
+
+SYSTEM_PROMPT = """你是一个 Telegram 资源搜索机器人的「意图理解」模块。用户会用自然语言描述想要的资源，可能包括：
+- AV 番号（如 ABP-123、FC2-PPV-12345）
+- 演员名字
+- 电影 / 电视剧名称
+- 一段剧情、台词或影片描述
+- 电影类型 / 题材（如「悬疑电影」「科幻片」）
+
+请解析用户输入，并只输出一个 JSON 对象（不要输出任何其他文字），字段如下：
+{
+ "intent": "av_id | actor | genre | keyword | plot | general",
+ "target": "解析后的搜索目标（番号 / 演员名 / 关键词 / 推测出的片名）",
+ "source": "javdb | javbus | sukebei | dmm | apibay | auto",
+ "explain": "用一句话说明你理解到的搜索目标"
+}
+
+规则：
+- intent=av_id：输入是明确的番号；target 保持番号原样（英文字母大写、连字符保留）
+- intent=actor：仅当输入是明确的日本 AV 演员 / 演员名（例如「波多野结衣」「桥本有菜」「三上悠亜」）；target 必须是最可能的正式演员名，不要直接原样照抄用户输入；如果用户把演员名打错、同音字、错别字、简写、日文假名近似、中文音译偏差，优先纠正成真正的日本 AV 演员名；如果用户说的是普通成人明星、非 AV 演员、或无法确定是日本 AV 演员，则不要使用 actor，改用 keyword / general
+- intent=genre：输入是电影类型 / 题材；target 为该类型关键词（如「悬疑」）
+- intent=plot：输入是剧情 / 台词 / 描述；target 为推测出的最可能片名（不确定也给一个最接近的）
+- intent=keyword：输入是电影 / 电视剧名称或其他明确关键词；target 为精简后的关键词
+- intent=general：无法明确分类时使用；target 为原始输入的精简版
+- source 表示最合适的数据源：AV 相关用 javdb/javbus/sukebei/dmm，电影 / 电视剧 / 通用关键词用 apibay，拿不准用 auto
+- target 尽量精简，去掉语气词和多余描述
+- 特别注意：演员搜索是“纠错 + 归一化”任务，不要因为用户拼错字或音近字就保留错误名字；例如“波多野结衣”可能写成“波多野结衣”“娑多野结衣”，应修正到正确正式名；“三上悠亚”可能写成“三上悠亜/三上悠亚”都应统一成正式名
+"""
+
+
+class AiClient:
+ def __init__(
+ self,
+ base_url: str = "",
+ api_key: str = "",
+ model: str = "gpt-4o-mini",
+ proxy: str = "",
+ timeout: int = 60,
+ ):
+ self.base_url = (base_url or "").rstrip("/")
+ self.api_key = api_key or ""
+ self.model = model or "gpt-4o-mini"
+ self.proxy = proxy or ""
+ self.proxies = {"http": self.proxy, "https": self.proxy} if self.proxy else None
+ self.timeout = timeout
+
+ @property
+ def enabled(self):
+ return bool(self.base_url and self.api_key)
+
+ def chat(self, messages, temperature=0.2, max_tokens=800):
+ if not self.enabled:
+ return None
+ url = f"{self.base_url}/chat/completions"
+ headers = {
+ "Authorization": f"Bearer {self.api_key}",
+ "Content-Type": "application/json",
+ }
+ payload = {
+ "model": self.model,
+ "messages": messages,
+ "temperature": temperature,
+ "max_tokens": max_tokens,
+ }
+ try:
+ resp = requests.post(
+ url,
+ headers=headers,
+ json=payload,
+ proxies=self.proxies,
+ timeout=self.timeout,
+ )
+ except Exception as e:
+ LOG.error(f"AI request failed: {e}")
+ return None
+ if resp.status_code != 200:
+ LOG.error(
+ f"AI request failed with status {resp.status_code}: {resp.text[:300]}"
+ )
+ return None
+ try:
+ data = resp.json()
+ return data["choices"][0]["message"]["content"]
+ except Exception as e:
+ LOG.error(f"AI response parse failed: {e}")
+ return None
+
+ def understand(self, text: str):
+ """将自然语言输入解析为搜索意图 dict，失败返回 None。"""
+ if not self.enabled:
+ return None
+ messages = [
+ {"role": "system", "content": SYSTEM_PROMPT},
+ {"role": "user", "content": text},
+ ]
+ content = self.chat(messages)
+ if not content:
+ return None
+ LOG.info(f"AI 原始返回: {content}")
+ intent = self._parse_json(content)
+ LOG.info(f"AI 解析意图: {intent}")
+ return intent
+
+ @staticmethod
+ def _parse_json(content: str):
+ content = (content or "").strip()
+ content = re.sub(r"^```(?:json)?\s*", "", content)
+ content = re.sub(r"\s*```$", "", content)
+ try:
+ return json.loads(content)
+ except Exception:
+ pass
+ start = content.find("{")
+ end = content.rfind("}")
+ if start == -1 or end == -1 or end <= start:
+ LOG.error(f"AI returned non-JSON content: {content[:300]}")
+ return None
+ try:
+ return json.loads(content[start : end + 1])
+ except Exception as e:
+ LOG.error(f"AI JSON parse failed: {e}, content: {content[:300]}")
+ return None
diff --git a/telegram/tg-search-bot/bot.py b/telegram/tg-search-bot/bot.py
new file mode 100644
index 0000000..3684659
--- /dev/null
+++ b/telegram/tg-search-bot/bot.py
@@ -0,0 +1,1466 @@
+import concurrent.futures
+import math
+import os
+import re
+import string
+import random
+import jvav as jv
+import yaml
+import threading
+import langdetect
+import html
+import telebot
+from telebot import apihelper, types
+from telebot.types import InlineKeyboardButton, InlineKeyboardMarkup, InputMediaPhoto
+from database import BotFileDb, BotCacheDb
+from ai import AiClient
+from pikpak import PikPakClient
+from requests import get
+from requests.compat import quote
+from bs4 import BeautifulSoup
+import logging
+from logging.handlers import RotatingFileHandler
+
+
+class Logger:
+
+ def __init__(self, path_log_file: str, log_level=logging.INFO):
+ self.logger = logging.getLogger()
+ formatter = logging.Formatter("[%(asctime)s] %(levelname)s: %(message)s")
+ stream_handler = logging.StreamHandler()
+ stream_handler.setFormatter(formatter)
+ r_file_handler = RotatingFileHandler(
+ path_log_file, maxBytes=1024 * 1024 * 16, backupCount=1
+ )
+ r_file_handler.setFormatter(formatter)
+ self.logger.addHandler(r_file_handler)
+ self.logger.addHandler(stream_handler)
+ self.logger.setLevel(log_level)
+
+
+class BotConfig:
+ def __init__(self, path_config_file: str):
+ with open(path_config_file, "r", encoding="utf8") as f:
+ config = yaml.safe_load(f)
+ self.tg_chat_id = str(config["tg_chat_id"]) if config["tg_chat_id"] else ""
+ self.tg_bot_token = (
+ str(config["tg_bot_token"]) if config["tg_bot_token"] else ""
+ )
+ self.enable_nsfw = str(config["enable_nsfw"]) if config["enable_nsfw"] else "0"
+ self.use_proxy = str(config["use_proxy"]) if config["use_proxy"] else "0"
+ self.proxy_addr = str(config["proxy_addr"]) if config["proxy_addr"] else ""
+ self.pikpak_username = (
+ str(config["pikpak_username"]) if config.get("pikpak_username") else ""
+ )
+ self.pikpak_password = (
+ str(config["pikpak_password"]) if config.get("pikpak_password") else ""
+ )
+ self.ai_base_url = (
+ str(config["ai_base_url"]) if config.get("ai_base_url") else ""
+ )
+ self.ai_api_key = str(config["ai_api_key"]) if config.get("ai_api_key") else ""
+ self.ai_model = (
+ str(config["ai_model"]) if config.get("ai_model") else "gpt-4o-mini"
+ )
+ # set
+ self.proxy_json = {"http": "", "https": ""}
+ if self.use_proxy == "1":
+ self.proxy_json = {"http": self.proxy_addr, "https": self.proxy_addr}
+ LOG.info(f'Set proxy: "{self.proxy_addr}"')
+ else:
+ self.proxy_addr = ""
+ LOG.info("Successfully read and loaded the configuration file.")
+
+
+# URL
+BASE_URL_TG = "https://t.me"
+PIKPAK_BOT_NAME = "PikPak6_Bot"
+URL_PROJECT_ADDRESS = "https://github.com/akynazh/tg-search-bot"
+URL_PIKPAK_BOT = f"{BASE_URL_TG}/{PIKPAK_BOT_NAME}"
+# PATH
+PATH_ROOT = f'{os.path.expanduser("~")}/.tg_search_bot'
+PATH_LOG_FILE = f"{PATH_ROOT}/log.txt"
+PATH_RECORD_FILE = f"{PATH_ROOT}/record.json"
+PATH_PIKPAK_TOKEN_FILE = f"{PATH_ROOT}/pikpak_token.json"
+PATH_CACHE_FILE = f"{PATH_ROOT}/cache.db"
+PATH_CONFIG_FILE = f"{PATH_ROOT}/config.yaml"
+# BASE
+LOG = Logger(path_log_file=PATH_LOG_FILE).logger
+BOT_CFG = BotConfig(PATH_CONFIG_FILE)
+apihelper.proxy = BOT_CFG.proxy_json
+BOT = telebot.TeleBot(BOT_CFG.tg_bot_token)
+BOT_DB = BotFileDb(PATH_RECORD_FILE)
+BOT_CACHE_DB = BotCacheDb(
+ path_cache_file=PATH_CACHE_FILE,
+ use_cache="1",
+)
+BASE_UTIL = jv.BaseUtil(BOT_CFG.proxy_addr)
+DMM_UTIL = jv.DmmUtil(BOT_CFG.proxy_addr)
+JBUS_UTIL = jv.JavBusUtil(BOT_CFG.proxy_addr)
+JDB_UTIL = jv.JavDbUtil(BOT_CFG.proxy_addr)
+JLIB_UTIL = jv.JavLibUtil(BOT_CFG.proxy_addr)
+SUKEBEI_UTIL = jv.SukebeiUtil(BOT_CFG.proxy_addr)
+TRANS_UTIL = jv.TransUtil(BOT_CFG.proxy_addr)
+WIKI_UTIL = jv.WikiUtil(BOT_CFG.proxy_addr)
+VGLE_UTIL = jv.AvgleUtil(BOT_CFG.proxy_addr)
+EXECUTOR = concurrent.futures.ThreadPoolExecutor()
+PIKPAK = PikPakClient(
+ username=BOT_CFG.pikpak_username,
+ password=BOT_CFG.pikpak_password,
+ proxy=BOT_CFG.proxy_addr,
+ token_path=PATH_PIKPAK_TOKEN_FILE,
+)
+AI = AiClient(
+ base_url=BOT_CFG.ai_base_url,
+ api_key=BOT_CFG.ai_api_key,
+ model=BOT_CFG.ai_model,
+ proxy=BOT_CFG.proxy_addr,
+)
+# BT 搜索结果分页会话：token -> {"q": ..., "bts": [...], "page": n}
+BT_SESSIONS = {}
+CHAT_BT_SESSION = {}
+ID_PAT = re.compile(r"[a-z0-9]+[-_](?:ppv-)?[a-z0-9]+")
+BOT_CMDS = {
+ "help": "查看帮助",
+ "stars": "查看已收藏的演职人员",
+ "ids": "查看已收藏的编号",
+ "record": "导出收藏记录文件",
+}
+MSG_HELP = f"""直接发送片名、关键词或编号，剩下的交给我！
+
+"""
+for cmd, content in BOT_CMDS.items():
+ MSG_HELP += f"""/{cmd} {content}
+"""
+MSG_HELP += f"""
+[NSFW: {"已开启" if BOT_CFG.enable_nsfw == "1" else "已关闭"}]"""
+
+
+class BotKey:
+ KEY_GET_SAMPLE_BY_ID = "k0_0"
+ KEY_GET_MORE_MAGNETS_BY_ID = "k0_1"
+ KEY_SEARCH_STAR_BY_NAME = "k0_2"
+ KEY_WATCH_PV_BY_ID = "k1_0"
+ KEY_WATCH_FV_BY_ID = "k1_1"
+ KEY_GET_V_BY_ID = "k2_0"
+ KEY_RANDOM_GET_V_BY_STAR_ID = "k2_1"
+ KEY_RANDOM_GET_V_NICE = "k2_2"
+ KEY_RANDOM_GET_V_NEW = "k2_3"
+ KEY_GET_NEW_VS_BY_STAR_NAME_ID = "k2_4"
+ KEY_GET_NICE_VS_BY_STAR_NAME = "k2_5"
+ KEY_RECORD_STAR_BY_STAR_NAME_ID = "k3_0"
+ KEY_RECORD_V_BY_ID_STAR_IDS = "k3_1"
+ KEY_GET_STARS_RECORD = "k3_2"
+ KEY_GET_VS_RECORD = "k3_3"
+ KEY_GET_STAR_DETAIL_RECORD_BY_STAR_NAME_ID = "k3_4"
+ KEY_GET_V_DETAIL_RECORD_BY_ID = "k3_5"
+ KEY_UNDO_RECORD_STAR_BY_STAR_NAME_ID = "k3_6"
+ KEY_UNDO_RECORD_V_BY_ID = "k3_7"
+ KEY_DEL_V_CACHE = "k4_1"
+ KEY_BT_PAGE = "k5_0"
+
+
+class BotUtils:
+ v_utils = [JDB_UTIL, JBUS_UTIL, SUKEBEI_UTIL]
+
+ def send_action_typing(self):
+ BOT.send_chat_action(chat_id=BOT_CFG.tg_chat_id, action="typing")
+
+ def send_msg(self, msg: str, pv=False, markup=None):
+ BOT.send_message(
+ chat_id=BOT_CFG.tg_chat_id,
+ text=msg,
+ disable_web_page_preview=not pv,
+ parse_mode="HTML",
+ reply_markup=markup,
+ )
+
+ def send_msg_code_op(self, code: int, op: str):
+ if code == 200:
+ self.send_msg(f"✅ 操作成功：{op}")
+ elif code == 404:
+ self.send_msg(f"没有找到相关结果，{op} 未完成 Q_Q")
+ elif code == 500:
+ self.send_msg(f"服务器出错，{op} 未完成，请稍后重试或查看日志 Q_Q")
+ elif code == 502:
+ self.send_msg(f"网络请求失败，{op} 未完成，请检查网络后重试 Q_Q")
+
+ def send_msg_success_op(self, op: str):
+ self.send_msg(f"✅ 操作成功：{op}")
+
+ def send_msg_fail_reason_op(self, reason: str, op: str):
+ self.send_msg(f"{op} 失败：{reason} Q_Q")
+
+ def send_msg_v_not_found(self, v_id: str):
+ self.send_msg(
+ f"没有找到 {html.escape(v_id)} 的相关结果，请稍后重试或检查编号是否正确。"
+ )
+
+ def check_success(self, code: int, op: str):
+ if code == 200:
+ return True
+ if code == 404:
+ self.send_msg_fail_reason_op(reason="未找到相关结果", op=op)
+ elif code == 500:
+ self.send_msg_fail_reason_op(reason="服务器错误", op=op)
+ elif code == 502:
+ self.send_msg_fail_reason_op(reason="网络请求失败", op=op)
+ return False
+
+ def create_btn_by_key(self, key_type: str, obj):
+ if key_type == BotKey.KEY_GET_STAR_DETAIL_RECORD_BY_STAR_NAME_ID:
+ return InlineKeyboardButton(
+ text=obj["name"], callback_data=f'{obj["name"]}|{obj["id"]}:{key_type}'
+ )
+ elif key_type == BotKey.KEY_GET_V_DETAIL_RECORD_BY_ID:
+ return InlineKeyboardButton(text=obj, callback_data=f"{obj}:{key_type}")
+ elif key_type == BotKey.KEY_SEARCH_STAR_BY_NAME:
+ return InlineKeyboardButton(text=obj, callback_data=f"{obj}:{key_type}")
+ elif key_type == BotKey.KEY_GET_V_BY_ID:
+ return InlineKeyboardButton(
+ text=f'{obj["id"]} | {obj["rate"]}',
+ callback_data=f'{obj["id"]}:{key_type}',
+ )
+
+ def send_msg_btns(
+ self,
+ max_btn_per_row: int,
+ max_row_per_msg: int,
+ key_type: str,
+ title: str,
+ objs: list,
+ extra_btns=[],
+ page_btns=[],
+ ):
+ markup = InlineKeyboardMarkup()
+ row_count = 0
+ btns = []
+ for obj in objs:
+ btns.append(self.create_btn_by_key(key_type, obj))
+ if len(btns) == max_btn_per_row:
+ markup.row(*btns)
+ row_count += 1
+ btns = []
+ if row_count == max_row_per_msg:
+ for extra_btn in extra_btns:
+ markup.row(*extra_btn)
+ if page_btns != []:
+ markup.row(*page_btns)
+ self.send_msg(msg=title, markup=markup)
+ row_count = 0
+ markup = InlineKeyboardMarkup()
+ if btns != []:
+ markup.row(*btns)
+ row_count += 1
+ if row_count != 0:
+ for extra_btn in extra_btns:
+ markup.row(*extra_btn)
+ if page_btns != []:
+ markup.row(*page_btns)
+ self.send_msg(msg=title, markup=markup)
+
+ def get_page_elements(
+ self, objs: list, page: int, col: int, row: int, key_type: str
+ ):
+ """
+ Get the list of objects on the current page, list of pagination buttons, and the title of quantity.
+
+ :param list objs: All objects
+ :param int page: Current page
+ :param int col: Number of columns on the current page
+ :param int row: Number of rows on the current page
+ :param str key_type: Key type
+ :return tuple[list, list, str]: List of objects on the current page, list of pagination buttons, title of quantity
+ """
+ record_count_total = len(objs)
+ record_count_per_page = col * row
+ if record_count_per_page > record_count_total:
+ page_count = 1
+ else:
+ page_count = math.ceil(record_count_total / record_count_per_page)
+ if page > page_count:
+ page = page_count
+ start_idx = (page - 1) * record_count_per_page
+ objs = objs[start_idx : start_idx + record_count_per_page]
+ if page == 1:
+ to_previous = 1
+ else:
+ to_previous = page - 1
+ if page == page_count:
+ to_next = page_count
+ else:
+ to_next = page + 1
+ btn_to_first = InlineKeyboardButton(text="<<", callback_data=f"1:{key_type}")
+ btn_to_previous = InlineKeyboardButton(
+ text="<", callback_data=f"{to_previous}:{key_type}"
+ )
+ btn_to_current = InlineKeyboardButton(
+ text=f"-{page}-", callback_data=f"{page}:{key_type}"
+ )
+ btn_to_next = InlineKeyboardButton(
+ text=">", callback_data=f"{to_next}:{key_type}"
+ )
+ btn_to_last = InlineKeyboardButton(
+ text=">>", callback_data=f"{page_count}:{key_type}"
+ )
+ # Get the title of quantity
+ title = f"共 {record_count_total} 条，共 {page_count} 页"
+ return (
+ objs,
+ [btn_to_first, btn_to_previous, btn_to_current, btn_to_next, btn_to_last],
+ title,
+ )
+
+ def check_if_enable_nsfw(self):
+ if BOT_CFG.enable_nsfw == "0":
+ self.send_msg("[NSFW] 已关闭，无法访问该内容。")
+ return False
+ return True
+
+ def get_stars_record(self, page=1):
+ record, is_star_exists, _ = BOT_DB.check_has_record()
+ if not record or not is_star_exists:
+ self.send_msg_fail_reason_op(
+ reason="还没有收藏任何演职人员", op="查看收藏的演职人员"
+ )
+ return
+ stars = record["stars"]
+ stars.reverse()
+ col, row = 4, 5
+ objs, page_btns, title = self.get_page_elements(
+ objs=stars,
+ page=page,
+ col=col,
+ row=row,
+ key_type=BotKey.KEY_GET_STARS_RECORD,
+ )
+ self.send_msg_btns(
+ max_btn_per_row=col,
+ max_row_per_msg=row,
+ key_type=BotKey.KEY_GET_STAR_DETAIL_RECORD_BY_STAR_NAME_ID,
+ title="已收藏的演职人员：" + title,
+ objs=objs,
+ page_btns=page_btns,
+ )
+
+ def get_star_detail_record_by_name_id(self, star_name: str, star_id: str):
+ record, is_stars_exists, is_vs_exists = BOT_DB.check_has_record()
+ if not record:
+ self.send_msg_fail_reason_op(
+ reason="还没有收藏该演职人员",
+ op=f"查看演职人员 {star_name} 的详情",
+ )
+ return
+ star_vs = []
+ cur_star_exists = False
+ if is_vs_exists:
+ vs = record["vs"]
+ vs.reverse()
+ for v in vs:
+ if star_id in v["stars"]:
+ star_vs.append(v["id"])
+ if is_stars_exists:
+ stars = record["stars"]
+ for star in stars:
+ if star["id"].lower() == star_id.lower():
+ cur_star_exists = True
+ extra_btn1 = InlineKeyboardButton(
+ text="随机一部",
+ callback_data=f"{star_name}|{star_id}:{BotKey.KEY_RANDOM_GET_V_BY_STAR_ID}",
+ )
+ extra_btn2 = InlineKeyboardButton(
+ text="最新作品",
+ callback_data=f"{star_name}|{star_id}:{BotKey.KEY_GET_NEW_VS_BY_STAR_NAME_ID}",
+ )
+ extra_btn3 = InlineKeyboardButton(
+ text="高分作品",
+ callback_data=f"{star_name}:{BotKey.KEY_GET_NICE_VS_BY_STAR_NAME}",
+ )
+ if cur_star_exists:
+ extra_btn4 = InlineKeyboardButton(
+ text="取消收藏",
+ callback_data=f"{star_name}|{star_id}:{BotKey.KEY_UNDO_RECORD_STAR_BY_STAR_NAME_ID}",
+ )
+ else:
+ extra_btn4 = InlineKeyboardButton(
+ text="收藏",
+ callback_data=f"{star_name}|{star_id}:{BotKey.KEY_RECORD_STAR_BY_STAR_NAME_ID}",
+ )
+ title = f'{star_name} | Wiki | Javbus'
+ if len(star_vs) == 0:
+ markup = InlineKeyboardMarkup()
+ markup.row(extra_btn1, extra_btn2, extra_btn3, extra_btn4)
+ self.send_msg(msg=title, markup=markup)
+ return
+ self.send_msg_btns(
+ max_btn_per_row=4,
+ max_row_per_msg=10,
+ key_type=BotKey.KEY_GET_V_DETAIL_RECORD_BY_ID,
+ title=title,
+ objs=star_vs,
+ extra_btns=[[extra_btn1, extra_btn2, extra_btn3, extra_btn4]],
+ )
+
+ def get_vs_record(self, page=1):
+ record, _, is_vs_exists = BOT_DB.check_has_record()
+ if not record or not is_vs_exists:
+ self.send_msg_fail_reason_op(
+ reason="还没有收藏任何编号",
+ op="查看收藏的编号",
+ )
+ return
+ vs = [v["id"] for v in record["vs"]]
+ vs.reverse()
+ extra_btn1 = InlineKeyboardButton(
+ text="随机高分",
+ callback_data=f"0:{BotKey.KEY_RANDOM_GET_V_NICE}",
+ )
+ extra_btn2 = InlineKeyboardButton(
+ text="随机最新", callback_data=f"0:{BotKey.KEY_RANDOM_GET_V_NEW}"
+ )
+ col, row = 4, 10
+ objs, page_btns, title = self.get_page_elements(
+ objs=vs, page=page, col=col, row=row, key_type=BotKey.KEY_GET_VS_RECORD
+ )
+ self.send_msg_btns(
+ max_btn_per_row=col,
+ max_row_per_msg=row,
+ key_type=BotKey.KEY_GET_V_DETAIL_RECORD_BY_ID,
+ title="已收藏的编号：" + title,
+ objs=objs,
+ extra_btns=[[extra_btn1, extra_btn2]],
+ page_btns=page_btns,
+ )
+
+ def get_v_detail_record_by_id(self, id: str):
+ record, _, is_vs_exists = BOT_DB.check_has_record()
+ vs = record["vs"]
+ cur_v_exists = False
+ for v in vs:
+ if id.lower() == v["id"].lower():
+ cur_v_exists = True
+ markup = InlineKeyboardMarkup()
+ btn = InlineKeyboardButton(
+ text=f"查看",
+ callback_data=f"{id}:{BotKey.KEY_GET_V_BY_ID}",
+ )
+ if cur_v_exists:
+ markup.row(
+ btn,
+ InlineKeyboardButton(
+ text=f"取消收藏",
+ callback_data=f"{id}:{BotKey.KEY_UNDO_RECORD_V_BY_ID}",
+ ),
+ )
+ else:
+ markup.row(btn)
+ self.send_msg(msg=f"{id}", markup=markup)
+
+ def search_bts(self, q):
+ def append_trackers():
+ """Returns the base tracker list"""
+ trackers = [
+ "udp://tracker.coppersurfer.tk:6969/announce",
+ "udp://tracker.openbittorrent.com:6969/announce",
+ "udp://9.rarbg.to:2710/announce",
+ "udp://9.rarbg.me:2780/announce",
+ "udp://9.rarbg.to:2730/announce",
+ "udp://tracker.opentrackr.org:1337",
+ "http://p4p.arenabg.com:1337/announce",
+ "udp://tracker.torrent.eu.org:451/announce",
+ "udp://tracker.tiny-vps.com:6969/announce",
+ "udp://open.stealth.si:80/announce",
+ ]
+ trackers = [quote(tr) for tr in trackers]
+ return "&tr=".join(trackers)
+
+ def category_name(category):
+ """Translates the category code to a name"""
+ names = ["", "audio", "video", "apps", "games", "nsfw", "other"]
+ category = int(category[0])
+ category = category if category < len(names) - 1 else -1
+ return names[category]
+
+ def size_as_str(size):
+ """Formats the file size in bytes to kb, mb or gb accordingly"""
+ size = int(size)
+ size_str = f"{size} b"
+ if size >= 1024:
+ size_str = f"{(size / 1024):.2f} kb"
+ if size >= 1024**2:
+ size_str = f"{(size / 1024 ** 2):.2f} mb"
+ if size >= 1024**3:
+ size_str = f"{(size / 1024 ** 3):.2f} gb"
+ return size_str
+
+ def magnet_link(ih, name):
+ """Creates the magnet URI"""
+ return f"magnet:?xt=urn:btih:{ih}&dn={quote(name)}&tr={append_trackers()}"
+
+ agent = BASE_UTIL.ua()
+ url = f"https://apibay.org/q.php?q={quote(q)}"
+ try:
+ results = get(url, headers={"agent": agent}, timeout=10)
+ except Exception as e:
+ LOG.error(f"apibay search request failed: {e}")
+ return None
+ if not results.status_code == 200:
+ return None
+ matches = []
+ data = results.json()
+ if data and "no results" in data[0]["name"].lower():
+ return matches
+ for d in data:
+ matches.append(
+ {
+ "seeders": d["seeders"],
+ "leechers": d["leechers"],
+ "name": d["name"],
+ "category": category_name(d["category"]),
+ "size": size_as_str(d["size"]),
+ "magnet": magnet_link(d["info_hash"], d["name"]),
+ }
+ )
+ return matches
+
+ def search_bts_torrentkitty(self, q):
+ """通过 TorrentKitty 搜索中文 BT 资源，返回与 search_bts 相同格式的列表或 None。
+
+ TorrentKitty 聚合国内多家 BT 站，中文关键词命中率远高于 apibay。
+ """
+ url = f"https://www.torrentkitty.tv/search/{quote(q)}"
+ try:
+ resp = get(
+ url,
+ headers={"user-agent": BASE_UTIL.ua()},
+ timeout=10,
+ )
+ except Exception as e:
+ LOG.error(f"TorrentKitty search request failed: {e}")
+ return None
+ if resp.status_code != 200:
+ return None
+ try:
+ soup = BeautifulSoup(resp.text, "html.parser")
+ matches = []
+ for tr in soup.find_all("tr"):
+ name_td = tr.find("td", class_="name")
+ size_td = tr.find("td", class_="size")
+ magnet_a = tr.find("a", rel="magnet")
+ if not name_td or not magnet_a:
+ continue
+ matches.append(
+ {
+ "seeders": 0,
+ "leechers": 0,
+ "name": name_td.get_text(strip=True),
+ "category": "general",
+ "size": size_td.get_text(strip=True) if size_td else "",
+ "magnet": magnet_a.get("href"),
+ }
+ )
+ return matches if matches else None
+ except Exception as e:
+ LOG.error(f"TorrentKitty search parse failed: {e}")
+ return None
+
+ def get_v_by_id(
+ self,
+ id: str,
+ send_to_pikpak=True,
+ is_nice=True,
+ is_uncensored=True,
+ magnet_max_count=3,
+ not_send=False,
+ ):
+ """
+ Get based on id
+
+ :param str id: Number
+ :param bool send_to_pikpak: Whether to send to pikpak, default is yes
+ :param bool is_nice: Whether to filter out HD, subtitled magnet links, default is yes
+ :param bool is_uncensored: Whether to filter out uncensored magnet links, default is yes
+ :param int magnet_max_count: Maximum id of magnet links after filtering, default is 3
+ :param not_send: Whether not to send results, default is to send
+ :return dict: When not sending results, return the obtained results (if any)
+ """
+ if not self.check_if_enable_nsfw():
+ return {}
+ op_get_v_by_id = f"搜索编号 {id}"
+ v = BOT_CACHE_DB.get_cache(key=id, type=BotCacheDb.TYPE_V)
+ v_score = None
+ is_cache = False
+ if not v or not_send:
+ v_util = None
+ for util in self.v_utils:
+ code, v = util.get_av_by_id(
+ id=id,
+ is_nice=is_nice,
+ is_uncensored=is_uncensored,
+ magnet_max_count=magnet_max_count,
+ )
+ if code == 200:
+ v_util = util
+ break
+ if not v_util:
+ if not not_send:
+ self.send_msg_v_not_found(v_id=id)
+ return
+ if "score" not in v.keys():
+ _, v["score"] = DMM_UTIL.get_score_by_id(id)
+ if not not_send:
+ if len(v["magnets"]) == 0:
+ BOT_CACHE_DB.set_cache(
+ key=id, value=v, type=BotCacheDb.TYPE_V, expire=3600 * 24 * 1
+ )
+ else:
+ BOT_CACHE_DB.set_cache(key=id, value=v, type=BotCacheDb.TYPE_V)
+ else:
+ v_score = v["score"]
+ is_cache = True
+ if not_send:
+ return v
+ v_id = id
+ v_title = v["title"]
+ v_img = v["img"]
+ v_date = v["date"]
+ v_tags = v["tags"]
+ v_stars = v["stars"]
+ v_magnets = v["magnets"]
+ v_url = v["url"]
+ LOG.info(f"AV 搜索结果: {v_id} | {v_title}")
+ for _m in v_magnets:
+ LOG.info(f" 磁力: {_m['link']}")
+ msg = ""
+ if v_title != "":
+ v_title = v_title.replace("<", "").replace(">", "")
+ msg += f"""【标题】{v_title}
+"""
+ msg += f"""【编号】{v_id}
+"""
+ if v_date != "":
+ msg += f"""【日期】{v_date}
+"""
+ if v_score:
+ msg += f"""【评分】{v_score}
+"""
+ if v_stars != []:
+ show_star_name = v_stars[0]["name"]
+ show_star_id = v_stars[0]["id"]
+ stars_msg = ""
+ for star in v_stars:
+ stars_msg += f"""【演职人员】{star["name"]}
+"""
+ msg += stars_msg
+ if v_tags:
+ v_tags = " ".join(v_tags).replace("<", "").replace(">", "")
+ msg += f"""【标签】{v_tags}
+"""
+ msg += f"""【其他】Pikpak | 项目主页
+"""
+ magnet_send_to_pikpak = ""
+ for i, magnet in enumerate(v_magnets):
+ if i == 0:
+ magnet_send_to_pikpak = magnet["link"]
+ magnet_tags = ""
+ if magnet["uc"] == "1":
+ magnet_tags += "无码 "
+ if magnet["hd"] == "1":
+ magnet_tags += "HD "
+ if magnet["zm"] == "1":
+ magnet_tags += "中字 "
+ msg_tmp = f"""【{magnet_tags}磁力-{string.ascii_letters[i].upper()} {magnet["size"]}】{magnet["link"]}
+"""
+ if len(msg + msg_tmp) >= 2000:
+ break
+ msg += msg_tmp
+ pv_btn = InlineKeyboardButton(
+ text="预览", callback_data=f"{v_id}:{BotKey.KEY_WATCH_PV_BY_ID}"
+ )
+ fv_btn = InlineKeyboardButton(
+ text="在线观看", callback_data=f"{v_id}:{BotKey.KEY_WATCH_FV_BY_ID}"
+ )
+ sample_btn = InlineKeyboardButton(
+ text="截图", callback_data=f"{v_id}:{BotKey.KEY_GET_SAMPLE_BY_ID}"
+ )
+ more_btn = InlineKeyboardButton(
+ text="更多磁力",
+ callback_data=f"{v_id}:{BotKey.KEY_GET_MORE_MAGNETS_BY_ID}",
+ )
+ if len(v_magnets) != 0:
+ markup = InlineKeyboardMarkup().row(sample_btn, pv_btn, fv_btn, more_btn)
+ else:
+ markup = InlineKeyboardMarkup().row(sample_btn, pv_btn, fv_btn)
+ star_record_btn = None
+ if len(v_stars) == 1:
+ if BOT_DB.check_star_exists_by_id(star_id=show_star_id):
+ star_record_btn = InlineKeyboardButton(
+ text="详情",
+ callback_data=f"{show_star_name}|{show_star_id}:{BotKey.KEY_GET_STAR_DETAIL_RECORD_BY_STAR_NAME_ID}",
+ )
+ else:
+ star_record_btn = InlineKeyboardButton(
+ text="收藏演职人员",
+ callback_data=f"{show_star_name}|{show_star_id}:{BotKey.KEY_RECORD_STAR_BY_STAR_NAME_ID}",
+ )
+ star_ids = ""
+ for i, star in enumerate(v_stars):
+ star_ids += star["id"] + "|"
+ if i >= 5:
+ star_ids += "...|"
+ break
+ if star_ids != "":
+ star_ids = star_ids[: len(star_ids) - 1]
+ v_record_btn = None
+ if BOT_DB.check_id_exists(id=v_id):
+ v_record_btn = InlineKeyboardButton(
+ text="详情",
+ callback_data=f"{v_id}:{BotKey.KEY_GET_V_DETAIL_RECORD_BY_ID}",
+ )
+ else:
+ v_record_btn = InlineKeyboardButton(
+ text="收藏",
+ callback_data=f"{v_id}|{star_ids}:{BotKey.KEY_RECORD_V_BY_ID_STAR_IDS}",
+ )
+ renew_btn = None
+ if is_cache:
+ renew_btn = InlineKeyboardButton(
+ text="刷新", callback_data=f"{v_id}:{BotKey.KEY_DEL_V_CACHE}"
+ )
+ if star_record_btn and renew_btn:
+ markup.row(v_record_btn, star_record_btn, renew_btn)
+ elif star_record_btn:
+ markup.row(v_record_btn, star_record_btn)
+ elif renew_btn:
+ markup.row(v_record_btn, renew_btn)
+ else:
+ markup.row(v_record_btn)
+ if v_img == "":
+ self.send_msg(msg=msg, markup=markup)
+ else:
+ try:
+ BOT.send_photo(
+ chat_id=BOT_CFG.tg_chat_id,
+ photo=v_img,
+ caption=msg,
+ parse_mode="HTML",
+ reply_markup=markup,
+ )
+ except Exception:
+ self.send_msg(msg=msg, markup=markup)
+ if magnet_send_to_pikpak != "" and send_to_pikpak:
+ self.send_magnet_to_pikpak(magnet_send_to_pikpak, v_id)
+
+ def send_magnet_to_pikpak(self, magnet: str, id: str):
+ if not PIKPAK.enabled:
+ self.send_msg_fail_reason_op(
+ reason="请在 config.yaml 中配置 PikPak 账号 (pikpak_username / pikpak_password)",
+ op=f"保存编号 {id} 的磁力链接到 PikPak",
+ )
+ return
+ self.send_msg("💾 正在保存到 PikPak 网盘 ...")
+ ok, msg = PIKPAK.offline_download(magnet)
+ if ok:
+ self.send_msg(f"✅ 保存结果：{msg}")
+ else:
+ self.send_msg_fail_reason_op(
+ reason=msg, op=f"保存编号 {id} 的磁力链接到 PikPak"
+ )
+
+ def get_bts_by_keyword(self, q):
+ """按关键词搜索 BT 资源（带缓存），返回资源列表或 None。
+
+ 并行搜索 TorrentKitty（中文）与 apibay（英文）两个来源的结果，按磁力 info_hash 去重。
+ """
+ bts = BOT_CACHE_DB.get_cache(key=q, type=BotCacheDb.TYPE_BT)
+ if bts:
+ return bts
+ merged = []
+ seen = set()
+ # 并行搜索两个来源，但每个来源最多等待短时间，避免长时间无响应。
+ with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
+ tk_future = executor.submit(self.search_bts_torrentkitty, q)
+ ab_future = executor.submit(self.search_bts, q)
+ srcs = []
+ for name, future in (("torrentkitty", tk_future), ("apibay", ab_future)):
+ try:
+ srcs.append(future.result(timeout=8))
+ except concurrent.futures.TimeoutError:
+ LOG.warning(f"BT source {name} timed out for query: {q}")
+ srcs.append([])
+ for src in srcs:
+ for bt in src or []:
+ if not bt.get("magnet"):
+ continue
+ m = re.search(r"urn:btih:([A-Za-z0-9]+)", bt["magnet"])
+ key = m.group(1).lower() if m else bt["magnet"]
+ if key in seen:
+ continue
+ seen.add(key)
+ merged.append(bt)
+ if not merged:
+ return None
+ if BOT_CFG.enable_nsfw == "0":
+ merged = list(filter(lambda bt: bt["category"] != "nsfw", merged))
+ merged = list(filter(lambda bt: "nsfw" not in bt["name"].lower(), merged))
+ merged = merged[:30]
+ BOT_CACHE_DB.set_cache(key=q, value=merged, type=BotCacheDb.TYPE_BT)
+ LOG.info(f"BT 搜索结果 {len(merged)} 条: {q}")
+ for _bt in merged:
+ LOG.info(f" {_bt['name']} | {_bt['size']} | {_bt['magnet']}")
+ return merged
+
+ def search_bts_and_save(self, q):
+ try:
+ bts = self.get_bts_by_keyword(q)
+ except Exception as e:
+ LOG.error(f"BT 搜索异常: {e}")
+ self.send_msg_fail_reason_op(
+ reason=f"搜索超时或网络错误：{str(e)[:50]}",
+ op=f"搜索 {html.escape(q)}",
+ )
+ return False
+ if not bts:
+ self.send_msg_fail_reason_op(
+ reason="没有找到相关资源", op=f"搜索 {html.escape(q)}"
+ )
+ return False
+ self.show_bt_results(q, bts, save_first=True)
+ return True
+
+ def search_av_keyword(self, q):
+ """按关键词搜索 AV（JavDB 标签搜索），找到则返回 True。"""
+ code, ids = JDB_UTIL.get_ids_by_tag(q)
+ if code == 200 and ids:
+ self.get_v_by_id(id=ids[0])
+ return True
+ return False
+
+ def handle_nlp_search(self, msg):
+ """AI 自然语言理解搜索流程。返回 True 表示已处理。"""
+ if not AI.enabled:
+ return False
+ LOG.info(f'AI 搜索字符: "{msg}"')
+ intent = AI.understand(msg)
+ LOG.info(f"AI 返回结果: {intent}")
+ if not intent:
+ return False
+ target = (intent.get("target") or msg).strip()
+ kind = (intent.get("intent") or "general").strip().lower()
+ explain = (intent.get("explain") or target).strip()
+ source = (intent.get("source") or "auto").strip().lower()
+
+ # AV 相关意图需先检查 NSFW 开关
+ if kind in ("av_id", "actor") and not self.check_if_enable_nsfw():
+ return True
+
+ self.send_msg(f"🔍 搜索：{html.escape(explain)}")
+ self.send_msg("⏳ 搜索中 ...")
+
+ if kind == "av_id":
+ self.get_v_by_id(id=target)
+ return True
+ if kind == "actor":
+ # 仅在 AI 明确判定为日本 AV 演员时，才走演员搜索；其他“演员/明星”类请求继续走一般搜索
+ self.search_star_by_name(target)
+ return True
+
+ # 类型 / 关键词 / 剧情 / 其他：优先 AV 关键词搜索，失败则 BT 搜索
+ if (
+ source in ("javdb", "javbus", "sukebei", "dmm")
+ and BOT_CFG.enable_nsfw == "1"
+ ):
+ if self.search_av_keyword(target):
+ return True
+ # BT 搜索（总是返回 True，表示已处理；搜索结果由 search_bts_and_save 内部反馈给用户）
+ self.search_bts_and_save(target)
+ return True
+
+ def get_sample_by_id(self, id: str):
+ op_get_sample = f"获取编号 {id} 的截图"
+ samples = BOT_CACHE_DB.get_cache(key=id, type=BotCacheDb.TYPE_SAMPLE)
+ if not samples:
+ code, samples = JBUS_UTIL.get_samples_by_id(id)
+ if not self.check_success(code, op_get_sample):
+ return
+ BOT_CACHE_DB.set_cache(key=id, value=samples, type=BotCacheDb.TYPE_SAMPLE)
+ samples_imp = []
+ sample_error = False
+ for sample in samples:
+ samples_imp.append(InputMediaPhoto(sample))
+ if len(samples_imp) == 10:
+ try:
+ BOT.send_media_group(chat_id=BOT_CFG.tg_chat_id, media=samples_imp)
+ samples_imp = []
+ except Exception:
+ sample_error = True
+ self.send_msg_fail_reason_op(
+ reason="图片解析失败", op=op_get_sample
+ )
+ break
+ if samples_imp != [] and not sample_error:
+ try:
+ BOT.send_media_group(chat_id=BOT_CFG.tg_chat_id, media=samples_imp)
+ except Exception:
+ self.send_msg_fail_reason_op(reason="图片解析失败", op=op_get_sample)
+
+ def watch_v_by_id(self, id: str, type: int):
+ id = id.lower()
+ if id.find("fc2") != -1 and id.find("ppv") == -1:
+ id = id.replace("fc2", "fc2-ppv")
+ if type == 0:
+ pv = BOT_CACHE_DB.get_cache(key=id, type=BotCacheDb.TYPE_PV)
+ if not pv:
+ op_watch_v = f"获取编号 {id} 的预览视频"
+ futures = {}
+ with concurrent.futures.ThreadPoolExecutor() as executor:
+ futures[executor.submit(DMM_UTIL.get_pv_by_id, id)] = 1
+ futures[executor.submit(VGLE_UTIL.get_pv_by_id, id)] = 2
+ for future in concurrent.futures.as_completed(futures):
+ if futures[future] == 1:
+ code_dmm, pv_dmm = future.result()
+ elif futures[future] == 2:
+ code_vgle, pv_vgle = future.result()
+ if code_dmm != 200 and code_vgle != 200:
+ if code_dmm == 502 or code_vgle == 502:
+ self.send_msg_code_op(502, op_watch_v)
+ else:
+ self.send_msg_code_op(404, op_watch_v)
+ return
+ from_site = ""
+ pv_src = ""
+ if code_dmm == 200:
+ from_site = "dmm"
+ pv_src = pv_dmm
+ elif code_vgle == 200:
+ from_site = "avgle"
+ pv_src = pv_vgle
+ pv_cache = {"from_site": from_site, "src": pv_src}
+ BOT_CACHE_DB.set_cache(key=id, value=pv_cache, type=BotCacheDb.TYPE_PV)
+ else:
+ from_site = pv["from_site"]
+ pv_src = pv["src"]
+ if from_site == "dmm":
+ try:
+ pv_src_nice = DMM_UTIL.get_nice_pv_by_src(pv_src)
+ BOT.send_video(
+ chat_id=BOT_CFG.tg_chat_id,
+ video=pv_src,
+ caption=f'来自 DMM 的搜索结果。点此观看更清晰的版本。',
+ parse_mode="HTML",
+ )
+ except Exception:
+ self.send_msg(
+ f'来自 DMM 的搜索结果，但视频解析失败：视频链接 Q_Q。'
+ )
+ elif from_site == "avgle":
+ try:
+ BOT.send_video(
+ chat_id=BOT_CFG.tg_chat_id,
+ video=pv_src,
+ caption=f'来自 Avgle 的搜索结果：视频链接。',
+ parse_mode="HTML",
+ )
+ except Exception:
+ self.send_msg(
+ f'来自 Avgle 的搜索结果，但视频解析失败：视频链接 Q_Q。'
+ )
+ elif type == 1:
+ video = BOT_CACHE_DB.get_cache(key=id, type=BotCacheDb.TYPE_FV)
+ if not video:
+ code, video = VGLE_UTIL.get_fv_by_id(id)
+ if code != 200:
+ self.send_msg("没有找到相关结果。")
+ return
+ BOT_CACHE_DB.set_cache(key=id, value=video, type=BotCacheDb.TYPE_FV)
+ self.send_msg(video)
+
+ def search_star_by_name(self, star_name: str):
+ if not self.check_if_enable_nsfw():
+ return False
+ op_search_star = f"搜索演职人员 {star_name}"
+ star = BOT_CACHE_DB.get_cache(key=star_name, type=BotCacheDb.TYPE_STAR)
+ if not star:
+ star_name_origin = star_name
+ star_name = self.get_star_ja_name_by_zh_name(star_name)
+ code, star = JBUS_UTIL.check_star_exists(star_name)
+ if not self.check_success(code, op_search_star):
+ return
+ BOT_CACHE_DB.set_cache(key=star_name, value=star, type=BotCacheDb.TYPE_STAR)
+ if star_name_origin != star_name:
+ BOT_CACHE_DB.set_cache(
+ key=star_name_origin,
+ value=star,
+ type=BotCacheDb.TYPE_STAR,
+ )
+ star_id = star["star_id"]
+ star_name = star["star_name"]
+ if BOT_DB.check_star_exists_by_id(star_id=star_id):
+ self.get_star_detail_record_by_name_id(star_name=star_name, star_id=star_id)
+ return True
+ markup = InlineKeyboardMarkup()
+ markup.row(
+ InlineKeyboardButton(
+ text="随机一部",
+ callback_data=f"{star_name}|{star_id}:{BotKey.KEY_RANDOM_GET_V_BY_STAR_ID}",
+ ),
+ InlineKeyboardButton(
+ text="最新作品",
+ callback_data=f"{star_name}|{star_id}:{BotKey.KEY_GET_NEW_VS_BY_STAR_NAME_ID}",
+ ),
+ InlineKeyboardButton(
+ text="高分作品",
+ callback_data=f"{star_name}:{BotKey.KEY_GET_NICE_VS_BY_STAR_NAME}",
+ ),
+ InlineKeyboardButton(
+ text=f"收藏 {star_name}",
+ callback_data=f"{star_name}|{star_id}:{BotKey.KEY_RECORD_STAR_BY_STAR_NAME_ID}",
+ ),
+ )
+ star_wiki = f"{WIKI_UTIL.BASE_URL_CHINA_WIKI}/{star_name}"
+ if langdetect.detect(star_name) == "ja":
+ star_wiki = f"{WIKI_UTIL.BASE_URL_JAPAN_WIKI}/{star_name}"
+ self.send_msg(
+ msg=f'{star_name} | Wiki | Javbus',
+ markup=markup,
+ )
+ return True
+
+ def get_more_magnets_by_id(self, id: str):
+ magnets = BOT_CACHE_DB.get_cache(key=id, type=BotCacheDb.TYPE_MAGNET)
+ if not magnets:
+ v = self.get_v_by_id(
+ id=id, is_nice=False, is_uncensored=False, not_send=True
+ )
+ if not v:
+ return
+ magnets = v["magnets"]
+ BOT_CACHE_DB.set_cache(key=id, value=magnets, type=BotCacheDb.TYPE_MAGNET)
+ msg = ""
+ for magnet in magnets:
+ magnet_tags = ""
+ if magnet["uc"] == "1":
+ magnet_tags += "无码 "
+ if magnet["hd"] == "1":
+ magnet_tags += "HD "
+ if magnet["zm"] == "1":
+ magnet_tags += "中字 "
+ star_tag = ""
+ if magnet["hd"] == "1" and magnet["zm"] == "1":
+ star_tag = "*"
+ msg_tmp = f"""【{star_tag}{magnet_tags}磁力 {magnet["size"]}】{magnet["link"]}
+ """
+ if len(msg + msg_tmp) >= 4000:
+ self.send_msg(msg)
+ msg = msg_tmp
+ else:
+ msg += msg_tmp
+ if msg != "":
+ self.send_msg(msg)
+
+ def get_star_new_vs_by_name_id(self, star_name: str, star_id: str):
+ op_get_star_new_vs = f"获取 {star_name} 的最新作品"
+ ids = BOT_CACHE_DB.get_cache(key=star_id, type=BotCacheDb.TYPE_NEW_VS_OF_STAR)
+ if not ids:
+ code, ids = JBUS_UTIL.get_new_ids_by_star_id(star_id=star_id)
+ if not self.check_success(code, op_get_star_new_vs):
+ return
+ BOT_CACHE_DB.set_cache(
+ key=star_id, value=ids, type=BotCacheDb.TYPE_NEW_VS_OF_STAR
+ )
+ title = f"{star_name} 的最新作品"
+ btns = [
+ InlineKeyboardButton(
+ text=id, callback_data=f"{id}:{BotKey.KEY_GET_V_BY_ID}"
+ )
+ for id in ids
+ ]
+ if len(btns) <= 4:
+ self.send_msg(msg=title, markup=InlineKeyboardMarkup().row(*btns))
+ else:
+ markup = InlineKeyboardMarkup()
+ markup.row(*btns[:4])
+ markup.row(*btns[4:])
+ self.send_msg(msg=title, markup=markup)
+
+ def get_star_ja_name_by_zh_name(self, star_name: str):
+ if langdetect.detect(star_name) == "ja":
+ return star_name
+ star_ja_name = BOT_CACHE_DB.get_cache(
+ key=star_name, type=BotCacheDb.TYPE_STAR_JA_NAME
+ )
+ if star_ja_name:
+ return star_ja_name
+ wiki_json = WIKI_UTIL.get_wiki_page_by_lang(
+ topic=star_name, from_lang="zh", to_lang="ja"
+ )
+ if wiki_json and wiki_json["lang"] == "ja":
+ BOT_CACHE_DB.set_cache(
+ key=star_name,
+ value=wiki_json["title"],
+ type=BotCacheDb.TYPE_STAR_JA_NAME,
+ )
+ return wiki_json["title"]
+ return star_name
+
+ def register_bt_session(self, q, bts):
+ token = "".join(random.choices(string.ascii_letters + string.digits, k=8))
+ BT_SESSIONS[token] = {"q": q, "bts": bts, "page": 1}
+ CHAT_BT_SESSION[BOT_CFG.tg_chat_id] = token
+ return token
+
+ def show_bt_results(self, q, bts, save_first=False):
+ token = self.register_bt_session(q, bts)
+ self.show_bt_page(token, 1)
+ if save_first and bts:
+ self.send_magnet_to_pikpak(bts[0]["magnet"], q)
+ return token
+
+ def show_bt_page(self, token, page):
+ sess = BT_SESSIONS.get(token)
+ if not sess:
+ self.send_msg("结果会话已过期，请重新搜索。")
+ return
+ bts = sess["bts"]
+ page_size = 5
+ page_count = math.ceil(len(bts) / page_size) if bts else 1
+ if page < 1:
+ page = 1
+ if page > page_count:
+ page = 1 # 换一批：最后一页回到第一页
+ sess["page"] = page
+ start = (page - 1) * page_size
+ items = bts[start : start + page_size]
+ res = (
+ f"「{html.escape(sess['q'])}」的搜索结果"
+ f"（第 {page}/{page_count} 页）：\n\n"
+ )
+ for i, bt in enumerate(items):
+ idx = start + i + 1
+ name = bt.get("name") or ""
+ if len(name) > 80:
+ name = name[:77] + "..."
+ magnet = bt.get("magnet") or ""
+ magnet_preview = magnet
+ if len(magnet_preview) > 80:
+ magnet_preview = magnet_preview[:77] + "..."
+ res += (
+ f"{idx}. {html.escape(name)}\n"
+ f" 大小: {bt.get('size', '')} | 分类: {bt.get('category', 'general')}\n"
+ f" 磁力: {html.escape(magnet_preview)}\n\n"
+ )
+ res = res[:3500] + "\n\n..." if len(res) > 3500 else res
+ markup = InlineKeyboardMarkup()
+ btns = []
+ if page > 1:
+ btns.append(
+ InlineKeyboardButton(
+ "◀ 上一页",
+ callback_data=f"{token}|{page - 1}:{BotKey.KEY_BT_PAGE}",
+ )
+ )
+ btns.append(
+ InlineKeyboardButton(
+ "🔄 换一批",
+ callback_data=f"{token}|{page + 1}:{BotKey.KEY_BT_PAGE}",
+ )
+ )
+ markup.row(*btns)
+ markup.row(InlineKeyboardButton("保存到 PikPak 网盘", url=URL_PIKPAK_BOT))
+ self.send_msg(res, markup=markup)
+
+ def handle_bt_followup(self, msg):
+ """处理 BT 搜索结果的多轮追问：换一批 / 上一页 / 保存第 N 个。"""
+ token = CHAT_BT_SESSION.get(BOT_CFG.tg_chat_id)
+ if not token or token not in BT_SESSIONS:
+ return False
+ sess = BT_SESSIONS[token]
+ m = msg.strip()
+ save_match = re.match(r"^(?:保存|save)\s*第?\s*(\d+)\s*个?$", m)
+ if save_match:
+ n = int(save_match.group(1))
+ bts = sess["bts"]
+ if 1 <= n <= len(bts):
+ self.send_magnet_to_pikpak(bts[n - 1]["magnet"], sess["q"])
+ else:
+ self.send_msg_fail_reason_op(
+ reason=f"没有第 {n} 个结果", op="保存到 PikPak"
+ )
+ return True
+ if m in ("下一页", "下一批", "next"):
+ self.show_bt_page(token, sess["page"] + 1)
+ return True
+ if m in ("上一页", "prev"):
+ self.show_bt_page(token, sess["page"] - 1)
+ return True
+ if m in ("换一批", "换一页", "还有吗", "再来", "more"):
+ self.show_bt_page(token, sess["page"] + 1)
+ return True
+ return False
+
+ def random_get_new_v(self):
+ page = random.randint(1, JLIB_UTIL.MAX_RANK_PAGE)
+ ids = BOT_CACHE_DB.get_cache(key=page, type=BotCacheDb.TYPE_JLIB_PAGE_NEW_VS)
+ if not ids:
+ code, ids = JLIB_UTIL.get_random_ids_from_rank_by_page(
+ page=page, list_type=1
+ )
+ if self.check_success(code, "随机获取最新影片"):
+ BOT_CACHE_DB.set_cache(
+ key=page,
+ value=ids,
+ type=BotCacheDb.TYPE_JLIB_PAGE_NEW_VS,
+ )
+ else:
+ return
+ self.get_v_by_id(id=random.choice(ids))
+
+ def random_get_nice_v(self):
+ page = random.randint(1, JLIB_UTIL.MAX_RANK_PAGE)
+ ids = BOT_CACHE_DB.get_cache(key=page, type=BotCacheDb.TYPE_JLIB_PAGE_NICE_VS)
+ if not ids:
+ code, ids = JLIB_UTIL.get_random_ids_from_rank_by_page(
+ page=page, list_type=0
+ )
+ if self.check_success(code, "随机获取高分影片"):
+ BOT_CACHE_DB.set_cache(
+ key=page,
+ value=ids,
+ type=BotCacheDb.TYPE_JLIB_PAGE_NICE_VS,
+ )
+ else:
+ return
+ self.get_v_by_id(id=random.choice(ids))
+
+ def random_get_nice_star_vs(self, star_name_ori):
+ vs = BOT_CACHE_DB.get_cache(
+ key=star_name_ori, type=BotCacheDb.TYPE_NICE_VS_OF_STAR
+ )
+ if not vs:
+ star_name_ja = self.get_star_ja_name_by_zh_name(star_name_ori)
+ code, vs = DMM_UTIL.get_nice_vs_by_star_name(star_name=star_name_ja)
+ if self.check_success(code, f"获取演职人员 {star_name_ori} 的高分作品"):
+ vs = vs[:60]
+ BOT_CACHE_DB.set_cache(
+ key=star_name_ori,
+ value=vs,
+ type=BotCacheDb.TYPE_NICE_VS_OF_STAR,
+ )
+ if star_name_ja != star_name_ori:
+ BOT_CACHE_DB.set_cache(
+ key=star_name_ja,
+ value=vs,
+ type=BotCacheDb.TYPE_NICE_VS_OF_STAR,
+ )
+ else:
+ return
+ self.send_msg_btns(
+ max_btn_per_row=3,
+ max_row_per_msg=20,
+ key_type=BotKey.KEY_GET_V_BY_ID,
+ title=f"{star_name_ori} 的高分作品",
+ objs=vs,
+ )
+
+
+def handle_callback(call):
+ bot_utils = BotUtils()
+ bot_utils.send_action_typing()
+ LOG.info(f"Handle callback: {call.data}")
+ s = call.data.rfind(":")
+ content = call.data[:s]
+ key_type = call.data[s + 1 :]
+ if key_type == BotKey.KEY_WATCH_PV_BY_ID:
+ bot_utils.watch_v_by_id(id=content, type=0)
+ elif key_type == BotKey.KEY_WATCH_FV_BY_ID:
+ bot_utils.watch_v_by_id(id=content, type=1)
+ elif key_type == BotKey.KEY_GET_SAMPLE_BY_ID:
+ bot_utils.get_sample_by_id(id=content)
+ elif key_type == BotKey.KEY_GET_MORE_MAGNETS_BY_ID:
+ bot_utils.get_more_magnets_by_id(id=content)
+ elif key_type == BotKey.KEY_RANDOM_GET_V_BY_STAR_ID:
+ tmp = content.split("|")
+ star_name = tmp[0]
+ star_id = tmp[1]
+ code, id = JBUS_UTIL.get_id_by_star_id(star_id=star_id)
+ if bot_utils.check_success(
+ code, f"从演职人员 {star_name} 随机选取作品"
+ ):
+ bot_utils.get_v_by_id(id=id)
+ elif key_type == BotKey.KEY_GET_NEW_VS_BY_STAR_NAME_ID:
+ tmp = content.split("|")
+ star_name = tmp[0]
+ star_id = tmp[1]
+ bot_utils.get_star_new_vs_by_name_id(star_name=star_name, star_id=star_id)
+ elif key_type == BotKey.KEY_RECORD_STAR_BY_STAR_NAME_ID:
+ s = content.find("|")
+ star_name = content[:s]
+ star_id = content[s + 1 :]
+ if BOT_DB.record_star_by_name_id(star_name=star_name, star_id=star_id):
+ bot_utils.get_star_detail_record_by_name_id(
+ star_name=star_name, star_id=star_id
+ )
+ else:
+ bot_utils.send_msg_code_op(500, f"收藏演职人员 {star_name}")
+ elif key_type == BotKey.KEY_RECORD_V_BY_ID_STAR_IDS:
+ res = content.split("|")
+ id = res[0]
+ stars = []
+ if res[1] != "":
+ stars = [s for s in res[1:]]
+ if BOT_DB.record_id_by_id_stars(id=id, stars=stars):
+ bot_utils.get_v_detail_record_by_id(id=id)
+ else:
+ bot_utils.send_msg_code_op(500, f"收藏编号 {id}")
+ elif key_type == BotKey.KEY_GET_STARS_RECORD:
+ bot_utils.get_stars_record(page=int(content))
+ elif key_type == BotKey.KEY_GET_VS_RECORD:
+ bot_utils.get_vs_record(page=int(content))
+ elif key_type == BotKey.KEY_GET_STAR_DETAIL_RECORD_BY_STAR_NAME_ID:
+ s = content.find("|")
+ bot_utils.get_star_detail_record_by_name_id(
+ star_name=content[:s], star_id=content[s + 1 :]
+ )
+ elif key_type == BotKey.KEY_GET_V_DETAIL_RECORD_BY_ID:
+ bot_utils.get_v_detail_record_by_id(id=content)
+ elif key_type == BotKey.KEY_GET_V_BY_ID:
+ bot_utils.get_v_by_id(id=content)
+ elif key_type == BotKey.KEY_RANDOM_GET_V_NICE:
+ code, id = JLIB_UTIL.get_random_id_from_rank(0)
+ if bot_utils.check_success(code, "随机获取高分影片"):
+ bot_utils.get_v_by_id(id=id)
+ elif key_type == BotKey.KEY_RANDOM_GET_V_NEW:
+ code, id = JLIB_UTIL.get_random_id_from_rank(1)
+ if bot_utils.check_success(code, "随机获取最新影片"):
+ bot_utils.get_v_by_id(id=id)
+ elif key_type == BotKey.KEY_UNDO_RECORD_V_BY_ID:
+ op_undo_record_v = f"取消收藏编号 {content}"
+ if BOT_DB.undo_record_id(id=content):
+ bot_utils.send_msg_success_op(op_undo_record_v)
+ else:
+ bot_utils.send_msg_fail_reason_op(
+ reason="文件解析错误", op=op_undo_record_v
+ )
+ elif key_type == BotKey.KEY_UNDO_RECORD_STAR_BY_STAR_NAME_ID:
+ s = content.find("|")
+ op_undo_record_star = f"取消收藏演职人员 {content[:s]}"
+ if BOT_DB.undo_record_star_by_id(star_id=content[s + 1 :]):
+ bot_utils.send_msg_success_op(op_undo_record_star)
+ else:
+ bot_utils.send_msg_fail_reason_op(
+ reason="文件解析错误", op=op_undo_record_star
+ )
+ elif key_type == BotKey.KEY_SEARCH_STAR_BY_NAME:
+ star_name = content
+ star_name_alias = ""
+ idx_alias = star_name.find("（")
+ if idx_alias != -1:
+ star_name_alias = star_name[idx_alias + 1 : -1]
+ star_name = star_name[:idx_alias]
+ if not bot_utils.search_star_by_name(star_name) and star_name_alias != "":
+ bot_utils.send_msg(
+ f"尝试搜索演职人员 {star_name} 的别名 {star_name_alias}..."
+ )
+ bot_utils.search_star_by_name(star_name_alias)
+ elif key_type == BotKey.KEY_GET_NICE_VS_BY_STAR_NAME:
+ bot_utils.random_get_nice_star_vs(content)
+ elif key_type == BotKey.KEY_DEL_V_CACHE:
+ BOT_CACHE_DB.remove_cache(key=content, type=BotCacheDb.TYPE_V)
+ BOT_CACHE_DB.remove_cache(key=content, type=BotCacheDb.TYPE_STARS_MSG)
+ bot_utils.get_v_by_id(id=content)
+ elif key_type == BotKey.KEY_BT_PAGE:
+ tmp = content.split("|")
+ bot_utils.show_bt_page(token=tmp[0], page=int(tmp[1]))
+
+
+def handle_message(message):
+ bot_utils = BotUtils()
+ bot_utils.send_action_typing()
+ chat_id = str(message.chat.id)
+ if chat_id.lower() != BOT_CFG.tg_chat_id.lower():
+ return
+ bot_utils = BotUtils()
+ if message.content_type != "text":
+ msg = message.caption
+ else:
+ msg = message.text
+ if not msg:
+ return
+ LOG.info(f'Get message: "{msg}"')
+ msg = msg.lower().strip()
+ msgs = msg.split(" ", 1)
+ msg_cmd = msgs[0]
+ msg_param = ""
+ if len(msgs) > 1:
+ msg_param = msgs[1].strip()
+ if msg_cmd == "/help" or msg_cmd == "/start":
+ bot_utils.send_msg(MSG_HELP)
+ elif msg_cmd == "/stars":
+ bot_utils.get_stars_record()
+ elif msg_cmd == "/ids":
+ bot_utils.get_vs_record()
+ elif msg_cmd == "/record":
+ if not os.path.exists(PATH_RECORD_FILE):
+ bot_utils.send_msg_fail_reason_op(
+ reason="还没有任何收藏记录", op="导出收藏记录"
+ )
+ return
+ BOT.send_document(
+ chat_id=BOT_CFG.tg_chat_id, document=types.InputFile(PATH_RECORD_FILE)
+ )
+ else:
+ # 多轮会话：换一批 / 上一页 / 保存第 N 个
+ if bot_utils.handle_bt_followup(msg):
+ return
+ ids = ID_PAT.findall(msg)
+ if not ids or len(ids) == 0:
+ # 优先走 AI 自然语言理解搜索，失败则退回 BT 关键词搜索
+ if bot_utils.handle_nlp_search(msg):
+ return
+ bts = bot_utils.get_bts_by_keyword(msg)
+ if not bts:
+ bot_utils.send_msg_fail_reason_op(
+ reason="没有找到相关资源",
+ op=f"搜索 {html.escape(msg)}",
+ )
+ return
+ bot_utils.show_bt_results(msg, bts)
+ else:
+ ids = [id.lower() for id in ids]
+ ids = set(ids)
+ ids_msg = ", ".join(ids)
+ bot_utils.send_msg(f"已识别到编号：{ids_msg}，开始搜索...")
+ for i, id in enumerate(ids):
+ threading.Thread(target=bot_utils.get_v_by_id, args=(id,)).start()
+
+
+@BOT.callback_query_handler(func=lambda call: True)
+def my_callback_handler(call):
+ EXECUTOR.submit(handle_callback, call)
+
+
+@BOT.message_handler(content_types=["text", "photo", "animation", "video", "document"])
+def my_message_handler(message):
+ EXECUTOR.submit(handle_message, message)
+
+
+def main():
+ try:
+ bot_info = BOT.get_me()
+ LOG.info(f"Connected to bot: @{bot_info.username} (ID: {bot_info.id})")
+ if PIKPAK.enabled:
+ ok, msg = PIKPAK.ensure_login()
+ if ok:
+ LOG.info(f"PikPak: {msg}")
+ else:
+ LOG.error(f"PikPak login failed: {msg}")
+ LOG.info("Connected to api")
+ except Exception as e:
+ LOG.error(f"Unable to connect to bot or api: {e}")
+ return
+ BOT.set_my_commands([types.BotCommand(cmd, BOT_CMDS[cmd]) for cmd in BOT_CMDS])
+ BOT.infinity_polling()
+
+
+if __name__ == "__main__":
+ main()
diff --git a/telegram/tg-search-bot/database.py b/telegram/tg-search-bot/database.py
new file mode 100644
index 0000000..1476147
--- /dev/null
+++ b/telegram/tg-search-bot/database.py
@@ -0,0 +1,304 @@
+import os
+import json
+import sqlite3
+import threading
+import time
+import logging
+
+LOG = logging.getLogger(__name__)
+
+
+class BotFileDb:
+ def __init__(self, path_record_file: str):
+ self.path_record_file = path_record_file
+ pass
+
+ def check_has_record(self):
+ record = {}
+ if os.path.exists(self.path_record_file):
+ try:
+ with open(self.path_record_file, "r", encoding="utf8") as f:
+ record = json.load(f)
+ except Exception as e:
+ LOG.error(f"Failed to load the saved records file: {e}")
+ return None, False, False
+ if not record or record == {}:
+ return None, False, False
+ is_stars_exists = False
+ is_vs_exists = False
+ if (
+ "stars" in record.keys()
+ and record["stars"] != []
+ and len(record["stars"]) > 0
+ ):
+ is_stars_exists = True
+ if "vs" in record.keys() and record["vs"] != [] and len(record["vs"]) > 0:
+ is_vs_exists = True
+ return record, is_stars_exists, is_vs_exists
+
+ def check_star_exists_by_id(self, star_id: str):
+ record, exists, _ = self.check_has_record()
+ if not record or not exists:
+ return False
+ stars = record["stars"]
+ for star in stars:
+ if star["id"].lower() == star_id.lower():
+ return True
+
+ def check_id_exists(self, id: str):
+ record, _, exists = self.check_has_record()
+ if not record or not exists:
+ return False
+ vs = record["vs"]
+ for v in vs:
+ if v["id"].lower() == id.lower():
+ return True
+
+ def renew_record(self, record: dict):
+ try:
+ with open(self.path_record_file, "w", encoding="utf8") as f:
+ json.dump(
+ record, f, separators=(",", ": "), indent=4, ensure_ascii=False
+ )
+ return True
+ except Exception as e:
+ LOG.error(f"Failed to update the saved records file: {e}")
+ return False
+
+ def record_star_by_name_id(self, star_name: str, star_id: str):
+ record, is_stars_exists, _ = self.check_has_record()
+ if not record:
+ record, stars = {}, []
+ else:
+ if not is_stars_exists:
+ stars = []
+ else:
+ stars = record["stars"]
+ for star in stars:
+ if star["id"].lower() == star_id.lower():
+ return True
+ stars.append({"name": star_name, "id": star_id.lower()})
+ record["stars"] = stars
+ return self.renew_record(record)
+
+ def record_id_by_id_stars(self, id: str, stars: list):
+ record, _, is_vs_exists = self.check_has_record()
+ if not record:
+ record, vs = {}, []
+ else:
+ if not is_vs_exists:
+ vs = []
+ else:
+ vs = record["vs"]
+ for v in vs:
+ if v["id"].lower() == id.lower():
+ return True
+ vs.append({"id": id.lower(), "stars": stars})
+ record["vs"] = vs
+ return self.renew_record(record)
+
+ def undo_record_star_by_id(self, star_id: str):
+ record, exists, _ = self.check_has_record()
+ if not record or not exists:
+ return False
+ stars = record["stars"]
+ exists = False
+ for i, star in enumerate(stars):
+ if star["id"].lower() == star_id.lower():
+ del stars[i]
+ exists = True
+ break
+ if exists:
+ record["stars"] = stars
+ return self.renew_record(record)
+ return True
+
+ def undo_record_id(self, id: str):
+ record, _, exists = self.check_has_record()
+ if not record or not exists:
+ return False
+ vs = record["vs"]
+ exists = False
+ for i, v in enumerate(vs):
+ if v["id"].lower() == id.lower():
+ del vs[i]
+ exists = True
+ break
+ if exists:
+ record["vs"] = vs
+ return self.renew_record(record)
+ return True
+
+
+class BotCacheDb:
+ CACHE_BT = {
+ "prefix": "bt-",
+ "expire": 3600 * 24 * 30,
+ }
+ CACHE_V = {
+ "prefix": "v-",
+ "expire": 3600 * 24 * 30,
+ }
+ CACHE_STAR = {
+ "prefix": "star-",
+ "expire": 0, # never expire
+ }
+ CACHE_RANK = {
+ "prefix": "rank-",
+ "expire": 3600 * 24 * 7,
+ }
+ CACHE_SAMPLE = {
+ "prefix": "sample-",
+ "expire": 3600 * 24 * 30,
+ }
+ CACHE_MAGNET = {
+ "prefix": "magnet-",
+ "expire": 3600 * 24 * 5,
+ }
+ CACHE_PV = {
+ "prefix": "pv-",
+ "expire": 3600 * 24 * 15,
+ }
+ CACHE_FV = {
+ "prefix": "fv-",
+ "expire": 3600 * 24 * 15,
+ }
+ CACHE_STARS_MSG = {
+ "prefix": "stars-msg-",
+ "expire": 3600 * 24 * 5,
+ }
+ CACHE_COMMENT = {"prefix": "comment-", "expire": 3600 * 24 * 30}
+ CACHE_NICE_VS_OF_STAR = {
+ "prefix": "nice-vs-of-star-",
+ "expire": 3600 * 24 * 15,
+ }
+ CACHE_JLIB_PAGE_NICE_VS = {
+ "prefix": "jlib-page-nice-vs-",
+ "expire": 3600 * 24 * 7,
+ }
+ CACHE_JLIB_PAGE_NEW_VS = {
+ "prefix": "jlib-page-new-vs-",
+ "expire": 3600 * 24 * 2,
+ }
+ CACHE_STAR_JA_NAME = {"prefix": "star-ja-name-", "expire": 3600 * 24 * 30 * 6}
+ CACHE_NEW_VS_OF_STAR = {
+ "prefix": "new-vs-of-star-",
+ "expire": 3600 * 24 * 12,
+ }
+
+ TYPE_V = 1
+ TYPE_STAR = 2
+ TYPE_RANK = 3
+ TYPE_SAMPLE = 4
+ TYPE_MAGNET = 5
+ TYPE_PV = 6
+ TYPE_FV = 7
+ TYPE_STARS_MSG = 8
+ TYPE_COMMENT = 10
+ TYPE_NICE_VS_OF_STAR = 11
+ TYPE_JLIB_PAGE_NICE_VS = 12
+ TYPE_JLIB_PAGE_NEW_VS = 13
+ TYPE_STAR_JA_NAME = 14
+ TYPE_NEW_VS_OF_STAR = 16
+ TYPE_BT = 17
+
+ TYPE_MAP = {
+ TYPE_V: CACHE_V,
+ TYPE_STAR: CACHE_STAR,
+ TYPE_RANK: CACHE_RANK,
+ TYPE_SAMPLE: CACHE_SAMPLE,
+ TYPE_MAGNET: CACHE_MAGNET,
+ TYPE_PV: CACHE_PV,
+ TYPE_FV: CACHE_FV,
+ TYPE_STARS_MSG: CACHE_STARS_MSG,
+ TYPE_COMMENT: CACHE_COMMENT,
+ TYPE_NICE_VS_OF_STAR: CACHE_NICE_VS_OF_STAR,
+ TYPE_JLIB_PAGE_NICE_VS: CACHE_JLIB_PAGE_NICE_VS,
+ TYPE_JLIB_PAGE_NEW_VS: CACHE_JLIB_PAGE_NEW_VS,
+ TYPE_STAR_JA_NAME: CACHE_STAR_JA_NAME,
+ TYPE_NEW_VS_OF_STAR: CACHE_NEW_VS_OF_STAR,
+ TYPE_BT: CACHE_BT,
+ }
+
+ def __init__(self, path_cache_file: str, use_cache: str = "1"):
+ self.use_cache = use_cache
+ self.path_cache_file = path_cache_file
+ self.cache = None
+ self._lock = threading.Lock()
+ if self.use_cache == "1":
+ try:
+ os.makedirs(os.path.dirname(self.path_cache_file), exist_ok=True)
+ self.cache = sqlite3.connect(
+ self.path_cache_file, check_same_thread=False
+ )
+ self.cache.execute(
+ "CREATE TABLE IF NOT EXISTS cache ("
+ "key TEXT PRIMARY KEY, value TEXT NOT NULL, expire_at INTEGER DEFAULT 0)"
+ )
+ self.cache.commit()
+ LOG.info(f"Connecting to the SQLite cache: {self.path_cache_file}")
+ except Exception as e:
+ self.cache = None
+ LOG.error(f"Unable to open SQLite cache: {self.path_cache_file}: {e}")
+
+ def remove_cache(self, key: str, type: int):
+ if self.use_cache == "0" or not self.cache:
+ return
+ key = str(key).lower()
+ cache_key = f"{BotCacheDb.TYPE_MAP[type]['prefix']}{key}"
+ try:
+ with self._lock:
+ self.cache.execute("DELETE FROM cache WHERE key=?", (cache_key,))
+ self.cache.commit()
+ except Exception as e:
+ LOG.error(f"Failed to delete cache: {cache_key}: {e}")
+
+ def set_cache(self, key: str, value, type: int, expire=None):
+ """
+ Set cache.
+
+ :param str key: Key
+ :param any value: Value
+ :param int type: Cache type
+ :param int expire: Cache expiration time (in seconds), defaults to using predefined time
+ """
+ if self.use_cache == "0" or not self.cache:
+ return
+ key = str(key).lower()
+ if not expire:
+ expire = BotCacheDb.TYPE_MAP[type]["expire"]
+ prefix = BotCacheDb.TYPE_MAP[type]["prefix"]
+ cache_key = f"{prefix}{key}"
+ expire_at = 0 if expire == 0 else int(time.time()) + expire
+ try:
+ with self._lock:
+ self.cache.execute(
+ "INSERT OR REPLACE INTO cache (key, value, expire_at) VALUES (?, ?, ?)",
+ (cache_key, json.dumps(value), expire_at),
+ )
+ self.cache.commit()
+ except Exception as e:
+ LOG.error(f"Failed to set cache: {cache_key}: {e}")
+
+ def get_cache(self, key, type: int):
+ if self.use_cache == "0" or not self.cache:
+ return
+ key = str(key).lower()
+ cache_key = f"{BotCacheDb.TYPE_MAP[type]['prefix']}{key}"
+ try:
+ with self._lock:
+ cur = self.cache.execute(
+ "SELECT value, expire_at FROM cache WHERE key=?", (cache_key,)
+ )
+ row = cur.fetchone()
+ if not row:
+ return
+ value, expire_at = row
+ if expire_at and expire_at < int(time.time()):
+ with self._lock:
+ self.cache.execute("DELETE FROM cache WHERE key=?", (cache_key,))
+ self.cache.commit()
+ return
+ return json.loads(value)
+ except Exception as e:
+ LOG.error(f"Failed to retrieve cache: {cache_key}: {e}")
diff --git a/telegram/tg-search-bot/docker-compose.yml b/telegram/tg-search-bot/docker-compose.yml
new file mode 100644
index 0000000..dd44c83
--- /dev/null
+++ b/telegram/tg-search-bot/docker-compose.yml
@@ -0,0 +1,16 @@
+version: "3"
+services:
+ tg_search_bot:
+ image: python:3.10
+ restart: always
+ working_dir: /app
+ environment:
+ - TZ=Asia/Shanghai
+ network_mode: "host"
+ volumes:
+ - .:/app
+ - ~/.tg_search_bot:/root/.tg_search_bot
+ command: >
+ sh -c "
+ pip3 install -U -r requirements.txt &&
+ python3 bot.py"
diff --git a/telegram/tg-search-bot/pikpak.py b/telegram/tg-search-bot/pikpak.py
new file mode 100644
index 0000000..37d5bbe
--- /dev/null
+++ b/telegram/tg-search-bot/pikpak.py
@@ -0,0 +1,296 @@
+# -*- coding: UTF-8 -*-
+"""PikPak 官方 OpenAPI 客户端。
+
+通过用户名 / 密码登录获取 token，再调用官方 drive API 直接提交磁力链接的
+离线下载任务，取代原先「向 PikPak6_Bot 发送消息」的保存方式。
+
+参考实现: https://pypi.org/project/pikpakapi/
+"""
+
+import hashlib
+import json
+import logging
+import os
+import time
+
+import requests
+
+LOG = logging.getLogger(__name__)
+
+CLIENT_ID = "YNxT9w7GMdWvEOKa"
+CLIENT_SECRET = "dbw2OtmVEeuUvIptb1Coyg"
+CLIENT_VERSION = "1.47.1"
+PACKAGE_NAME = "com.pikcloud.pikpak"
+
+API_HOST = "https://api-drive.mypikpak.com"
+USER_HOST = "https://user.mypikpak.com"
+
+# PikPak 客户端内置的验签盐值
+SALTS = [
+ "Gez0T9ijiI9WCeTsKSg3SMlx",
+ "zQdbalsolyb1R/",
+ "ftOjr52zt51JD68C3s",
+ "yeOBMH0JkbQdEFNNwQ0RI9T3wU/v",
+ "BRJrQZiTQ65WtMvwO",
+ "je8fqxKPdQVJiy1DM6Bc9Nb1",
+ "niV",
+ "9hFCW2R1",
+ "sHKHpe2i96",
+ "p7c5E6AcXQ/IJUuAEC9W6",
+ "",
+ "aRv9hjc9P+Pbn+u3krN6",
+ "BzStcgE8qVdqjEH16l4",
+ "SqgeZvL5j9zoHP95xWHt",
+ "zVof5yaJkPe3VFpadPof",
+]
+
+
+def captcha_sign(device_id: str, timestamp: str) -> str:
+ """生成登录所需的 captcha_sign。"""
+ sign = CLIENT_ID + CLIENT_VERSION + PACKAGE_NAME + device_id + timestamp
+ for salt in SALTS:
+ sign = hashlib.md5((sign + salt).encode()).hexdigest()
+ return "1." + sign
+
+
+class PikPakClient:
+ def __init__(
+ self,
+ username: str = "",
+ password: str = "",
+ proxy: str = "",
+ token_path: str = "",
+ ):
+ self.username = username or ""
+ self.password = password or ""
+ self.proxy = proxy or ""
+ self.proxies = {"http": self.proxy, "https": self.proxy} if self.proxy else None
+ self.token_path = token_path or os.path.expanduser(
+ "~/.tg_search_bot/pikpak_token.json"
+ )
+ self.access_token = None
+ self.refresh_token = None
+ self.user_id = None
+ self.device_id = None
+ self.expires_at = 0
+ self._load_token()
+
+ @property
+ def enabled(self):
+ return bool(self.username and self.password)
+
+ # ---- token 持久化 ----
+ def _load_token(self):
+ try:
+ if os.path.exists(self.token_path):
+ with open(self.token_path, "r", encoding="utf8") as f:
+ data = json.load(f)
+ self.access_token = data.get("access_token")
+ self.refresh_token = data.get("refresh_token")
+ self.user_id = data.get("user_id")
+ self.device_id = data.get("device_id")
+ self.expires_at = data.get("expires_at", 0)
+ except Exception as e:
+ LOG.error(f"Failed to load PikPak token: {e}")
+
+ def _save_token(self):
+ try:
+ os.makedirs(os.path.dirname(self.token_path), exist_ok=True)
+ with open(self.token_path, "w", encoding="utf8") as f:
+ json.dump(
+ {
+ "access_token": self.access_token,
+ "refresh_token": self.refresh_token,
+ "user_id": self.user_id,
+ "device_id": self.device_id,
+ "expires_at": self.expires_at,
+ },
+ f,
+ ensure_ascii=False,
+ indent=4,
+ )
+ except Exception as e:
+ LOG.error(f"Failed to save PikPak token: {e}")
+
+ # ---- 基础请求 ----
+ def _device_id(self) -> str:
+ if not self.device_id:
+ self.device_id = hashlib.md5(
+ f"{self.username}{self.password}".encode()
+ ).hexdigest()
+ self._save_token()
+ return self.device_id
+
+ def _post_json(self, url: str, data: dict, headers: dict = None, timeout: int = 30):
+ h = {"Content-Type": "application/json; charset=utf-8"}
+ if headers:
+ h.update(headers)
+ return requests.post(
+ url, json=data, headers=h, proxies=self.proxies, timeout=timeout
+ )
+
+ def _post_form(self, url: str, data: dict, headers: dict = None, timeout: int = 30):
+ h = {"Content-Type": "application/x-www-form-urlencoded"}
+ if headers:
+ h.update(headers)
+ return requests.post(
+ url, data=data, headers=h, proxies=self.proxies, timeout=timeout
+ )
+
+ def _authorized_headers(self) -> dict:
+ return {
+ "Authorization": f"Bearer {self.access_token}",
+ "X-Device-Id": self._device_id(),
+ }
+
+ # ---- 登录 / 刷新 ----
+ def _captcha_init(self, action: str):
+ device_id = self._device_id()
+ ts = str(int(time.time() * 1000))
+ meta = {
+ "captcha_sign": captcha_sign(device_id, ts),
+ "client_version": CLIENT_VERSION,
+ "package_name": PACKAGE_NAME,
+ "user_id": self.user_id or "",
+ "timestamp": ts,
+ "username": self.username,
+ }
+ payload = {
+ "client_id": CLIENT_ID,
+ "action": action,
+ "device_id": device_id,
+ "meta": meta,
+ }
+ resp = self._post_json(f"{USER_HOST}/v1/shield/captcha/init", payload)
+ if resp.status_code != 200:
+ LOG.error(
+ f"PikPak captcha init failed: {resp.status_code} {resp.text[:300]}"
+ )
+ return None
+ data = resp.json()
+ token = data.get("captcha_token")
+ if not token:
+ LOG.error(f"PikPak captcha init returned no token: {data}")
+ return None
+ return token
+
+ def login(self):
+ if not self.enabled:
+ return False, "未配置 PikPak 账号"
+ captcha_token = self._captcha_init(f"POST:{USER_HOST}/v1/auth/signin")
+ if not captcha_token:
+ return False, "验证码初始化失败"
+ data = {
+ "client_id": CLIENT_ID,
+ "client_secret": CLIENT_SECRET,
+ "password": self.password,
+ "username": self.username,
+ "captcha_token": captcha_token,
+ }
+ resp = self._post_form(f"{USER_HOST}/v1/auth/signin", data)
+ if resp.status_code != 200:
+ reason = "登录失败（账号或密码错误）"
+ try:
+ j = resp.json()
+ reason = j.get("error_description") or j.get("error") or reason
+ except Exception:
+ pass
+ LOG.error(f"PikPak login failed: {resp.status_code} {resp.text[:300]}")
+ return False, reason
+ j = resp.json()
+ if "access_token" not in j:
+ reason = j.get("error_description") or j.get("error") or "登录失败"
+ LOG.error(f"PikPak login returned error: {j}")
+ return False, reason
+ self.access_token = j["access_token"]
+ self.refresh_token = j["refresh_token"]
+ self.user_id = j.get("sub")
+ self.expires_at = int(time.time()) + int(j.get("expires_in", 7200))
+ self._save_token()
+ LOG.info("PikPak login success")
+ return True, "登录成功"
+
+ def refresh(self):
+ if not self.refresh_token:
+ return False, "缺少 refresh_token"
+ data = {
+ "client_id": CLIENT_ID,
+ "refresh_token": self.refresh_token,
+ "grant_type": "refresh_token",
+ }
+ resp = self._post_json(f"{USER_HOST}/v1/auth/token", data)
+ if resp.status_code != 200:
+ LOG.error(f"PikPak refresh failed: {resp.status_code} {resp.text[:300]}")
+ return False, "令牌刷新失败"
+ j = resp.json()
+ if "access_token" not in j:
+ return False, j.get("error_description") or j.get("error") or "令牌刷新失败"
+ self.access_token = j["access_token"]
+ self.refresh_token = j["refresh_token"]
+ self.user_id = j.get("sub")
+ self.expires_at = int(time.time()) + int(j.get("expires_in", 7200))
+ self._save_token()
+ return True, "刷新成功"
+
+ def ensure_login(self):
+ if not self.enabled:
+ return False, "未配置 PikPak 账号"
+ if not self.access_token:
+ return self.login()
+ if self.expires_at and time.time() > self.expires_at - 60:
+ ok, msg = self.refresh()
+ if ok:
+ return True, msg
+ return self.login()
+ return True, "已登录"
+
+ # ---- 离线下载 ----
+ def offline_download(
+ self, magnet_url: str, name: str = None, parent_id: str = None
+ ):
+ """将磁力链接提交为 PikPak 离线下载任务。
+
+ :return tuple[bool, str]: (是否成功, 提示信息)
+ """
+ if not self.enabled:
+ return False, "未配置 PikPak 账号"
+ ok, msg = self.ensure_login()
+ if not ok:
+ return False, msg
+ data = {
+ "kind": "drive#file",
+ "upload_type": "UPLOAD_TYPE_URL",
+ "url": {"url": magnet_url},
+ "folder_type": "DOWNLOAD" if not parent_id else "",
+ }
+ if name:
+ data["name"] = name
+ if parent_id:
+ data["parent_id"] = parent_id
+ for attempt in range(2):
+ resp = self._post_json(
+ f"{API_HOST}/drive/v1/files", data, headers=self._authorized_headers()
+ )
+ if resp.status_code == 200:
+ j = resp.json()
+ if "error" in j:
+ if j.get("error_code") == 16 and attempt == 0:
+ self.refresh()
+ continue
+ return (
+ False,
+ j.get("error_description") or j.get("error") or "未知错误",
+ )
+ return True, "已提交离线任务到 PikPak"
+ if resp.status_code in (401, 403) and attempt == 0:
+ self.refresh()
+ continue
+ reason = f"HTTP {resp.status_code}"
+ try:
+ j = resp.json()
+ reason = j.get("error_description") or j.get("error") or reason
+ except Exception:
+ pass
+ LOG.error(f"PikPak offline download failed: {resp.text[:300]}")
+ return False, reason
+ return False, "离线任务提交失败"
diff --git a/telegram/tg-search-bot/requirements.txt b/telegram/tg-search-bot/requirements.txt
new file mode 100644
index 0000000..67a8328
--- /dev/null
+++ b/telegram/tg-search-bot/requirements.txt
@@ -0,0 +1,6 @@
+langdetect==1.0.9
+pyTelegramBotAPI==4.14.0
+PyYAML==6.0.1
+Requests==2.31.0
+beautifulsoup4==4.15.0
+Jvav
diff --git a/telegram/tg-search-bot/start_windows.bat b/telegram/tg-search-bot/start_windows.bat
new file mode 100644
index 0000000..bc90647
--- /dev/null
+++ b/telegram/tg-search-bot/start_windows.bat
@@ -0,0 +1,61 @@
+@echo off
+chcp 65001 >nul
+echo ========================================
+echo tg-search-bot 部署脚本 (Windows)
+echo ========================================
+echo.
+
+cd /d "%~dp0"
+
+echo [1/4] 检查Python环境...
+python --version
+if errorlevel 1 (
+ echo ❌ 未找到Python，请先安装Python 3.9+
+ pause
+ exit /b 1
+)
+
+echo.
+echo [2/4] 安装依赖...
+pip install -r requirements.txt
+if errorlevel 1 (
+ echo ❌ 依赖安装失败
+ pause
+ exit /b 1
+)
+
+echo.
+echo [3/4] 检查配置文件...
+set CONFIG_DIR=%USERPROFILE%\.tg_search_bot
+if not exist "%CONFIG_DIR%" (
+ mkdir "%CONFIG_DIR%"
+ echo 已创建配置目录: %CONFIG_DIR%
+)
+
+if not exist "%CONFIG_DIR%\config.yaml" (
+ echo.
+ echo ⚠️ 配置文件不存在，请先创建: %CONFIG_DIR%\config.yaml
+ echo.
+ echo 配置模板:
+ echo tg_chat_id: 你的Telegram用户ID
+ echo tg_bot_token: 你的Bot Token
+ echo use_proxy: 0
+ echo enable_nsfw: 0
+ echo.
+ pause
+ exit /b 1
+)
+echo ✅ 配置文件存在
+
+echo.
+echo [4/4] 启动机器人...
+echo.
+echo ========================================
+echo 机器人启动中...
+echo 按 Ctrl+C 停止
+echo ========================================
+echo.
+
+python bot.py
+
+pause