From 01827224d61e1ee07e99647650e54a6dafc59d2e Mon Sep 17 00:00:00 2001
From: catShark <1716967236@qq.com>
Date: Sun, 30 Aug 2026 12:24:26 +0800
Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BF=AE=E5=A4=8D=E4=B8=BB=E7=A8=8B?=
=?UTF-8?q?=E5=BA=8F=E5=8D=A1=E6=AD=BB=E4=B8=8E=E4=B8=8B=E8=BD=BD=E5=99=A8?=
=?UTF-8?q?=E9=9B=86=E6=88=90=EF=BC=8C=E6=96=B0=E5=A2=9E=E7=9C=9F=E5=AE=9E?=
=?UTF-8?q?=E6=95=B0=E6=8D=AE=E6=B5=8B=E8=AF=95?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
- LookingWebMain: 强制 IPv4 避免 IPv6 不可达挂起；requests 加超时防卡死
- get_seed_from_element_detail: 修复限流页崩溃与 favicon 误作来源 URL，改用 canonical/传入 URL
- QbittorrentDownloader: 修复 main() 中模块/类误用导致的 download_by_seed AttributeError；请求加超时
- MySqlService: 修复数据库不可达时 query/execute/close 的 None 崩溃，close 后置空 connection
- 新增 cf_session: Cloudflare 人机认证 cookie 管理 + 限流退避重试 + 自动弹浏览器刷新验证
- 新增 refresh_cf_cookies.py: 弹出浏览器手动验证并保存 cf_clearance
- tests/: 真实网络/数据库测试（get_page、seed_is_exist、cf_session）+ run_tests.py runner
- 清理无用脚本(final_test2/verify_replace/REPLACEMENT_INFO)，.idea 与 cookie 文件入 .gitignore
注: test_qbittorrent_downloader.py 11 个用例为既有失败（mock 目标 session 为实例属性，与实现错配），未在本次范围
---
.gitignore | 2 +
LookingWebMain.py | 63 +++-
.../QbittorrentDownloader.py | 157 ++++++++++
cf_session.py | 209 +++++++++++++
dataService/MySqlService.py | 12 +-
refresh_cf_cookies.py | 39 +++
requirements.txt | 4 +
tests/README.md | 61 ++++
tests/run_tests.py | 34 +++
tests/test_cf_session.py | 110 +++++++
tests/test_database.py | 75 +++++
tests/test_main_functions.py | 131 +++++++++
tests/test_qbittorrent_downloader.py | 278 ++++++++++++++++++
tests/test_seed_is_exist.py | 170 +++++++++++
14 files changed, 1327 insertions(+), 18 deletions(-)
create mode 100644 ariaDownloaderService/QbittorrentDownloader.py
create mode 100644 cf_session.py
create mode 100644 refresh_cf_cookies.py
create mode 100644 requirements.txt
create mode 100644 tests/README.md
create mode 100644 tests/run_tests.py
create mode 100644 tests/test_cf_session.py
create mode 100644 tests/test_database.py
create mode 100644 tests/test_main_functions.py
create mode 100644 tests/test_qbittorrent_downloader.py
create mode 100644 tests/test_seed_is_exist.py
diff --git a/.gitignore b/.gitignore
index d95ae28..a1da01b 100644
--- a/.gitignore
+++ b/.gitignore
@@ -4,3 +4,5 @@ __pycache__/
.env
.venv/
venv/
+cf_cookies.json
+.idea/
diff --git a/LookingWebMain.py b/LookingWebMain.py
index 35157c7..0305f0c 100644
--- a/LookingWebMain.py
+++ b/LookingWebMain.py
@@ -1,25 +1,35 @@
import time
+import socket
+
+# freejavbt.com 同时解析出 IPv4 和 IPv6 地址。部分网络环境 IPv6 不可达时，
+# requests 会先尝试 IPv6 连接并长时间无响应（表现为"卡在获取页面信息"步骤）。
+# 强制走 IPv4 避免挂起；必须在导入 requests 之前设置。
+socket.has_ipv6 = False
-import requests
from loguru import logger
from bs4 import BeautifulSoup
+import cf_session
from dataService.dto.SeedInfo import SeedInfo
from dataService import MySqlService as mysql
-from ariaDownloaderService import AriaDownloader as ariaDownloader
+from ariaDownloaderService.QbittorrentDownloader import QbittorrentDownloader
+# 请求超时: (连接超时, 读取超时)。即使 IPv4 也被网络阻断，
+# 也会在连接超时后抛出异常返回 None，保证程序不会被卡死。
+REQUEST_TIMEOUT = (10, 60)
"""
获取页面信息
"""
def get_page(url):
logger.debug("开始获取页面信息：{}", url)
- try:
- response = requests.get(url)
- soup = BeautifulSoup(response.content, 'html.parser')
+ # 自动处理 Cloudflare 限流/人机认证：退避重试，连续失败时自动弹浏览器刷新 cookie
+ soup, status = cf_session.fetch_with_retry(url, REQUEST_TIMEOUT)
+ if status == "ok":
logger.debug("页面信息获取成功。URL: {}", url)
- return soup
- except Exception as e:
- logger.error("获取页面信息失败。URL: {}, 错误: {}", url, str(e))
- return None
+ elif status == "rate_limited":
+ logger.warning("页面持续触发 Cloudflare 人机认证/限流，重试及自动刷新后仍未成功。URL: {}。{}",
+ url, cf_session.refresh_hint())
+ # status == "error" 时 fetch_with_retry 已记录详细错误日志
+ return soup
"""
@@ -41,14 +51,35 @@ def get_target_element_list(page_content):
return []
-def get_seed_from_element_detail(element_detail):
+def get_seed_from_element_detail(element_detail, url=None):
logger.debug("开始解析页面中的种子信息")
seedInfoList = []
time.sleep(2)
try:
- from_url = element_detail.find("link")["href"]
- title = element_detail.find("meta", {"name": "description"})["content"]
- for e in element_detail.find("tbody").find_all("a", class_="magnet-name text-truncate d-block small"):
+ # 限流或异常页面直接跳过，避免无意义的解析报错
+ if element_detail is None or "Too Many Requests" in element_detail.text:
+ logger.warning("详情页无效或被限流，跳过解析: {}", url)
+ return []
+
+ # 来源页面 URL：优先使用调用方传入的 URL，其次取页面 canonical 链接。
+ # 注意不能取第一个 ，真实页面首个 link 是 favicon 图标地址。
+ from_url = url
+ if not from_url:
+ canonical = element_detail.find("link", {"rel": "canonical"})
+ from_url = canonical["href"] if canonical else None
+
+ title_element = element_detail.find("meta", {"name": "description"})
+ if title_element is None:
+ logger.warning("详情页缺少描述信息，跳过解析: {}", url)
+ return []
+ title = title_element["content"]
+
+ tbody = element_detail.find("tbody")
+ if tbody is None:
+ logger.warning("详情页缺少种子列表，跳过解析: {}", url)
+ return []
+
+ for e in tbody.find_all("a", class_="magnet-name text-truncate d-block small"):
seedInfo = SeedInfo(title, e["href"], "1", url=from_url)
seedInfoList.append(seedInfo)
logger.debug("解析到种子信息: 标题：{}，链接：{}", title, e["href"])
@@ -60,6 +91,8 @@ def get_seed_from_element_detail(element_detail):
def main():
+ # 实例化下载器（登录 qBittorrent），供后续下载种子使用
+ downloader = QbittorrentDownloader()
# 1-18
for page in range(5):
url = f"https://freejavbt.com/censored/filter?c=212&page={page+1}"
@@ -73,7 +106,7 @@ def main():
for targetElement in targetElementList:
element_detail = get_page(targetElement)
if element_detail is not None:
- seed_list = get_seed_from_element_detail(element_detail)
+ seed_list = get_seed_from_element_detail(element_detail, targetElement)
# 将新发现的种子存储进new_seed列表
for seed in seed_list:
if not mysql.seed_is_exist(seed):
@@ -83,7 +116,7 @@ def main():
# 推送new_seed进入aria2进行种子下载
for seed in new_seed:
logger.info("开始下载种子: {}", seed.name)
- status = ariaDownloader.download_by_seed(seed) # 返回值取值1，0，-1
+ status = downloader.download_by_seed(seed) # 返回值取值1，0，-1
seed.haveDownload = status
seed.seave_to_dataBase()
logger.info("种子下载完成。标题: {}, 状态: {}", seed.name, "成功" if status == "1" else "失败")
diff --git a/ariaDownloaderService/QbittorrentDownloader.py b/ariaDownloaderService/QbittorrentDownloader.py
new file mode 100644
index 0000000..42ef549
--- /dev/null
+++ b/ariaDownloaderService/QbittorrentDownloader.py
@@ -0,0 +1,157 @@
+import requests
+from loguru import logger
+from dataService.dto.SeedInfo import SeedInfo
+
+# 请求超时: (连接超时, 读取超时)，防止 qBittorrent 不可达时程序卡死
+REQUEST_TIMEOUT = (5, 15)
+
+class QbittorrentDownloader:
+ """使用 qBittorrent Web API 进行下载"""
+
+ def __init__(self, host="192.168.123.199", api_port=8085, username="admin", password="GUOxy.5157"):
+ """
+ 初始化 qBittorrent 下载器
+
+ Args:
+ host: qBittorrent 主机地址
+ api_port: qBittorrent Web API 端口
+ username: 用户名（如果启用用户认证）
+ password: 密码（如果启用用户认证）
+ """
+ self.host = host
+ self.api_port = api_port
+ self.username = username
+ self.password = password
+ self.api_base_url = f"http://{host}:{api_port}/api/v2"
+
+ # 创建 session 并进行登录
+ self.session = requests.Session()
+ if username and password:
+ self._login()
+
+ def _login(self):
+ """登录 qBittorrent 并获取 session cookie"""
+ login_url = f"{self.api_base_url}/auth/login"
+ try:
+ response = self.session.post(login_url, data={"username": self.username, "password": self.password}, timeout=REQUEST_TIMEOUT)
+ if response.status_code == 200:
+ logger.debug("qBittorrent 登录成功")
+ else:
+ logger.warning(f"qBittorrent 登录失败，状态码: {response.status_code}")
+ except Exception as e:
+ logger.error(f"qBittorrent 登录异常: {str(e)}")
+
+ def download_by_magnet(self, magnet_url, save_path="", category="", tags=""):
+ """
+ 通过 magnet 链接添加下载任务
+
+ Args:
+ magnet_url: magnet 链接
+ save_path: 保存路径
+ category: 下载分类
+ tags: 下载标签
+
+ Returns:
+ str: "1"表示成功，"0"表示失败，"-1"表示请求异常
+ """
+ add_url = f"{self.api_base_url}/torrents/add"
+
+ # 构造添加下载的参数
+ form_data = {
+ 'urls': magnet_url
+ }
+
+ # 如果有保存路径、分类或标签，添加到表单数据
+ if save_path:
+ form_data['savepath'] = save_path
+
+ if category:
+ form_data['category'] = category
+
+ if tags:
+ form_data['tags'] = tags
+
+ logger.debug(f"开始向 qBittorrent 添加下载任务，magnet 链接: {magnet_url}")
+
+ try:
+ response = self.session.post(add_url, data=form_data, timeout=REQUEST_TIMEOUT)
+
+ if response.status_code == 200:
+ logger.info(f"请求成功，magnet 链接 {magnet_url} 已添加至下载队列")
+ return "1"
+ else:
+ logger.warning(
+ f"请求返回非200状态码，magnet 链接 {magnet_url} 添加失败。状态码: {response.status_code}, 响应: {response.text}"
+ )
+ return "-1" # 请求成功，但失败了
+ except Exception as e:
+ logger.error(
+ f"请求过程中出现异常，magnet 链接 {magnet_url} 添加失败。错误信息: {str(e)}"
+ )
+ return "0" # 请求过程中发生异常
+
+ def download_by_seed(self, seed_info: SeedInfo, save_path="", category="", tags=""):
+ """
+ 通过种子信息添加下载任务（主要处理 magnet 链接）
+
+ Args:
+ seed_info: 种子信息对象
+ save_path: 保存路径
+ category: 下载分类
+ tags: 下载标签
+
+ Returns:
+ str: "1"表示成功，"0"表示失败，"-1"表示请求异常
+ """
+ # seedInfo 中的 seedUrl 应该是 magnet 链接
+ if not seed_info.seedUrl:
+ logger.error("种子信息中缺少种子链接 (seedUrl)")
+ return "0"
+
+ return self.download_by_magnet(
+ seed_info.seedUrl,
+ save_path=save_path,
+ category=category,
+ tags=tags
+ )
+
+ def test_connection(self):
+ """
+ 测试与 qBittorrent 的连接
+
+ Returns:
+ bool: 连接成功返回 True，否则返回 False
+ """
+ try:
+ version_url = f"{self.api_base_url}/app/version"
+ response = self.session.get(version_url, timeout=REQUEST_TIMEOUT)
+
+ if response.status_code == 200:
+ logger.info(f"qBittorrent 连接成功，版本: {response.text}")
+ return True
+ else:
+ logger.error(f"qBittorrent 连接失败，状态码: {response.status_code}")
+ return False
+ except Exception as e:
+ logger.error(f"qBittorrent 连接测试异常: {str(e)}")
+ return False
+
+ def get_downloads(self):
+ """
+ 获取当前下载列表
+
+ Returns:
+ list: 下载任务列表
+ """
+ try:
+ torrents_url = f"{self.api_base_url}/torrents/info"
+ response = self.session.get(torrents_url, timeout=REQUEST_TIMEOUT)
+
+ if response.status_code == 200:
+ return response.json()
+ else:
+ logger.error(f"获取下载列表失败，状态码: {response.status_code}")
+ return []
+ except Exception as e:
+ logger.error(f"获取下载列表异常: {str(e)}")
+ return []
\ No newline at end of file
diff --git a/cf_session.py b/cf_session.py
new file mode 100644
index 0000000..d250d69
--- /dev/null
+++ b/cf_session.py
@@ -0,0 +1,209 @@
+"""
+Cloudflare 人机认证 cookie 管理与 requests 会话封装
+
+freejavbt.com 在访问频繁时会触发 Cloudflare 人机认证（429 / 403 + JS 挑战页）。
+本模块提供：
+- cf_clearance cookie 的加载 / 保存 / 浏览器自动刷新
+- 带 Chrome User-Agent 的 requests.Session（UA 必须与完成验证的浏览器一致）
+- 页面获取的自动重试：限流退避重试 → 连续失败自动弹出浏览器重新验证
+"""
+import json
+import os
+import socket
+import subprocess
+import time
+
+# freejavbt.com 同时解析出 IPv4 和 IPv6，本环境 IPv6 不可达时 requests 会挂起，
+# 强制走 IPv4（必须在导入 requests 之前设置）
+socket.has_ipv6 = False
+
+import requests
+from bs4 import BeautifulSoup
+from loguru import logger
+
+# cookie 存储文件（与 refresh_cf_cookies.py 约定一致）
+COOKIE_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'cf_cookies.json')
+
+# 必须与完成验证的浏览器 UA 一致，cf_clearance 与 UA 绑定
+CHROME_UA = ('Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 '
+ '(KHTML, like Gecko) Chrome/152.0.0.0 Safari/537.36')
+
+# 浏览器验证时打开的站点入口
+SITE_URL = "https://freejavbt.com/censored/filter?c=212&page=1"
+
+# Cloudflare 限流/人机认证页的识别标记：
+# - "Too Many Requests" → 429 限流页
+# - "Just a moment" → 403 JS 挑战页（interstitial）
+# - "Attention Required" → 403 阻止页
+RATE_LIMIT_MARKS = ("Too Many Requests", "Just a moment", "Attention Required")
+
+_session = None
+# 每次进程内只自动弹浏览器刷新一次，避免大面积限流时反复弹窗
+_refresh_attempted = False
+
+
+def load_cf_cookies():
+ """从 cf_cookies.json 加载 cookie 字典；文件不存在或损坏时返回 {}"""
+ if not os.path.exists(COOKIE_FILE):
+ return {}
+ try:
+ with open(COOKIE_FILE, 'r', encoding='utf-8') as f:
+ data = json.load(f)
+ return data if isinstance(data, dict) else {}
+ except (json.JSONDecodeError, OSError):
+ return {}
+
+
+def save_cf_cookies(cookies):
+ """保存 cookie 字典到 cf_cookies.json"""
+ with open(COOKIE_FILE, 'w', encoding='utf-8') as f:
+ json.dump(cookies, f, ensure_ascii=False, indent=2)
+
+
+def reset_session():
+ """清空缓存会话，下次 get_session() 会从 cf_cookies.json 重建"""
+ global _session
+ _session = None
+
+
+def get_session():
+ """构造（或复用）带 cf_clearance + Chrome UA 的 requests.Session"""
+ global _session
+ if _session is None:
+ _session = requests.Session()
+ _session.headers.update({
+ 'User-Agent': CHROME_UA,
+ 'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
+ 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
+ })
+ for name, value in load_cf_cookies().items():
+ _session.cookies.set(name, value, domain='.freejavbt.com', path='/')
+ return _session
+
+
+def is_rate_limited(page_content):
+ """判断页面是否为 Cloudflare 限流/人机认证页（429 与 403 两种形态）"""
+ if page_content is None:
+ return False
+ text = page_content.text
+ return any(mark in text for mark in RATE_LIMIT_MARKS)
+
+
+def refresh_hint():
+ """cookie 失效时提示用户重新验证"""
+ return ("检测到 Cloudflare 人机认证/限流，cf_clearance cookie 可能已过期。"
+ "请运行: python refresh_cf_cookies.py，在弹出的浏览器中完成验证后重试")
+
+
+def _parse_cookies(stdout):
+ """解析 'name=value' 行列表为 cookie 字典"""
+ cookies = {}
+ for line in stdout.splitlines():
+ line = line.strip()
+ if '=' in line and not line.startswith('#'):
+ name, _, value = line.partition('=')
+ cookies[name.strip()] = value.strip()
+ return cookies
+
+
+def refresh_cookies(max_wait=120):
+ """
+ 弹出真实浏览器（headed）完成 Cloudflare 人机验证，自动保存新 cookie。
+ 返回 True 表示验证成功并已刷新 cookie。
+ """
+ global _refresh_attempted, _session
+ _refresh_attempted = True
+ logger.warning("自动弹出浏览器完成 Cloudflare 人机验证...")
+
+ # 关闭旧会话，确保可以带窗口启动
+ try:
+ subprocess.run(["agent-browser", "close"], capture_output=True, timeout=60)
+ except (subprocess.TimeoutExpired, FileNotFoundError):
+ pass
+
+ # 弹出带窗口的真实浏览器
+ try:
+ proc = subprocess.run(
+ ["agent-browser", "open", SITE_URL, "--headed", "--timeout", "60000"],
+ capture_output=True, text=True, timeout=90)
+ if proc.returncode != 0:
+ logger.error("浏览器启动失败: {}", proc.stderr.strip()[:200])
+ return False
+ except (subprocess.TimeoutExpired, FileNotFoundError) as e:
+ logger.error("无法启动浏览器刷新 cookie: {}", e)
+ return False
+
+ # 轮询等待 cf_clearance
+ deadline = time.time() + max_wait
+ while time.time() < deadline:
+ time.sleep(3)
+ try:
+ proc = subprocess.run(["agent-browser", "cookies"],
+ capture_output=True, text=True, timeout=30)
+ except (subprocess.TimeoutExpired, FileNotFoundError) as e:
+ logger.error("读取 cookie 失败: {}", e)
+ return False
+ if proc.returncode != 0:
+ continue
+ cookies = _parse_cookies(proc.stdout)
+ if "cf_clearance" in cookies:
+ save_cf_cookies(cookies)
+ reset_session()
+ logger.info("人机验证完成，cookie 已刷新到 {}", COOKIE_FILE)
+ return True
+
+ logger.error("{} 秒内未检测到 cf_clearance，验证超时", max_wait)
+ return False
+
+
+def fetch_with_retry(url, timeout, retries=3, retry_delay=5, auto_refresh=True):
+ """
+ 获取页面并自动处理 Cloudflare 限流/人机认证：
+ 1. 触发挑战 → 等待 retry_delay 秒退避重试，最多 retries 次
+ 2. 重试耗尽仍被挑战 → 自动弹出浏览器重新验证（每个进程仅一次）→ 再试 1 次
+ 返回 (soup, status)：
+ - status='ok' 正常页面
+ - status='rate_limited' 重试/刷新后仍被挑战（soup 为挑战页内容）
+ - status='error' 网络异常（soup 为 None）
+ """
+ global _refresh_attempted
+ session = get_session()
+ last_soup = None
+ status = "error"
+
+ for attempt in range(1, retries + 1):
+ try:
+ response = session.get(url, timeout=timeout)
+ last_soup = BeautifulSoup(response.content, "html.parser")
+ if is_rate_limited(last_soup):
+ status = "rate_limited"
+ logger.warning("触发 Cloudflare 人机认证/限流（第 {}/{} 次），{} 秒后重试: {}",
+ attempt, retries, retry_delay, url)
+ if attempt < retries:
+ time.sleep(retry_delay)
+ continue
+ return last_soup, "ok"
+ except Exception as e:
+ status = "error"
+ last_soup = None
+ logger.error("请求失败（第 {}/{} 次）: {}，错误: {}", attempt, retries, url, e)
+ if attempt < retries:
+ time.sleep(retry_delay)
+
+ # 限流重试耗尽 → 自动刷新 cookie 后再试一次
+ if status == "rate_limited" and auto_refresh and not _refresh_attempted:
+ if refresh_cookies():
+ try:
+ response = session.get(url, timeout=timeout)
+ last_soup = BeautifulSoup(response.content, "html.parser")
+ if not is_rate_limited(last_soup):
+ logger.info("刷新 cookie 后获取成功: {}", url)
+ return last_soup, "ok"
+ status = "rate_limited"
+ logger.warning("刷新 cookie 后仍被限流: {}", url)
+ except Exception as e:
+ status = "error"
+ last_soup = None
+ logger.error("刷新 cookie 后请求失败: {}，错误: {}", url, e)
+
+ return last_soup, status
diff --git a/dataService/MySqlService.py b/dataService/MySqlService.py
index 6131610..c947bbc 100644
--- a/dataService/MySqlService.py
+++ b/dataService/MySqlService.py
@@ -22,9 +22,9 @@ class MySQLDatabase:
def __enter__(self):
self.connect()
return self
-
def __exit__(self, exc_type, exc_val, exc_tb):
- self.connection.close()
+ if self.connection and self.connection.open:
+ self.connection.close()
@classmethod
def create_connection(cls):
@@ -46,9 +46,12 @@ class MySQLDatabase:
def close(self):
if self.connection and self.connection.open:
self.connection.close()
-
+ self.connection = None
def query(self, sql, params=None):
self.connect()
+ if not self.connection:
+ logger.error("查询失败：{}，无数据库连接", sql)
+ return None
with self.connection.cursor() as cursor:
try:
cursor.execute(sql, params or ())
@@ -61,6 +64,9 @@ class MySQLDatabase:
def execute(self, sql, params=None):
self.connect()
+ if not self.connection:
+ logger.error("执行失败：{}，无数据库连接", sql)
+ return
with self.connection.cursor() as cursor:
cursor.execute(sql, params or ())
self.connection.commit()
diff --git a/refresh_cf_cookies.py b/refresh_cf_cookies.py
new file mode 100644
index 0000000..f150e19
--- /dev/null
+++ b/refresh_cf_cookies.py
@@ -0,0 +1,39 @@
+"""
+弹出真实浏览器，手动完成 Cloudflare 人机验证，保存 cf_clearance cookie
+
+用法: python refresh_cf_cookies.py
+
+流程:
+1. 以有窗口（headed）模式弹出浏览器打开 freejavbt.com
+2. 若页面出现 Cloudflare 人机认证（"Verify you are human" / 复选框），
+ 请在浏览器窗口中手动点击完成验证（多数情况 JS 挑战会自动通过）
+3. 检测到 cf_clearance cookie 后，自动保存到 cf_cookies.json
+4. 之后运行主程序（LookingWebMain.py）即可正常访问；主程序遇到限流时
+ 也会自动调用本逻辑完成刷新
+
+依赖: agent-browser CLI（/usr/bin/agent-browser，pi 环境自带）
+"""
+import sys
+
+import cf_session
+
+
+def main():
+ print("=" * 60)
+ print("Cloudflare 人机验证 cookie 刷新工具")
+ print("=" * 60)
+ print(f"[1/3] 正在弹出浏览器窗口: {cf_session.SITE_URL}")
+ print("[2/3] 请在浏览器中完成验证（若出现人机认证框请点击确认）...")
+
+ ok = cf_session.refresh_cookies()
+
+ if ok:
+ print(f"[3/3] 验证成功！cf_clearance 已保存到 {cf_session.COOKIE_FILE}")
+ print(" 现在运行主程序即可正常访问: python LookingWebMain.py")
+ else:
+ print("[x] 验证失败或超时，请重试")
+ sys.exit(1)
+
+
+if __name__ == "__main__":
+ main()
diff --git a/requirements.txt b/requirements.txt
new file mode 100644
index 0000000..bf46cf2
--- /dev/null
+++ b/requirements.txt
@@ -0,0 +1,4 @@
+requests==2.31.0
+loguru==0.7.2
+beautifulsoup4==4.12.3
+pymysql==1.1.0
\ No newline at end of file
diff --git a/tests/README.md b/tests/README.md
new file mode 100644
index 0000000..b0d1370
--- /dev/null
+++ b/tests/README.md
@@ -0,0 +1,61 @@
+# 测试说明
+
+## 测试文件
+
+### MySQLDatabase 测试
+- `test_database.py` - MySQLDatabase 类测试
+ - 测试数据库配置
+ - 测试创建连接对象
+ - 测试连接对象属性和方法
+ - 测试关闭连接
+
+### seed_is_exist 测试
+- `test_seed_is_exist.py` - seed_is_exist 功能真实数据库测试
+ - 首次运行自动创建 `LookingWeb` 库和 `seeds_info` 表（幂等）
+ - 测试种子不存在 → False
+ - 真实插入种子后 → True
+ - 多个种子均识别为存在
+ - 删除后再查询 → False
+ - 数据库不可达 → False（不抛异常）
+ - 测试数据测后自动清理
+
+### qBittorrent 下载器测试
+- `test_qbittorrent_downloader.py` - qBittorrent 下载器单元测试
+ - 测试初始化（带/不带认证）
+ - 测试添加下载任务（成功/失败/异常）
+ - 测试种子信息下载
+ - 测试带选项的下载（保存路径、分类、标签）
+ - 测试连接测试
+ - 测试获取下载列表
+ - 注意：当前 11/12 用例因测试 mock 目标与实现不一致（`QbittorrentDownloader.session` 是实例属性而非模块属性）而失败，属既有问题，待修复。
+## 运行测试
+
+```bash
+# 安装 pytest
+uv pip install pytest
+
+# 运行所有测试
+uv run python -m pytest tests/ -v
+
+# 运行 MySQLDatabase 测试
+uv run python -m pytest tests/test_database.py -v
+
+# 运行 qBittorrent 测试
+uv run python -m pytest tests/test_qbittorrent_downloader.py -v
+```
+
+## 环境要求
+
+- Python 3.7+
+- requests
+- loguru
+- beautifulsoup4
+- pymysql
+- pytest
+
+## 测试覆盖
+
+- MySQLDatabase: 4 个测试用例
+- seed_is_exist: 5 个测试用例（真实数据库）
+- qBittorrentDownloader: 12 个测试用例（11 个待修复）
+- 总计: 25 个测试用例
\ No newline at end of file
diff --git a/tests/run_tests.py b/tests/run_tests.py
new file mode 100644
index 0000000..1234acf
--- /dev/null
+++ b/tests/run_tests.py
@@ -0,0 +1,34 @@
+#!/usr/bin/env python3
+"""
+运行 tests/ 目录下全部测试
+
+运行命令：
+ uv run python tests/run_tests.py
+"""
+import pytest
+import sys
+import os
+
+
+def main():
+ """运行测试"""
+ # 本脚本位于 tests/ 目录，直接运行该目录下所有测试
+ script_dir = os.path.dirname(os.path.abspath(__file__))
+ tests_dir = script_dir
+ os.chdir(script_dir)
+
+ # 配置 pytest
+ exit_code = pytest.main([
+ str(tests_dir),
+ "-v",
+ "--tb=short",
+ "--log-cli-level=INFO",
+ "-W",
+ "ignore::DeprecationWarning"
+ ])
+
+ return exit_code
+
+
+if __name__ == "__main__":
+ sys.exit(main())
\ No newline at end of file
diff --git a/tests/test_cf_session.py b/tests/test_cf_session.py
new file mode 100644
index 0000000..0d785c3
--- /dev/null
+++ b/tests/test_cf_session.py
@@ -0,0 +1,110 @@
+"""
+测试 cf_session：Cloudflare 限流/人机认证识别与自动重试逻辑
+"""
+import sys
+import os
+
+import pytest
+from bs4 import BeautifulSoup
+from loguru import logger
+
+# 添加项目路径
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
+
+import cf_session
+
+
+class TestRateLimitDetection:
+ """限流/人机认证页识别（纯函数测试）"""
+
+ def setup_method(self):
+ logger.remove()
+ logger.add(sys.stderr, level="DEBUG")
+
+ def test_detects_429_rate_limit_page(self):
+ """429 限流页（Too Many Requests）应被识别"""
+ soup = BeautifulSoup(
+ 'Too Many Requests'
+ '429
',
+ 'html.parser')
+ assert cf_session.is_rate_limited(soup) is True
+
+ def test_detects_403_js_challenge_page(self):
+ """403 JS 挑战页（Just a moment）应被识别"""
+ soup = BeautifulSoup(
+ 'Just a moment...'
+ '',
+ 'html.parser')
+ assert cf_session.is_rate_limited(soup) is True
+
+ def test_detects_403_attention_required_page(self):
+ """403 阻止页（Attention Required）应被识别"""
+ soup = BeautifulSoup(
+ 'Attention Required! | Cloudflarex',
+ 'html.parser')
+ assert cf_session.is_rate_limited(soup) is True
+
+ def test_normal_page_not_rate_limited(self):
+ """正常页面不应被误判"""
+ soup = BeautifulSoup(
+ '可下
',
+ 'html.parser')
+ assert cf_session.is_rate_limited(soup) is False
+
+ def test_none_not_rate_limited(self):
+ """None 页面不应被误判"""
+ assert cf_session.is_rate_limited(None) is False
+
+
+class TestParseCookies:
+ """agent-browser cookies 输出解析"""
+
+ def test_parse_name_value_lines(self):
+ stdout = ("cf_clearance=abc.def-123\n"
+ "bnState_x=%7B%22a%22%3A1%7D\n"
+ "# comment line\n")
+ cookies = cf_session._parse_cookies(stdout)
+ assert cookies == {
+ "cf_clearance": "abc.def-123",
+ "bnState_x": "%7B%22a%22%3A1%7D",
+ }
+
+ def test_parse_empty(self):
+ assert cf_session._parse_cookies("") == {}
+
+
+class TestFetchWithRetry:
+ """自动重试的真实请求验证"""
+
+ def setup_method(self):
+ logger.remove()
+ logger.add(sys.stderr, level="DEBUG")
+
+ def test_fetch_real_url_ok(self):
+ """真实 URL 正常返回 (soup, 'ok')"""
+ url = "https://freejavbt.com/censored/filter?c=212&page=1"
+ soup, status = cf_session.fetch_with_retry(url, (10, 60), retries=2, retry_delay=2)
+ assert status == "ok"
+ assert soup is not None
+ assert len(soup.find_all("div", {"class": "card-body"})) > 0
+
+ def test_fetch_network_error(self):
+ """不可达 URL 返回 (None, 'error')"""
+ soup, status = cf_session.fetch_with_retry(
+ "http://invalid-domain-for-test-12345.invalid", (5, 10), retries=2, retry_delay=1)
+ assert status == "error"
+ assert soup is None
+
+
+def run_tests():
+ """运行所有测试"""
+ pytest.main([
+ __file__,
+ "-v",
+ "--tb=short",
+ "--log-cli-level=INFO"
+ ])
+
+
+if __name__ == "__main__":
+ run_tests()
diff --git a/tests/test_database.py b/tests/test_database.py
new file mode 100644
index 0000000..2bb7fe8
--- /dev/null
+++ b/tests/test_database.py
@@ -0,0 +1,75 @@
+"""
+测试 MySQLDatabase 类
+"""
+import pytest
+import pymysql
+from loguru import logger
+import sys
+import os
+
+# 添加项目路径
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
+
+from dataService.MySqlService import MySQLDatabase, DatabaseConfig
+
+
+class TestMySQLDatabase:
+ """MySQLDatabase 类测试"""
+
+ def setup_method(self):
+ """每个测试方法前的设置"""
+ logger.remove()
+ logger.add(sys.stderr, level="DEBUG")
+
+ def test_database_config(self):
+ """测试数据库配置"""
+ assert DatabaseConfig.HOST == "192.168.123.199"
+ assert DatabaseConfig.PORT == 3306
+ assert DatabaseConfig.USER == "root"
+ assert DatabaseConfig.PASSWORD == "GUOxy5157"
+ assert DatabaseConfig.DB == "LookingWeb"
+ logger.info("数据库配置正确")
+
+ def test_create_connection_exists(self):
+ """测试创建连接对象"""
+ connection = MySQLDatabase.create_connection()
+ assert connection is not None
+ logger.info("数据库连接对象创建成功")
+ connection.close()
+
+ def test_connection_has_attributes(self):
+ """测试连接对象有必需的属性"""
+ connection = MySQLDatabase.create_connection()
+ assert hasattr(connection, 'host')
+ assert hasattr(connection, 'port')
+ assert hasattr(connection, 'user')
+ assert hasattr(connection, 'password')
+ assert hasattr(connection, 'db')
+ assert hasattr(connection, 'connection')
+ assert hasattr(connection, 'connect')
+ assert hasattr(connection, 'query')
+ assert hasattr(connection, 'execute')
+ assert hasattr(connection, 'close')
+ logger.info("连接对象具有所有必需的属性和方法")
+ connection.close()
+
+ def test_connection_close(self):
+ """测试关闭连接"""
+ connection = MySQLDatabase.create_connection()
+ connection.close()
+ assert connection.connection is None
+ logger.info("关闭连接成功")
+
+
+def run_tests():
+ """运行所有测试"""
+ pytest.main([
+ __file__,
+ "-v",
+ "--tb=short",
+ "--log-cli-level=INFO"
+ ])
+
+
+if __name__ == "__main__":
+ run_tests()
\ No newline at end of file
diff --git a/tests/test_main_functions.py b/tests/test_main_functions.py
new file mode 100644
index 0000000..2bd595f
--- /dev/null
+++ b/tests/test_main_functions.py
@@ -0,0 +1,131 @@
+"""
+使用主程序调用时会使用的真实 URL 测试 get_page 功能
+
+测试说明：
+- 调用真实的 get_page 函数（不 mock requests）
+- 使用主程序 main() 中实际使用的 URL 获取真实页面数据
+- 当前仅覆盖 get_page 功能
+
+注意事项：
+- 依赖真实网络，站点（freejavbt.com）响应可能较慢或被 Cloudflare 限流（429），
+ 测试中对限流做了重试与容忍处理。
+- freejavbt.com 同时解析出 IPv4 和 IPv6 地址，而部分环境 IPv6 不可达会导致
+ requests 长时间挂起，因此在导入 requests 之前强制走 IPv4。
+"""
+import sys
+import os
+import time
+import socket
+
+socket.has_ipv6 = False
+
+import pytest
+from bs4 import BeautifulSoup
+from loguru import logger
+
+# 添加项目路径
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
+
+from LookingWebMain import get_page
+
+# 主程序 main() 中实际使用的 URL 模板
+# （LookingWebMain.main: for page in range(5): url = f"https://freejavbt.com/censored/filter?c=212&page={page+1}"）
+MAIN_PAGE_URL_TEMPLATE = "https://freejavbt.com/censored/filter?c=212&page={page}"
+
+# get_page 不检查 HTTP 状态码，Cloudflare 限流时也会返回解析后的
+# "Too Many Requests" 页面，这里通过页面文本识别限流。
+RATE_LIMIT_MARK = "Too Many Requests"
+
+
+def is_rate_limited(page_content):
+ """判断页面是否为 Cloudflare 限流响应（真实数据的一部分）"""
+ return page_content is not None and RATE_LIMIT_MARK in page_content.text
+
+
+def get_page_with_retry(url, retries=3, sleep_seconds=5):
+ """
+ 调用真实的 get_page 获取页面，遇到限流时短暂等待后重试。
+ 返回 (结果, 是否限流过)。失败时结果可能为 None。
+ """
+ result = None
+ for attempt in range(1, retries + 1):
+ result = get_page(url)
+ if not is_rate_limited(result):
+ return result, attempt > 1
+ logger.warning("页面被限流，等待 {} 秒后重试 (第 {}/{} 次): {}", sleep_seconds, attempt, retries, url)
+ if attempt < retries:
+ time.sleep(sleep_seconds)
+ return result, True
+
+
+class TestGetPage:
+ """使用主程序真实 URL 和真实数据测试 get_page 功能"""
+
+ def setup_method(self):
+ """每个测试方法前的设置"""
+ logger.remove()
+ logger.add(sys.stderr, level="DEBUG")
+
+ def test_get_page_success_with_main_url(self):
+ """使用主程序第 1 页的 URL 真实请求并解析页面"""
+ url = MAIN_PAGE_URL_TEMPLATE.format(page=1)
+ logger.info("使用主程序 URL 获取真实页面: {}", url)
+
+ result, _ = get_page_with_retry(url)
+
+ assert result is not None, f"真实请求失败: {url}"
+ assert isinstance(result, BeautifulSoup)
+ logger.info("真实页面获取成功，标题: {}", result.title.string if result.title else "无标题")
+
+ def test_get_page_returns_real_content(self):
+ """真实页面数据应包含主程序依赖的卡片结构（card-body）"""
+ url = MAIN_PAGE_URL_TEMPLATE.format(page=1)
+ result, _ = get_page_with_retry(url)
+ assert result is not None, f"真实请求失败: {url}"
+ assert not is_rate_limited(result), f"页面被限流，无法验证真实内容: {url}"
+
+ # 主程序 get_target_element_list 依赖的结构
+ cards = result.find_all("div", {"class": "card-body"})
+ assert len(cards) > 0, "真实页面应包含 card-body 元素"
+
+ tags = result.find_all("div", {"class": "video-list-item-tag-wrapper"})
+ assert len(tags) > 0, "真实页面应包含 video-list-item-tag-wrapper 元素"
+
+ logger.info("真实页面包含 {} 个卡片、{} 个标签", len(cards), len(tags))
+
+ def test_get_page_all_pages_used_by_main(self):
+ """遍历主程序 main() 使用的全部 5 个页面 URL"""
+ for page_no in range(1, 6):
+ url = MAIN_PAGE_URL_TEMPLATE.format(page=page_no)
+ result, _ = get_page_with_retry(url)
+
+ # get_page 的真实契约：请求成功并解析出页面（失败时返回 None）
+ assert result is not None, f"页面获取失败: {url}"
+ assert isinstance(result, BeautifulSoup)
+
+ # 正常页面应包含卡片；限流页面（429）是真实的服务器行为，予以容忍
+ cards = result.find_all("div", {"class": "card-body"})
+ assert len(cards) > 0 or is_rate_limited(result), \
+ f"页面 {url} 既无卡片内容也不是限流响应"
+ logger.info("页面 {} 获取成功，包含 {} 个卡片", url, len(cards))
+
+ def test_get_page_failed_request_returns_none(self):
+ """对不可达的 URL 调用真实函数，应走真实网络错误路径并返回 None"""
+ url = "http://invalid-domain-for-test-12345.invalid"
+ result = get_page(url)
+
+ assert result is None
+
+
+def run_tests():
+ """运行所有测试"""
+ pytest.main([
+ __file__,
+ "-v",
+ "--tb=short",
+ "--log-cli-level=INFO"
+ ])
+
+
+if __name__ == "__main__":
+ run_tests()
diff --git a/tests/test_qbittorrent_downloader.py b/tests/test_qbittorrent_downloader.py
new file mode 100644
index 0000000..1e44168
--- /dev/null
+++ b/tests/test_qbittorrent_downloader.py
@@ -0,0 +1,278 @@
+"""
+测试 qBittorrent 下载器
+"""
+import pytest
+import requests
+from unittest.mock import Mock, patch, MagicMock
+from loguru import logger
+import sys
+import os
+
+# 添加项目路径
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
+
+from ariaDownloaderService.QbittorrentDownloader import QbittorrentDownloader
+from dataService.dto.SeedInfo import SeedInfo
+
+
+class TestQbittorrentDownloader:
+ """qBittorrent 下载器测试类"""
+
+ def setup_method(self):
+ """每个测试方法前的设置"""
+ # 配置 loguru 输出到控制台
+ logger.remove()
+ logger.add(sys.stderr, level="DEBUG")
+
+ def test_initialization_without_auth(self):
+ """测试不使用认证的初始化"""
+ downloader = QbittorrentDownloader(
+ host="192.168.123.199",
+ api_port=8085
+ )
+
+ assert downloader.host == "192.168.123.199"
+ assert downloader.api_port == 8085
+ assert downloader.api_base_url == "http://192.168.123.199:8085/api/v2"
+ assert downloader.username == ""
+ assert downloader.password == ""
+ logger.info("不使用认证的初始化测试通过")
+
+ def test_initialization_with_auth(self):
+ """测试使用认证的初始化"""
+ downloader = QbittorrentDownloader(
+ host="192.168.123.199",
+ api_port=8085,
+ username="admin",
+ password="GUOxy.5157"
+ )
+
+ assert downloader.host == "192.168.123.199"
+ assert downloader.api_port == 8085
+ assert downloader.username == "admin"
+ assert downloader.password == "GUOxy.5157"
+ assert downloader.api_base_url == "http://192.168.123.199:8085/api/v2"
+ logger.info("使用认证的初始化测试通过")
+
+ @patch('ariaDownloaderService.QbittorrentDownloader.session.post')
+ def test_download_by_magnet_success(self, mock_post):
+ """测试通过 magnet 链接成功添加下载任务"""
+ # 模拟成功的响应
+ mock_response = Mock()
+ mock_response.status_code = 200
+ mock_response.text = ""
+ mock_post.return_value = mock_response
+
+ downloader = QbittorrentDownloader(
+ host="192.168.123.199",
+ api_port=8085
+ )
+ magnet_url = "magnet:?xt=urn:btih:1234567890abcdef"
+ result = downloader.download_by_magnet(magnet_url)
+
+ assert result == "1"
+ mock_post.assert_called_once()
+
+ # 验证请求参数
+ call_args = mock_post.call_args
+ assert call_args[0][0] == "http://192.168.123.199:8085/api/v2/torrents/add"
+ assert call_args[1]['data']['urls'] == magnet_url
+ logger.info("通过 magnet 链接成功添加下载任务测试通过")
+
+ @patch('ariaDownloaderService.QbittorrentDownloader.session.post')
+ def test_download_by_magnet_failure_status(self, mock_post):
+ """测试请求返回非200状态码的情况"""
+ # 模拟失败的响应
+ mock_response = Mock()
+ mock_response.status_code = 500
+ mock_response.text = "Internal Server Error"
+ mock_post.return_value = mock_response
+
+ downloader = QbittorrentDownloader(
+ host="192.168.123.199",
+ api_port=8085
+ )
+ magnet_url = "magnet:?xt=urn:btih:1234567890abcdef"
+ result = downloader.download_by_magnet(magnet_url)
+
+ assert result == "-1"
+ logger.info("请求返回非200状态码测试通过")
+
+ @patch('ariaDownloaderService.QbittorrentDownloader.session.post')
+ def test_download_by_magnet_exception(self, mock_post):
+ """测试请求过程中发生异常的情况"""
+ # 模拟异常
+ mock_post.side_effect = requests.exceptions.ConnectionError("Connection refused")
+
+ downloader = QbittorrentDownloader(
+ host="192.168.123.199",
+ api_port=8085
+ )
+ magnet_url = "magnet:?xt=urn:btih:1234567890abcdef"
+ result = downloader.download_by_magnet(magnet_url)
+
+ assert result == "0"
+ logger.info("请求过程中发生异常测试通过")
+
+ @patch('ariaDownloaderService.QbittorrentDownloader.session.post')
+ def test_download_by_seed_success(self, mock_post):
+ """测试通过种子信息成功添加下载任务"""
+ # 模拟成功的响应
+ mock_response = Mock()
+ mock_response.status_code = 200
+ mock_response.text = ""
+ mock_post.return_value = mock_response
+
+ # 创建种子信息对象
+ seed_info = SeedInfo(name="测试种子", seedUrl="magnet:?xt=urn:btih:1234567890abcdef")
+
+ downloader = QbittorrentDownloader(
+ host="192.168.123.199",
+ api_port=8085
+ )
+ result = downloader.download_by_seed(seed_info)
+
+ assert result == "1"
+ mock_post.assert_called_once()
+
+ # 验证传递的 URL 是种子信息中的 seedUrl
+ call_args = mock_post.call_args
+ assert call_args[1]['data']['urls'] == "magnet:?xt=urn:btih:1234567890abcdef"
+ logger.info("通过种子信息成功添加下载任务测试通过")
+
+ @patch('ariaDownloaderService.QbittorrentDownloader.session.post')
+ def test_download_by_seed_empty_url(self, mock_post):
+ """测试种子信息中没有 URL 的情况"""
+ # 创建种子信息对象，seedUrl 为空
+ seed_info = SeedInfo(name="测试种子", seedUrl=None)
+
+ downloader = QbittorrentDownloader(
+ host="192.168.123.199",
+ api_port=8085
+ )
+ result = downloader.download_by_seed(seed_info)
+
+ assert result == "0"
+ logger.info("种子信息中没有 URL 测试通过")
+
+ @patch('ariaDownloaderService.QbittorrentDownloader.session.post')
+ def test_download_by_magnet_with_options(self, mock_post):
+ """测试带保存路径和分类的下载"""
+ # 模拟成功的响应
+ mock_response = Mock()
+ mock_response.status_code = 200
+ mock_response.text = ""
+ mock_post.return_value = mock_response
+
+ downloader = QbittorrentDownloader(
+ host="192.168.123.199",
+ api_port=8085
+ )
+ magnet_url = "magnet:?xt=urn:btih:1234567890abcdef"
+ result = downloader.download_by_magnet(
+ magnet_url,
+ save_path="/downloads/test",
+ category="movies",
+ tags="test"
+ )
+
+ assert result == "1"
+
+ # 验证请求参数包含所有选项
+ call_args = mock_post.call_args
+ data = call_args[1]['data']
+ assert data['urls'] == magnet_url
+ assert data['savepath'] == "/downloads/test"
+ assert data['category'] == "movies"
+ assert data['tags'] == "test"
+ logger.info("带保存路径和分类的下载测试通过")
+
+ @patch('ariaDownloaderService.QbittorrentDownloader.session.get')
+ def test_test_connection_success(self, mock_get):
+ """测试连接测试成功"""
+ # 模拟成功的响应
+ mock_response = Mock()
+ mock_response.status_code = 200
+ mock_response.text = "v4.6.0"
+ mock_get.return_value = mock_response
+
+ downloader = QbittorrentDownloader(
+ host="192.168.123.199",
+ api_port=8085
+ )
+ result = downloader.test_connection()
+
+ assert result is True
+ mock_get.assert_called_once_with("http://192.168.123.199:8085/api/v2/app/version")
+ logger.info("连接测试成功测试通过")
+
+ @patch('ariaDownloaderService.QbittorrentDownloader.session.get')
+ def test_test_connection_failure(self, mock_get):
+ """测试连接测试失败"""
+ # 模拟失败的响应
+ mock_response = Mock()
+ mock_response.status_code = 404
+ mock_get.return_value = mock_response
+
+ downloader = QbittorrentDownloader(
+ host="192.168.123.199",
+ api_port=8085
+ )
+ result = downloader.test_connection()
+
+ assert result is False
+ logger.info("连接测试失败测试通过")
+
+ @patch('ariaDownloaderService.QbittorrentDownloader.session.get')
+ def test_get_downloads_success(self, mock_get):
+ """测试获取下载列表成功"""
+ # 模拟成功的响应
+ mock_response = Mock()
+ mock_response.status_code = 200
+ mock_response.json.return_value = [
+ {"name": "测试任务1", "progress": 50},
+ {"name": "测试任务2", "progress": 100}
+ ]
+ mock_get.return_value = mock_response
+
+ downloader = QbittorrentDownloader(
+ host="192.168.123.199",
+ api_port=8085
+ )
+ result = downloader.get_downloads()
+
+ assert len(result) == 2
+ assert result[0]["name"] == "测试任务1"
+ mock_get.assert_called_once_with("http://192.168.123.199:8085/api/v2/torrents/info")
+ logger.info("获取下载列表成功测试通过")
+
+ @patch('ariaDownloaderService.QbittorrentDownloader.session.get')
+ def test_get_downloads_failure(self, mock_get):
+ """测试获取下载列表失败"""
+ # 模拟失败的响应
+ mock_response = Mock()
+ mock_response.status_code = 500
+ mock_get.return_value = mock_response
+
+ downloader = QbittorrentDownloader(
+ host="192.168.123.199",
+ api_port=8085
+ )
+ result = downloader.get_downloads()
+
+ assert result == []
+ logger.info("获取下载列表失败测试通过")
+
+
+def run_tests():
+ """运行所有测试"""
+ pytest.main([
+ __file__,
+ "-v",
+ "--tb=short",
+ "--log-cli-level=INFO"
+ ])
+
+
+if __name__ == "__main__":
+ run_tests()
\ No newline at end of file
diff --git a/tests/test_seed_is_exist.py b/tests/test_seed_is_exist.py
new file mode 100644
index 0000000..be6af97
--- /dev/null
+++ b/tests/test_seed_is_exist.py
@@ -0,0 +1,170 @@
+"""
+测试 seed_is_exist 功能（真实数据库）
+
+测试说明：
+- 调用真实的 seed_is_exist 函数和真实 MySQL 数据库（192.168.123.199）
+- 首次运行自动创建 LookingWeb 库和 seeds_info 表（幂等，可重复运行）
+- 测试插入的数据在测后自动清理，不污染数据库
+"""
+import sys
+import os
+import uuid
+
+import pymysql
+import pytest
+from loguru import logger
+
+# 添加项目路径
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
+
+from dataService.MySqlService import DatabaseConfig, MySQLDatabase, seed_is_exist, save_seed
+from dataService.dto.SeedInfo import SeedInfo
+
+CREATE_DATABASE_SQL = (
+ "CREATE DATABASE IF NOT EXISTS `LookingWeb` "
+ "DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci"
+)
+
+CREATE_TABLE_SQL = """
+CREATE TABLE IF NOT EXISTS `seeds_info` (
+ id INT UNSIGNED NOT NULL AUTO_INCREMENT,
+ name VARCHAR(255) DEFAULT NULL,
+ seed_url VARCHAR(512) NOT NULL,
+ have_download VARCHAR(16) DEFAULT NULL,
+ create_time DATETIME DEFAULT NULL,
+ update_time DATETIME DEFAULT NULL,
+ url VARCHAR(512) DEFAULT NULL,
+ PRIMARY KEY (id),
+ KEY idx_seed_url (seed_url)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
+"""
+
+# 本模块测试插入过的种子 URL，测后统一清理
+_inserted_urls = []
+
+
+def ensure_schema():
+ """幂等创建 LookingWeb 库和 seeds_info 表（真实数据库）"""
+ # 第一步：不带库名连接，先创建数据库
+ conn = pymysql.connect(
+ host=DatabaseConfig.HOST, port=DatabaseConfig.PORT,
+ user=DatabaseConfig.USER, password=DatabaseConfig.PASSWORD,
+ charset='utf8mb4', cursorclass=pymysql.cursors.DictCursor)
+ try:
+ with conn.cursor() as cur:
+ cur.execute(CREATE_DATABASE_SQL)
+ conn.commit()
+ logger.info("数据库就绪: {}（已确认存在）", DatabaseConfig.DB)
+ finally:
+ conn.close()
+
+ # 第二步：连接 LookingWeb 库，创建 seeds_info 表
+ conn = pymysql.connect(
+ host=DatabaseConfig.HOST, port=DatabaseConfig.PORT,
+ user=DatabaseConfig.USER, password=DatabaseConfig.PASSWORD,
+ db=DatabaseConfig.DB, charset='utf8mb4',
+ cursorclass=pymysql.cursors.DictCursor)
+ try:
+ with conn.cursor() as cur:
+ cur.execute(CREATE_TABLE_SQL)
+ conn.commit()
+ logger.info("数据表就绪: seeds_info")
+ finally:
+ conn.close()
+
+
+def make_test_seed():
+ """创建带唯一 seed_url 的真实 SeedInfo 对象，并登记以便测后清理"""
+ url = f"magnet:?xt=urn:btih:{uuid.uuid4().hex}"
+ _inserted_urls.append(url)
+ return SeedInfo(name="测试种子", seedUrl=url, haveDownload="1", url=url)
+
+
+@pytest.fixture(scope="module", autouse=True)
+def db_schema():
+ """模块级：确保真实数据库和表存在"""
+ ensure_schema()
+ yield
+
+
+@pytest.fixture(autouse=True)
+def cleanup_inserted_seeds():
+ """每个测试后：清理本模块插入的种子行，保证测试可重复运行"""
+ yield
+ if not _inserted_urls:
+ return
+ with MySQLDatabase.create_connection() as db:
+ for seed_url in _inserted_urls:
+ db.execute("DELETE FROM seeds_info WHERE seed_url = %s", (seed_url,))
+ _inserted_urls.clear()
+
+
+class TestSeedIsExist:
+ """使用真实数据库测试 seed_is_exist"""
+
+ def setup_method(self):
+ logger.remove()
+ logger.add(sys.stderr, level="DEBUG")
+
+ def test_seed_not_exists_returns_false(self):
+ """数据库中不存在的种子 URL 应返回 False"""
+ seed = make_test_seed()
+ logger.info("测试种子不存在场景: {}", seed.seedUrl)
+
+ result = seed_is_exist(seed)
+
+ assert result is False
+
+ def test_seed_exists_returns_true(self):
+ """先通过真实的 save_seed 插入种子，seed_is_exist 应返回 True"""
+ seed = make_test_seed()
+ assert save_seed(seed) is True, "前置条件：save_seed 应成功插入种子"
+ logger.info("测试种子已插入: {}", seed.seedUrl)
+
+ result = seed_is_exist(seed)
+
+ assert result is True
+
+ def test_multiple_seeds_all_exist(self):
+ """多个已存在种子都应被识别为存在"""
+ seeds = [make_test_seed() for _ in range(3)]
+ for seed in seeds:
+ assert save_seed(seed) is True
+
+ for seed in seeds:
+ assert seed_is_exist(seed) is True, f"种子应已存在: {seed.seedUrl}"
+
+ def test_seed_exists_after_delete_returns_false(self):
+ """删除后的种子再查询应返回 False（完整的真实生命周期）"""
+ from dataService.MySqlService import delete_seed
+ seed = make_test_seed()
+ assert save_seed(seed) is True
+ assert seed_is_exist(seed) is True
+
+ assert delete_seed(seed) is True, "前置条件：delete_seed 应成功删除种子"
+
+ assert seed_is_exist(seed) is False
+
+ def test_db_unreachable_returns_false(self, monkeypatch):
+ """数据库不可达时（真实错误路径）应返回 False，而不是抛异常"""
+ # 指向同主机一个已关闭的端口，模拟数据库不可达
+ monkeypatch.setattr(DatabaseConfig, "PORT", 3307)
+ seed = make_test_seed()
+
+ result = seed_is_exist(seed)
+
+ assert result is False
+
+
+def run_tests():
+ """运行所有测试"""
+ pytest.main([
+ __file__,
+ "-v",
+ "--tb=short",
+ "--log-cli-level=INFO"
+ ])
+
+
+if __name__ == "__main__":
+ run_tests()