bwadmin/cookie
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

重构项目结构，优化模块划分，修复数据结构和登录验证问题

Browse Source
This commit is contained in:
bwadmin
2026-03-05 17:05:35 +08:00
parent e090f546c6
commit a70af1b0da
13 changed files with 625 additions and 1262 deletions
Download Patch File Download Diff File Expand all files Collapse all files
browser_login.py
-548
View File
@@ -1,548 +0,0 @@
"""浏览器登录模块 - 使用 DrissionPage 控制浏览器进行登录检测和 cookie 管理"""
from typing import Dict, List, Optional
import json
import time
from DrissionPage import Chromium, ChromiumOptions
from applogger import get_logger
logger = get_logger(__name__)
class BrowserLogin:
"""浏览器登录客户端"""
def __init__(self, browser_type: str = "edge", headless: bool = False):
"""
初始化浏览器登录客户端
Args:
browser_type: 浏览器类型 (chrome 或 edge)
headless: 是否无头模式
"""
self.browser_type = browser_type.lower()
self.headless = headless
self.browser = None
self.tab = None
def _create_browser(self):
"""创建浏览器实例"""
co = ChromiumOptions()
# 设置浏览器路径
if self.browser_type == "edge":
co.set_browser_path("C:\\Program Files (x86)\\Microsoft\\Edge\\Application\\msedge.exe")
elif self.browser_type == "chrome":
co.set_browser_path("chrome")
# 设置无头模式
if self.headless:
co.headless = True
# 设置其他选项
co.set_argument('--no-sandbox')
co.set_argument('--disable-dev-shm-usage')
# 创建浏览器
self.browser = Chromium(co)
self.tab = self.browser.new_tab()
def _close_browser(self):
"""关闭浏览器"""
if self.browser:
try:
self.browser.quit()
except Exception as e:
logger.error(f"关闭浏览器失败: {e}")
finally:
self.browser = None
self.tab = None
def _get_domain_from_url(self, url: str) -> str:
"""
从 URL 中提取域名
Args:
url: 网站地址
Returns:
域名
"""
from urllib.parse import urlparse
parsed = urlparse(url)
return parsed.netloc
def get_all_cookies(self) -> List[Dict]:
"""
获取当前浏览器实例的所有 cookie
Returns:
所有 cookie 的列表
"""
# 先获取 cookies 对象，然后调用 as_dict 方法
cookies_obj = self.tab.cookies
cookies = cookies_obj() if callable(cookies_obj) else cookies_obj
cookies = cookies.as_dict() if hasattr(cookies, 'as_dict') else cookies
# 确保返回完整的 cookie 列表，包含所有属性
formatted_cookies = []
for cookie in cookies:
formatted_cookie = {
'name': cookie.get('name', ''),
'value': cookie.get('value', ''),
'domain': cookie.get('domain', ''),
'path': cookie.get('path', '/'),
'expires': cookie.get('expirationDate', None),
'secure': cookie.get('secure', False),
'httpOnly': cookie.get('httpOnly', False),
'sameSite': cookie.get('sameSite', 'Lax')
}
formatted_cookies.append(formatted_cookie)
return formatted_cookies
def check_login(self, url: str, check_selector: str, success_text: str) -> bool:
"""
检查是否已登录
Args:
url: 网站地址
check_selector: 登录检测选择器
success_text: 登录成功时显示的文本
Returns:
是否已登录
"""
if not self.browser or not self.tab:
self._create_browser()
try:
# 访问网站
self.tab.get(url)
# 等待页面加载
time.sleep(2)
# 检查登录状态
element = self.tab.ele(check_selector, timeout=10)
if element and element.text and success_text in element.text:
return True
return False
except Exception as e:
logger.error(f"检查登录状态失败: {e}")
return False
def import_all_cookies(self, cookies: List[Dict]) -> bool:
"""
一次性导入所有 cookies 到浏览器
Args:
cookies: cookie 列表
Returns:
是否成功
"""
if not self.browser or not self.tab:
self._create_browser()
success_count = 0
fail_count = 0
logger.info(f"开始导入 {len(cookies)} 个 cookie")
try:
# 准备所有 cookie 数据
formatted_cookies = []
for cookie in cookies:
cookie_dict = {
'name': cookie.get('name', ''),
'value': cookie.get('value', ''),
'domain': cookie.get('domain', ''),
'path': cookie.get('path', '/'),
}
if cookie.get('secure'):
cookie_dict['secure'] = True
if cookie.get('httpOnly'):
cookie_dict['httpOnly'] = True
if cookie.get('sameSite'):
cookie_dict['sameSite'] = cookie.get('sameSite')
formatted_cookies.append(cookie_dict)
# 根据 DrissionPage 文档，set.cookies() 方法可以直接接收列表格式的多个 cookie
logger.info("使用一次性注入方式导入 cookie")
self.tab.set.cookies(formatted_cookies)
success_count = len(formatted_cookies)
except Exception as e:
logger.error(f"导入 cookie 失败: {e}")
# 回退到逐个注入
logger.info("回退到逐个注入方式导入 cookie")
for cookie_dict in formatted_cookies:
try:
self.tab.set.cookies(cookie_dict)
success_count += 1
except Exception as e:
fail_count += 1
logger.info(f"Cookie 导入完成: 成功 {success_count}, 失败 {fail_count}")
return fail_count == 0
def verify_login(self, url: str, check_selector: str = "", success_text: str = "") -> bool:
"""
验证登录状态（cookies已预先导入）
Args:
url: 网站地址
check_selector: 登录检测选择器（可选）
success_text: 登录成功时显示的文本（可选）
Returns:
是否已登录
"""
if not self.browser or not self.tab:
self._create_browser()
try:
self.tab.get(url)
time.sleep(5)
# 如果没有指定选择器，只检查页面是否成功加载
if not check_selector:
logger.info(" 未指定登录检测选择器，跳过登录验证")
return True
# 尝试查找元素
try:
element = self.tab.ele(check_selector, timeout=10)
if element:
element_text = element.text or ""
logger.info(f" 找到元素，文本内容: {element_text[:50]}...")
if not success_text:
logger.info(" 未指定成功文本，找到元素即认为登录成功")
return True
if success_text in element_text:
logger.info(f" 检测到成功文本: {success_text}")
return True
else:
logger.warning(f" 未检测到成功文本 '{success_text}'，元素文本: {element_text[:100]}")
return False
else:
logger.warning(f" 未找到元素: {check_selector}")
return False
except Exception as ele_error:
logger.warning(f" 查找元素失败: {ele_error}")
return False
except Exception as e:
logger.error(f"验证登录状态失败: {e}")
return False
def login_with_cookies(
self,
url: str,
cookies: List[Dict],
check_selector: str,
success_text: str
) -> bool:
"""
使用 cookie 登录
Args:
url: 网站地址
cookies: cookie 列表
check_selector: 登录检测选择器
success_text: 登录成功时显示的文本
Returns:
是否登录成功
"""
if not self.browser or not self.tab:
self._create_browser()
try:
from urllib.parse import urlparse
parsed = urlparse(url)
base_url = f"{parsed.scheme}://{parsed.netloc}"
self.tab.get(base_url)
time.sleep(1)
logger.info(f"开始设置 {len(cookies)} 个 cookie")
for cookie in cookies:
try:
cookie_dict = {
'name': cookie.get('name', ''),
'value': cookie.get('value', ''),
'domain': cookie.get('domain', parsed.netloc),
'path': cookie.get('path', '/'),
}
if cookie.get('secure'):
cookie_dict['secure'] = True
if cookie.get('httpOnly'):
cookie_dict['httpOnly'] = True
if cookie.get('sameSite'):
cookie_dict['sameSite'] = cookie.get('sameSite')
self.tab.set.cookies(cookie_dict)
logger.info(f"设置 cookie: {cookie_dict['name']}")
except Exception as e:
logger.error(f"设置 cookie 失败 {cookie.get('name')}: {e}")
self.tab.get(url)
time.sleep(3)
element = self.tab.ele(check_selector, timeout=10)
if element and element.text and success_text in element.text:
return True
return False
except Exception as e:
logger.error(f"使用 cookie 登录失败: {e}")
return False
def refresh_and_save_cookies(self, url: str) -> List[Dict]:
"""
刷新页面并保存新的 cookie
Args:
url: 网站地址
Returns:
cookie 列表
"""
if not self.browser or not self.tab:
self._create_browser()
try:
# 刷新页面
self.tab.refresh()
# 等待页面加载
time.sleep(3)
# 获取 cookies
cookies_obj = self.browser.cookies
cookies = cookies_obj() if callable(cookies_obj) else cookies_obj
cookies = cookies.as_dict() if hasattr(cookies, 'as_dict') else cookies
# 确保返回的是列表
if isinstance(cookies, list):
return cookies
elif isinstance(cookies, dict):
# 如果是字典，转换为列表
return [cookies]
else:
# 如果是其他类型，返回空列表
return []
except Exception as e:
logger.error(f"刷新页面并保存 cookie 失败: {e}")
return []
def close(self):
"""关闭浏览器"""
self._close_browser()
class CookieManager:
"""Cookie 管理器 - 实现浏览器实例级别的 Cookie 共享"""
def __init__(self, cookie_file: str):
"""
初始化 Cookie 管理器
Args:
cookie_file: cookie 文件路径
"""
self.cookie_file = cookie_file
self.cookies = self._load_cookies()
def _load_cookies(self) -> Dict[str, List[Dict]]:
"""加载 cookie 文件"""
import os
if os.path.exists(self.cookie_file):
try:
with open(self.cookie_file, 'r', encoding='utf-8') as f:
data = json.load(f)
# 兼容旧格式
if isinstance(data, dict) and any(isinstance(v, dict) for v in data.values()):
# 转换旧格式为新格式
new_data = {}
for user, websites in data.items():
all_cookies = []
for site_cookies in websites.values():
all_cookies.extend(site_cookies)
new_data[user] = all_cookies
return new_data
return data
except Exception as e:
logger.error(f"加载 cookie 文件失败: {e}")
return {}
def _save_cookies(self):
"""保存 cookie 文件"""
try:
with open(self.cookie_file, 'w', encoding='utf-8') as f:
json.dump(self.cookies, f, indent=2, ensure_ascii=False)
except Exception as e:
logger.error(f"保存 cookie 文件失败: {e}")
def get_all_cookies(self, user_name: str) -> List[Dict]:
"""
获取指定用户的所有 cookie
Args:
user_name: 用户名
Returns:
cookie 列表
"""
return self.cookies.get(user_name, [])
def save_all_cookies(self, user_name: str, cookies: List[Dict]):
"""
保存指定用户的所有 cookie
Args:
user_name: 用户名
cookies: cookie 列表
"""
self.cookies[user_name] = cookies
self._save_cookies()
def save_cookies(self, user_name: str, website_name: str, cookies: List[Dict]):
"""
保存指定用户和网站的 cookie（兼容旧接口）
Args:
user_name: 用户名
website_name: 网站名（未使用，保持兼容）
cookies: cookie 列表
"""
# 保持向后兼容，将网站的cookie添加到用户的所有cookie中
if user_name not in self.cookies:
self.cookies[user_name] = []
# 合并新cookie到现有cookie中
existing_cookies = self.cookies[user_name]
# 创建现有cookie的索引，用于快速查找
cookie_index = {}
for i, cookie in enumerate(existing_cookies):
if isinstance(cookie, dict):
key = (cookie.get('name'), cookie.get('domain'), cookie.get('path'))
cookie_index[key] = i
# 添加或更新cookie
if isinstance(cookies, list):
valid_cookies = []
for cookie in cookies:
if isinstance(cookie, dict):
valid_cookies.append(cookie)
for cookie in valid_cookies:
key = (cookie.get('name'), cookie.get('domain'), cookie.get('path'))
if key in cookie_index:
# 更新现有cookie
existing_cookies[cookie_index[key]] = cookie
else:
# 添加新cookie
existing_cookies.append(cookie)
if valid_cookies:
logger.info(f" 已保存 {len(valid_cookies)} 个新 cookie")
self._save_cookies()
def add_cookie(self, user_name: str, cookie: Dict):
"""
为指定用户添加单个 cookie
Args:
user_name: 用户名
cookie: cookie 字典
"""
if user_name not in self.cookies:
self.cookies[user_name] = []
# 检查是否已存在相同的 cookie
existing_index = -1
for i, existing_cookie in enumerate(self.cookies[user_name]):
if (existing_cookie.get('name') == cookie.get('name') and
existing_cookie.get('domain') == cookie.get('domain') and
existing_cookie.get('path') == cookie.get('path')):
existing_index = i
break
if existing_index >= 0:
# 更新现有 cookie
self.cookies[user_name][existing_index] = cookie
else:
# 添加新 cookie
self.cookies[user_name].append(cookie)
self._save_cookies()
def remove_cookie(self, user_name: str, cookie_name: str, domain: str):
"""
从指定用户中移除指定的 cookie
Args:
user_name: 用户名
cookie_name: cookie 名称
domain: cookie 域名
"""
if user_name in self.cookies:
self.cookies[user_name] = [
cookie for cookie in self.cookies[user_name]
if not (cookie.get('name') == cookie_name and cookie.get('domain') == domain)
]
self._save_cookies()
def clear_cookies(self, user_name: str):
"""
清空指定用户的所有 cookie
Args:
user_name: 用户名
"""
if user_name in self.cookies:
self.cookies[user_name] = []
self._save_cookies()
def get_cookies_for_domain(self, user_name: str, domain: str) -> List[Dict]:
"""
获取指定用户的指定域名的 cookie
Args:
user_name: 用户名
domain: 域名
Returns:
cookie 列表
"""
user_cookies = self.get_all_cookies(user_name)
filtered_cookies = []
for cookie in user_cookies:
cookie_domain = cookie.get('domain', '')
# 统一域名格式
if cookie_domain.startswith('.'):
cookie_domain = cookie_domain[1:]
if domain in cookie_domain or cookie_domain in domain:
filtered_cookies.append(cookie)
return filtered_cookies
browser_tool.py
-179
View File
@@ -1,179 +0,0 @@
import tkinter as tk
from tkinter import ttk, messagebox
from DrissionPage import ChromiumPage, ChromiumOptions
import json
import threading
import time
class BrowserTool:
def __init__(self, root):
self.root = root
self.root.title('浏览器Cookie管理工具')
self.root.geometry('500x200')
self.root.resizable(False, False)
self.page = None
self.cookie_file = 'd:\\test\\openUrl\\cookies.json'
self.setup_ui()
def setup_ui(self):
frame = ttk.Frame(self.root, padding=20)
frame.pack(fill=tk.BOTH, expand=True)
ttk.Label(frame, text='网址:').grid(row=0, column=0, sticky=tk.W, pady=5)
self.url_entry = ttk.Entry(frame, width=60)
self.url_entry.grid(row=0, column=1, columnspan=3, pady=5, sticky=tk.W)
self.url_entry.insert(0, 'https://yunlu.sinopec.com/AddressBook/AddressBookCheck')
btn_frame = ttk.Frame(frame)
btn_frame.grid(row=1, column=0, columnspan=4, pady=20)
self.open_btn = ttk.Button(btn_frame, text='打开', width=15, command=self.open_browser)
self.open_btn.pack(side=tk.LEFT, padx=10)
self.save_btn = ttk.Button(btn_frame, text='保存', width=15, command=self.save_cookies)
self.save_btn.pack(side=tk.LEFT, padx=10)
self.import_btn = ttk.Button(btn_frame, text='导入', width=15, command=self.import_cookies)
self.import_btn.pack(side=tk.LEFT, padx=10)
self.status_var = tk.StringVar(value='就绪')
ttk.Label(frame, textvariable=self.status_var).grid(row=2, column=0, columnspan=4, pady=10)
def log(self, msg):
self.status_var.set(msg)
self.root.update()
def init_browser(self):
co = ChromiumOptions()
co.set_browser_path(r'C:\Program Files (x86)\Microsoft\Edge\Application\msedge.exe')
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 open_browser(self):
def run():
try:
self.log('正在启动浏览器...')
self.init_browser()
url = self.url_entry.get().strip()
if not url:
messagebox.showerror('错误', '请输入网址')
return
self.log(f'正在打开: {url}')
self.page.get(url)
self.wait_page_loaded()
self.log('页面加载完成')
except Exception as e:
self.log(f'错误: {str(e)}')
messagebox.showerror('错误', str(e))
threading.Thread(target=run, daemon=True).start()
def save_cookies(self):
def run():
try:
if not self.page:
messagebox.showwarning('提示', '请先打开浏览器')
return
self.log('正在保存Cookies...')
cookies = self.page.cookies()
with open(self.cookie_file, 'w', encoding='utf-8') as f:
json.dump(cookies, f, ensure_ascii=False, indent=2)
self.log(f'Cookies已保存到 {self.cookie_file}')
self.log('正在关闭浏览器...')
self.page.quit()
self.page = None
self.log('浏览器已关闭')
messagebox.showinfo('成功', 'Cookies保存成功，浏览器已关闭')
except Exception as e:
self.log(f'错误: {str(e)}')
messagebox.showerror('错误', str(e))
threading.Thread(target=run, daemon=True).start()
def import_cookies(self):
def run():
try:
import os
if not os.path.exists(self.cookie_file):
messagebox.showwarning('提示', 'Cookie文件不存在，请先保存Cookies')
return
self.log('正在启动浏览器...')
self.init_browser()
self.log('正在加载Cookies...')
with open(self.cookie_file, 'r', encoding='utf-8') as f:
cookies = json.load(f)
url = self.url_entry.get().strip()
if not url:
messagebox.showerror('错误', '请输入网址')
return
from urllib.parse import urlparse
parsed = urlparse(url)
domain = parsed.netloc
self.page.get('about:blank')
for cookie in cookies:
cookie_dict = {
'name': cookie.get('name'),
'value': cookie.get('value'),
'domain': cookie.get('domain', domain),
'path': cookie.get('path', '/')
}
try:
self.page.set.cookies(cookie_dict)
except Exception as ce:
self.log(f'设置Cookie失败: {cookie.get("name")} - {str(ce)}')
self.log(f'已导入 {len(cookies)} 个Cookie')
self.log(f'正在打开: {url}')
self.page.get(url)
self.wait_page_loaded()
self.log('刷新页面验证登录状态...')
for i in range(3):
self.page.refresh()
self.wait_page_loaded()
time.sleep(1)
self.log(f'刷新第 {i+1} 次...')
self.log('页面加载完成')
except Exception as e:
self.log(f'错误: {str(e)}')
messagebox.showerror('错误', str(e))
threading.Thread(target=run, daemon=True).start()
def on_closing(self):
if self.page:
try:
self.page.quit()
except:
pass
self.root.destroy()
if __name__ == '__main__':
root = tk.Tk()
app = BrowserTool(root)
root.protocol('WM_DELETE_WINDOW', app.on_closing)
root.mainloop()
config.json
+43 -6
View File
@@ -1,7 +1,7 @@
{
"users": [
{
"name": "用户A",
"name": "小余",
"cookie_cloud": {
"uuid": "hu6n2vcUmzpu7mqUN2rVCg",
"password": "jtDM5dV9AyqXkZdQVeA9f6",
@@ -18,7 +18,7 @@
},
"websites": [
{
"name": "网站A",
"name": "柴油联名卡",
"url": "https://lmkbi.95155.com/bi-system/#/carUnpetrifiedComAll",
"login_check_selector": "",
"success_text": "彭峰"
@@ -26,7 +26,7 @@
]
},
{
"name": "用户B",
"name": "彭峰(个人)",
"cookie_cloud": {
"uuid": "qyZpTrgiP5mVwiRZwfBhjz",
"password": "iGn1B4FnWftp3oj4Ko3jxA",
@@ -43,14 +43,51 @@
},
"websites": [
{
"name": "网站A",
"name": "网站B",
"url": "https://192.168.1.200:8006/",
"login_check_selector": "",
"success_text": "彭峰"
}
]
},
{
"name": "彭峰(公司)",
"cookie_cloud": {
"uuid": "aCgAWN5DsYhP88FnTmrr3H",
"password": "bK8dfh1a6ZZ79jWeeMedjN",
"api_url": "https://movie-pilot.org/cookiecloud"
},
"notification": {
"iyuu_token": "IYUU37629Tc1d371c7ce99a49ff9778e196286b7e4592be191",
"max_fail_count": 3,
"notify_on_success": true
},
"browser": {
"type": "edge",
"headless": false
},
"websites": [
{
"name": "柴油联名卡",
"url": "https://lmkbi.95155.com/bi-system/#/carUnpetrifiedComAll",
"login_check_selector": "",
"success_text": "彭峰"
},
{
"name": "网站B",
"url": "https://192.168.1.200:8006/",
"name": "加油卡交易",
"url": "https://cardweb.salecard.sinopec.com/card/index.jsp",
"login_check_selector": "",
"success_text": "彭峰"
},
{
"name": "数字销售门户",
"url": "https://ewcc.sinopec.com/?authConfigs=37d5aed075525d4fa0fe635231cba447",
"login_check_selector": "",
"success_text": "彭峰"
},
{
"name": "飞牛存储",
"url": "http://10.190.20.156:5666/",
"login_check_selector": "",
"success_text": "彭峰"
}
config_manager.py
+292
View File
@@ -0,0 +1,292 @@
"""配置管理模块"""
import json
import os
from logger import get_logger
logger = get_logger(__name__)
DEFAULT_CONFIG_FILE = "config.json"
DEFAULT_COOKIE_DATA_FILE = "cookie_data.json"
class ConfigManager:
"""配置管理器"""
def __init__(self, config_file: str = DEFAULT_CONFIG_FILE):
"""
初始化配置管理器
Args:
config_file: 配置文件路径
"""
self.config_file = config_file
self.config = self._load_config()
def _load_config(self) -> dict:
"""加载配置文件"""
try:
if os.path.exists(self.config_file):
with open(self.config_file, 'r', encoding='utf-8') as f:
config = json.load(f)
logger.info(f"配置文件加载成功: {self.config_file}")
return config
else:
logger.warning(f"配置文件不存在: {self.config_file}")
return {}
except Exception as e:
logger.error(f"加载配置文件失败: {e}")
return {}
def save_config(self, config: dict) -> bool:
"""
保存配置文件
Args:
config: 配置字典
Returns:
是否保存成功
"""
try:
with open(self.config_file, 'w', encoding='utf-8') as f:
json.dump(config, f, indent=2, ensure_ascii=False)
logger.info(f"配置文件保存成功: {self.config_file}")
return True
except Exception as e:
logger.error(f"保存配置文件失败: {e}")
return False
def get(self, key: str, default=None):
"""
获取配置项
Args:
key: 配置键
default: 默认值
Returns:
配置值
"""
return self.config.get(key, default)
def set(self, key: str, value):
"""
设置配置项
Args:
key: 配置键
value: 配置值
"""
self.config[key] = value
self.save_config(self.config)
def get_users(self) -> list:
"""
获取所有用户配置
Returns:
用户配置列表
"""
return self.config.get('users', [])
def get_user_config(self, user_name: str) -> dict:
"""
获取指定用户的配置
Args:
user_name: 用户名
Returns:
用户配置字典
"""
users = self.get_users()
for user in users:
if user.get('name') == user_name:
return user
return {}
def get_cookie_cloud_config(self, user_name: str) -> dict:
"""
获取用户的Cookie Cloud配置
Args:
user_name: 用户名
Returns:
Cookie Cloud配置字典
"""
user_config = self.get_user_config(user_name)
return user_config.get('cookie_cloud', {})
def get_notification_config(self, user_name: str) -> dict:
"""
获取用户的通知配置
Args:
user_name: 用户名
Returns:
通知配置字典
"""
user_config = self.get_user_config(user_name)
return user_config.get('notification', {})
def get_browser_config(self, user_name: str) -> dict:
"""
获取用户的浏览器配置
Args:
user_name: 用户名
Returns:
浏览器配置字典
"""
user_config = self.get_user_config(user_name)
return user_config.get('browser', {})
def get_websites(self, user_name: str) -> list:
"""
获取用户的网站配置
Args:
user_name: 用户名
Returns:
网站配置列表
"""
user_config = self.get_user_config(user_name)
return user_config.get('websites', [])
class CookieDataManager:
"""Cookie数据管理器"""
def __init__(self, data_file: str = DEFAULT_COOKIE_DATA_FILE):
"""
初始化Cookie数据管理器
Args:
data_file: Cookie数据文件路径
"""
self.data_file = data_file
self.data = self._load_data()
def _load_data(self) -> dict:
"""加载Cookie数据文件"""
try:
if os.path.exists(self.data_file):
with open(self.data_file, 'r', encoding='utf-8') as f:
data = json.load(f)
logger.info(f"Cookie数据文件加载成功: {self.data_file}")
return data
else:
logger.warning(f"Cookie数据文件不存在: {self.data_file}")
return {}
except Exception as e:
logger.error(f"加载Cookie数据文件失败: {e}")
return {}
def save_data(self, data: dict) -> bool:
"""
保存Cookie数据文件
Args:
data: Cookie数据字典
Returns:
是否保存成功
"""
try:
with open(self.data_file, 'w', encoding='utf-8') as f:
json.dump(data, f, indent=2, ensure_ascii=False)
logger.info(f"Cookie数据文件保存成功: {self.data_file}")
return True
except Exception as e:
logger.error(f"保存Cookie数据文件失败: {e}")
return False
def get_user_cookies(self, user_name: str) -> dict:
"""
获取用户的Cookie数据
Args:
user_name: 用户名
Returns:
用户Cookie数据字典
"""
return self.data.get(user_name, {})
def save_user_cookies(self, user_name: str, cookies: list, url: str, website_name: str = ""):
"""
保存用户的Cookie数据
Args:
user_name: 用户名
cookies: Cookie列表
url: 网站URL
website_name: 网站名称
"""
if user_name not in self.data:
self.data[user_name] = {}
from urllib.parse import urlparse
parsed = urlparse(url)
domain = parsed.netloc
cookie_entry = {
"cookies": cookies,
"url": url,
"website_name": website_name
}
self.data[user_name][domain] = cookie_entry
self.save_data(self.data)
logger.info(f"已保存 {len(cookies)} 个 cookie ({user_name} -> {domain})")
def get_domain_cookies(self, user_name: str, domain: str) -> dict:
"""
获取指定域名的Cookie数据
Args:
user_name: 用户名
domain: 域名
Returns:
域名Cookie数据字典
"""
user_cookies = self.get_user_cookies(user_name)
for cookie_domain, cookie_data in user_cookies.items():
clean_domain = cookie_domain.lstrip('.')
if domain in clean_domain or clean_domain in domain:
return cookie_data
return {}
def save_simple_cookies(self, cookies: list, url: str):
"""
保存简单的Cookie数据（用于Cookie提取器）
Args:
cookies: Cookie列表
url: 网站URL
"""
data = {
"cookies": cookies,
"url": url
}
self.save_data(data)
logger.info(f"已保存 {len(cookies)} 个 cookie")
def load_simple_cookies(self) -> dict:
"""
加载简单的Cookie数据（用于Cookie提取器）
Returns:
Cookie数据字典
"""
return {
"cookies": self.data.get("cookies", []),
"url": self.data.get("url", "")
}
cookie_cloud.py
+1 -4
View File
@@ -7,7 +7,7 @@ import urllib.request
import urllib.error
from Crypto.Cipher import AES
from dataclasses import dataclass
from applogger import get_logger
from logger import get_logger
logger = get_logger(__name__)
@@ -253,6 +253,3 @@ class CookieCloud:
logger.info(f"获取到所有数据: {len(result['cookies'])} 个域名的 cookies, {len(result['local_storage'])} 个域名的 local storage")
return result
cookie_data.json
+25 -8
View File
@@ -1,21 +1,38 @@
{
"用户A": {
"小余": {
"lmkbi.95155.com": {
"cookies": [],
"url": "https://lmkbi.95155.com/bi-system/#/carUnpetrifiedComAll",
"website_name": "网站A"
"website_name": "柴油联名卡"
}
},
"用户B": {
"lmkbi.95155.com": {
"cookies": [],
"url": "https://lmkbi.95155.com/bi-system/#/carUnpetrifiedComAll",
"website_name": "网站A"
},
"彭峰(个人)": {
"192.168.1.200:8006": {
"cookies": [],
"url": "https://192.168.1.200:8006/",
"website_name": "网站B"
}
},
"彭峰(公司)": {
"lmkbi.95155.com": {
"cookies": [],
"url": "https://lmkbi.95155.com/bi-system/#/carUnpetrifiedComAll",
"website_name": "柴油联名卡"
},
"cardweb.salecard.sinopec.com": {
"cookies": [],
"url": "https://cardweb.salecard.sinopec.com/card/index.jsp",
"website_name": "加油卡交易"
},
"ewcc.sinopec.com": {
"cookies": [],
"url": "https://ewcc.sinopec.com/?authConfigs=37d5aed075525d4fa0fe635231cba447",
"website_name": "数字销售门户"
},
"10.190.20.156:5666": {
"cookies": [],
"url": "http://10.190.20.156:5666/",
"website_name": "飞牛存储"
}
}
}
cookie_extractor.py
+41 -105
View File
@@ -1,35 +1,35 @@
"""Cookie提取器 - 使用DrissionPage库"""
import tkinter as tk
from tkinter import ttk, messagebox
import json
import os
import logging
from DrissionPage import ChromiumPage, ChromiumOptions
from browser_manager import BrowserManager
from config_manager import CookieDataManager
from logger import get_logger
logger = get_logger(__name__)
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('cookie_extractor.log'),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
class CookieExtractorApp:
"""Cookie提取器应用"""
def __init__(self, root):
"""
初始化应用
Args:
root: 根窗口
"""
self.root = root
self.root.title("Cookie提取器")
self.root.geometry("600x250")
self.root.resizable(True, True)
self.page = None
self.browser_manager = BrowserManager()
self.cookie_data_manager = CookieDataManager()
self._create_widgets()
def _create_widgets(self):
"""创建界面控件"""
main_frame = ttk.Frame(self.root, padding="20")
main_frame.pack(fill=tk.BOTH, expand=True)
@@ -63,18 +63,6 @@ class CookieExtractorApp:
ttk.Label(status_frame, text="状态:").pack(side=tk.LEFT, padx=5)
ttk.Label(status_frame, textvariable=self.status_var, foreground="blue").pack(side=tk.LEFT, padx=5)
def _init_browser(self):
"""初始化浏览器"""
co = ChromiumOptions()
co.set_browser_path(r"C:\Program Files (x86)\Microsoft\Edge\Application\msedge.exe")
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 open_website(self):
"""打开指定网站"""
url = self.url_var.get().strip()
@@ -93,13 +81,13 @@ class CookieExtractorApp:
self.status_var.set("正在打开浏览器...")
self.root.update()
self._init_browser()
self.browser_manager.init_browser()
self.status_var.set(f"正在访问: {url}")
self.root.update()
self.page.get(url)
self._wait_page_loaded()
self.browser_manager.page.get(url)
self.browser_manager.wait_page_loaded()
self.status_var.set(f"已打开: {url}")
self.save_button.config(state=tk.NORMAL)
@@ -113,30 +101,25 @@ class CookieExtractorApp:
def save_cookies(self):
"""保存当前的cookie信息，并强制关闭浏览器"""
if not self.page:
if not self.browser_manager.page:
messagebox.showerror("错误", "请先打开网站")
return
try:
self.status_var.set("正在保存Cookies...")
self.status_var.set("正在提取cookies...")
self.root.update()
cookies = self.page.cookies()
cookies = self.browser_manager.get_cookies()
logger.info(f"成功提取 {len(cookies)} 个 cookies")
data = {
"cookies": cookies,
"url": self.url_var.get()
}
self.cookie_data_manager.save_simple_cookies(cookies, self.url_var.get())
output_file = "cookie_data.json"
with open(output_file, 'w', encoding='utf-8') as f:
json.dump(data, f, indent=2, ensure_ascii=False)
self.status_var.set("正在关闭浏览器...")
self.root.update()
self.page.quit()
self.page = None
self.save_button.config(state=tk.DISABLED)
self._cleanup_browser()
self.status_var.set(f"Cookies已保存到 {output_file}")
self.status_var.set(f"Cookies已保存到 cookie_data.json")
messagebox.showinfo('成功', 'Cookies保存成功，浏览器已关闭')
except Exception as e:
@@ -146,23 +129,13 @@ class CookieExtractorApp:
self._cleanup_browser()
def import_cookies(self):
"""打开浏览器，导入cookie和localStorage信息，然后打开指定网站"""
"""打开浏览器，导入cookie信息，然后打开指定网站"""
import time
try:
output_file = "cookie_data.json"
if not os.path.exists(output_file):
messagebox.showerror("错误", f"文件不存在: {output_file}")
return
self.status_var.set("正在读取cookies...")
self.root.update()
with open(output_file, 'r', encoding='utf-8') as f:
data = json.load(f)
data = self.cookie_data_manager.load_simple_cookies()
url = data.get("url", "")
cookies = data.get("cookies", [])
local_storage = data.get("local_storage", {})
if not url:
url = self.url_var.get().strip()
@@ -179,11 +152,11 @@ class CookieExtractorApp:
self.status_var.set("正在打开浏览器...")
self.root.update()
self._init_browser()
self.browser_manager.init_browser()
self.status_var.set("等待浏览器就绪...")
self.root.update()
self.page.get("about:blank")
self.browser_manager.page.get("about:blank")
time.sleep(1)
from urllib.parse import urlparse
@@ -193,55 +166,19 @@ class CookieExtractorApp:
if cookies:
self.status_var.set("正在导入cookies...")
self.root.update()
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:
logger.error(f"导入cookie失败: {cookie.get('name')} - {e}")
success_count = self.browser_manager.import_cookies(cookies, domain)
logger.info(f"成功导入 {success_count} 个 cookies")
if local_storage:
self.status_var.set("正在导入localStorage...")
self.root.update()
ls_count = 0
for ls_domain, items in local_storage.items():
try:
for key, value in items.items():
try:
self.page.set.local_storage(key, value)
ls_count += 1
except Exception as e:
logger.error(f"导入localStorage失败: {key} - {e}")
except Exception as e:
logger.error(f"导入localStorage域失败: {ls_domain} - {e}")
logger.info(f"成功导入 {ls_count} 个 localStorage")
self.status_var.set(f"正在访问: {url}")
self.root.update()
self.page.get(url)
self._wait_page_loaded()
self.browser_manager.page.get(url)
self.browser_manager.wait_page_loaded()
self.status_var.set("刷新页面验证登录状态...")
self.root.update()
for i in range(3):
self.page.refresh()
self._wait_page_loaded()
self.browser_manager.page.refresh()
self.browser_manager.wait_page_loaded()
time.sleep(1)
self.status_var.set(f"刷新第 {i+1} 次...")
self.root.update()
@@ -258,21 +195,19 @@ class CookieExtractorApp:
self._cleanup_browser()
def close_app(self):
"""关闭应用"""
self._cleanup_browser()
self.root.destroy()
def _cleanup_browser(self):
if self.page:
try:
self.page.quit()
except Exception as e:
logger.error(f"关闭浏览器失败: {e}")
finally:
self.page = None
"""清理浏览器资源"""
self.browser_manager.close()
self.save_button.config(state=tk.DISABLED)
self.status_var.set("就绪")
def main():
"""主函数"""
root = tk.Tk()
app = CookieExtractorApp(root)
@@ -283,5 +218,6 @@ def main():
root.protocol("WM_DELETE_WINDOW", on_closing)
root.mainloop()
if __name__ == "__main__":
main()
applogger.py → logger.py
+1 -1
View File
@@ -1,4 +1,4 @@
"""日志配置模块"""
"""日志管理模块"""
import os
import logging
from logging.handlers import TimedRotatingFileHandler
monitor.py
+84 -216
View File
@@ -1,13 +1,12 @@
"""Cookie 监控主程序"""
import json
import time
import os
from datetime import datetime, time as dt_time
from datetime import datetime
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
from notification import create_notification_manager
from config_manager import ConfigManager, CookieDataManager
from browser_manager import BrowserManager
from logger import get_logger
logger = get_logger(__name__)
@@ -16,6 +15,9 @@ STATE_FILE = "notify_state.json"
def load_notify_state() -> dict:
"""加载通知状态"""
import json
import os
if os.path.exists(STATE_FILE):
try:
with open(STATE_FILE, 'r', encoding='utf-8') as f:
@@ -26,6 +28,8 @@ def load_notify_state() -> dict:
def save_notify_state(state: dict):
"""保存通知状态"""
import json
try:
with open(STATE_FILE, 'w', encoding='utf-8') as f:
json.dump(state, f, indent=2, ensure_ascii=False)
@@ -34,10 +38,12 @@ def save_notify_state(state: dict):
def should_send_notification(user_name: str) -> bool:
"""判断是否应该发送通知"""
return False
def mark_notification_sent(user_name: str):
"""标记通知已发送"""
now = datetime.now()
state = load_notify_state()
@@ -50,148 +56,37 @@ def mark_notification_sent(user_name: str):
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()
"""
初始化监控器
Args:
config_file: 配置文件路径
"""
self.config_manager = ConfigManager(config_file)
self.cookie_data_manager = CookieDataManager()
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):
"""处理单个用户：依次验证各网站登录状态"""
"""
处理单个用户：依次验证各网站登录状态
Args:
user_config: 用户配置字典
"""
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
browser_config = user_config.get('browser', {})
browser = BrowserManager(
browser_type=browser_config.get('type', 'edge'),
headless=browser_config.get('headless', False)
@@ -199,7 +94,7 @@ class CookieMonitor:
user_results = {
'user_name': user_name,
'notification_config': notification_config,
'notification_config': user_config.get('notification', {}),
'websites': [],
'success_count': 0,
'fail_count': 0
@@ -207,9 +102,10 @@ class CookieMonitor:
try:
logger.info(f"步骤1: 从本地 cookie_data.json 读取用户 cookies")
local_cookies = self._load_local_cookies(user_name)
local_cookies = self.cookie_data_manager.get_user_cookies(user_name)
logger.info(f"步骤1.5: 从 Cookie Cloud 获取所有 cookies")
cookie_cloud_config = user_config.get('cookie_cloud', {})
cookie_cloud = CookieCloud(
api_url=cookie_cloud_config.get('api_url', ''),
uuid=cookie_cloud_config.get('uuid', ''),
@@ -249,19 +145,6 @@ class CookieMonitor:
finally:
browser.close()
def _load_local_cookies(self, user_name: str) -> dict:
"""从本地 cookie_data.json 读取用户的 cookies"""
filename = "cookie_data.json"
if not os.path.exists(filename):
return {}
try:
with open(filename, 'r', encoding='utf-8') as f:
cookie_data = json.load(f)
return cookie_data.get(user_name, {})
except:
return {}
def process_website(
self,
user_name: str,
@@ -270,7 +153,19 @@ class CookieMonitor:
all_cookies: dict,
local_cookies: dict
) -> dict:
"""验证网站登录状态"""
"""
验证网站登录状态
Args:
user_name: 用户名
website_config: 网站配置
browser: 浏览器管理器
all_cookies: 所有cookies
local_cookies: 本地cookies
Returns:
验证结果字典
"""
website_name = website_config.get('name', '未知网站')
url = website_config.get('url', '')
check_selector = website_config.get('login_check_selector', '')
@@ -296,9 +191,15 @@ class CookieMonitor:
domain = parsed.netloc
domain_cookies = []
for cookie_domain, cookies in local_cookies.items():
for cookie_domain, cookie_data in local_cookies.items():
clean_domain = cookie_domain.lstrip('.')
if domain in clean_domain or clean_domain in domain:
if isinstance(cookie_data, dict):
cookies = cookie_data.get('cookies', [])
elif isinstance(cookie_data, list):
cookies = cookie_data
else:
cookies = []
domain_cookies.extend(cookies)
if domain_cookies:
@@ -318,14 +219,29 @@ class CookieMonitor:
result['success'] = True
logger.info(f" ✓ 使用本地 cookie 登录验证成功")
self._save_cookies_to_file(domain_cookies, url, website_name, user_name)
if domain_cookies:
logger.info(f" 保存 {len(domain_cookies)} 个 cookie 到本地文件")
self.cookie_data_manager.save_user_cookies(
user_name=user_name,
cookies=domain_cookies,
url=url,
website_name=website_name
)
else:
logger.warning(f" 登录验证成功但 cookies 为空，跳过保存")
else:
logger.warning(f" ✗ 本地 cookie 登录失败，尝试使用服务器 cookie")
domain_cookies = []
for cookie_domain, cookies in all_cookies.items():
for cookie_domain, cookie_data in all_cookies.items():
clean_domain = cookie_domain.lstrip('.')
if domain in clean_domain or clean_domain in domain:
if isinstance(cookie_data, dict):
cookies = cookie_data.get('cookies', [])
elif isinstance(cookie_data, list):
cookies = cookie_data
else:
cookies = []
domain_cookies.extend(cookies)
if domain_cookies:
@@ -345,7 +261,16 @@ class CookieMonitor:
result['success'] = True
logger.info(f" ✓ 使用服务器 cookie 登录验证成功")
self._save_cookies_to_file(domain_cookies, url, website_name, user_name)
if domain_cookies:
logger.info(f" 保存 {len(domain_cookies)} 个 cookie 到本地文件")
self.cookie_data_manager.save_user_cookies(
user_name=user_name,
cookies=domain_cookies,
url=url,
website_name=website_name
)
else:
logger.warning(f" 登录验证成功但 cookies 为空，跳过保存")
else:
result['error'] = '登录状态验证失败（本地和服务器 cookie 都无效）'
logger.warning(f" ✗ {result['error']}")
@@ -356,49 +281,6 @@ class CookieMonitor:
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'])
@@ -418,13 +300,13 @@ class CookieMonitor:
"""发送单个用户的通知"""
user_name = user_result['user_name']
notification_config = user_result.get('notification_config', {})
iyuu_token = notification_config.get('iyuu_token', '')
if not iyuu_token:
notifier = create_notification_manager(notification_config)
if not notifier:
logger.warning(f"用户 {user_name} 未配置通知")
return
notifier = IYUUNotifier(iyuu_token)
success_count = user_result['success_count']
fail_count = user_result['fail_count']
@@ -445,26 +327,13 @@ class CookieMonitor:
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):
@@ -474,7 +343,7 @@ class CookieMonitor:
logger.info(f"时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
logger.info("="*50)
users = self.config.get('users', [])
users = self.config_manager.get_users()
if not users:
logger.error("未找到用户配置")
@@ -486,15 +355,14 @@ class CookieMonitor:
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():
"""主函数"""
from logger import setup_logging
setup_logging()
monitor = CookieMonitor()
monitor.run()
notification.py
+101
View File
@@ -0,0 +1,101 @@
"""信息发送管理模块"""
import requests
from logger import get_logger
logger = get_logger(__name__)
class NotificationManager:
"""通知管理器"""
def __init__(self, token: str = "", provider: str = "iyuu"):
"""
初始化通知管理器
Args:
token: 通知token
provider: 通知服务提供商 (iyuu, wechat, dingtalk等)
"""
self.token = token
self.provider = provider
self.api_url = f"https://iyuu.cn/{token}.send" if provider == "iyuu" else ""
def send(self, title: str, content: str) -> bool:
"""
发送通知
Args:
title: 通知标题
content: 通知内容
Returns:
是否发送成功
"""
if not self.token:
logger.warning("未配置通知token，跳过发送")
return False
if self.provider == "iyuu":
return self._send_iyuu(title, content)
else:
logger.warning(f"不支持的通知提供商: {self.provider}")
return False
def _send_iyuu(self, title: str, content: str) -> bool:
"""
通过爱语飞飞发送通知
Args:
title: 通知标题
content: 通知内容
Returns:
是否发送成功
"""
try:
logger.info(f"发送通知: {title}")
response = requests.post(
self.api_url,
json={
'text': title,
'desp': content
},
timeout=10
)
response.raise_for_status()
result = response.json()
if result.get('errcode') == 0:
logger.info("通知发送成功")
return True
else:
logger.error(f"通知发送失败: {result.get('errmsg')}")
return False
except requests.RequestException as e:
logger.error(f"通知请求失败: {e}")
return False
except Exception as e:
logger.error(f"通知发送异常: {e}")
return False
def create_notification_manager(config: dict) -> NotificationManager:
"""
根据配置创建通知管理器
Args:
config: 通知配置字典
Returns:
通知管理器实例
"""
token = config.get('iyuu_token', '')
provider = config.get('provider', 'iyuu')
if token:
return NotificationManager(token=token, provider=provider)
return None
notifier.py
-54
View File
@@ -1,54 +0,0 @@
"""消息通知模块 - 通过爱语飞飞发送通知"""
import requests
from applogger import get_logger
logger = get_logger(__name__)
class IYUUNotifier:
"""爱语飞飞通知客户端"""
def __init__(self, token: str):
self.token = token
self.api_url = f"https://iyuu.cn/{token}.send"
def send(self, title: str, content: str) -> bool:
print(f"\n{'='*50}")
print(f"[测试] 准备发送通知")
print(f"[测试] API URL: {self.api_url}")
print(f"[测试] Title: {title}")
print(f"[测试] Content: {content[:100]}...")
print(f"{'='*50}\n")
try:
response = requests.get(
self.api_url,
params={
'text': title,
'desp': content
},
timeout=10
)
print(f"[测试] 响应状态码: {response.status_code}")
print(f"[测试] 响应内容: {response.text}")
response.raise_for_status()
result = response.json()
print(f"[测试] 解析结果: {result}")
if result.get('errcode') == 0:
print(f"[测试] ✓ 发送成功!")
return True
else:
print(f"[测试] ✗ 发送失败: {result.get('errmsg')}")
logger.error(f"发送通知失败: {result.get('errmsg')}")
return False
except requests.RequestException as e:
logger.error(f"发送通知请求失败: {e}")
return False
except Exception as e:
logger.error(f"发送通知异常: {e}")
return False
send_test.py
-62
View File
@@ -1,62 +0,0 @@
"""直接发送测试消息给两个用户"""
from notifier import IYUUNotifier
from datetime import datetime
iyuu_token = "IYUU37629Tc1d371c7ce99a49ff9778e196286b7e4592be191"
notifier = IYUUNotifier(iyuu_token)
print("="*60)
print("发送测试消息给用户A")
print("="*60)
title_a = "【测试】用户A - Cookie监控测试"
content_a = f"""**用户A Cookie监控测试报告**
测试时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
测试类型: 直接发送测试消息
这是一条测试消息，用于验证爱语飞飞通知功能是否正常工作。
如果您收到此消息，说明通知功能正常。"""
print(f"\n发送内容:")
print(f"Title: {title_a}")
print(f"Content:\n{content_a}")
print(f"\n发送方式: 爱语飞飞 API (GET 请求)")
print(f"API URL: https://iyuu.cn/{iyuu_token}.send")
print(f"参数: text={title_a}, desp={content_a}")
print("\n正在发送...")
result_a = notifier.send(title_a, content_a)
print(f"\n发送结果: {'✓ 成功' if result_a else '✗ 失败'}")
print("\n" + "="*60)
print("发送测试消息给用户B")
print("="*60)
title_b = "【测试】用户B - Cookie监控测试"
content_b = f"""**用户B Cookie监控测试报告**
测试时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
测试类型: 直接发送测试消息
这是一条测试消息，用于验证爱语飞飞通知功能是否正常工作。
如果您收到此消息，说明通知功能正常。"""
print(f"\n发送内容:")
print(f"Title: {title_b}")
print(f"Content:\n{content_b}")
print(f"\n发送方式: 爱语飞飞 API (GET 请求)")
print(f"API URL: https://iyuu.cn/{iyuu_token}.send")
print(f"参数: text={title_b}, desp={content_b}")
print("\n正在发送...")
result_b = notifier.send(title_b, content_b)
print(f"\n发送结果: {'✓ 成功' if result_b else '✗ 失败'}")
print("\n" + "="*60)
print("测试完成")
print("="*60)
print(f"用户A: {'✓ 发送成功' if result_a else '✗ 发送失败'}")
print(f"用户B: {'✓ 发送成功' if result_b else '✗ 发送失败'}")
test_notify.py
-42
View File
@@ -1,42 +0,0 @@
"""测试爱语飞飞通知"""
import requests
iyuu_token = "IYUU37629Tc1d371c7ce99a49ff9778e196286b7e4592be191"
api_url = f"https://iyuu.cn/{iyuu_token}.send"
title = "【测试】Cookie监控通知测试"
content = """这是一条测试消息。
测试时间: 2026-03-04
测试内容: 验证爱语飞飞通知是否正常工作
如果您收到此消息，说明通知功能正常。"""
print(f"API URL: {api_url}")
print(f"Title: {title}")
print(f"Content: {content}")
print("\n正在发送请求...")
try:
response = requests.get(
api_url,
params={
'text': title,
'desp': content
},
timeout=10
)
print(f"状态码: {response.status_code}")
print(f"响应内容: {response.text}")
result = response.json()
print(f"\n解析结果: {result}")
if result.get('errcode') == 0:
print("\n✓ 发送成功！")
else:
print(f"\n✗ 发送失败: {result.get('errmsg')}")
except Exception as e:
print(f"\n✗ 请求异常: {e}")