重构Cookie监控系统:添加Web管理后台、Vue3前端界面、Cookie Cloud参数支持、5秒倒计时自动刷新、专业UI配色

This commit is contained in:
bwstudio
2026-04-17 22:37:28 +08:00
parent ed95ce5b23
commit 70802b9b50
44 changed files with 2923 additions and 3654 deletions
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
+83
View File
@@ -0,0 +1,83 @@
"""
检测结果路由
"""
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):
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': self.created_at.isoformat() if self.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
+237
View File
@@ -0,0 +1,237 @@
"""
网站管理路由
"""
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():
"""检测所有网站的网络可达性"""
websites = Website.query.all()
result = {}
for website in websites:
try:
resp = http_requests.get(website.url, timeout=5, allow_redirects=True)
result[website.id] = {
'status': resp.status_code,
'reachable': resp.status_code < 400,
'response_time': resp.elapsed.total_seconds() * 1000
}
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