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()