189 lines
5.3 KiB
Python
189 lines
5.3 KiB
Python
"""消息通知模块 - 通过爱语飞飞发送通知"""
|
|
import requests
|
|
from typing import Optional
|
|
from applogger import get_logger
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
|
|
class IYUUNotifier:
|
|
"""爱语飞飞通知客户端"""
|
|
|
|
def __init__(self, token: str):
|
|
"""
|
|
初始化爱语飞飞通知客户端
|
|
|
|
Args:
|
|
token: 爱语飞飞令牌
|
|
"""
|
|
self.token = token
|
|
self.api_url = f"https://iyuu.cn/{token}.send"
|
|
|
|
def send(self, title: str, content: str) -> bool:
|
|
"""
|
|
发送通知
|
|
|
|
Args:
|
|
title: 通知标题
|
|
content: 通知内容
|
|
|
|
Returns:
|
|
是否发送成功
|
|
"""
|
|
try:
|
|
response = requests.get(
|
|
self.api_url,
|
|
params={
|
|
'text': title,
|
|
'desp': content
|
|
},
|
|
timeout=10
|
|
)
|
|
response.raise_for_status()
|
|
result = response.json()
|
|
|
|
if result.get('errcode') == 0:
|
|
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
|
|
|
|
|
|
class FailureTracker:
|
|
"""失败次数追踪器"""
|
|
|
|
def __init__(self, state_file: str):
|
|
"""
|
|
初始化失败次数追踪器
|
|
|
|
Args:
|
|
state_file: 状态文件路径
|
|
"""
|
|
self.state_file = state_file
|
|
self.state = self._load_state()
|
|
|
|
def _load_state(self) -> dict:
|
|
"""加载状态文件"""
|
|
import json
|
|
import os
|
|
|
|
if os.path.exists(self.state_file):
|
|
try:
|
|
with open(self.state_file, 'r', encoding='utf-8') as f:
|
|
return json.load(f)
|
|
except Exception as e:
|
|
logger.error(f"加载状态文件失败: {e}")
|
|
|
|
return {}
|
|
|
|
def _save_state(self):
|
|
"""保存状态文件"""
|
|
import json
|
|
|
|
try:
|
|
with open(self.state_file, 'w', encoding='utf-8') as f:
|
|
json.dump(self.state, f, indent=2, ensure_ascii=False)
|
|
except Exception as e:
|
|
logger.error(f"保存状态文件失败: {e}")
|
|
|
|
def get_fail_count(self, user_name: str, website_name: str) -> int:
|
|
"""
|
|
获取指定网站的失败次数
|
|
|
|
Args:
|
|
user_name: 用户名
|
|
website_name: 网站名
|
|
|
|
Returns:
|
|
失败次数
|
|
"""
|
|
key = f"{user_name}_{website_name}"
|
|
return self.state.get(key, 0)
|
|
|
|
def increment_fail_count(self, user_name: str, website_name: str) -> int:
|
|
"""
|
|
增加失败次数
|
|
|
|
Args:
|
|
user_name: 用户名
|
|
website_name: 网站名
|
|
|
|
Returns:
|
|
增加后的失败次数
|
|
"""
|
|
key = f"{user_name}_{website_name}"
|
|
self.state[key] = self.state.get(key, 0) + 1
|
|
self._save_state()
|
|
return self.state[key]
|
|
|
|
def reset_fail_count(self, user_name: str, website_name: str):
|
|
"""
|
|
重置失败次数
|
|
|
|
Args:
|
|
user_name: 用户名
|
|
website_name: 网站名
|
|
"""
|
|
key = f"{user_name}_{website_name}"
|
|
self.state[key] = 0
|
|
self._save_state()
|
|
|
|
def check_and_notify(
|
|
self,
|
|
user_name: str,
|
|
website_name: str,
|
|
max_fail_count: int,
|
|
notifier: Optional[IYUUNotifier],
|
|
error_msg: str
|
|
) -> bool:
|
|
"""
|
|
检查失败次数并在达到阈值时发送通知
|
|
|
|
Args:
|
|
user_name: 用户名
|
|
website_name: 网站名
|
|
max_fail_count: 最大失败次数
|
|
notifier: 爱语飞飞通知客户端
|
|
error_msg: 错误信息
|
|
|
|
Returns:
|
|
是否应该停止重试
|
|
"""
|
|
fail_count = self.increment_fail_count(user_name, website_name)
|
|
|
|
if fail_count >= max_fail_count:
|
|
# 发送通知
|
|
if notifier:
|
|
title = f"【Cookie监控】{user_name} - {website_name} 登录失败"
|
|
content = f"网站: {website_name}\n用户: {user_name}\n连续失败次数: {fail_count}\n错误信息: {error_msg}"
|
|
|
|
if notifier.send(title, content):
|
|
logger.info(f"已发送失败通知: {user_name} - {website_name}")
|
|
else:
|
|
logger.error(f"发送失败通知失败: {user_name} - {website_name}")
|
|
|
|
# 重置失败计数
|
|
self.reset_fail_count(user_name, website_name)
|
|
return True
|
|
|
|
return False
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# 测试代码
|
|
tracker = FailureTracker("state.json")
|
|
|
|
# 增加失败次数
|
|
count = tracker.increment_fail_count("测试用户", "测试网站")
|
|
logger.info(f"当前失败次数: {count}")
|
|
|
|
# 检查并发送通知（需要真实的爱语飞飞令牌）
|
|
# notifier = IYUUNotifier("您的爱语飞飞令牌")
|
|
# tracker.check_and_notify("测试用户", "测试网站", 3, notifier, "测试错误信息") |