Compare commits

14 Commits
Author SHA1 Message Date
bwstudio 9fb210970f 修复:后端to_dict转换时间为本地时区,兼容新旧数据 2026-04-18 22:34:15 +08:00
bwstudio c08ca148e7 优化:监控页面加载时先显示所有用户网站初始状态,再检测网络状态实时更新 2026-04-18 22:24:54 +08:00
bwstudio 6a2abe8681 修复:检测时间时区转换,使用UTC+0时区确保前端正确显示北京时间 2026-04-18 22:17:14 +08:00
bwstudio be1709b622 项目整理:添加Cookie Cloud下载功能、优化前端界面、配置改为config.json、打包脚本、清理无用文件 2026-04-18 21:48:29 +08:00
bwstudio 70802b9b50 重构Cookie监控系统:添加Web管理后台、Vue3前端界面、Cookie Cloud参数支持、5秒倒计时自动刷新、专业UI配色 2026-04-17 22:37:28 +08:00
bwadmin ed95ce5b23 移除监控报告功能,清理日志输出 2026-03-08 09:20:03 +08:00
bwadmin 3e963d1a52 添加浏览器管理模块,修复登录验证逻辑 2026-03-05 21:46:39 +08:00
bwadmin a70af1b0da 重构项目结构,优化模块划分,修复数据结构和登录验证问题 2026-03-05 17:05:35 +08:00
bwadmin e090f546c6 实现双重验证机制:优先使用本地cookies,失败时回退到服务器cookies 2026-03-05 08:33:24 +08:00
bwadmin 7a896f56ba 修改通知时间为晚上10点,添加cookie保存功能,按用户和域名区分保存 2026-03-04 22:45:31 +08:00
bwadmin 918ac92230 优化cookie提取器,移除Playwright依赖,重新设计界面和功能 2026-03-04 17:36:09 +08:00
bwadmin c3d07f7325 完善 cookie 监控系统:
1. 修复 browser_login.py 中的 cookie 处理错误
2. 重构 cookie_cloud.py 实现浏览器实例级 cookie 共享
3. 创建 cookie_extractor.py GUI 应用程序,支持打开、保存和恢复功能
4. 优化 cookie 导入方式,支持批量导入和回退机制
5. 修复 localStorage 注入问题,确保特殊字符正确处理
2026-03-03 17:10:50 +08:00
bwadmin 894acc3191 add-files 2026-03-03 08:58:16 +08:00
bwadmin 0fbbbce2ee update 2026-03-03 08:53:50 +08:00
38 changed files with 3253 additions and 1845 deletions
+43
View File
@@ -0,0 +1,43 @@
# Python
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
*.egg-info/
.installed.cfg
*.egg
# Virtual Environment
venv/
env/
ENV/
.venv
# IDE
.vscode/
.idea/
*.swp
*.swo
*~
# Logs
*.log
log/
# Data files (optional - uncomment if you don't want to commit these)
# cookie_data.json
# cookies.json
# config.json
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
-62
View File
@@ -1,62 +0,0 @@
"""日志配置模块"""
import os
import logging
from logging.handlers import TimedRotatingFileHandler
def setup_logging(
log_dir: str = "log",
log_file: str = "app.log",
level: int = logging.INFO,
format_string: str = '%(asctime)s - %(levelname)s - %(message)s',
when: str = "H",
interval: int = 12,
backup_count: int = 7
) -> None:
"""
配置日志系统
Args:
log_dir: 日志目录
log_file: 日志文件名
level: 日志级别
format_string: 日志格式
when: 轮转时间单位 (S:秒, M:分, H:时, D:天, W:周)
interval: 轮转间隔
backup_count: 保留的备份文件数量
"""
# 确保日志目录存在
if not os.path.exists(log_dir):
os.makedirs(log_dir)
# 配置日志处理器
handlers = [
TimedRotatingFileHandler(
os.path.join(log_dir, log_file),
when=when,
interval=interval,
backupCount=backup_count,
encoding='utf-8'
),
logging.StreamHandler()
]
# 配置日志
logging.basicConfig(
level=level,
format=format_string,
handlers=handlers
)
def get_logger(name: str) -> logging.Logger:
"""
获取日志记录器
Args:
name: 日志记录器名称,通常使用 __name__
Returns:
Logger 实例
"""
return logging.getLogger(name)
+52
View File
@@ -0,0 +1,52 @@
"""
Cookie监控管理系统 - 后台服务
"""
from flask import Flask, jsonify
from flask_cors import CORS
import os
# 导入配置和数据库
from config.settings import Config
from models.database import db
from routes.users import user_bp
from routes.websites import website_bp
from routes.cookies import cookie_bp
from routes.detections import detection_bp
from routes.notifications import notification_bp
from routes.admin import admin_bp
def create_app(config_class=Config):
"""创建Flask应用"""
app = Flask(__name__)
app.config.from_object(config_class)
# 配置CORS(允许session cookie
CORS(app, supports_credentials=True)
# 初始化数据库
db.init_app(app)
# 注册蓝图
app.register_blueprint(admin_bp)
app.register_blueprint(user_bp, url_prefix='/api/users')
app.register_blueprint(website_bp, url_prefix='/api/websites')
app.register_blueprint(cookie_bp, url_prefix='/api/cookies')
app.register_blueprint(detection_bp, url_prefix='/api/detections')
app.register_blueprint(notification_bp, url_prefix='/api/notifications')
# 健康检查接口
@app.route('/api/health', methods=['GET'])
def health_check():
return jsonify({'status': 'ok', 'message': '服务运行中'})
# 创建数据库表
with app.app_context():
db.create_all()
return app
if __name__ == '__main__':
app = create_app()
app.run(host='0.0.0.0', port=5000, debug=False)
View File
+12
View File
@@ -0,0 +1,12 @@
"""
配置文件
"""
import os
class Config:
"""基础配置"""
SECRET_KEY = os.environ.get('SECRET_KEY') or 'cookie-monitor-secret-key'
SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or 'sqlite:///cookie_monitor.db'
SQLALCHEMY_TRACK_MODIFICATIONS = False
JSON_AS_ASCII = False
ADMIN_PASSWORD = os.environ.get('ADMIN_PASSWORD') or 'admin'
Binary file not shown.
View File
+92
View File
@@ -0,0 +1,92 @@
"""
数据库模型定义
"""
from flask_sqlalchemy import SQLAlchemy
from datetime import datetime, timezone
db = SQLAlchemy()
class User(db.Model):
"""用户表"""
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True, autoincrement=True)
name = db.Column(db.String(100), nullable=False)
iyuu_token = db.Column(db.String(255))
max_fail_count = db.Column(db.Integer, default=3)
cookie_cloud_uuid = db.Column(db.String(255))
cookie_cloud_password = db.Column(db.String(255))
cookie_cloud_api_url = db.Column(db.String(255))
created_at = db.Column(db.DateTime(timezone=True), default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime(timezone=True), default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
# 关联关系
websites = db.relationship('Website', backref='user', lazy=True, cascade='all, delete-orphan')
notifications = db.relationship('Notification', backref='user', lazy=True)
class Website(db.Model):
"""网站表"""
__tablename__ = 'websites'
id = db.Column(db.Integer, primary_key=True, autoincrement=True)
user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
name = db.Column(db.String(100), nullable=False)
url = db.Column(db.String(255), nullable=False)
login_check_selector = db.Column(db.String(255))
success_text = db.Column(db.String(255))
interval_minutes = db.Column(db.Integer, default=30, name='interval')
status = db.Column(db.Integer, default=1)
created_at = db.Column(db.DateTime(timezone=True), default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime(timezone=True), default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
# 关联关系
cookies = db.relationship('Cookie', backref='website', lazy=True, cascade='all, delete-orphan')
detections = db.relationship('Detection', backref='website', lazy=True, cascade='all, delete-orphan')
notifications = db.relationship('Notification', backref='website', lazy=True)
class Cookie(db.Model):
"""Cookie表"""
__tablename__ = 'cookies'
id = db.Column(db.Integer, primary_key=True, autoincrement=True)
website_id = db.Column(db.Integer, db.ForeignKey('websites.id'), nullable=False)
name = db.Column(db.String(100), nullable=False)
value = db.Column(db.Text, nullable=False)
domain = db.Column(db.String(255), nullable=False)
path = db.Column(db.String(255), default='/')
expiry = db.Column(db.DateTime)
secure = db.Column(db.Integer, default=0)
http_only = db.Column(db.Integer, default=0)
created_at = db.Column(db.DateTime(timezone=True), default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime(timezone=True), default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
class Detection(db.Model):
"""检测结果表"""
__tablename__ = 'detections'
id = db.Column(db.Integer, primary_key=True, autoincrement=True)
website_id = db.Column(db.Integer, db.ForeignKey('websites.id'), nullable=False)
status = db.Column(db.Integer, nullable=False)
response_time = db.Column(db.Float)
http_status = db.Column(db.Integer)
message = db.Column(db.Text)
created_at = db.Column(db.DateTime(timezone=True), default=lambda: datetime.now(timezone.utc))
class Notification(db.Model):
"""通知表"""
__tablename__ = 'notifications'
id = db.Column(db.Integer, primary_key=True, autoincrement=True)
user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
website_id = db.Column(db.Integer, db.ForeignKey('websites.id'), nullable=False)
title = db.Column(db.String(100), nullable=False)
content = db.Column(db.Text, nullable=False)
status = db.Column(db.String(20), default='pending')
send_time = db.Column(db.DateTime)
created_at = db.Column(db.DateTime(timezone=True), default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime(timezone=True), default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
+5
View File
@@ -0,0 +1,5 @@
Flask==3.0.0
Flask-SQLAlchemy==3.1.1
Flask-CORS==4.0.0
requests==2.31.0
DrissionPage==4.1.1.2
View File
+40
View File
@@ -0,0 +1,40 @@
"""
后台管理路由
"""
from flask import Blueprint, render_template, request, jsonify, session
from config.settings import Config
admin_bp = Blueprint('admin', __name__)
@admin_bp.route('/')
@admin_bp.route('/<path:path>')
def admin_page(path=None):
"""管理后台页面"""
return render_template('index.html')
@admin_bp.route('/api/admin/login', methods=['POST'])
def admin_login():
"""后台登录验证"""
data = request.json
password = data.get('password', '')
if password == Config.ADMIN_PASSWORD:
session['admin_logged_in'] = True
return jsonify({'success': True, 'message': '登录成功'})
else:
return jsonify({'success': False, 'message': '密码错误'}), 401
@admin_bp.route('/api/admin/logout', methods=['POST'])
def admin_logout():
"""后台退出登录"""
session.pop('admin_logged_in', None)
return jsonify({'success': True, 'message': '已退出'})
@admin_bp.route('/api/admin/status', methods=['GET'])
def admin_status():
"""检查后台登录状态"""
return jsonify({'logged_in': session.get('admin_logged_in', False)})
+114
View File
@@ -0,0 +1,114 @@
"""
Cookie管理路由
"""
from flask import Blueprint, request, jsonify
from models.database import db, Cookie
from datetime import datetime
cookie_bp = Blueprint('cookies', __name__)
@cookie_bp.route('/websites/<int:website_id>', methods=['GET'])
def get_cookies(website_id):
"""获取网站的cookie"""
cookies = Cookie.query.filter_by(website_id=website_id).all()
return jsonify([cookie.to_dict() for cookie in cookies])
@cookie_bp.route('/websites/<int:website_id>', methods=['POST'])
def create_cookie(website_id):
"""创建cookie"""
data = request.json
cookie = Cookie(
website_id=website_id,
name=data.get('name'),
value=data.get('value'),
domain=data.get('domain'),
path=data.get('path', '/'),
expiry=datetime.fromisoformat(data.get('expiry')) if data.get('expiry') else None,
secure=data.get('secure', 0),
http_only=data.get('http_only', 0)
)
db.session.add(cookie)
db.session.commit()
return jsonify(cookie.to_dict()), 201
@cookie_bp.route('/<int:cookie_id>', methods=['PUT'])
def update_cookie(cookie_id):
"""更新cookie"""
cookie = Cookie.query.get_or_404(cookie_id)
data = request.json
cookie.name = data.get('name', cookie.name)
cookie.value = data.get('value', cookie.value)
cookie.domain = data.get('domain', cookie.domain)
cookie.path = data.get('path', cookie.path)
cookie.expiry = datetime.fromisoformat(data.get('expiry')) if data.get('expiry') else cookie.expiry
cookie.secure = data.get('secure', cookie.secure)
cookie.http_only = data.get('http_only', cookie.http_only)
db.session.commit()
return jsonify(cookie.to_dict())
@cookie_bp.route('/<int:cookie_id>', methods=['DELETE'])
def delete_cookie(cookie_id):
"""删除cookie"""
cookie = Cookie.query.get_or_404(cookie_id)
db.session.delete(cookie)
db.session.commit()
return jsonify({'message': 'Cookie已删除'})
@cookie_bp.route('/websites/<int:website_id>', methods=['PUT'])
def batch_update_cookies(website_id):
"""批量更新网站的cookie"""
data = request.json
cookies_data = data.get('cookies', [])
# 删除该网站的所有现有cookie
Cookie.query.filter_by(website_id=website_id).delete()
# 添加新的cookie
for cookie_data in cookies_data:
cookie = Cookie(
website_id=website_id,
name=cookie_data.get('name'),
value=cookie_data.get('value'),
domain=cookie_data.get('domain'),
path=cookie_data.get('path', '/'),
expiry=datetime.fromisoformat(cookie_data.get('expiry')) if cookie_data.get('expiry') else None,
secure=cookie_data.get('secure', 0),
http_only=cookie_data.get('http_only', 0)
)
db.session.add(cookie)
db.session.commit()
return jsonify({'message': 'Cookies已批量更新'})
# 为Cookie模型添加to_dict方法
def cookie_to_dict(self):
return {
'id': self.id,
'website_id': self.website_id,
'name': self.name,
'value': self.value,
'domain': self.domain,
'path': self.path,
'expiry': self.expiry.isoformat() if self.expiry else None,
'secure': self.secure,
'http_only': self.http_only,
'created_at': self.created_at.isoformat() if self.created_at else None,
'updated_at': self.updated_at.isoformat() if self.updated_at else None
}
Cookie.to_dict = cookie_to_dict
+89
View File
@@ -0,0 +1,89 @@
"""
检测结果路由
"""
from flask import Blueprint, request, jsonify
from models.database import db, Detection
detection_bp = Blueprint('detections', __name__)
@detection_bp.route('/', methods=['GET'])
def get_detections():
"""获取检测结果"""
website_id = request.args.get('website_id')
if website_id:
detections = Detection.query.filter_by(website_id=website_id).order_by(Detection.created_at.desc()).all()
else:
detections = Detection.query.order_by(Detection.created_at.desc()).all()
return jsonify([detection.to_dict() for detection in detections])
@detection_bp.route('/', methods=['POST'])
def create_detection():
"""保存检测结果"""
data = request.json
detection = Detection(
website_id=data.get('website_id'),
status=data.get('status'),
response_time=data.get('response_time'),
http_status=data.get('http_status'),
message=data.get('message')
)
db.session.add(detection)
db.session.commit()
return jsonify(detection.to_dict()), 201
@detection_bp.route('/latest', methods=['GET'])
def get_latest_detections():
"""获取每个网站的最新检测结果"""
from sqlalchemy import func
subquery = db.session.query(
Detection.website_id,
func.max(Detection.id).label('max_id')
).group_by(Detection.website_id).subquery()
detections = db.session.query(Detection).join(
subquery,
(Detection.website_id == subquery.c.website_id) &
(Detection.id == subquery.c.max_id)
).all()
result = {}
for d in detections:
result[d.website_id] = d.to_dict()
return jsonify(result)
@detection_bp.route('/<int:detection_id>', methods=['GET'])
def get_detection(detection_id):
"""获取单个检测结果"""
detection = Detection.query.get_or_404(detection_id)
return jsonify(detection.to_dict())
# 为Detection模型添加to_dict方法
def detection_to_dict(self):
created_at = self.created_at
if created_at:
if created_at.tzinfo is None:
from datetime import timezone
created_at = created_at.replace(tzinfo=timezone.utc)
created_at = created_at.astimezone()
return {
'id': self.id,
'website_id': self.website_id,
'status': self.status,
'response_time': self.response_time,
'http_status': self.http_status,
'message': self.message,
'created_at': created_at.isoformat() if created_at else None
}
Detection.to_dict = detection_to_dict
+102
View File
@@ -0,0 +1,102 @@
"""
通知管理路由
"""
import requests
from flask import Blueprint, request, jsonify
from models.database import db, Notification, User
from datetime import datetime
notification_bp = Blueprint('notifications', __name__)
@notification_bp.route('/', methods=['GET'])
def get_notifications():
"""获取通知列表"""
user_id = request.args.get('user_id')
if user_id:
notifications = Notification.query.filter_by(user_id=user_id).order_by(Notification.created_at.desc()).all()
else:
notifications = Notification.query.order_by(Notification.created_at.desc()).all()
return jsonify([notification.to_dict() for notification in notifications])
@notification_bp.route('/', methods=['POST'])
def send_notification():
"""发送通知"""
data = request.json
user_id = data.get('user_id')
website_id = data.get('website_id')
title = data.get('title')
content = data.get('content')
# 获取用户的爱语飞飞令牌
user = User.query.get_or_404(user_id)
if not user.iyuu_token:
return jsonify({'message': '用户未配置爱语飞飞令牌'}), 400
# 发送爱语飞飞通知
send_success = _send_iyuu_notification(user.iyuu_token, title, content)
# 创建通知记录
notification = Notification(
user_id=user_id,
website_id=website_id,
title=title,
content=content,
status='sent' if send_success else 'failed',
send_time=datetime.utcnow() if send_success else None
)
db.session.add(notification)
db.session.commit()
return jsonify(notification.to_dict()), 201
def _send_iyuu_notification(token, title, content):
"""通过爱语飞飞发送通知"""
try:
url = f"https://iyuu.cn/{token}.send"
response = requests.post(
url,
json={
'text': title,
'desp': content
},
timeout=10
)
response.raise_for_status()
result = response.json()
if result.get('errcode') == 0:
return True
else:
return False
except requests.RequestException:
return False
except Exception:
return False
# 为Notification模型添加to_dict方法
def notification_to_dict(self):
return {
'id': self.id,
'user_id': self.user_id,
'website_id': self.website_id,
'title': self.title,
'content': self.content,
'status': self.status,
'send_time': self.send_time.isoformat() if self.send_time else None,
'created_at': self.created_at.isoformat() if self.created_at else None,
'updated_at': self.updated_at.isoformat() if self.updated_at else None
}
Notification.to_dict = notification_to_dict
+78
View File
@@ -0,0 +1,78 @@
"""
用户管理路由
"""
from flask import Blueprint, request, jsonify
from models.database import db, User
user_bp = Blueprint('users', __name__)
@user_bp.route('/', methods=['GET'])
def get_users():
"""获取所有用户"""
users = User.query.all()
return jsonify([user.to_dict() for user in users])
@user_bp.route('/', methods=['POST'])
def create_user():
"""创建用户"""
data = request.json
user = User(
name=data.get('name'),
iyuu_token=data.get('iyuu_token'),
max_fail_count=data.get('max_fail_count', 3),
cookie_cloud_uuid=data.get('cookie_cloud_uuid'),
cookie_cloud_password=data.get('cookie_cloud_password'),
cookie_cloud_api_url=data.get('cookie_cloud_api_url')
)
db.session.add(user)
db.session.commit()
return jsonify(user.to_dict()), 201
@user_bp.route('/<int:user_id>', methods=['GET'])
def get_user(user_id):
"""获取单个用户"""
user = User.query.get_or_404(user_id)
return jsonify(user.to_dict())
@user_bp.route('/<int:user_id>', methods=['PUT'])
def update_user(user_id):
"""更新用户"""
user = User.query.get_or_404(user_id)
data = request.json
user.name = data.get('name', user.name)
user.iyuu_token = data.get('iyuu_token', user.iyuu_token)
user.max_fail_count = data.get('max_fail_count', user.max_fail_count)
user.cookie_cloud_uuid = data.get('cookie_cloud_uuid', user.cookie_cloud_uuid)
user.cookie_cloud_password = data.get('cookie_cloud_password', user.cookie_cloud_password)
user.cookie_cloud_api_url = data.get('cookie_cloud_api_url', user.cookie_cloud_api_url)
db.session.commit()
return jsonify(user.to_dict())
@user_bp.route('/<int:user_id>', methods=['DELETE'])
def delete_user(user_id):
"""删除用户"""
user = User.query.get_or_404(user_id)
db.session.delete(user)
db.session.commit()
return jsonify({'message': '用户已删除'})
def user_to_dict(self):
return {
'id': self.id,
'name': self.name,
'iyuu_token': self.iyuu_token,
'max_fail_count': self.max_fail_count,
'cookie_cloud_uuid': self.cookie_cloud_uuid,
'cookie_cloud_password': self.cookie_cloud_password,
'cookie_cloud_api_url': self.cookie_cloud_api_url,
'created_at': self.created_at.isoformat() if self.created_at else None,
'updated_at': self.updated_at.isoformat() if self.updated_at else None
}
User.to_dict = user_to_dict
+254
View File
@@ -0,0 +1,254 @@
"""
网站管理路由
"""
from flask import Blueprint, request, jsonify
from models.database import db, Website
from models.database import Cookie as CookieModel
import requests as http_requests
from datetime import datetime
website_bp = Blueprint('websites', __name__)
@website_bp.route('/check-network', methods=['GET'])
def check_all_network():
"""检测所有网站的网络可达性"""
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
websites = Website.query.all()
result = {}
for website in websites:
try:
resp = http_requests.get(website.url, timeout=5, allow_redirects=True, verify=False)
result[website.id] = {
'status': resp.status_code,
'reachable': resp.status_code < 500,
'response_time': resp.elapsed.total_seconds() * 1000
}
except http_requests.exceptions.SSLError:
try:
resp = http_requests.get(website.url, timeout=5, allow_redirects=True, verify=False)
result[website.id] = {
'status': resp.status_code,
'reachable': resp.status_code < 500,
'response_time': resp.elapsed.total_seconds() * 1000
}
except Exception:
result[website.id] = {
'status': 0,
'reachable': False,
'response_time': 0
}
except Exception:
result[website.id] = {
'status': 0,
'reachable': False,
'response_time': 0
}
return jsonify(result)
@website_bp.route('/check-login/<int:website_id>', methods=['POST'])
def check_website_login(website_id):
"""检查网站登录状态"""
from DrissionPage import ChromiumPage, ChromiumOptions
website = Website.query.get_or_404(website_id)
try:
# 获取网站的cookie
cookies = CookieModel.query.filter_by(website_id=website_id).all()
if not cookies:
return jsonify({'success': False, 'message': '没有可用的cookie'})
# 初始化浏览器
co = ChromiumOptions()
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')
browser = ChromiumPage(addr_or_opts=co)
# 导入cookie
for cookie in cookies:
cookie_dict = {
'name': cookie.name,
'value': cookie.value,
'domain': cookie.domain,
'path': cookie.path
}
browser.set.cookies(cookie_dict)
# 访问网站
browser.get(website.url)
browser.wait.doc_loaded(timeout=30)
import time
time.sleep(2)
browser.refresh()
browser.wait.doc_loaded(timeout=30)
time.sleep(2)
# 检查登录状态
login_success = False
check_selector = website.login_check_selector
success_text = website.success_text
if check_selector:
try:
element = browser.ele(check_selector, timeout=10)
if element:
element_text = element.text or ""
if not success_text:
login_success = True
elif success_text in element_text:
login_success = True
except:
pass
if not login_success and success_text:
try:
page_text = browser.html or ""
if success_text in page_text:
login_success = True
except:
pass
# 如果登录成功,获取最新cookie
if login_success:
new_cookies = browser.cookies()
# 更新cookie
CookieModel.query.filter_by(website_id=website_id).delete()
for c in new_cookies:
new_cookie = CookieModel(
website_id=website_id,
name=c.get('name', ''),
value=c.get('value', ''),
domain=c.get('domain', ''),
path=c.get('path', '/')
)
db.session.add(new_cookie)
# 保存检测结果
from models.database import Detection
detection = Detection(
website_id=website_id,
status=1,
response_time=0,
http_status=200,
message='登录验证成功'
)
db.session.add(detection)
db.session.commit()
result = {'success': True, 'message': '登录成功'}
else:
# 保存检测结果
from models.database import Detection
detection = Detection(
website_id=website_id,
status=0,
response_time=0,
http_status=200,
message='登录验证失败'
)
db.session.add(detection)
db.session.commit()
result = {'success': False, 'message': '登录失败'}
browser.quit()
return jsonify(result)
except Exception as e:
return jsonify({'success': False, 'message': f'检测异常: {str(e)}'})
@website_bp.route('/', methods=['GET'])
def get_websites():
"""获取所有网站"""
user_id = request.args.get('user_id')
if user_id:
websites = Website.query.filter_by(user_id=user_id).all()
else:
websites = Website.query.all()
return jsonify([website.to_dict() for website in websites])
@website_bp.route('/', methods=['POST'])
def create_website():
"""创建网站"""
data = request.json
website = Website(
user_id=data.get('user_id'),
name=data.get('name'),
url=data.get('url'),
login_check_selector=data.get('login_check_selector'),
success_text=data.get('success_text'),
interval_minutes=data.get('interval_minutes', 30),
status=data.get('status', 1)
)
db.session.add(website)
db.session.commit()
return jsonify(website.to_dict()), 201
@website_bp.route('/<int:website_id>', methods=['GET'])
def get_website(website_id):
"""获取单个网站"""
website = Website.query.get_or_404(website_id)
return jsonify(website.to_dict())
@website_bp.route('/<int:website_id>', methods=['PUT'])
def update_website(website_id):
"""更新网站"""
website = Website.query.get_or_404(website_id)
data = request.json
website.name = data.get('name', website.name)
website.url = data.get('url', website.url)
website.login_check_selector = data.get('login_check_selector', website.login_check_selector)
website.success_text = data.get('success_text', website.success_text)
website.interval_minutes = data.get('interval_minutes', website.interval_minutes)
website.status = data.get('status', website.status)
db.session.commit()
return jsonify(website.to_dict())
@website_bp.route('/<int:website_id>', methods=['DELETE'])
def delete_website(website_id):
"""删除网站"""
website = Website.query.get_or_404(website_id)
db.session.delete(website)
db.session.commit()
return jsonify({'message': '网站已删除'})
# 为Website模型添加to_dict方法
def website_to_dict(self):
return {
'id': self.id,
'user_id': self.user_id,
'name': self.name,
'url': self.url,
'login_check_selector': self.login_check_selector,
'success_text': self.success_text,
'interval_minutes': self.interval_minutes,
'status': self.status,
'created_at': self.created_at.isoformat() if self.created_at else None,
'updated_at': self.updated_at.isoformat() if self.updated_at else None
}
Website.to_dict = website_to_dict
+564
View File
@@ -0,0 +1,564 @@
/* Cookie监控管理系统 - 专业版样式 */
:root {
--primary-color: #4f6ef7;
--primary-dark: #3b5bdb;
--primary-light: #e8edff;
--success-color: #10b981;
--success-light: #d1fae5;
--danger-color: #ef4444;
--danger-light: #fee2e2;
--warning-color: #f59e0b;
--warning-light: #fef3c7;
--info-color: #6b7280;
--info-light: #f3f4f6;
--header-bg: linear-gradient(135deg, #1a1a2e 0%, #16213e 100%);
--card-bg: #ffffff;
--page-bg: #f0f2f5;
--text-primary: #1f2937;
--text-secondary: #6b7280;
--text-light: #9ca3af;
--border-color: #e5e7eb;
--shadow-sm: 0 1px 3px rgba(0, 0, 0, 0.08);
--shadow-md: 0 4px 12px rgba(0, 0, 0, 0.1);
--shadow-lg: 0 8px 24px rgba(0, 0, 0, 0.12);
--radius: 8px;
--radius-lg: 12px;
}
* {
margin: 0;
padding: 0;
box-sizing: border-box;
}
body {
font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', 'PingFang SC', 'Hiragino Sans GB', 'Microsoft YaHei', sans-serif;
background-color: var(--page-bg);
color: var(--text-primary);
-webkit-font-smoothing: antialiased;
-moz-osx-font-smoothing: grayscale;
}
.container {
min-height: 100vh;
background: var(--page-bg);
}
.el-header {
background: var(--header-bg);
color: #fff;
display: flex;
justify-content: space-between;
align-items: center;
padding: 0 32px;
height: 64px;
box-shadow: 0 2px 8px rgba(0, 0, 0, 0.15);
}
.header-left {
display: flex;
align-items: center;
gap: 20px;
}
.el-header h1 {
font-size: 20px;
font-weight: 600;
letter-spacing: 0.5px;
margin: 0;
background: linear-gradient(135deg, #fff 0%, #e0e7ff 100%);
-webkit-background-clip: text;
-webkit-text-fill-color: transparent;
background-clip: text;
}
.header-stat {
color: rgba(255, 255, 255, 0.8);
font-size: 14px;
font-weight: 500;
}
.header-stat b {
color: #fff;
font-size: 18px;
margin-left: 4px;
}
.header-right {
display: flex;
align-items: center;
gap: 12px;
}
.header-right .el-button {
background: rgba(255, 255, 255, 0.1);
border-color: rgba(255, 255, 255, 0.2);
color: rgba(255, 255, 255, 0.9);
font-weight: 500;
transition: all 0.2s;
}
.header-right .el-button:hover {
background: rgba(255, 255, 255, 0.2);
border-color: rgba(255, 255, 255, 0.3);
color: #fff;
}
.auto-refresh-indicator {
display: flex;
align-items: center;
gap: 6px;
color: rgba(255, 255, 255, 0.8);
font-size: 13px;
padding: 6px 12px;
background: rgba(255, 255, 255, 0.08);
border-radius: 20px;
backdrop-filter: blur(4px);
}
.rotating-icon {
animation: rotate 5s linear infinite;
color: var(--primary-color);
}
.refresh-text {
font-size: 13px;
font-weight: 500;
}
@keyframes rotate {
from { transform: rotate(0deg); }
to { transform: rotate(360deg); }
}
.el-main {
background: var(--page-bg);
padding: 24px 32px;
min-height: calc(100vh - 64px);
}
/* 仪表板样式 */
.dashboard-header {
display: flex;
justify-content: space-between;
align-items: center;
margin-bottom: 28px;
padding-bottom: 20px;
border-bottom: 1px solid var(--border-color);
}
.dashboard-header h2 {
font-size: 24px;
font-weight: 700;
color: var(--text-primary);
margin: 0;
letter-spacing: -0.5px;
}
.dashboard-stats {
display: flex;
gap: 24px;
}
.dashboard-stats .el-statistic {
padding: 16px 24px;
background: var(--card-bg);
border-radius: var(--radius-lg);
box-shadow: var(--shadow-sm);
min-width: 140px;
text-align: center;
transition: all 0.2s;
}
.dashboard-stats .el-statistic:hover {
transform: translateY(-2px);
box-shadow: var(--shadow-md);
}
.dashboard-stats .el-statistic__title {
font-size: 12px;
color: var(--text-secondary);
text-transform: uppercase;
letter-spacing: 0.5px;
font-weight: 600;
}
.dashboard-stats .el-statistic__content {
font-size: 28px;
font-weight: 700;
color: var(--text-primary);
}
/* 用户区域卡片 */
.user-section {
background: var(--card-bg);
border-radius: var(--radius-lg);
padding: 24px;
margin-bottom: 20px;
box-shadow: var(--shadow-sm);
border: 1px solid var(--border-color);
transition: box-shadow 0.2s;
}
.user-section:hover {
box-shadow: var(--shadow-md);
}
.user-header {
display: flex;
justify-content: space-between;
align-items: center;
margin-bottom: 20px;
padding-bottom: 16px;
border-bottom: 2px solid var(--primary-light);
}
.user-header h3 {
margin: 0;
color: var(--text-primary);
font-size: 18px;
font-weight: 600;
display: flex;
align-items: center;
gap: 8px;
}
.user-header h3::before {
content: '';
display: inline-block;
width: 4px;
height: 20px;
background: var(--primary-color);
border-radius: 2px;
}
.user-header .el-tag {
padding: 6px 14px;
border-radius: 20px;
font-weight: 600;
font-size: 13px;
border: none;
}
/* 表格样式优化 */
.el-table {
border-radius: var(--radius);
overflow: hidden;
}
.el-table th.el-table__cell {
background: var(--page-bg);
color: var(--text-secondary);
font-weight: 600;
font-size: 13px;
text-transform: uppercase;
letter-spacing: 0.3px;
border-bottom: 2px solid var(--border-color);
}
.el-table td.el-table__cell {
font-size: 14px;
color: var(--text-primary);
border-bottom: 1px solid var(--border-color);
}
.el-table--enable-row-hover .el-table__body tr:hover > td.el-table__cell {
background-color: var(--primary-light);
}
.el-table::before {
display: none;
}
/* 状态标签样式 */
.el-tag {
border-radius: 20px;
padding: 4px 12px;
font-weight: 500;
border: none;
}
.el-tag--success {
background: var(--success-light);
color: var(--success-color);
}
.el-tag--danger {
background: var(--danger-light);
color: var(--danger-color);
}
.el-tag--warning {
background: var(--warning-light);
color: var(--warning-color);
}
.el-tag--info {
background: var(--info-light);
color: var(--info-color);
}
/* 按钮样式 */
.el-button--primary {
background: var(--primary-color);
border-color: var(--primary-color);
font-weight: 500;
border-radius: var(--radius);
}
.el-button--primary:hover {
background: var(--primary-dark);
border-color: var(--primary-dark);
}
.el-button--success {
background: var(--success-color);
border-color: var(--success-color);
font-weight: 500;
border-radius: var(--radius);
}
.el-button--success:hover {
background: #059669;
border-color: #059669;
}
.el-button--warning {
background: var(--warning-color);
border-color: var(--warning-color);
font-weight: 500;
}
.el-button--primary.is-plain {
background: transparent;
color: var(--primary-color);
border-color: var(--primary-color);
}
.el-button--primary.is-plain:hover {
background: var(--primary-light);
}
/* 链接样式 */
.el-link--primary {
color: var(--primary-color);
font-weight: 500;
}
.el-link:hover {
opacity: 0.8;
}
/* 设置页面样式 */
.website-manage-header,
.admin-header {
display: flex;
align-items: center;
gap: 16px;
margin-bottom: 24px;
padding-bottom: 16px;
border-bottom: 2px solid var(--primary-light);
}
.website-manage-header h2,
.admin-header h2 {
margin: 0;
font-size: 24px;
font-weight: 700;
color: var(--text-primary);
}
.el-tabs {
background: var(--card-bg);
border-radius: var(--radius-lg);
padding: 20px;
box-shadow: var(--shadow-sm);
border: 1px solid var(--border-color);
}
.el-tabs__header {
margin-bottom: 20px;
}
.el-tabs__item {
font-weight: 500;
font-size: 14px;
color: var(--text-secondary);
}
.el-tabs__item.is-active {
color: var(--primary-color);
font-weight: 600;
}
.el-tabs__active-bar {
background: var(--primary-color);
height: 3px;
border-radius: 2px;
}
/* 表单样式 */
.el-form-item__label {
font-weight: 500;
color: var(--text-secondary);
}
.el-input__inner {
border-radius: var(--radius);
border: 1px solid var(--border-color);
transition: all 0.2s;
}
.el-input__inner:focus {
border-color: var(--primary-color);
box-shadow: 0 0 0 3px var(--primary-light);
}
.el-input-number {
border-radius: var(--radius);
}
/* 对话框样式 */
.el-dialog {
border-radius: var(--radius-lg);
box-shadow: var(--shadow-lg);
}
.el-dialog__header {
border-bottom: 1px solid var(--border-color);
padding-bottom: 16px;
margin-bottom: 20px;
}
.el-dialog__title {
font-weight: 600;
color: var(--text-primary);
}
.el-dialog__footer {
border-top: 1px solid var(--border-color);
padding-top: 16px;
}
.el-divider__text {
font-weight: 600;
color: var(--primary-color);
background: var(--card-bg);
padding: 0 12px;
}
/* Cookie值样式 */
.cookie-value {
font-family: 'SF Mono', 'Monaco', 'Inconsolata', 'Fira Code', monospace;
font-size: 12px;
color: var(--text-secondary);
word-break: break-all;
background: var(--page-bg);
padding: 4px 8px;
border-radius: 4px;
}
/* 筛选表单 */
.filter-form {
margin-bottom: 20px;
display: flex;
align-items: center;
gap: 12px;
}
/* 空状态 */
.el-empty {
padding: 60px 0;
}
.el-empty__description {
color: var(--text-secondary);
font-size: 14px;
}
/* 仪表板页脚 */
.dashboard-footer {
margin-top: 40px;
padding: 16px 20px;
background: var(--card-bg);
border-radius: var(--radius);
border: 1px solid var(--border-color);
box-shadow: var(--shadow-sm);
}
.dashboard-footer p {
margin: 0;
color: var(--text-secondary);
font-size: 13px;
text-align: center;
line-height: 1.6;
}
/* 统计数字卡片颜色 */
.stat-total .el-statistic__content {
color: var(--primary-color);
}
.refresh-hint {
font-size: 12px;
color: var(--text-light);
font-weight: 400;
margin-left: 4px;
}
.stat-online .el-statistic__content {
color: var(--success-color);
}
.stat-offline .el-statistic__content {
color: var(--danger-color);
}
.stat-unknown .el-statistic__content {
color: var(--info-color);
}
/* 响应式适配 */
@media (max-width: 768px) {
.el-header {
padding: 0 16px;
}
.el-main {
padding: 16px;
}
.dashboard-header {
flex-direction: column;
gap: 20px;
align-items: flex-start;
}
.dashboard-stats {
flex-wrap: wrap;
}
.user-section {
padding: 16px;
}
}
/* 滚动条美化 */
::-webkit-scrollbar {
width: 8px;
height: 8px;
}
::-webkit-scrollbar-track {
background: var(--page-bg);
}
::-webkit-scrollbar-thumb {
background: var(--text-light);
border-radius: 4px;
}
::-webkit-scrollbar-thumb:hover {
background: var(--text-secondary);
}
+614
View File
@@ -0,0 +1,614 @@
const { createApp, ref, onMounted, computed } = Vue;
createApp({
setup() {
const activeTab = ref('dashboard');
const settingsTab = ref('users');
const manageTab = ref('cookies');
const currentWebsite = ref(null);
const currentUser = ref(null);
const isAdmin = ref(false);
const users = ref([]);
const websites = ref([]);
const dashboardData = ref([]);
const websiteCookies = ref([]);
const websiteDetections = ref([]);
const detections = ref([]);
const notifications = ref([]);
const stats = ref({ totalWebsites: 0, onlineCount: 0, offlineCount: 0, unknownCount: 0 });
const countdown = ref(5);
const adminLoginVisible = ref(false);
const adminLoginForm = ref({ password: '' });
const userDialogVisible = ref(false);
const userDialogTitle = ref('添加用户');
const userForm = ref({ id: null, name: '', iyuu_token: '', max_fail_count: 3, cookie_cloud_uuid: '', cookie_cloud_password: '', cookie_cloud_api_url: '' });
const websiteDialogVisible = ref(false);
const websiteDialogTitle = ref('添加网站');
const websiteFormDialog = ref({
id: null, user_id: null, name: '', url: '',
login_check_selector: '', success_text: '',
interval_minutes: 30, status: 1
});
const cookieDialogVisible = ref(false);
const cookieDialogTitle = ref('添加Cookie');
const cookieForm = ref({ id: null, name: '', value: '', domain: '', path: '/' });
const websiteForm = ref({
id: null, user_id: null, name: '', url: '',
login_check_selector: '', success_text: '',
interval_minutes: 30
});
const filterWebsiteUser = ref(null);
const currentUserWebsites = computed(() => {
if (!currentUser.value) return [];
return websites.value.filter(w => w.user_id === currentUser.value.id);
});
const filteredWebsites = computed(() => {
if (!filterWebsiteUser.value) return websites.value;
return websites.value.filter(w => w.user_id === filterWebsiteUser.value);
});
const checkAdminStatus = async () => {
try {
const response = await axios.get('/api/admin/status');
isAdmin.value = response.data.logged_in;
} catch (error) {
isAdmin.value = false;
}
};
const handleDropdownCommand = async (command) => {
if (command === 'login') {
showAdminLogin();
}
};
const showAdminLogin = () => {
adminLoginForm.value = { password: '' };
adminLoginVisible.value = true;
};
const adminLogin = async () => {
try {
const response = await axios.post('/api/admin/login', adminLoginForm.value);
if (response.data.success) {
isAdmin.value = true;
adminLoginVisible.value = false;
ElementPlus.ElMessage.success('登录成功');
activeTab.value = 'settings';
settingsTab.value = 'users';
await loadUsersWithWebsites();
}
} catch (error) {
ElementPlus.ElMessage.error('密码错误');
}
};
const handleAdminCommand = async (command) => {
if (command === 'settings') {
stopAutoRefresh();
activeTab.value = 'settings';
settingsTab.value = 'users';
await loadUsersWithWebsites();
} else if (command === 'dashboard') {
activeTab.value = 'dashboard';
await loadDashboard();
startAutoRefresh();
} else if (command === 'logout') {
await axios.post('/api/admin/logout');
isAdmin.value = false;
activeTab.value = 'dashboard';
ElementPlus.ElMessage.success('已退出');
}
};
const loadUsers = async () => {
try {
const response = await axios.get('/api/users');
users.value = response.data;
} catch (error) {
console.error('加载用户失败:', error);
}
};
const loadUsersWithWebsites = async () => {
try {
const usersRes = await axios.get('/api/users');
const websitesRes = await axios.get('/api/websites');
users.value = usersRes.data;
websites.value = websitesRes.data;
users.value = users.value.map(user => ({
...user,
website_count: websitesRes.data.filter(w => w.user_id === user.id).length
}));
} catch (error) {
console.error('加载用户失败:', error);
}
};
const loadWebsites = async () => {
try {
const response = await axios.get('/api/websites');
websites.value = response.data;
} catch (error) {
console.error('加载网站失败:', error);
}
};
const loadDashboard = async () => {
try {
const usersRes = await axios.get('/api/users');
const websitesRes = await axios.get('/api/websites');
const usersData = usersRes.data;
const allWebsites = websitesRes.data;
dashboardData.value = usersData.map(user => {
const userWebsites = allWebsites.filter(w => w.user_id === user.id);
return {
id: user.id,
name: user.name,
websites: userWebsites.map(w => ({
...w,
networkStatus: 'unknown',
loginStatus: 'unknown',
lastCheckTime: '-'
}))
};
});
const latestDetectionsRes = await axios.get('/api/detections/latest');
const networkRes = await axios.get('/api/websites/check-network');
const latestDetections = latestDetectionsRes.data;
const networkStatus = networkRes.data;
let onlineCount = 0, offlineCount = 0, unknownCount = 0, totalWebsites = allWebsites.length;
dashboardData.value = dashboardData.value.map(user => {
return {
...user,
websites: user.websites.map(w => {
const detection = latestDetections[w.id];
const net = networkStatus[w.id];
let loginStatus = 'unknown';
let lastCheckTime = '-';
let networkStatusValue = 'unknown';
if (net) {
networkStatusValue = net.reachable ? 'online' : 'offline';
}
if (detection) {
if (detection.status === 1) {
loginStatus = 'success';
onlineCount++;
} else {
loginStatus = 'failed';
offlineCount++;
}
lastCheckTime = new Date(detection.created_at).toLocaleString();
} else {
unknownCount++;
}
return {
...w,
networkStatus: networkStatusValue,
loginStatus: loginStatus,
lastCheckTime: lastCheckTime
};
})
};
});
stats.value = { totalWebsites, onlineCount, offlineCount, unknownCount };
} catch (error) {
console.error('加载仪表板失败:', error);
}
};
const refreshDashboardStatus = async () => {
if (activeTab.value !== 'dashboard') return;
if (isRefreshing) return;
isRefreshing = true;
try {
const [detectionsRes, networkRes] = await Promise.all([
axios.get('/api/detections/latest'),
axios.get('/api/websites/check-network')
]);
const latestDetections = detectionsRes.data;
const networkStatus = networkRes.data;
let onlineCount = 0, offlineCount = 0, unknownCount = 0;
dashboardData.value = dashboardData.value.map(user => {
return {
...user,
websites: user.websites.map(w => {
const detection = latestDetections[w.id];
const net = networkStatus[w.id];
let loginStatus = w.loginStatus;
let lastCheckTime = w.lastCheckTime;
let networkStatusValue = 'unknown';
if (net) {
networkStatusValue = net.reachable ? 'online' : 'offline';
}
if (detection) {
if (detection.status === 1) {
loginStatus = 'success';
onlineCount++;
} else {
loginStatus = 'failed';
offlineCount++;
}
lastCheckTime = new Date(detection.created_at).toLocaleString();
} else {
unknownCount++;
}
return {
...w,
networkStatus: networkStatusValue,
loginStatus: loginStatus,
lastCheckTime: lastCheckTime
};
})
};
});
stats.value = {
totalWebsites: onlineCount + offlineCount + unknownCount,
onlineCount,
offlineCount,
unknownCount
};
} catch (error) {
// 静默失败,不显示错误提示
} finally {
isRefreshing = false;
if (activeTab.value === 'dashboard') {
countdown.value = 5;
startCountdown();
}
}
};
let refreshTimer = null;
let countdownTimer = null;
let isRefreshing = false;
const startAutoRefresh = () => {
stopAutoRefresh();
countdown.value = 5;
startCountdown();
};
const startCountdown = () => {
if (countdownTimer) clearInterval(countdownTimer);
countdownTimer = setInterval(() => {
countdown.value--;
if (countdown.value <= 0) {
clearInterval(countdownTimer);
countdownTimer = null;
refreshDashboardStatus();
}
}, 1000);
};
const stopAutoRefresh = () => {
if (refreshTimer) {
clearInterval(refreshTimer);
refreshTimer = null;
}
if (countdownTimer) {
clearInterval(countdownTimer);
countdownTimer = null;
}
countdown.value = 5;
};
const checkLogin = async (website) => {
website.loginStatus = 'checking';
try {
const response = await axios.post(`/api/websites/check-login/${website.id}`);
if (response.data.success) {
website.loginStatus = 'success';
updateStats(website.id, 'success');
ElementPlus.ElMessage.success(response.data.message);
} else {
website.loginStatus = 'failed';
updateStats(website.id, 'failed');
ElementPlus.ElMessage.error(response.data.message);
}
website.lastCheckTime = new Date().toLocaleString();
} catch (error) {
website.loginStatus = 'failed';
updateStats(website.id, 'failed');
ElementPlus.ElMessage.error('检测失败');
}
};
const updateStats = (websiteId, status) => {
let onlineCount = 0, offlineCount = 0, unknownCount = 0;
dashboardData.value.forEach(user => {
user.websites.forEach(w => {
if (w.id === websiteId) {
if (status === 'success') onlineCount++;
else if (status === 'failed') offlineCount++;
} else {
if (w.loginStatus === 'success') onlineCount++;
else if (w.loginStatus === 'failed') offlineCount++;
else unknownCount++;
}
});
});
stats.value = { totalWebsites: onlineCount + offlineCount + unknownCount, onlineCount, offlineCount, unknownCount };
};
const manageWebsite = async (website) => {
stopAutoRefresh();
currentWebsite.value = website;
websiteForm.value = { ...website };
activeTab.value = 'website_manage';
manageTab.value = 'cookies';
await loadWebsiteCookies();
};
const manageUserWebsites = (user) => {
stopAutoRefresh();
currentUser.value = user;
activeTab.value = 'user_websites';
};
const loadWebsiteCookies = async () => {
if (!currentWebsite.value) return;
try {
const response = await axios.get(`/api/cookies/websites/${currentWebsite.value.id}`);
websiteCookies.value = response.data;
} catch (error) {
console.error('加载Cookie失败:', error);
}
};
const loadWebsiteDetections = async () => {
if (!currentWebsite.value) return;
try {
const response = await axios.get(`/api/detections?website_id=${currentWebsite.value.id}`);
websiteDetections.value = response.data;
} catch (error) {
console.error('加载检测记录失败:', error);
}
};
const showAddUserDialog = () => {
userDialogTitle.value = '添加用户';
userForm.value = { id: null, name: '', iyuu_token: '', max_fail_count: 3, cookie_cloud_uuid: '', cookie_cloud_password: '', cookie_cloud_api_url: '' };
userDialogVisible.value = true;
};
const editUser = (user) => {
userDialogTitle.value = '编辑用户';
userForm.value = {
id: user.id,
name: user.name,
iyuu_token: user.iyuu_token,
max_fail_count: user.max_fail_count,
cookie_cloud_uuid: user.cookie_cloud_uuid || '',
cookie_cloud_password: user.cookie_cloud_password || '',
cookie_cloud_api_url: user.cookie_cloud_api_url || ''
};
userDialogVisible.value = true;
};
const saveUser = async () => {
try {
if (userForm.value.id) {
await axios.put(`/api/users/${userForm.value.id}`, userForm.value);
ElementPlus.ElMessage.success('更新成功');
} else {
await axios.post('/api/users', userForm.value);
ElementPlus.ElMessage.success('添加成功');
}
userDialogVisible.value = false;
await loadUsersWithWebsites();
} catch (error) {
ElementPlus.ElMessage.error('保存失败');
}
};
const deleteUser = async (user) => {
try {
await ElementPlus.ElMessageBox.confirm('确定要删除该用户吗?', '提示', { type: 'warning' });
await axios.delete(`/api/users/${user.id}`);
ElementPlus.ElMessage.success('删除成功');
await loadUsersWithWebsites();
} catch (error) {
if (error !== 'cancel') {
ElementPlus.ElMessage.error('删除失败');
}
}
};
const showAddWebsiteDialog = () => {
websiteDialogTitle.value = '添加网站';
websiteFormDialog.value = {
id: null, user_id: filterWebsiteUser.value || users.value[0]?.id,
name: '', url: '', login_check_selector: '', success_text: '',
interval_minutes: 30, status: 1
};
websiteDialogVisible.value = true;
};
const showAddWebsiteDialogForUser = () => {
websiteDialogTitle.value = '添加网站';
websiteFormDialog.value = {
id: null, user_id: currentUser.value?.id,
name: '', url: '', login_check_selector: '', success_text: '',
interval_minutes: 30, status: 1
};
websiteDialogVisible.value = true;
};
const editWebsite = (website) => {
websiteDialogTitle.value = '编辑网站';
websiteFormDialog.value = { ...website };
websiteDialogVisible.value = true;
};
const saveWebsite = async () => {
try {
if (websiteFormDialog.value.id) {
await axios.put(`/api/websites/${websiteFormDialog.value.id}`, websiteFormDialog.value);
ElementPlus.ElMessage.success('更新成功');
} else {
await axios.post('/api/websites', websiteFormDialog.value);
ElementPlus.ElMessage.success('添加成功');
}
websiteDialogVisible.value = false;
await loadWebsites();
await loadUsersWithWebsites();
} catch (error) {
ElementPlus.ElMessage.error('保存失败');
}
};
const deleteWebsite = async (website) => {
try {
await ElementPlus.ElMessageBox.confirm('确定要删除该网站吗?', '提示', { type: 'warning' });
await axios.delete(`/api/websites/${website.id}`);
ElementPlus.ElMessage.success('删除成功');
await loadWebsites();
await loadUsersWithWebsites();
} catch (error) {
if (error !== 'cancel') {
ElementPlus.ElMessage.error('删除失败');
}
}
};
const showAddCookieDialog = () => {
cookieDialogTitle.value = '添加Cookie';
cookieForm.value = { id: null, name: '', value: '', domain: currentWebsite.value?.url ? new URL(currentWebsite.value.url).hostname : '', path: '/' };
cookieDialogVisible.value = true;
};
const editCookie = (cookie) => {
cookieDialogTitle.value = '编辑Cookie';
cookieForm.value = { ...cookie };
cookieDialogVisible.value = true;
};
const saveCookie = async () => {
try {
if (cookieForm.value.id) {
await axios.put(`/api/cookies/${cookieForm.value.id}`, cookieForm.value);
ElementPlus.ElMessage.success('更新成功');
} else {
await axios.post(`/api/cookies/websites/${currentWebsite.value.id}`, cookieForm.value);
ElementPlus.ElMessage.success('添加成功');
}
cookieDialogVisible.value = false;
await loadWebsiteCookies();
} catch (error) {
ElementPlus.ElMessage.error('保存失败');
}
};
const deleteCookie = async (cookie) => {
try {
await ElementPlus.ElMessageBox.confirm('确定要删除该Cookie吗?', '提示', { type: 'warning' });
await axios.delete(`/api/cookies/${cookie.id}`);
ElementPlus.ElMessage.success('删除成功');
await loadWebsiteCookies();
} catch (error) {
if (error !== 'cancel') {
ElementPlus.ElMessage.error('删除失败');
}
}
};
const saveWebsiteSettings = async () => {
try {
await axios.put(`/api/websites/${currentWebsite.value.id}`, websiteForm.value);
ElementPlus.ElMessage.success('保存成功');
} catch (error) {
ElementPlus.ElMessage.error('保存失败');
}
};
const maskCookieValue = (value) => {
if (!value) return '';
return value.length > 30 ? value.substring(0, 30) + '...' : value;
};
const loadDetections = async () => {
try {
const response = await axios.get('/api/detections');
detections.value = response.data;
} catch (error) {
console.error('加载检测结果失败:', error);
}
};
const loadNotifications = async () => {
try {
const response = await axios.get('/api/notifications');
notifications.value = response.data;
} catch (error) {
console.error('加载通知失败:', error);
}
};
const viewNotificationDetail = (notification) => {
ElementPlus.ElMessageBox.alert(notification.content, notification.title, { confirmButtonText: '确定' });
};
onMounted(async () => {
await checkAdminStatus();
await loadUsers();
await loadWebsites();
await loadDashboard();
startAutoRefresh();
});
return {
activeTab, settingsTab, manageTab, currentWebsite, currentUser, isAdmin,
users, websites, dashboardData, websiteCookies, websiteDetections, detections, notifications,
stats, countdown,
adminLoginVisible, adminLoginForm,
userDialogVisible, userDialogTitle, userForm,
websiteDialogVisible, websiteDialogTitle, websiteFormDialog,
cookieDialogVisible, cookieDialogTitle, cookieForm,
websiteForm, filterWebsiteUser,
currentUserWebsites, filteredWebsites,
checkAdminStatus, showAdminLogin, adminLogin, handleAdminCommand, handleDropdownCommand,
loadUsers, loadUsersWithWebsites, loadWebsites, loadDashboard, refreshDashboardStatus,
startAutoRefresh, stopAutoRefresh,
checkLogin, manageWebsite, manageUserWebsites,
loadWebsiteCookies, loadWebsiteDetections,
showAddUserDialog, editUser, saveUser, deleteUser,
showAddWebsiteDialog, showAddWebsiteDialogForUser, editWebsite, saveWebsite, deleteWebsite,
showAddCookieDialog, editCookie, saveCookie, deleteCookie,
saveWebsiteSettings, maskCookieValue,
loadDetections, loadNotifications, viewNotificationDetail
};
}
}).use(ElementPlus).mount('#app');
+405
View File
@@ -0,0 +1,405 @@
{% raw %}<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Cookie监控管理系统</title>
<link rel="stylesheet" href="/static/css/style.css">
<link rel="stylesheet" href="https://unpkg.com/element-plus/dist/index.css">
<script src="https://unpkg.com/vue@3/dist/vue.global.js"></script>
<script src="https://unpkg.com/element-plus"></script>
<script src="https://unpkg.com/axios/dist/axios.min.js"></script>
</head>
<body>
<div id="app">
<div class="container">
<el-container>
<el-header>
<div class="header-left">
<h1>网站监控仪表板</h1>
<el-button v-if="activeTab === 'settings'" @click="activeTab = 'dashboard'" type="primary" plain>返回监控首页</el-button>
</div>
<div class="header-right">
<div v-if="activeTab === 'dashboard'" class="auto-refresh-indicator">
<el-icon class="rotating-icon"><Refresh /></el-icon>
<span class="refresh-text">{{ countdown }}秒后自动刷新</span>
</div>
<span class="header-stat" v-if="activeTab === 'dashboard'">网站总数:<b>{{ stats.totalWebsites }}</b></span>
<el-dropdown @command="handleDropdownCommand" v-if="!isAdmin">
<el-button link type="text" style="color: rgba(255,255,255,0.8); font-size: 18px">
<el-icon><arrow-down /></el-icon>
</el-button>
<template #dropdown>
<el-dropdown-menu>
<el-dropdown-item command="login"><el-icon><Lock /></el-icon> 登录后台</el-dropdown-item>
</el-dropdown-menu>
</template>
</el-dropdown>
<el-dropdown @command="handleAdminCommand" v-else>
<el-button type="success" plain>后台管理 <el-icon><arrow-down /></el-icon></el-button>
<template #dropdown>
<el-dropdown-menu>
<el-dropdown-item command="settings">系统设置</el-dropdown-item>
<el-dropdown-item command="dashboard">监控首页</el-dropdown-item>
<el-dropdown-item command="logout" divided>退出登录</el-dropdown-item>
</el-dropdown-menu>
</template>
</el-dropdown>
</div>
</el-header>
<el-main>
<!-- 监控仪表板 -->
<div v-if="activeTab === 'dashboard'">
<div v-for="user in dashboardData" :key="user.id" class="user-section">
<div class="user-header">
<h3>{{ user.name }}</h3>
<el-tag :type="user.websites.some(w => w.loginStatus === 'success') ? 'success' : user.websites.some(w => w.loginStatus === 'failed') ? 'danger' : 'info'">
{{ user.websites.filter(w => w.loginStatus === 'success').length }}/{{ user.websites.length }} 在线
</el-tag>
</div>
<el-table :data="user.websites" style="width: 100%; margin-bottom: 20px">
<el-table-column label="启用状态" width="90">
<template #default="scope">
<el-tag :type="scope.row.status === 1 ? 'success' : 'danger'" size="small">
{{ scope.row.status === 1 ? '启用' : '禁用' }}
</el-tag>
</template>
</el-table-column>
<el-table-column prop="name" label="网站名称" width="150"></el-table-column>
<el-table-column prop="url" label="URL" width="250">
<template #default="scope">
<el-link :href="scope.row.url" target="_blank" type="primary">{{ scope.row.url }}</el-link>
</template>
</el-table-column>
<el-table-column label="网络状态" width="100">
<template #default="scope">
<el-tag :type="scope.row.networkStatus === 'online' ? 'success' : scope.row.networkStatus === 'offline' ? 'danger' : 'info'">
{{ scope.row.networkStatus === 'online' ? '在线' : scope.row.networkStatus === 'offline' ? '离线' : '未知' }}
</el-tag>
</template>
</el-table-column>
<el-table-column label="登录状态" width="100">
<template #default="scope">
<el-tag :type="scope.row.loginStatus === 'success' ? 'success' : scope.row.loginStatus === 'failed' ? 'danger' : scope.row.loginStatus === 'checking' ? 'warning' : 'info'">
{{ scope.row.loginStatus === 'success' ? '已登录' : scope.row.loginStatus === 'failed' ? '未登录' : scope.row.loginStatus === 'checking' ? '检测中' : '未检测' }}
</el-tag>
</template>
</el-table-column>
<el-table-column prop="lastCheckTime" label="最后检测" width="180"></el-table-column>
<el-table-column label="操作" width="200">
<template #default="scope">
<el-button size="small" type="primary" @click="checkLogin(scope.row)" :loading="scope.row.loginStatus === 'checking'" :disabled="scope.row.loginStatus === 'checking'">
{{ scope.row.loginStatus === 'checking' ? '检测中' : '检测登录' }}
</el-button>
<el-button size="small" type="success" @click="manageWebsite(scope.row)" v-if="scope.row.loginStatus === 'success'">管理</el-button>
</template>
</el-table-column>
</el-table>
</div>
<el-empty v-if="dashboardData.length === 0" description="暂无网站数据"></el-empty>
<div class="dashboard-footer">
<p>免责声明:本系统仅用于Cookie状态监控和自动化管理,不对任何网站数据承担责任。请确保您有权访问和管理相关网站账号。</p>
</div>
</div>
<!-- 网站管理 -->
<div v-if="activeTab === 'website_manage'">
<div class="website-manage-header">
<el-button @click="activeTab = 'dashboard'" type="info">返回监控首页</el-button>
<h2>网站管理 - {{ currentWebsite?.name }}</h2>
</div>
<el-tabs v-model="manageTab">
<el-tab-pane label="Cookie管理" name="cookies">
<el-button @click="showAddCookieDialog" type="primary" size="small">添加Cookie</el-button>
<el-button @click="loadWebsiteCookies" size="small">刷新</el-button>
<el-table :data="websiteCookies" style="width: 100%; margin-top: 10px">
<el-table-column prop="name" label="名称" width="200"></el-table-column>
<el-table-column prop="value" label="值" width="300">
<template #default="scope">
<span class="cookie-value">{{ maskCookieValue(scope.row.value) }}</span>
</template>
</el-table-column>
<el-table-column prop="domain" label="域名" width="150"></el-table-column>
<el-table-column label="操作" width="150">
<template #default="scope">
<el-button size="small" @click="editCookie(scope.row)">编辑</el-button>
<el-button size="small" type="danger" @click="deleteCookie(scope.row)">删除</el-button>
</template>
</el-table-column>
</el-table>
</el-tab-pane>
<el-tab-pane label="网站设置" name="settings">
<el-form :model="websiteForm" label-width="120px" style="max-width: 500px">
<el-form-item label="网站名称">
<el-input v-model="websiteForm.name"></el-input>
</el-form-item>
<el-form-item label="URL">
<el-input v-model="websiteForm.url"></el-input>
</el-form-item>
<el-form-item label="登录检测选择器">
<el-input v-model="websiteForm.login_check_selector"></el-input>
</el-form-item>
<el-form-item label="登录成功文本">
<el-input v-model="websiteForm.success_text"></el-input>
</el-form-item>
<el-form-item label="检测间隔(分钟)">
<el-input-number v-model="websiteForm.interval_minutes" :min="1" :max="1440"></el-input-number>
</el-form-item>
<el-form-item>
<el-button type="primary" @click="saveWebsiteSettings">保存设置</el-button>
</el-form-item>
</el-form>
</el-tab-pane>
<el-tab-pane label="检测记录" name="history">
<el-button @click="loadWebsiteDetections" type="primary" size="small">刷新</el-button>
<el-table :data="websiteDetections" style="width: 100%; margin-top: 10px">
<el-table-column prop="status" label="状态" width="80">
<template #default="scope">
<el-tag :type="scope.row.status === 1 ? 'success' : 'danger'">
{{ scope.row.status === 1 ? '成功' : '失败' }}
</el-tag>
</template>
</el-table-column>
<el-table-column prop="message" label="消息"></el-table-column>
<el-table-column prop="created_at" label="时间" width="180"></el-table-column>
</el-table>
</el-tab-pane>
</el-tabs>
</div>
<!-- 系统设置页面 -->
<div v-if="activeTab === 'settings'">
<h2>系统设置</h2>
<el-tabs v-model="settingsTab">
<el-tab-pane label="用户管理" name="users">
<el-button @click="showAddUserDialog" type="primary" size="small" style="margin-bottom: 10px">添加用户</el-button>
<el-table :data="users" style="width: 100%">
<el-table-column prop="id" label="ID" width="80"></el-table-column>
<el-table-column prop="name" label="用户名" width="150"></el-table-column>
<el-table-column prop="iyuu_token" label="爱语飞飞令牌" width="200"></el-table-column>
<el-table-column prop="max_fail_count" label="最大失败次数" width="120"></el-table-column>
<el-table-column prop="website_count" label="网站数量" width="100"></el-table-column>
<el-table-column label="操作">
<template #default="scope">
<el-button size="small" @click="editUser(scope.row)">编辑</el-button>
<el-button size="small" type="warning" @click="manageUserWebsites(scope.row)">管理网站</el-button>
<el-button size="small" type="danger" @click="deleteUser(scope.row)">删除</el-button>
</template>
</el-table-column>
</el-table>
</el-tab-pane>
<el-tab-pane label="网站管理" name="websites">
<div class="filter-form">
<el-select v-model="filterWebsiteUser" placeholder="选择用户" @change="loadWebsites">
<el-option label="全部用户" :value="null"></el-option>
<el-option v-for="user in users" :key="user.id" :label="user.name" :value="user.id"></el-option>
</el-select>
<el-button @click="showAddWebsiteDialog" type="primary" size="small" style="margin-left: 10px">添加网站</el-button>
</div>
<el-table :data="filteredWebsites" style="width: 100%">
<el-table-column prop="id" label="ID" width="80"></el-table-column>
<el-table-column prop="user_id" label="用户ID" width="100"></el-table-column>
<el-table-column prop="name" label="网站名称" width="150"></el-table-column>
<el-table-column prop="url" label="URL" width="250"></el-table-column>
<el-table-column prop="interval_minutes" label="间隔(分钟)" width="100"></el-table-column>
<el-table-column prop="status" label="状态" width="80">
<template #default="scope">
<el-tag :type="scope.row.status === 1 ? 'success' : 'danger'">
{{ scope.row.status === 1 ? '启用' : '禁用' }}
</el-tag>
</template>
</el-table-column>
<el-table-column label="操作">
<template #default="scope">
<el-button size="small" @click="editWebsite(scope.row)">编辑</el-button>
<el-button size="small" type="danger" @click="deleteWebsite(scope.row)">删除</el-button>
</template>
</el-table-column>
</el-table>
</el-tab-pane>
<el-tab-pane label="检测记录" name="detections">
<el-button @click="loadDetections" type="primary" size="small" style="margin-bottom: 10px">刷新</el-button>
<el-table :data="detections" style="width: 100%">
<el-table-column prop="id" label="ID" width="80"></el-table-column>
<el-table-column prop="website_id" label="网站ID" width="100"></el-table-column>
<el-table-column prop="status" label="状态" width="80">
<template #default="scope">
<el-tag :type="scope.row.status === 1 ? 'success' : 'danger'">
{{ scope.row.status === 1 ? '成功' : '失败' }}
</el-tag>
</template>
</el-table-column>
<el-table-column prop="response_time" label="响应时间(ms)" width="120"></el-table-column>
<el-table-column prop="http_status" label="HTTP状态" width="100"></el-table-column>
<el-table-column prop="message" label="消息"></el-table-column>
<el-table-column prop="created_at" label="检测时间" width="180"></el-table-column>
</el-table>
</el-tab-pane>
<el-tab-pane label="通知记录" name="notifications">
<el-button @click="loadNotifications" type="primary" size="small" style="margin-bottom: 10px">刷新</el-button>
<el-table :data="notifications" style="width: 100%">
<el-table-column prop="id" label="ID" width="80"></el-table-column>
<el-table-column prop="user_id" label="用户ID" width="100"></el-table-column>
<el-table-column prop="website_id" label="网站ID" width="100"></el-table-column>
<el-table-column prop="title" label="标题" width="200"></el-table-column>
<el-table-column prop="status" label="状态" width="80">
<template #default="scope">
<el-tag :type="scope.row.status === 'sent' ? 'success' : scope.row.status === 'failed' ? 'danger' : 'warning'">
{{ scope.row.status === 'sent' ? '已发送' : scope.row.status === 'failed' ? '失败' : '待发送' }}
</el-tag>
</template>
</el-table-column>
<el-table-column prop="send_time" label="发送时间" width="180"></el-table-column>
<el-table-column prop="created_at" label="创建时间" width="180"></el-table-column>
<el-table-column label="操作">
<template #default="scope">
<el-button size="small" @click="viewNotificationDetail(scope.row)">查看</el-button>
</template>
</el-table-column>
</el-table>
</el-tab-pane>
</el-tabs>
</div>
<!-- 用户网站管理 -->
<div v-if="activeTab === 'user_websites'">
<div class="website-manage-header">
<el-button @click="activeTab = 'settings'; settingsTab = 'users'" type="info">返回设置</el-button>
<h2>网站管理 - {{ currentUser?.name }}</h2>
<el-button @click="showAddWebsiteDialogForUser" type="primary">添加网站</el-button>
</div>
<el-table :data="currentUserWebsites" style="width: 100%">
<el-table-column prop="id" label="ID" width="80"></el-table-column>
<el-table-column prop="name" label="网站名称" width="150"></el-table-column>
<el-table-column prop="url" label="URL" width="250"></el-table-column>
<el-table-column prop="interval_minutes" label="间隔(分钟)" width="100"></el-table-column>
<el-table-column prop="status" label="状态" width="80">
<template #default="scope">
<el-tag :type="scope.row.status === 1 ? 'success' : 'danger'">
{{ scope.row.status === 1 ? '启用' : '禁用' }}
</el-tag>
</template>
</el-table-column>
<el-table-column label="操作">
<template #default="scope">
<el-button size="small" @click="editWebsite(scope.row)">编辑</el-button>
<el-button size="small" type="danger" @click="deleteWebsite(scope.row)">删除</el-button>
</template>
</el-table-column>
</el-table>
</div>
</el-main>
</el-container>
</div>
<!-- 后台登录对话框 -->
<el-dialog v-model="adminLoginVisible" title="登录后台" width="400px">
<el-form :model="adminLoginForm" label-width="80px">
<el-form-item label="密码">
<el-input v-model="adminLoginForm.password" type="password" show-password @keyup.enter="adminLogin"></el-input>
</el-form-item>
</el-form>
<template #footer>
<el-button @click="adminLoginVisible = false">取消</el-button>
<el-button type="primary" @click="adminLogin">登录</el-button>
</template>
</el-dialog>
<!-- 用户对话框 -->
<el-dialog v-model="userDialogVisible" :title="userDialogTitle" width="600px">
<el-form :model="userForm" label-width="120px">
<el-form-item label="用户名">
<el-input v-model="userForm.name"></el-input>
</el-form-item>
<el-form-item label="爱语飞飞令牌">
<el-input v-model="userForm.iyuu_token"></el-input>
</el-form-item>
<el-form-item label="最大失败次数">
<el-input-number v-model="userForm.max_fail_count" :min="1" :max="10"></el-input-number>
</el-form-item>
<el-divider content-position="left">Cookie Cloud 设置</el-divider>
<el-form-item label="UUID">
<el-input v-model="userForm.cookie_cloud_uuid" placeholder="hu6n2vcUmzpu7mqUN2rVCg"></el-input>
</el-form-item>
<el-form-item label="密码">
<el-input v-model="userForm.cookie_cloud_password" show-password></el-input>
</el-form-item>
<el-form-item label="API地址">
<el-input v-model="userForm.cookie_cloud_api_url" placeholder="https://movie-pilot.org/cookiecloud"></el-input>
</el-form-item>
</el-form>
<template #footer>
<el-button @click="userDialogVisible = false">取消</el-button>
<el-button type="primary" @click="saveUser">保存</el-button>
</template>
</el-dialog>
<!-- 网站对话框 -->
<el-dialog v-model="websiteDialogVisible" :title="websiteDialogTitle">
<el-form :model="websiteFormDialog" label-width="120px">
<el-form-item label="所属用户">
<el-select v-model="websiteFormDialog.user_id" placeholder="选择用户">
<el-option v-for="user in users" :key="user.id" :label="user.name" :value="user.id"></el-option>
</el-select>
</el-form-item>
<el-form-item label="网站名称">
<el-input v-model="websiteFormDialog.name"></el-input>
</el-form-item>
<el-form-item label="URL">
<el-input v-model="websiteFormDialog.url"></el-input>
</el-form-item>
<el-form-item label="登录检测选择器">
<el-input v-model="websiteFormDialog.login_check_selector"></el-input>
</el-form-item>
<el-form-item label="登录成功文本">
<el-input v-model="websiteFormDialog.success_text"></el-input>
</el-form-item>
<el-form-item label="检测间隔(分钟)">
<el-input-number v-model="websiteFormDialog.interval_minutes" :min="1" :max="1440"></el-input-number>
</el-form-item>
<el-form-item label="状态">
<el-switch v-model="websiteFormDialog.status" :active-value="1" :inactive-value="0"></el-switch>
</el-form-item>
</el-form>
<template #footer>
<el-button @click="websiteDialogVisible = false">取消</el-button>
<el-button type="primary" @click="saveWebsite">保存</el-button>
</template>
</el-dialog>
<!-- Cookie对话框 -->
<el-dialog v-model="cookieDialogVisible" :title="cookieDialogTitle">
<el-form :model="cookieForm" label-width="100px">
<el-form-item label="名称">
<el-input v-model="cookieForm.name"></el-input>
</el-form-item>
<el-form-item label="值">
<el-input v-model="cookieForm.value" type="textarea" :rows="3"></el-input>
</el-form-item>
<el-form-item label="域名">
<el-input v-model="cookieForm.domain"></el-input>
</el-form-item>
<el-form-item label="路径">
<el-input v-model="cookieForm.path"></el-input>
</el-form-item>
</el-form>
<template #footer>
<el-button @click="cookieDialogVisible = false">取消</el-button>
<el-button type="primary" @click="saveCookie">保存</el-button>
</template>
</el-dialog>
</div>
<script src="/static/js/app.js"></script>
</body>
</html>{% endraw %}
-320
View File
@@ -1,320 +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_cookies_dict(self) -> Dict[str, List[Dict]]:
"""
获取当前浏览器的所有 cookie,按域名分组
Returns:
按域名分组的 cookie 字典
"""
cookies = self.tab.cookies.as_dict
# 按域名分组
cookie_dict = {}
for cookie in cookies:
domain = cookie.get('domain', '')
# 统一域名格式(去掉开头的点)
if domain.startswith('.'):
domain = domain[1:]
if domain not in cookie_dict:
cookie_dict[domain] = []
cookie_dict[domain].append(cookie)
return cookie_dict
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 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:
# 访问网站
self.tab.get(url)
# 添加 cookies
for cookie in cookies:
self.tab.set.cookies(cookie)
# 刷新页面
self.tab.refresh()
# 等待页面加载
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 = self.tab.cookies.as_dict
return cookies
except Exception as e:
logger.error(f"刷新页面并保存 cookie 失败: {e}")
return []
def close(self):
"""关闭浏览器"""
self._close_browser()
class CookieManager:
"""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, 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:
return json.load(f)
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_cookies(self, user_name: str, website_name: str) -> List[Dict]:
"""
获取指定用户和网站的 cookie
Args:
user_name: 用户名
website_name: 网站名
Returns:
cookie 列表
"""
if user_name not in self.cookies:
return []
if website_name not in self.cookies[user_name]:
return []
return self.cookies[user_name][website_name]
def save_cookies(self, user_name: str, website_name: str, cookies: List[Dict]):
"""
保存指定用户和网站的 cookie
Args:
user_name: 用户名
website_name: 网站名
cookies: cookie 列表
"""
if user_name not in self.cookies:
self.cookies[user_name] = {}
self.cookies[user_name][website_name] = cookies
self._save_cookies()
def get_cookies_for_domain(self, domain: str) -> List[Dict]:
"""
获取指定域名的 cookie(从所有用户的 cookie 中查找)
Args:
domain: 域名
Returns:
cookie 列表
"""
all_cookies = []
for user_name, websites in self.cookies.items():
for website_name, cookies in websites.items():
for cookie in 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:
all_cookies.append(cookie)
return all_cookies
if __name__ == "__main__":
# 测试代码
manager = CookieManager("cookies.json")
# 保存测试 cookie
test_cookies = [
{"name": "test", "value": "123", "domain": "example.com", "path": "/"}
]
manager.save_cookies("测试用户", "测试网站", test_cookies)
# 获取 cookie
cookies = manager.get_cookies("测试用户", "测试网站")
logger.info(f"获取到的 cookie: {cookies}")
-1
View File
@@ -1 +0,0 @@
{"users": [{"name": "用户A", "cookie_cloud": {"uuid": "hu6n2vcUmzpu7mqUN2rVCg", "password": "jtDM5dV9AyqXkZdQVeA9f6", "api_url": "https://movie-pilot.org/cookiecloud"}, "notification": {"iyuu_token": "IYUU37629Tc1d371c7ce99a49ff9778e196286b7e4592be191", "max_fail_count": 3}, "browser": {"type": "edge", "headless": false}, "websites": [{"name": "网站A", "url": "https://lmkbi.95155.com/bi-system/#/carUnpetrifiedComAll", "login_check_selector": "#user-info", "success_text": "彭峰"}]}]}
-183
View File
@@ -1,183 +0,0 @@
"""Cookie Cloud 模块 - 从 Cookie Cloud 服务器获取并解密 cookie 数据"""
import hashlib
import json
import base64
from typing import Dict, List, Optional
import requests
from Crypto.Cipher import AES
from Crypto.Util.Padding import unpad
from applogger import get_logger
logger = get_logger(__name__)
class CookieCloud:
"""Cookie Cloud 客户端"""
def __init__(self, api_url: str, uuid: str, password: str):
"""
初始化 Cookie Cloud 客户端
Args:
api_url: Cookie Cloud API 地址
uuid: 用户 UUID
password: 用户密码
"""
self.api_url = api_url.rstrip('/')
self.uuid = uuid
self.password = password
def decrypt(self, encrypted_data: str) -> Dict:
"""
解密 Cookie Cloud 数据
Args:
encrypted_data: 加密的数据字符串
Returns:
解密后的数据字典,包含 cookie_data 和 local_storage_data
"""
# 生成解密密钥: md5(uuid+password) 取前 16 个字符
key_str = f"{self.uuid}{self.password}"
key = hashlib.md5(key_str.encode()).hexdigest()[:16].encode()
print(f"生成的密钥: {key}")
print(f"密钥长度: {len(key)}")
# 解码 Base64
encrypted = base64.b64decode(encrypted_data)
print(f"Base64 解码后长度: {len(encrypted)}")
print(f"解码后前 20 字节: {encrypted[:20]}")
# 检查是否有盐值(OpenSSL 格式)
if encrypted.startswith(b'Salted__'):
print("检测到 OpenSSL 格式数据")
# 对于 OpenSSL 格式,我们需要跳过盐值部分
# 但根据用户要求,我们直接使用16字符密钥,不考虑盐值
ciphertext = encrypted[16:] # 跳过 "Salted__" 和盐值
print(f"密文长度: {len(ciphertext)}")
else:
print("未检测到 OpenSSL 格式,使用整个数据作为密文")
ciphertext = encrypted
# 使用固定的 IV (16 个 0 字节)
iv = b'\x00' * 16
print(f"使用的 IV: {iv}")
# AES-128-CBC 解密(16字节密钥)
cipher = AES.new(key, AES.MODE_CBC, iv)
decrypted = cipher.decrypt(ciphertext)
print(f"解密后长度: {len(decrypted)}")
print(f"解密后前 100 字节: {decrypted[:100]}")
try:
# 移除 PKCS#7 填充
decrypted = unpad(decrypted, AES.block_size)
print(f"移除填充后长度: {len(decrypted)}")
print(f"移除填充后前 100 字节: {decrypted[:100]}")
# 尝试解析 JSON
result = json.loads(decrypted.decode('utf-8'))
print("JSON 解析成功")
return result
except Exception as e:
print(f"解密失败: {e}")
# 尝试直接截取可能的有效数据
try:
# 尝试找到 JSON 开始的位置
for i in range(len(decrypted)):
try:
test_data = decrypted[i:]
test_str = test_data.decode('utf-8', errors='ignore')
if test_str.strip().startswith('{'):
print(f"找到可能的 JSON 开始位置: {i}")
result = json.loads(test_str)
print("JSON 解析成功(直接截取)")
return result
except:
continue
except:
pass
raise
def get_cookies(self) -> Dict[str, List[Dict]]:
"""
从 Cookie Cloud 服务器获取 cookie 数据
Returns:
按域名分组的 cookie 数据字典
格式: {"domain.com": [{"name": "cookie1", "value": "val1", ...}, ...], ...}
"""
url = f"{self.api_url}/get/{self.uuid}"
try:
response = requests.get(url, timeout=30)
response.raise_for_status()
data = response.json()
logger.info(f"Cookie Cloud 响应: {json.dumps(data, indent=2)}")
if not data or 'encrypted' not in data:
logger.error("响应中没有 encrypted 字段")
return {}
# 打印完整的加密数据长度和前200个字符
logger.info(f"加密数据长度: {len(data['encrypted'])}")
logger.info(f"加密数据前 200 个字符: {data['encrypted'][:200]}...")
try:
decrypted_data = self.decrypt(data['encrypted'])
logger.info(f"解密后的数据: {json.dumps(decrypted_data, indent=2)}")
except Exception as e:
logger.error(f"解密失败: {e}")
return {}
# 解析 cookie_data
cookie_dict = {}
if 'cookie_data' in decrypted_data:
for domain, cookies in decrypted_data['cookie_data'].items():
cookie_dict[domain] = []
for cookie in cookies:
# 处理 sameSite 字段
if cookie.get('sameSite') == 'unspecified':
cookie['sameSite'] = 'Lax'
cookie_dict[domain].append(cookie)
return cookie_dict
except requests.RequestException as e:
logger.error(f"请求 Cookie Cloud 失败: {e}")
return {}
except Exception as e:
logger.error(f"处理 Cookie Cloud 数据失败: {e}")
return {}
def get_cookies_for_domain(self, domain: str) -> List[Dict]:
"""
获取指定域名的 cookie
Args:
domain: 域名
Returns:
cookie 列表
"""
all_cookies = self.get_cookies()
# 查找匹配的域名
for cookie_domain, cookies in all_cookies.items():
if domain in cookie_domain or cookie_domain in domain:
return cookies
return []
if __name__ == "__main__":
# 测试代码
cc = CookieCloud(
api_url="https://movie-pilot.org/cookiecloud",
uuid="hu6n2vcUmzpu7mqUN2rVCg",
password="jtDM5dV9AyqXkZdQVeA9f6"
)
cookies = cc.get_cookies()
logger.info(json.dumps(cookies, indent=2, ensure_ascii=False))
-1
View File
@@ -1 +0,0 @@
{}
+621
View File
@@ -0,0 +1,621 @@
"""
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()
+6
View File
@@ -0,0 +1,6 @@
{
"backend_url": "http://localhost:5000",
"browser_type": "edge",
"headless_mode": false,
"page_load_timeout": 30
}
+36
View File
@@ -0,0 +1,36 @@
"""
配置文件 - 从 config.json 读取配置,支持环境变量覆盖
"""
import json
import os
_config = {}
def _load_config():
global _config
config_path = os.path.join(os.path.dirname(__file__), 'config.json')
defaults = {
'backend_url': 'http://localhost:5000',
'browser_type': 'edge',
'headless_mode': False,
'page_load_timeout': 30
}
if os.path.exists(config_path):
with open(config_path, 'r', encoding='utf-8') as f:
_config = {**defaults, **json.load(f)}
else:
_config = defaults
def get(key, default=None):
if not _config:
_load_config()
return _config.get(key.upper(), _config.get(key, default))
# 兼容旧接口
BACKEND_URL = os.environ.get('BACKEND_URL') or get('backend_url', 'http://localhost:5000')
BROWSER_TYPE = os.environ.get('BROWSER_TYPE') or get('browser_type', 'edge')
HEADLESS_MODE = os.environ.get('HEADLESS_MODE') or get('headless_mode', False)
if isinstance(HEADLESS_MODE, str):
HEADLESS_MODE = HEADLESS_MODE.lower() in ('true', '1', 'yes')
CHECK_INTERVAL = int(os.environ.get('CHECK_INTERVAL') or get('check_interval', 30))
PAGE_LOAD_TIMEOUT = int(os.environ.get('PAGE_LOAD_TIMEOUT') or get('page_load_timeout', 30))
+126
View File
@@ -0,0 +1,126 @@
"""
Cookie Cloud 客户端(独立前端使用)
"""
import json
import hashlib
import logging
import requests
from base64 import b64decode
logger = logging.getLogger('cookie_cloud')
try:
from Cryptodome.Cipher import AES
from Cryptodome.Util.Padding import unpad
except ImportError:
from Crypto.Cipher import AES
from Crypto.Util.Padding import unpad
def cookie_decrypt(uuid: str, encrypted_b64: str, password: str) -> dict:
"""解密Cookie Cloud数据"""
the_key = hashlib.md5(f"{uuid}-{password}".encode()).hexdigest()[:16].encode()
encrypted_data = b64decode(encrypted_b64)
logger.info("尝试AES-ECB解密...")
try:
cipher = AES.new(the_key, AES.MODE_ECB)
decrypted = cipher.decrypt(encrypted_data)
result = unpad(decrypted, AES.block_size)
data = json.loads(result.decode('utf-8'))
logger.info("AES-ECB解密成功")
return data
except Exception as e:
logger.info(f"AES-ECB解密失败: {e},尝试Salted__格式...")
pass
if encrypted_data[:8] == b'Salted__':
logger.info("检测到Salted__格式,尝试CBC解密...")
salt = encrypted_data[8:16]
ciphertext = encrypted_data[16:]
logger.info(f"Salt: {salt.hex()}")
d = [b'']
while len(b''.join(d)) < 48:
d.append(hashlib.md5(d[-1] + the_key + salt).digest())
key_iv = b''.join(d)
cipher = AES.new(key_iv[:32], AES.MODE_CBC, key_iv[32:48])
decrypted = unpad(cipher.decrypt(ciphertext), AES.block_size)
data = json.loads(decrypted.decode('utf-8'))
logger.info("CBC解密成功")
return data
logger.error("所有解密方式均失败")
raise ValueError('无法解密Cookie数据')
def download_cookies(api_url: str, uuid: str, password: str) -> dict:
"""从Cookie Cloud服务器下载并解密Cookie"""
api_url = api_url.rstrip('/')
url = f"{api_url}/get/{uuid}"
logger.info(f"[1/3] 请求Cookie Cloud: {url}")
logger.info(f"[1/3] UUID: {uuid}")
response = requests.get(url, params={'password': password}, timeout=10)
response.raise_for_status()
result = response.json()
logger.info(f"[1/3] HTTP状态码: {response.status_code}")
logger.info(f"[1/3] 返回数据keys: {list(result.keys())}")
logger.info(f"[1/3] 返回数据大小: {len(json.dumps(result))} bytes")
if isinstance(result, dict) and result.get('encrypted'):
logger.info("[2/3] 检测到加密数据,开始解密...")
result = cookie_decrypt(uuid, result['encrypted'], password)
logger.info(f"[2/3] 解密完成,数据类型: {type(result)}")
if isinstance(result, dict):
logger.info(f"[2/3] Cookie数据包含域名: {list(result.keys())}")
if isinstance(result, dict) and 'cookie_data' in result:
cookies_data = result.get('cookie_data', {})
logger.info(f"[3/3] 新版CookieCloud格式,提取cookie_data")
logger.info(f"[3/3] cookie_data包含域名数: {len(cookies_data)}")
return cookies_data
logger.info(f"[3/3] 返回数据已处理完成")
return result
def get_cookies_for_domain(cookies_data: dict, domain: str) -> list:
"""提取指定域名对应的Cookie列表"""
result = []
matched_hosts = []
domain_parts = domain.lower().split('.')
for host, cookie_list in cookies_data.items():
host_clean = host.lower().lstrip('.')
domain_clean = domain.lower()
is_match = False
if host_clean == domain_clean:
is_match = True
elif host_clean.endswith('.' + domain_clean):
is_match = True
elif domain_clean.endswith('.' + host_clean):
is_match = True
if is_match:
matched_hosts.append(host)
if isinstance(cookie_list, list):
for cookie in cookie_list:
if cookie.get('name') and cookie.get('value'):
result.append({
'name': cookie['name'],
'value': cookie['value'],
'domain': cookie.get('domain', domain),
'path': cookie.get('path', '/')
})
logger.info(f"域名匹配结果: 目标={domain}, 匹配hosts={matched_hosts}")
logger.info(f"提取Cookie数量: {len(result)}")
if result:
cookie_names = [c['name'] for c in result[:5]]
logger.info(f"前5个Cookie名称: {cookie_names}")
return result
-345
View File
File diff suppressed because one or more lines are too long
-251
View File
@@ -1,251 +0,0 @@
"""Cookie 监控主程序"""
import json
import os
import logging
from datetime import datetime
from cookie_cloud import CookieCloud
from notifier import IYUUNotifier, FailureTracker
from browser_login import BrowserLogin, CookieManager
from applogger import setup_logging, get_logger
logger = get_logger(__name__)
class CookieMonitor:
"""Cookie 监控器"""
def __init__(self, config_file: str = "config.json"):
"""
初始化 Cookie 监控器
Args:
config_file: 配置文件路径
"""
self.config_file = config_file
self.config = self._load_config()
self.cookie_manager = CookieManager("cookies.json")
self.failure_tracker = FailureTracker("state.json")
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:
print(f"加载配置文件失败: {e}")
return {}
def _log(self, message: str):
"""记录日志"""
logger.info(message)
def _get_domain_from_url(self, url: str) -> str:
"""从 URL 中提取域名"""
from urllib.parse import urlparse
parsed = urlparse(url)
return parsed.netloc
def process_user(self, user_config: dict):
"""
处理单个用户的所有网站
Args:
user_config: 用户配置
"""
user_name = user_config.get('name', '未知用户')
self._log(f"开始处理用户: {user_name}")
# 获取用户配置
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', [])
# 初始化爱语飞飞通知
iyuu_token = notification_config.get('iyuu_token', '')
max_fail_count = notification_config.get('max_fail_count', 3)
notifier = IYUUNotifier(iyuu_token) if iyuu_token else None
# 初始化 Cookie Cloud
cookie_cloud = CookieCloud(
api_url=cookie_cloud_config.get('api_url', ''),
uuid=cookie_cloud_config.get('uuid', ''),
password=cookie_cloud_config.get('password', '')
)
# 初始化浏览器
browser = BrowserLogin(
browser_type=browser_config.get('type', 'edge'),
headless=browser_config.get('headless', False)
)
try:
# 处理每个网站
for website in websites:
self.process_website(
user_name=user_name,
website_config=website,
cookie_cloud=cookie_cloud,
notifier=notifier,
max_fail_count=max_fail_count,
browser=browser
)
finally:
# 关闭浏览器
browser.close()
self._log(f"用户 {user_name} 处理完成")
def process_website(
self,
user_name: str,
website_config: dict,
cookie_cloud: CookieCloud,
notifier: IYUUNotifier,
max_fail_count: int,
browser: BrowserLogin
):
"""
处理单个网站
Args:
user_name: 用户名
website_config: 网站配置
cookie_cloud: Cookie Cloud 客户端
notifier: 爱语飞飞通知客户端
max_fail_count: 最大失败次数
browser: 浏览器客户端
"""
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', '')
self._log(f"处理网站: {user_name} - {website_name}")
# 获取域名
domain = self._get_domain_from_url(url)
try:
# 第一步:使用本地 cookie 登录
self._log(f"尝试使用本地 cookie 登录: {website_name}")
local_cookies = self.cookie_manager.get_cookies(user_name, website_name)
if local_cookies:
login_success = browser.login_with_cookies(
url=url,
cookies=local_cookies,
check_selector=check_selector,
success_text=success_text
)
if login_success:
self._log(f"本地 cookie 登录成功: {website_name}")
# 刷新页面并保存新的 cookie
new_cookies = browser.refresh_and_save_cookies(url)
if new_cookies:
self.cookie_manager.save_cookies(user_name, website_name, new_cookies)
self._log(f"已更新 cookie: {website_name}")
# 重置失败计数
self.failure_tracker.reset_fail_count(user_name, website_name)
return
self._log(f"本地 cookie 登录失败: {website_name}")
# 第二步:从 Cookie Cloud 获取 cookie 并重试
self._log(f"从 Cookie Cloud 获取 cookie: {website_name}")
cloud_cookies = cookie_cloud.get_cookies_for_domain(domain)
if cloud_cookies:
login_success = browser.login_with_cookies(
url=url,
cookies=cloud_cookies,
check_selector=check_selector,
success_text=success_text
)
if login_success:
self._log(f"Cookie Cloud 登录成功: {website_name}")
# 刷新页面并保存新的 cookie
new_cookies = browser.refresh_and_save_cookies(url)
if new_cookies:
self.cookie_manager.save_cookies(user_name, website_name, new_cookies)
self._log(f"已更新 cookie: {website_name}")
# 重置失败计数
self.failure_tracker.reset_fail_count(user_name, website_name)
return
self._log(f"Cookie Cloud 登录失败: {website_name}")
else:
self._log(f"未从 Cookie Cloud 获取到 cookie: {website_name}")
# 第三步:登录失败,增加失败计数并检查是否需要通知
self._log(f"登录失败: {website_name}")
error_msg = f"{website_name} 登录失败,请检查 Cookie Cloud 是否有最新的 cookie"
should_stop = self.failure_tracker.check_and_notify(
user_name=user_name,
website_name=website_name,
max_fail_count=max_fail_count,
notifier=notifier,
error_msg=error_msg
)
if should_stop:
self._log(f"已达到最大失败次数,已发送通知并重置计数: {website_name}")
else:
fail_count = self.failure_tracker.get_fail_count(user_name, website_name)
self._log(f"当前失败次数: {fail_count}/{max_fail_count}")
except Exception as e:
self._log(f"处理网站 {website_name} 时发生异常: {e}")
error_msg = f"{website_name} 处理异常: {str(e)}"
should_stop = self.failure_tracker.check_and_notify(
user_name=user_name,
website_name=website_name,
max_fail_count=max_fail_count,
notifier=notifier,
error_msg=error_msg
)
def run(self):
"""运行监控"""
self._log("===== Cookie 监控开始 =====")
users = self.config.get('users', [])
if not users:
self._log("未找到用户配置")
return
success_count = 0
fail_count = 0
for user_config in users:
try:
self.process_user(user_config)
success_count += 1
except Exception as e:
self._log(f"处理用户失败: {e}")
fail_count += 1
self._log(f"===== Cookie 监控结束 =====")
self._log(f"成功处理: {success_count} 个用户,失败: {fail_count} 个用户")
def main():
"""主函数"""
# 配置日志
setup_logging()
monitor = CookieMonitor()
monitor.run()
if __name__ == "__main__":
main()
-189
View File
@@ -1,189 +0,0 @@
"""消息通知模块 - 通过爱语飞飞发送通知"""
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, "测试错误信息")
-3
View File
@@ -1,3 +0,0 @@
DrissionPage
requests
pycryptodome
-3
View File
@@ -1,3 +0,0 @@
{
"用户A_网站A": 2
}
-314
View File
@@ -1,314 +0,0 @@
# Cookie 自动监控功能需求文档
## 1. 功能概述
本系统提供 Cookie 自动监控、管理和通知功能,确保用户能够持续访问多个网站。
## 2. 详细功能需求
### 2.1 Cookie Cloud 集成
#### 2.1.1 从 Cookie Cloud 获取 Cookie
- **功能描述**: 从 Cookie Cloud 服务器获取并解密 Cookie 数据
- **输入**: Cookie Cloud API 地址、UUID、密码
- **输出**: 按域名分组的 Cookie 数据
- **处理逻辑**:
1. 发送 GET 请求到 `/get/{uuid}` 接口
2. 获取加密的 Cookie 数据
3. 使用 AES 算法解密数据
4. 解析 JSON 格式的 Cookie 数据
5. 按域名组织 Cookie
#### 2.1.2 按域名获取 Cookie
- **功能描述**: 获取指定域名的 Cookie
- **输入**: 域名
- **输出**: 该域名的 Cookie 列表
- **处理逻辑**: 遍历所有 Cookie,匹配域名
### 2.2 浏览器控制
#### 2.2.1 创建浏览器实例
- **功能描述**: 创建浏览器实例并启动浏览器
- **输入**: 浏览器类型(chrome/edge)、是否无头模式
- **输出**: 浏览器对象
- **支持浏览器**: Chrome、Edge
- **运行模式**: 有头模式、无头模式
#### 2.2.2 检测登录状态
- **功能描述**: 检测网站是否已登录
- **输入**: 网站 URL、登录检测选择器、成功文本
- **输出**: 是否已登录(True/False
- **处理逻辑**:
1. 访问网站
2. 查找登录检测元素
3. 检查元素文本是否包含成功文本
#### 2.2.3 使用 Cookie 登录
- **功能描述**: 使用 Cookie 登录网站
- **输入**: 网站 URL、Cookie 列表、登录检测选择器、成功文本
- **输出**: 是否登录成功(True/False
- **处理逻辑**:
1. 访问网站
2. 设置 Cookie
3. 刷新页面
4. 检测登录状态
#### 2.2.4 刷新并保存 Cookie
- **功能描述**: 刷新页面并获取最新的 Cookie
- **输入**: 网站 URL
- **输出**: 最新的 Cookie 列表
- **处理逻辑**:
1. 刷新页面
2. 获取当前所有 Cookie
3. 返回 Cookie 列表
### 2.3 Cookie 管理
#### 2.3.1 保存 Cookie
- **功能描述**: 保存指定用户和网站的 Cookie
- **输入**: 用户名、网站名、Cookie 列表
- **输出**: 无
- **存储位置**: `cookies.json` 文件
- **存储格式**:
```json
{
"用户名": {
"网站名": [
{
"name": "cookie名称",
"value": "cookie值",
"domain": "域名",
"path": "路径",
"expiry": "过期时间"
}
]
}
}
```
#### 2.3.2 读取 Cookie
- **功能描述**: 读取指定用户和网站的 Cookie
- **输入**: 用户名、网站名
- **输出**: Cookie 列表
- **数据来源**: `cookies.json` 文件
#### 2.3.3 按域名获取 Cookie
- **功能描述**: 从所有 Cookie 中查找指定域名的 Cookie
- **输入**: 域名
- **输出**: Cookie 列表
- **匹配规则**: 域名完全匹配或包含关系
### 2.4 失败追踪
#### 2.4.1 获取失败次数
- **功能描述**: 获取指定用户和网站的失败次数
- **输入**: 用户名、网站名
- **输出**: 失败次数
- **数据来源**: `state.json` 文件
#### 2.4.2 增加失败次数
- **功能描述**: 增加指定用户和网站的失败次数
- **输入**: 用户名、网站名
- **输出**: 增加后的失败次数
- **存储位置**: `state.json` 文件
#### 2.4.3 重置失败次数
- **功能描述**: 重置指定用户和网站的失败次数
- **输入**: 用户名、网站名
- **输出**: 无
#### 2.4.4 检查并发送通知
- **功能描述**: 检查失败次数,达到阈值时发送通知
- **输入**: 用户名、网站名、最大失败次数、通知器、错误信息
- **输出**: 是否应该停止重试
- **处理逻辑**:
1. 增加失败次数
2. 判断是否达到阈值
3. 达到阈值则发送通知
4. 发送通知后重置失败计数
5. 返回是否停止重试
### 2.5 消息通知
#### 2.5.1 发送爱语飞飞通知
- **功能描述**: 通过爱语飞飞 API 发送通知
- **输入**: 令牌、标题、内容
- **输出**: 是否发送成功(True/False
- **API 地址**: `https://iyuu.cn/{令牌}.send`
- **请求方式**: GET
- **请求参数**:
- `text`: 通知标题
- `desp`: 通知内容
### 2.6 主程序流程
#### 2.6.1 读取配置
- **功能描述**: 读取配置文件
- **输入**: 配置文件路径
- **输出**: 配置字典
- **配置文件**: `config.json`
#### 2.6.2 处理用户
- **功能描述**: 处理单个用户的所有网站
- **输入**: 用户配置
- **输出**: 无
- **处理逻辑**:
1. 初始化 Cookie Cloud 客户端
2. 初始化爱语飞飞通知器
3. 初始化浏览器
4. 遍历该用户的所有网站
5. 调用处理网站功能
6. 关闭浏览器
#### 2.6.3 处理网站
- **功能描述**: 处理单个网站的监控
- **输入**: 用户名、网站配置、Cookie Cloud 客户端、通知器、最大失败次数、浏览器
- **输出**: 无
- **处理流程**:
1. **步骤 1**: 尝试使用本地 Cookie 登录
- 读取本地 Cookie
- 登录网站
- 检测登录状态
- 登录成功 → 跳转到步骤 3
- 登录失败 → 跳转到步骤 2
2. **步骤 2**: 从 Cookie Cloud 获取 Cookie 并重试
- 从 Cookie Cloud 获取 Cookie
- 登录网站
- 检测登录状态
- 登录成功 → 跳转到步骤 3
- 登录失败 → 跳转到步骤 4
3. **步骤 3**: 刷新页面并保存新 Cookie
- 刷新页面
- 获取最新 Cookie
- 保存到 Cookie 文件
- 重置失败计数
- 结束
4. **步骤 4**: 登录失败处理
- 增加失败计数
- 判断是否达到阈值
- 达到阈值 → 发送通知 → 重置计数
- 未达到阈值 → 记录日志
#### 2.6.4 运行监控
- **功能描述**: 运行整个监控流程
- **输入**: 无
- **输出**: 无
- **处理逻辑**:
1. 读取配置文件
2. 遍历所有用户
3. 处理每个用户
4. 记录成功和失败数量
5. 输出日志
## 3. 配置管理功能
### 3.1 配置文件管理
- 支持多用户配置
- 支持多网站配置
- 支持 Cookie Cloud 配置
- 支持通知配置
- 支持浏览器配置
### 3.2 状态文件管理
- 记录失败次数
- 持久化存储
- 自动加载和保存
### 3.3 Cookie 文件管理
- 统一管理所有 Cookie
- 按用户和网站组织
- 持久化存储
## 4. 日志功能
### 4.1 日志记录
- 记录操作开始和结束
- 记录用户和网站处理状态
- 记录登录成功和失败
- 记录 Cookie 更新
- 记录通知发送
- 记录异常信息
### 4.2 日志格式
```
[时间戳] 日志内容
```
## 5. 异常处理
### 5.1 网络异常
- Cookie Cloud 请求失败
- 网站访问失败
- 通知发送失败
### 5.2 浏览器异常
- 浏览器启动失败
- 元素查找失败
- Cookie 设置失败
### 5.3 文件异常
- 配置文件读取失败
- Cookie 文件读写失败
- 状态文件读写失败
### 5.4 加密异常
- Cookie 解密失败
- 密钥生成失败
## 6. 性能要求
- Cookie Cloud 请求超时:30 秒
- 网站访问超时:10 秒
- 元素查找超时:10 秒
- 通知发送超时:10 秒
## 7. 安全要求
- Cookie Cloud 使用端到端加密
- 配置文件包含敏感信息,需妥善保管
- 爱语飞飞令牌不应泄露
- Cookie 数据不应明文传输
## 8. 扩展功能
### 8.1 多用户支持
- 每个用户独立的 Cookie Cloud 配置
- 每个用户独立的爱语飞飞令牌
- 每个用户独立的浏览器配置
### 8.2 多网站支持
- 每个用户可以配置多个网站
- 每个网站独立的登录检测
- 每个网站独立的失败计数
### 8.3 浏览器配置
- 支持 Chrome 和 Edge
- 支持有头和无头模式
- 每个用户可以独立配置浏览器
### 8.4 通知配置
- 支持自定义最大失败次数
- 支持自定义通知内容
- 每个用户独立的通知配置
## 9. 非功能需求
### 9.1 可靠性
- 程序异常不影响下次执行
- 状态文件确保数据不丢失
- 浏览器异常自动关闭
### 9.2 可维护性
- 模块化设计
- 清晰的日志输出
- 详细的错误信息
### 9.3 可扩展性
- 易于添加新用户
- 易于添加新网站
- 易于添加新通知方式
### 9.4 易用性
- 配置文件简单明了
- 日志输出清晰易懂
- 通知信息详细准确
-173
View File
@@ -1,173 +0,0 @@
# Cookie 自动监控项目需求文档
## 1. 项目背景
本项目旨在实现一个自动化 Cookie 管理和监控系统,通过定时任务检查多个网站的登录状态,自动刷新和更新 Cookie,确保用户能够持续访问目标网站。当登录失败时,系统会自动从 Cookie Cloud 服务器获取最新的 Cookie 进行重试,并在连续失败达到阈值时通过爱语飞飞消息服务发送通知。
## 2. 项目目标
- 自动监控多个网站的登录状态
- 自动刷新和更新 Cookie
- 支持 Cookie Cloud 备份和恢复
- 支持多用户、多网站配置
- 支持多种浏览器(Chrome、Edge
- 支持有头/无头模式切换
- 失败通知机制,避免消息过于频繁
## 3. 技术栈
- **编程语言**: Python 3.14.2
- **浏览器控制**: DrissionPage
- **HTTP 请求**: requests
- **加密解密**: pycryptodome
- **消息通知**: 爱语飞飞 API
- **定时任务**: Windows 任务计划程序
- **配置格式**: JSON
## 4. 运行环境
- **操作系统**: Windows 10/11
- **Python 版本**: 3.14.2
- **浏览器**: Chrome 或 Edge(系统自带)
## 5. 系统架构
### 5.1 模块划分
1. **Cookie Cloud 模块** (`cookie_cloud.py`)
- 从 Cookie Cloud 服务器获取 Cookie 数据
- 解密 Cookie 数据
- 按域名组织 Cookie
2. **消息通知模块** (`notifier.py`)
- 集成爱语飞飞 API
- 失败次数追踪
- 通知发送控制
3. **浏览器登录模块** (`browser_login.py`)
- 使用 DrissionPage 控制浏览器
- 检测登录状态
- 管理 Cookie
4. **Cookie 管理器** (`browser_login.py`)
- 统一管理所有 Cookie
- 支持按用户和网站存储
- 持久化存储
5. **主程序** (`monitor.py`)
- 读取配置文件
- 协调各模块工作
- 处理异常和日志
### 5.2 数据流
```
配置文件 → 主程序 → 浏览器登录模块 → 网站检测
Cookie 管理器 ← Cookie Cloud 模块
消息通知模块 ← 失败追踪器
```
## 6. 核心流程
### 6.1 监控流程
1. 读取配置文件
2. 遍历每个用户
3. 遍历该用户的每个网站
4. 尝试使用本地 Cookie 登录
5. 登录失败 → 从 Cookie Cloud 获取 Cookie → 重试登录
6. 登录成功 → 刷新页面 → 保存新 Cookie → 重置失败计数
7. 登录失败 → 失败计数 +1
8. 失败计数达到阈值 → 发送提醒 → 重置计数
9. 保存状态到状态文件
### 6.2 失败重试策略
- 每个网站独立计数失败次数
- 连续失败 3 次后发送通知
- 发送通知后重置计数
- 下次执行重新开始计数
## 7. 配置管理
### 7.1 配置文件结构
```json
{
"users": [
{
"name": "用户名",
"cookie_cloud": {
"uuid": "Cookie Cloud UUID",
"password": "Cookie Cloud 密码",
"api_url": "Cookie Cloud API 地址"
},
"notification": {
"iyuu_token": "爱语飞飞令牌",
"max_fail_count": 3
},
"browser": {
"type": "edge/chrome",
"headless": true/false
},
"websites": [
{
"name": "网站名称",
"url": "网站地址",
"login_check_selector": "登录检测选择器",
"success_text": "登录成功文本"
}
]
}
]
}
```
### 7.2 数据文件
- `config.json`: 配置文件
- `cookies.json`: Cookie 存储文件
- `state.json`: 失败计数状态文件
## 8. 定时任务
使用 Windows 任务计划程序设置定时任务:
- 执行频率:每 30 分钟
- 执行命令:`python G:\test\cookie\monitor.py`
- 运行账户:SYSTEM
## 9. 依赖安装
```bash
pip install DrissionPage requests pycryptodome
```
## 10. 项目文件结构
```
G:\test\cookie\
├── config.json # 配置文件
├── monitor.py # 主程序
├── cookie_cloud.py # Cookie Cloud 模块
├── notifier.py # 消息通知模块
├── browser_login.py # 浏览器登录模块
├── state.json # 失败计数状态文件
└── cookies.json # Cookie 存储文件
```
## 11. 安全考虑
- Cookie Cloud 使用端到端加密
- 配置文件包含敏感信息,需妥善保管
- 定时任务使用 SYSTEM 账户运行,需谨慎授权
- 爱语飞飞令牌不应泄露
## 12. 扩展性
- 支持添加更多用户
- 支持添加更多网站
- 支持自定义失败阈值
- 支持自定义通知消息内容
- 支持切换浏览器类型和运行模式