461 lines
15 KiB
Python
461 lines
15 KiB
Python
"""Cookie 监控主程序"""
|
|
import json
|
|
import time
|
|
import os
|
|
from datetime import datetime, time as dt_time
|
|
from urllib.parse import urlparse
|
|
from cookie_cloud import CookieCloud
|
|
from notifier import IYUUNotifier
|
|
from applogger import setup_logging, get_logger
|
|
from DrissionPage import ChromiumPage, ChromiumOptions
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
NOTIFY_TIMES = [6, 22]
|
|
STATE_FILE = "notify_state.json"
|
|
|
|
|
|
def load_notify_state() -> dict:
|
|
if os.path.exists(STATE_FILE):
|
|
try:
|
|
with open(STATE_FILE, 'r', encoding='utf-8') as f:
|
|
return json.load(f)
|
|
except:
|
|
pass
|
|
return {}
|
|
|
|
|
|
def save_notify_state(state: dict):
|
|
try:
|
|
with open(STATE_FILE, 'w', encoding='utf-8') as f:
|
|
json.dump(state, f, indent=2, ensure_ascii=False)
|
|
except Exception as e:
|
|
logger.error(f"保存通知状态失败: {e}")
|
|
|
|
|
|
def should_send_notification(user_name: str) -> bool:
|
|
return False
|
|
|
|
|
|
def mark_notification_sent(user_name: str):
|
|
now = datetime.now()
|
|
state = load_notify_state()
|
|
|
|
if user_name not in state:
|
|
state[user_name] = {}
|
|
|
|
state[user_name]['date'] = now.strftime('%Y-%m-%d')
|
|
state[user_name]['hour'] = now.hour
|
|
|
|
save_notify_state(state)
|
|
|
|
|
|
class BrowserManager:
|
|
"""浏览器管理器"""
|
|
|
|
def __init__(self, browser_type: str = 'edge', headless: bool = False):
|
|
self.browser_type = browser_type
|
|
self.headless = headless
|
|
self.page = None
|
|
|
|
def init_browser(self):
|
|
"""初始化浏览器"""
|
|
co = ChromiumOptions()
|
|
if self.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 self.headless:
|
|
co.set_argument('--headless')
|
|
self.page = ChromiumPage(addr_or_opts=co)
|
|
|
|
def wait_page_loaded(self, timeout=30):
|
|
"""等待页面加载完成"""
|
|
if self.page:
|
|
self.page.wait.doc_loaded(timeout=timeout)
|
|
self.page.ele('tag:body', timeout=timeout)
|
|
|
|
def ensure_connection(self):
|
|
"""确保浏览器连接正常"""
|
|
try:
|
|
if self.page:
|
|
self.page.get("about:blank")
|
|
return True
|
|
except:
|
|
pass
|
|
|
|
try:
|
|
self.init_browser()
|
|
self.page.get("about:blank")
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"无法建立浏览器连接: {e}")
|
|
return False
|
|
|
|
def import_cookies(self, cookies: list, domain: str):
|
|
"""导入cookies"""
|
|
if not self.ensure_connection():
|
|
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.page.set.cookies(cookie_dict)
|
|
success_count += 1
|
|
except Exception as e:
|
|
pass
|
|
|
|
return success_count
|
|
|
|
def verify_login(self, url: str, check_selector: str = "", success_text: str = "") -> bool:
|
|
"""验证登录状态"""
|
|
if not self.ensure_connection():
|
|
return False
|
|
|
|
try:
|
|
self.page.get(url)
|
|
self.wait_page_loaded()
|
|
|
|
for i in range(3):
|
|
time.sleep(1)
|
|
self.page.refresh()
|
|
self.wait_page_loaded()
|
|
|
|
if not check_selector:
|
|
return True
|
|
|
|
try:
|
|
element = self.page.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
|
|
|
|
return False
|
|
|
|
except Exception as e:
|
|
logger.error(f"验证登录状态失败: {e}")
|
|
return False
|
|
|
|
def close(self):
|
|
"""关闭浏览器"""
|
|
if self.page:
|
|
try:
|
|
self.page.quit()
|
|
except:
|
|
pass
|
|
finally:
|
|
self.page = None
|
|
|
|
|
|
class CookieMonitor:
|
|
"""Cookie 监控器"""
|
|
|
|
def __init__(self, config_file: str = "config.json"):
|
|
self.config_file = config_file
|
|
self.config = self._load_config()
|
|
self.all_results = []
|
|
|
|
def _load_config(self) -> dict:
|
|
try:
|
|
with open(self.config_file, 'r', encoding='utf-8') as f:
|
|
return json.load(f)
|
|
except Exception as e:
|
|
logger.error(f"加载配置文件失败: {e}")
|
|
return {}
|
|
|
|
def process_user(self, user_config: dict):
|
|
"""处理单个用户：依次验证各网站登录状态"""
|
|
user_name = user_config.get('name', '未知用户')
|
|
logger.info(f"{'='*20} 开始处理用户: {user_name} {'='*20}")
|
|
|
|
cookie_cloud_config = user_config.get('cookie_cloud', {})
|
|
notification_config = user_config.get('notification', {})
|
|
browser_config = user_config.get('browser', {})
|
|
websites = user_config.get('websites', [])
|
|
|
|
if not websites:
|
|
logger.info(f"用户 {user_name} 没有配置网站，跳过")
|
|
return
|
|
|
|
cookie_cloud = CookieCloud(
|
|
api_url=cookie_cloud_config.get('api_url', ''),
|
|
uuid=cookie_cloud_config.get('uuid', ''),
|
|
password=cookie_cloud_config.get('password', '')
|
|
)
|
|
|
|
logger.info(f"步骤1: 从 Cookie Cloud 获取所有 cookies")
|
|
all_cookies = cookie_cloud.get_cookies()
|
|
total_domains = len(all_cookies)
|
|
total_cookies = sum(len(c) for c in all_cookies.values())
|
|
logger.info(f"获取到 {total_domains} 个域名，共 {total_cookies} 个 cookie")
|
|
|
|
browser = BrowserManager(
|
|
browser_type=browser_config.get('type', 'edge'),
|
|
headless=browser_config.get('headless', False)
|
|
)
|
|
|
|
user_results = {
|
|
'user_name': user_name,
|
|
'notification_config': notification_config,
|
|
'websites': [],
|
|
'success_count': 0,
|
|
'fail_count': 0
|
|
}
|
|
|
|
try:
|
|
logger.info(f"步骤2: 初始化浏览器")
|
|
browser.init_browser()
|
|
|
|
logger.info(f"步骤3: 依次验证各网站登录状态")
|
|
for website in websites:
|
|
result = self.process_website(
|
|
user_name=user_name,
|
|
website_config=website,
|
|
browser=browser,
|
|
all_cookies=all_cookies
|
|
)
|
|
user_results['websites'].append(result)
|
|
|
|
if result['success']:
|
|
user_results['success_count'] += 1
|
|
else:
|
|
user_results['fail_count'] += 1
|
|
|
|
self.all_results.append(user_results)
|
|
self._print_summary(user_name, user_results['websites'])
|
|
|
|
logger.info(f"\n检查用户 {user_name} 是否需要发送通知")
|
|
if should_send_notification(user_name):
|
|
self._send_user_notification(user_results)
|
|
|
|
finally:
|
|
browser.close()
|
|
|
|
def process_website(
|
|
self,
|
|
user_name: str,
|
|
website_config: dict,
|
|
browser: BrowserManager,
|
|
all_cookies: dict
|
|
) -> dict:
|
|
"""验证网站登录状态"""
|
|
website_name = website_config.get('name', '未知网站')
|
|
url = website_config.get('url', '')
|
|
check_selector = website_config.get('login_check_selector', '')
|
|
success_text = website_config.get('success_text', '')
|
|
|
|
logger.info(f"\n验证网站: {website_name}")
|
|
logger.info(f" URL: {url}")
|
|
|
|
result = {
|
|
'name': website_name,
|
|
'url': url,
|
|
'success': False,
|
|
'error': ''
|
|
}
|
|
|
|
if not url:
|
|
result['error'] = 'URL 未配置'
|
|
logger.error(f" 失败: {result['error']}")
|
|
return result
|
|
|
|
try:
|
|
parsed = urlparse(url)
|
|
domain = parsed.netloc
|
|
|
|
domain_cookies = []
|
|
for cookie_domain, cookies in all_cookies.items():
|
|
clean_domain = cookie_domain.lstrip('.')
|
|
if domain in clean_domain or clean_domain in domain:
|
|
domain_cookies.extend(cookies)
|
|
|
|
if domain_cookies:
|
|
logger.info(f" 找到 {len(domain_cookies)} 个相关 cookie")
|
|
success_count = browser.import_cookies(domain_cookies, domain)
|
|
logger.info(f" 成功导入 {success_count} 个 cookie")
|
|
else:
|
|
logger.warning(f" 未找到相关 cookie")
|
|
|
|
login_success = browser.verify_login(
|
|
url=url,
|
|
check_selector=check_selector,
|
|
success_text=success_text
|
|
)
|
|
|
|
if login_success:
|
|
result['success'] = True
|
|
logger.info(f" ✓ 登录验证成功")
|
|
|
|
self._save_cookies_to_file(domain_cookies, url, website_name, user_name)
|
|
else:
|
|
result['error'] = '登录状态验证失败（cookies可能已过期或选择器配置错误）'
|
|
logger.warning(f" ✗ {result['error']}")
|
|
|
|
except Exception as e:
|
|
result['error'] = f'处理异常: {str(e)}'
|
|
logger.error(f" 异常: {e}")
|
|
|
|
return result
|
|
|
|
def _save_cookies_to_file(self, cookies: list, url: str, website_name: str, user_name: str):
|
|
"""保存 cookies 到文件（按用户和域名区分）"""
|
|
try:
|
|
parsed = urlparse(url)
|
|
domain = parsed.netloc
|
|
|
|
cookie_entry = {
|
|
"cookies": [],
|
|
"url": url,
|
|
"website_name": website_name
|
|
}
|
|
|
|
for cookie in cookies:
|
|
if cookie.get('name') and cookie.get('value'):
|
|
cookie_entry["cookies"].append({
|
|
"name": cookie.get('name'),
|
|
"value": cookie.get('value'),
|
|
"domain": cookie.get('domain', domain)
|
|
})
|
|
|
|
filename = "cookie_data.json"
|
|
|
|
cookie_data = {}
|
|
if os.path.exists(filename):
|
|
with open(filename, 'r', encoding='utf-8') as f:
|
|
try:
|
|
cookie_data = json.load(f)
|
|
except:
|
|
cookie_data = {}
|
|
|
|
if user_name not in cookie_data:
|
|
cookie_data[user_name] = {}
|
|
|
|
cookie_data[user_name][domain] = cookie_entry
|
|
|
|
with open(filename, 'w', encoding='utf-8') as f:
|
|
json.dump(cookie_data, f, indent=2, ensure_ascii=False)
|
|
|
|
logger.info(f" 已保存 {len(cookie_entry['cookies'])} 个 cookie 到 {filename} ({user_name} -> {domain})")
|
|
|
|
except Exception as e:
|
|
logger.error(f" 保存 cookie 失败: {e}")
|
|
|
|
def _print_summary(self, user_name: str, results: list):
|
|
"""打印处理结果汇总"""
|
|
success_count = sum(1 for r in results if r['success'])
|
|
fail_count = len(results) - success_count
|
|
|
|
logger.info(f"\n{'='*20} 用户 {user_name} 处理结果汇总 {'='*20}")
|
|
logger.info(f"总网站数: {len(results)}")
|
|
logger.info(f"成功: {success_count}, 失败: {fail_count}")
|
|
|
|
if fail_count > 0:
|
|
logger.info("\n失败详情:")
|
|
for r in results:
|
|
if not r['success']:
|
|
logger.info(f" - {r['name']}: {r['error']}")
|
|
|
|
def _send_user_notification(self, user_result: dict):
|
|
"""发送单个用户的通知"""
|
|
user_name = user_result['user_name']
|
|
notification_config = user_result.get('notification_config', {})
|
|
iyuu_token = notification_config.get('iyuu_token', '')
|
|
|
|
if not iyuu_token:
|
|
return
|
|
|
|
notifier = IYUUNotifier(iyuu_token)
|
|
|
|
success_count = user_result['success_count']
|
|
fail_count = user_result['fail_count']
|
|
|
|
title = f"【Cookie监控】{user_name} - 成功{success_count} 失败{fail_count}"
|
|
|
|
content_lines = [
|
|
f"**{user_name} Cookie监控报告**",
|
|
f"",
|
|
f"监控时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
|
|
f"成功: {success_count} 个网站",
|
|
f"失败: {fail_count} 个网站",
|
|
f"",
|
|
]
|
|
|
|
for website in user_result['websites']:
|
|
status = "✓ 成功" if website['success'] else f"✗ 失败: {website['error']}"
|
|
content_lines.append(f" {website['name']}: {status}")
|
|
|
|
content = "\n".join(content_lines)
|
|
|
|
print(f"\n{'='*60}")
|
|
print(f"发送用户通知: {user_name}")
|
|
print(f"{'='*60}")
|
|
print(f"\n发送内容:")
|
|
print(f"Title: {title}")
|
|
print(f"Content:\n{content}")
|
|
print(f"\n发送方式: 爱语飞飞 API (GET 请求)")
|
|
print(f"API URL: https://iyuu.cn/{iyuu_token}.send")
|
|
print(f"参数: text={title}, desp={content[:50]}...")
|
|
print(f"\n正在发送...")
|
|
|
|
logger.info(f"\n发送用户通知: {user_name}")
|
|
logger.info(f"Title: {title}")
|
|
|
|
if notifier.send(title, content):
|
|
print(f"\n发送结果: ✓ 成功")
|
|
logger.info(f"✓ 用户 {user_name} 通知发送成功")
|
|
mark_notification_sent(user_name)
|
|
else:
|
|
print(f"\n发送结果: ✗ 失败")
|
|
logger.error(f"✗ 用户 {user_name} 通知发送失败")
|
|
|
|
def run(self):
|
|
"""运行监控"""
|
|
logger.info("="*50)
|
|
logger.info("Cookie 监控开始")
|
|
logger.info(f"时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
|
logger.info("="*50)
|
|
|
|
users = self.config.get('users', [])
|
|
|
|
if not users:
|
|
logger.error("未找到用户配置")
|
|
return
|
|
|
|
for user_config in users:
|
|
try:
|
|
self.process_user(user_config)
|
|
except Exception as e:
|
|
logger.error(f"处理用户失败: {e}")
|
|
|
|
for user_result in self.all_results:
|
|
pass
|
|
|
|
logger.info("\n" + "="*50)
|
|
logger.info("Cookie 监控结束")
|
|
logger.info("="*50)
|
|
|
|
|
|
def main():
|
|
setup_logging()
|
|
monitor = CookieMonitor()
|
|
monitor.run()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|