项目整理:添加Cookie Cloud下载功能、优化前端界面、配置改为config.json、打包脚本、清理无用文件

This commit is contained in:
bwstudio
2026-04-18 21:48:29 +08:00
parent 70802b9b50
commit be1709b622
14 changed files with 373 additions and 395 deletions
+123 -14
View File
@@ -5,11 +5,20 @@ Cookie监控管理系统 - 检测客户端
然后分用户、分网站登录。
"""
import time
import logging
import requests
from datetime import datetime
from typing import Dict, List, Optional
from DrissionPage import ChromiumPage, ChromiumOptions
from config.config import BACKEND_URL, BROWSER_TYPE, HEADLESS_MODE, PAGE_LOAD_TIMEOUT
from cookie_cloud_client import download_cookies, get_cookies_for_domain
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s [%(name)s] %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
logger = logging.getLogger('client')
class DetectionClient:
@@ -239,15 +248,40 @@ class DetectionClient:
# 获取网站的cookie
cookies = self.get_website_cookies(website_id)
if not cookies:
print(f" 警告: 未找到网站 {website_name} 的cookie")
self.save_detection_result(
website_id=website_id,
status=0,
response_time=time.time() - start_time,
http_status=0,
message="未找到cookie"
)
return False
print(f" 警告: 未找到网站 {website_name} 的cookie,尝试从Cookie Cloud获取...")
login_success = self._try_cookie_cloud_and_retry(user, website, url, check_selector, success_text)
response_time = time.time() - start_time
if login_success:
print(f" [OK] 登录验证成功")
current_cookies = self.get_current_cookies()
if current_cookies:
self.update_website_cookies(website_id, current_cookies)
print(f" 已更新cookie ({len(current_cookies)} 个)")
self.save_detection_result(
website_id=website_id,
status=1,
response_time=response_time,
http_status=200,
message="登录验证成功(Cookie Cloud首次导入)"
)
return True
else:
print(f" [FAIL] Cookie Cloud重试失败")
self.failed_websites.append({
'user': user,
'website': website,
'reason': '未找到cookie且Cookie Cloud重试失败'
})
self.save_detection_result(
website_id=website_id,
status=0,
response_time=response_time,
http_status=0,
message="未找到cookie且Cookie Cloud重试失败"
)
return False
# 导入cookie到浏览器
from urllib.parse import urlparse
@@ -264,8 +298,14 @@ class DetectionClient:
login_success = self._check_login_with_retry(url, check_selector, success_text)
response_time = time.time() - start_time
# 如果仍然失败,尝试从Cookie Cloud下载新cookie并重试
if not login_success:
print(f" 常规检测全部失败,尝试从Cookie Cloud更新cookie...")
login_success = self._try_cookie_cloud_and_retry(user, website, url, check_selector, success_text)
response_time = time.time() - start_time
if login_success:
print(f" 登录验证成功")
print(f" [OK] 登录验证成功")
# 获取最新cookie并保存到后台
current_cookies = self.get_current_cookies()
@@ -283,13 +323,13 @@ class DetectionClient:
)
return True
else:
print(f" 登录验证失败(连续6次检测未登录")
print(f" [FAIL] 登录验证失败(含Cookie Cloud重试")
# 记录失败信息
self.failed_websites.append({
'user': user,
'website': website,
'reason': '登录验证失败(连续6次检测未登录'
'reason': '登录验证失败(含Cookie Cloud重试'
})
# 保存检测结果
@@ -298,13 +338,13 @@ class DetectionClient:
status=0,
response_time=response_time,
http_status=200,
message="登录验证失败(连续6次检测未登录"
message="登录验证失败(含Cookie Cloud重试"
)
return False
except Exception as e:
response_time = time.time() - start_time
print(f" 检测异常: {e}")
print(f" [FAIL] 检测异常: {e}")
# 记录失败信息
self.failed_websites.append({
@@ -323,6 +363,75 @@ class DetectionClient:
)
return False
def _try_cookie_cloud_and_retry(self, user: Dict, website: Dict, url: str,
check_selector: str, success_text: str) -> bool:
"""
尝试从Cookie Cloud下载新cookie并重新检测登录
Returns:
是否登录成功
"""
from urllib.parse import urlparse
parsed = urlparse(url)
domain = parsed.netloc
cc_uuid = user.get('cookie_cloud_uuid')
cc_password = user.get('cookie_cloud_password')
cc_api_url = user.get('cookie_cloud_api_url', 'https://movie-pilot.org/cookiecloud')
if not cc_uuid or not cc_password:
logger.info(f" 用户未配置Cookie Cloud,跳过 (UUID: {cc_uuid})")
return False
logger.info(f" ==================== Cookie Cloud 下载开始 ====================")
logger.info(f" 目标域名: {domain}")
logger.info(f" API地址: {cc_api_url}")
logger.info(f" UUID: {cc_uuid}")
logger.info(f" Password: {'*' * len(cc_password)}")
try:
cookies_data = download_cookies(cc_api_url, cc_uuid, cc_password)
logger.info(f" 下载完成,总域名数: {len(cookies_data) if isinstance(cookies_data, dict) else 0}")
domain_cookies = get_cookies_for_domain(cookies_data, domain)
if not domain_cookies:
logger.info(f" Cookie Cloud中未找到 {domain} 的cookie")
logger.info(f" ==================== Cookie Cloud 下载结束 ====================")
return False
logger.info(f" 从Cookie Cloud获取到 {len(domain_cookies)} 个cookie")
logger.info(f" 导入并重试...")
try:
self.browser.clear_cache()
logger.info(f" 已清除浏览器缓存")
except Exception as e:
logger.warning(f" 清除缓存失败: {e}")
imported = self.import_cookies_to_browser(domain_cookies, domain)
logger.info(f" 导入cookie: {imported}/{len(domain_cookies)} 个成功")
logger.info(f" 访问网站: {url}")
self.browser.get(url)
self.browser.wait.doc_loaded(timeout=PAGE_LOAD_TIMEOUT)
time.sleep(3)
if self._is_logged_in(check_selector, success_text):
logger.info(f" Cookie Cloud cookie登录成功")
logger.info(f" ==================== Cookie Cloud 下载结束 ====================")
return True
logger.info(f" Cookie Cloud cookie登录失败")
logger.info(f" ==================== Cookie Cloud 下载结束 ====================")
return False
except Exception as e:
logger.error(f" Cookie Cloud异常: {e}")
logger.info(f" ==================== Cookie Cloud 下载结束 ====================")
return False
def _check_login_with_retry(self, url: str, check_selector: str, success_text: str,
retry_count: int = 6, retry_interval: int = 30) -> bool:
"""