622 lines
22 KiB
Python
622 lines
22 KiB
Python
"""
|
|
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:
|
|
"""检测客户端"""
|
|
|
|
def __init__(self):
|
|
self.backend_url = BACKEND_URL
|
|
self.browser = None
|
|
self.failed_websites = [] # 记录失败的网站
|
|
|
|
def init_browser(self):
|
|
"""初始化浏览器"""
|
|
co = ChromiumOptions()
|
|
if BROWSER_TYPE == 'edge':
|
|
co.set_browser_path(r"C:\Program Files (x86)\Microsoft\Edge\Application\msedge.exe")
|
|
co.set_argument('--no-sandbox')
|
|
co.set_argument('--disable-dev-shm-usage')
|
|
if HEADLESS_MODE:
|
|
co.set_argument('--headless')
|
|
self.browser = ChromiumPage(addr_or_opts=co)
|
|
print("浏览器初始化成功")
|
|
|
|
def close_browser(self):
|
|
"""关闭浏览器"""
|
|
if self.browser:
|
|
try:
|
|
self.browser.quit()
|
|
except Exception as e:
|
|
print(f"关闭浏览器失败: {e}")
|
|
finally:
|
|
self.browser = None
|
|
|
|
def get_all_users(self) -> List[Dict]:
|
|
"""获取所有用户"""
|
|
try:
|
|
response = requests.get(f"{self.backend_url}/api/users")
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except requests.RequestException as e:
|
|
print(f"获取用户失败: {e}")
|
|
return []
|
|
|
|
def get_user_websites(self, user_id: int) -> List[Dict]:
|
|
"""获取用户的网站列表"""
|
|
try:
|
|
response = requests.get(f"{self.backend_url}/api/websites?user_id={user_id}")
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except requests.RequestException as e:
|
|
print(f"获取用户 {user_id} 的网站失败: {e}")
|
|
return []
|
|
|
|
def get_website_cookies(self, website_id: int) -> List[Dict]:
|
|
"""获取网站的cookie"""
|
|
try:
|
|
response = requests.get(f"{self.backend_url}/api/cookies/websites/{website_id}")
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except requests.RequestException as e:
|
|
print(f"获取网站 {website_id} 的cookie失败: {e}")
|
|
return []
|
|
|
|
def update_website_cookies(self, website_id: int, cookies: List[Dict]):
|
|
"""更新网站的cookie"""
|
|
try:
|
|
response = requests.put(
|
|
f"{self.backend_url}/api/cookies/websites/{website_id}",
|
|
json={'cookies': cookies}
|
|
)
|
|
response.raise_for_status()
|
|
print(f"网站 {website_id} 的cookie已更新")
|
|
except requests.RequestException as e:
|
|
print(f"更新网站 {website_id} 的cookie失败: {e}")
|
|
|
|
def save_detection_result(self, website_id: int, status: int, response_time: float,
|
|
http_status: int, message: str):
|
|
"""保存检测结果"""
|
|
try:
|
|
response = requests.post(
|
|
f"{self.backend_url}/api/detections",
|
|
json={
|
|
'website_id': website_id,
|
|
'status': status,
|
|
'response_time': response_time,
|
|
'http_status': http_status,
|
|
'message': message
|
|
}
|
|
)
|
|
response.raise_for_status()
|
|
except requests.RequestException as e:
|
|
print(f"保存检测结果失败: {e}")
|
|
|
|
def send_notification(self, user_id: int, website_id: int, title: str, content: str):
|
|
"""发送通知"""
|
|
try:
|
|
response = requests.post(
|
|
f"{self.backend_url}/api/notifications",
|
|
json={
|
|
'user_id': user_id,
|
|
'website_id': website_id,
|
|
'title': title,
|
|
'content': content
|
|
}
|
|
)
|
|
response.raise_for_status()
|
|
print(f"通知已发送")
|
|
except requests.RequestException as e:
|
|
print(f"发送通知失败: {e}")
|
|
|
|
def import_cookies_to_browser(self, cookies: List[Dict], domain: str) -> int:
|
|
"""导入cookies到浏览器"""
|
|
if not self.browser or not cookies:
|
|
return 0
|
|
|
|
success_count = 0
|
|
for cookie in cookies:
|
|
try:
|
|
if not cookie.get('name') or not cookie.get('value'):
|
|
continue
|
|
|
|
cookie_dict = {
|
|
'name': cookie.get('name'),
|
|
'value': cookie.get('value'),
|
|
'domain': cookie.get('domain', domain),
|
|
'path': cookie.get('path', '/')
|
|
}
|
|
|
|
self.browser.set.cookies(cookie_dict)
|
|
success_count += 1
|
|
except Exception as e:
|
|
print(f"导入cookie失败: {e}")
|
|
|
|
return success_count
|
|
|
|
def verify_login(self, url: str, check_selector: str = "", success_text: str = "") -> bool:
|
|
"""
|
|
验证登录状态
|
|
|
|
Args:
|
|
url: 网站地址
|
|
check_selector: 登录检测选择器
|
|
success_text: 登录成功时显示的文本
|
|
|
|
Returns:
|
|
是否已登录
|
|
"""
|
|
if not self.browser:
|
|
return False
|
|
|
|
try:
|
|
self.browser.get(url)
|
|
self.browser.wait.doc_loaded(timeout=PAGE_LOAD_TIMEOUT)
|
|
time.sleep(3)
|
|
|
|
# 刷新页面几次确保登录状态
|
|
for i in range(3):
|
|
time.sleep(3)
|
|
self.browser.refresh()
|
|
self.browser.wait.doc_loaded(timeout=PAGE_LOAD_TIMEOUT)
|
|
time.sleep(3)
|
|
|
|
if not check_selector and not success_text:
|
|
return False
|
|
|
|
# 检查选择器
|
|
if check_selector:
|
|
try:
|
|
element = self.browser.ele(check_selector, timeout=10)
|
|
if element:
|
|
element_text = element.text or ""
|
|
if not success_text:
|
|
return True
|
|
if success_text in element_text:
|
|
return True
|
|
except:
|
|
pass
|
|
|
|
# 检查成功文本
|
|
if success_text:
|
|
try:
|
|
page_text = self.browser.html or ""
|
|
if success_text in page_text:
|
|
return True
|
|
except:
|
|
pass
|
|
|
|
return False
|
|
|
|
except Exception as e:
|
|
print(f"验证登录状态失败: {e}")
|
|
return False
|
|
|
|
def get_current_cookies(self) -> List[Dict]:
|
|
"""获取当前浏览器的所有cookies"""
|
|
if not self.browser:
|
|
return []
|
|
|
|
try:
|
|
return self.browser.cookies()
|
|
except Exception as e:
|
|
print(f"获取cookies失败: {e}")
|
|
return []
|
|
|
|
def check_website(self, user: Dict, website: Dict) -> bool:
|
|
"""
|
|
检测单个网站
|
|
|
|
Args:
|
|
user: 用户信息
|
|
website: 网站信息
|
|
|
|
Returns:
|
|
是否登录成功
|
|
"""
|
|
website_id = website.get('id')
|
|
website_name = website.get('name')
|
|
url = website.get('url')
|
|
check_selector = website.get('login_check_selector', '')
|
|
success_text = website.get('success_text', '')
|
|
|
|
print(f"\n检测网站: {website_name} ({url})")
|
|
print(f"用户: {user.get('name')}")
|
|
|
|
start_time = time.time()
|
|
|
|
try:
|
|
# 获取网站的cookie
|
|
cookies = self.get_website_cookies(website_id)
|
|
if not cookies:
|
|
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
|
|
parsed = urlparse(url)
|
|
domain = parsed.netloc
|
|
self.import_cookies_to_browser(cookies, domain)
|
|
|
|
# 访问网站
|
|
self.browser.get(url)
|
|
self.browser.wait.doc_loaded(timeout=PAGE_LOAD_TIMEOUT)
|
|
time.sleep(3)
|
|
|
|
# 验证登录状态，如果未登录则以30秒为周期连续检测6次
|
|
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" [OK] 登录验证成功")
|
|
|
|
# 获取最新cookie并保存到后台
|
|
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="登录验证成功"
|
|
)
|
|
return True
|
|
else:
|
|
print(f" [FAIL] 登录验证失败（含Cookie Cloud重试）")
|
|
|
|
# 记录失败信息
|
|
self.failed_websites.append({
|
|
'user': user,
|
|
'website': website,
|
|
'reason': '登录验证失败（含Cookie Cloud重试）'
|
|
})
|
|
|
|
# 保存检测结果
|
|
self.save_detection_result(
|
|
website_id=website_id,
|
|
status=0,
|
|
response_time=response_time,
|
|
http_status=200,
|
|
message="登录验证失败（含Cookie Cloud重试）"
|
|
)
|
|
return False
|
|
|
|
except Exception as e:
|
|
response_time = time.time() - start_time
|
|
print(f" [FAIL] 检测异常: {e}")
|
|
|
|
# 记录失败信息
|
|
self.failed_websites.append({
|
|
'user': user,
|
|
'website': website,
|
|
'reason': str(e)
|
|
})
|
|
|
|
# 保存检测结果
|
|
self.save_detection_result(
|
|
website_id=website_id,
|
|
status=0,
|
|
response_time=response_time,
|
|
http_status=0,
|
|
message=f"检测异常: {e}"
|
|
)
|
|
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:
|
|
"""
|
|
检查登录状态，如果未登录则重复检测
|
|
|
|
Args:
|
|
url: 网站地址
|
|
check_selector: 登录检测选择器
|
|
success_text: 登录成功文本
|
|
retry_count: 最大重试次数
|
|
retry_interval: 重试间隔（秒）
|
|
|
|
Returns:
|
|
是否登录成功
|
|
"""
|
|
# 第一次检测
|
|
if self._is_logged_in(check_selector, success_text):
|
|
return True
|
|
|
|
print(f" 首次检测未登录，开始周期性检测（最多{retry_count}次，间隔{retry_interval}秒）")
|
|
|
|
for i in range(1, retry_count + 1):
|
|
print(f" 第 {i}/{retry_count} 次检测...")
|
|
time.sleep(retry_interval)
|
|
|
|
# 刷新页面
|
|
self.browser.refresh()
|
|
self.browser.wait.doc_loaded(timeout=PAGE_LOAD_TIMEOUT)
|
|
time.sleep(3)
|
|
|
|
if self._is_logged_in(check_selector, success_text):
|
|
print(f" 第 {i} 次检测登录成功")
|
|
return True
|
|
|
|
return False
|
|
|
|
def _is_logged_in(self, check_selector: str, success_text: str) -> bool:
|
|
"""
|
|
判断是否已登录
|
|
|
|
Returns:
|
|
是否已登录
|
|
"""
|
|
if not check_selector and not success_text:
|
|
return False
|
|
|
|
# 检查选择器
|
|
if check_selector:
|
|
try:
|
|
element = self.browser.ele(check_selector, timeout=10)
|
|
if element:
|
|
element_text = element.text or ""
|
|
if not success_text:
|
|
return True
|
|
if success_text in element_text:
|
|
return True
|
|
except:
|
|
pass
|
|
|
|
# 检查成功文本
|
|
if success_text:
|
|
try:
|
|
page_text = self.browser.html or ""
|
|
if success_text in page_text:
|
|
return True
|
|
except:
|
|
pass
|
|
|
|
return False
|
|
|
|
def send_failed_notifications(self):
|
|
"""发送失败网站的通知"""
|
|
if not self.failed_websites:
|
|
return
|
|
|
|
print(f"\n发现 {len(self.failed_websites)} 个失败的网站，发送通知")
|
|
|
|
# 按用户分组
|
|
user_notifications = {}
|
|
for failed in self.failed_websites:
|
|
user_id = failed['user'].get('id')
|
|
if user_id not in user_notifications:
|
|
user_notifications[user_id] = {
|
|
'user': failed['user'],
|
|
'websites': []
|
|
}
|
|
user_notifications[user_id]['websites'].append(failed)
|
|
|
|
# 为每个用户发送通知
|
|
for user_id, notif_data in user_notifications.items():
|
|
user = notif_data['user']
|
|
websites = notif_data['websites']
|
|
|
|
# 构建通知内容
|
|
content_lines = [
|
|
f"**{user.get('name')} Cookie登录报告**",
|
|
f"",
|
|
f"检测时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
|
|
f"失败数量: {len(websites)} 个网站",
|
|
f"",
|
|
"失败详情:"
|
|
]
|
|
|
|
for failed in websites:
|
|
website = failed['website']
|
|
reason = failed['reason']
|
|
content_lines.append(f"- {website.get('name')}: {reason}")
|
|
|
|
content = "\n".join(content_lines)
|
|
|
|
# 发送通知
|
|
if websites:
|
|
self.send_notification(
|
|
user_id=user_id,
|
|
website_id=websites[0]['website'].get('id'),
|
|
title=f"Cookie登录失败 ({len(websites)} 个网站)",
|
|
content=content
|
|
)
|
|
|
|
def run_detection_cycle(self):
|
|
"""执行一轮检测"""
|
|
print(f"\n{'='*50}")
|
|
print(f"开始执行检测任务 - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
|
print(f"{'='*50}")
|
|
|
|
self.failed_websites = []
|
|
|
|
# 初始化浏览器
|
|
self.init_browser()
|
|
|
|
try:
|
|
# 获取所有用户
|
|
users = self.get_all_users()
|
|
if not users:
|
|
print("未找到用户，跳过检测")
|
|
return
|
|
|
|
print(f"找到 {len(users)} 个用户")
|
|
|
|
# 分用户处理
|
|
for user in users:
|
|
print(f"\n处理用户: {user.get('name')}")
|
|
|
|
# 获取用户的网站
|
|
websites = self.get_user_websites(user.get('id'))
|
|
if not websites:
|
|
print(f" 用户 {user.get('name')} 没有网站，跳过")
|
|
continue
|
|
|
|
print(f" 找到 {len(websites)} 个网站")
|
|
|
|
# 分网站处理
|
|
for website in websites:
|
|
if website.get('status', 1) != 1:
|
|
print(f" 网站 {website.get('name')} 已禁用，跳过")
|
|
continue
|
|
|
|
self.check_website(user, website)
|
|
|
|
# 发送失败通知
|
|
self.send_failed_notifications()
|
|
|
|
except Exception as e:
|
|
print(f"检测任务执行异常: {e}")
|
|
finally:
|
|
self.close_browser()
|
|
|
|
print(f"\n{'='*50}")
|
|
print(f"检测任务完成 - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
|
print(f"{'='*50}")
|
|
|
|
def run(self):
|
|
"""运行检测客户端"""
|
|
print("Cookie监控检测客户端启动")
|
|
|
|
# 执行一轮检测
|
|
self.run_detection_cycle()
|
|
|
|
|
|
def main():
|
|
"""主函数"""
|
|
client = DetectionClient()
|
|
client.run()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|