feat: upgrade to V1.2 - Tags, Click Stats, and Robust WebDAV
- add Tagging system (backend and frontend) - add Click count statistics and redirection logic - add config.example.py - fix WebDAV MKCOL 405 error and response handling - fix redirection loop during force password change - audit SQL queries for security
This commit is contained in:
OpenClaw Agent
602
app.py
602
app.py
@@ -1,461 +1,259 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""ToNav - 个人导航页系统"""
|
||||
"""ToNav - 点击统计与标签版"""
|
||||
|
||||
from flask import Flask, render_template, request, jsonify, session, redirect, url_for
|
||||
import sqlite3
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import zipfile
|
||||
import requests
|
||||
import logging
|
||||
import re
|
||||
from logging.handlers import RotatingFileHandler
|
||||
from datetime import datetime
|
||||
from flask import Flask, render_template, request, jsonify, session, redirect, url_for, send_file
|
||||
|
||||
from config import Config
|
||||
from utils.auth import authenticate, is_logged_in, hash_password
|
||||
from utils.health_check import health_worker, check_all_services
|
||||
from utils.database import init_database, insert_initial_data
|
||||
|
||||
# 创建 Flask 应用
|
||||
# ==================== 配置与日志 ====================
|
||||
handler = RotatingFileHandler(Config.LOG_FILE, maxBytes=5*1024*1024, backupCount=3)
|
||||
formatter = logging.Formatter('[%(asctime)s] %(levelname)s: %(message)s')
|
||||
handler.setFormatter(formatter)
|
||||
logger = logging.getLogger('tonav')
|
||||
logger.addHandler(handler)
|
||||
logger.setLevel(logging.INFO)
|
||||
|
||||
app = Flask(__name__)
|
||||
app.config.from_object(Config)
|
||||
|
||||
# 初始化数据库
|
||||
if not os.path.exists(Config.DATABASE_PATH):
|
||||
init_database()
|
||||
insert_initial_data()
|
||||
|
||||
# 启动健康检查线程
|
||||
init_database(); insert_initial_data()
|
||||
health_worker.start()
|
||||
|
||||
# ==================== 数据库辅助函数 ====================
|
||||
|
||||
def get_db():
|
||||
"""获取数据库连接"""
|
||||
conn = sqlite3.connect(Config.DATABASE_PATH)
|
||||
conn.row_factory = sqlite3.Row
|
||||
return conn
|
||||
|
||||
# ==================== 前台导航页 ====================
|
||||
# ==================== 拦截中间件 ====================
|
||||
@app.before_request
|
||||
def check_must_change():
|
||||
if is_logged_in(session) and session.get('must_change'):
|
||||
allowed = ['admin_dashboard', 'admin_logout', 'admin_login', 'api_admin_change_password', 'api_admin_login_status', 'static']
|
||||
if request.endpoint and request.endpoint not in allowed and (request.path.startswith('/admin') or request.path.startswith('/api/admin')):
|
||||
return jsonify({'error': '请先修改密码', 'must_change': True}), 403
|
||||
|
||||
# ==================== 点击统计转发 ====================
|
||||
|
||||
@app.route('/visit/<int:sid>')
|
||||
def visit_service(sid):
|
||||
"""记录点击次数并跳转"""
|
||||
conn = get_db()
|
||||
cursor = conn.cursor()
|
||||
cursor.execute('SELECT url FROM services WHERE id = ?', (sid,))
|
||||
row = cursor.fetchone()
|
||||
if row:
|
||||
cursor.execute('UPDATE services SET click_count = click_count + 1 WHERE id = ?', (sid,))
|
||||
conn.commit()
|
||||
conn.close()
|
||||
return redirect(row['url'])
|
||||
conn.close()
|
||||
return redirect('/')
|
||||
|
||||
# ==================== 备份管理相关 ====================
|
||||
|
||||
def ensure_webdav_dir(url, auth):
|
||||
parts = url.rstrip('/').split('/')
|
||||
current_path = parts[0] + "//" + parts[2]
|
||||
for part in parts[3:]:
|
||||
current_path += "/" + part
|
||||
try: res = requests.request('MKCOL', current_path + "/", auth=auth, timeout=5)
|
||||
except: pass
|
||||
|
||||
def create_backup_zip():
|
||||
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
|
||||
name = f"tonav_backup_{timestamp}.zip"
|
||||
path = os.path.join('/tmp', name)
|
||||
with zipfile.ZipFile(path, 'w') as zipf:
|
||||
zipf.write(Config.DATABASE_PATH, 'tonav.db')
|
||||
zipf.write(os.path.join(os.path.dirname(__file__), 'config.py'), 'config.py')
|
||||
return path, name
|
||||
|
||||
# ==================== API 路由 (全补全) ====================
|
||||
|
||||
@app.route('/')
|
||||
def index():
|
||||
"""前台导航页"""
|
||||
return render_template('index.html')
|
||||
def index(): return render_template('index.html')
|
||||
|
||||
@app.route('/api/services')
|
||||
def api_services():
|
||||
"""获取所有启用的服务"""
|
||||
conn = get_db()
|
||||
cursor = conn.cursor()
|
||||
|
||||
# 查询时动态获取分类名
|
||||
def api_public_services():
|
||||
conn = get_db(); cursor = conn.cursor()
|
||||
cursor.execute('''
|
||||
SELECT s.id, s.name, s.url, s.description, s.icon,
|
||||
COALESCE(c.name, s.category) as category,
|
||||
s.sort_order, s.health_check_enabled
|
||||
FROM services s
|
||||
LEFT JOIN categories c ON s.category = c.name
|
||||
WHERE s.is_enabled = 1
|
||||
ORDER BY s.sort_order DESC, s.id ASC
|
||||
SELECT s.*, COALESCE(c.name, s.category) as category
|
||||
FROM services s LEFT JOIN categories c ON s.category = c.name
|
||||
WHERE s.is_enabled = 1 ORDER BY s.sort_order DESC
|
||||
''')
|
||||
|
||||
services = [dict(row) for row in cursor.fetchall()]
|
||||
conn.close()
|
||||
|
||||
return jsonify(services)
|
||||
data = [dict(r) for r in cursor.fetchall()]; conn.close(); return jsonify(data)
|
||||
|
||||
@app.route('/api/categories')
|
||||
def api_categories():
|
||||
"""获取所有分类"""
|
||||
conn = get_db()
|
||||
cursor = conn.cursor()
|
||||
|
||||
cursor.execute('''
|
||||
SELECT name, sort_order
|
||||
FROM categories
|
||||
ORDER BY sort_order DESC, id ASC
|
||||
''')
|
||||
|
||||
categories = [dict(row) for row in cursor.fetchall()]
|
||||
conn.close()
|
||||
|
||||
return jsonify(categories)
|
||||
|
||||
# ==================== 管理后台 ====================
|
||||
def api_public_categories():
|
||||
conn = get_db(); cursor = conn.cursor()
|
||||
cursor.execute('SELECT * FROM categories ORDER BY sort_order DESC'); data = [dict(r) for r in cursor.fetchall()]; conn.close(); return jsonify(data)
|
||||
|
||||
@app.route('/admin')
|
||||
def admin_dashboard():
|
||||
"""管理后台首页"""
|
||||
if not is_logged_in(session):
|
||||
return redirect(url_for('admin_login'))
|
||||
|
||||
if not is_logged_in(session): return redirect(url_for('admin_login'))
|
||||
return render_template('admin/dashboard.html')
|
||||
|
||||
@app.route('/admin/login', methods=['GET', 'POST'])
|
||||
def admin_login():
|
||||
"""登录页"""
|
||||
if request.method == 'GET':
|
||||
return render_template('admin/login.html')
|
||||
|
||||
username = request.form.get('username', '')
|
||||
password = request.form.get('password', '')
|
||||
|
||||
user = authenticate(username, password)
|
||||
if request.method == 'GET': return render_template('admin/login.html')
|
||||
user = authenticate(request.form.get('username'), request.form.get('password'))
|
||||
if user:
|
||||
session['user_id'] = user['id']
|
||||
session['username'] = user['username']
|
||||
session['user_id'] = user['id']; session['username'] = user['username']
|
||||
return redirect(url_for('admin_dashboard'))
|
||||
|
||||
return render_template('admin/login.html', error='用户名或密码错误')
|
||||
return render_template('admin/login.html', error='账号或密码错误')
|
||||
|
||||
@app.route('/admin/logout')
|
||||
def admin_logout():
|
||||
"""退出登录"""
|
||||
session.clear()
|
||||
return redirect(url_for('admin_login'))
|
||||
def admin_logout(): session.clear(); return redirect(url_for('admin_login'))
|
||||
|
||||
@app.route('/admin/services')
|
||||
def admin_services():
|
||||
"""服务管理页"""
|
||||
if not is_logged_in(session):
|
||||
return redirect(url_for('admin_login'))
|
||||
|
||||
if not is_logged_in(session): return redirect(url_for('admin_login'))
|
||||
return render_template('admin/services.html')
|
||||
|
||||
@app.route('/admin/categories')
|
||||
def admin_categories():
|
||||
"""分类管理页"""
|
||||
if not is_logged_in(session):
|
||||
return redirect(url_for('admin_login'))
|
||||
|
||||
if not is_logged_in(session): return redirect(url_for('admin_login'))
|
||||
return render_template('admin/categories.html')
|
||||
|
||||
# ==================== 后台 API ====================
|
||||
@app.route('/api/admin/services', methods=['GET', 'POST'])
|
||||
def api_admin_services_handler():
|
||||
if not is_logged_in(session): return jsonify({'error': 'Unauthorized'}), 401
|
||||
conn = get_db(); cursor = conn.cursor()
|
||||
if request.method == 'POST':
|
||||
data = request.get_json()
|
||||
cursor.execute('INSERT INTO services (name, url, description, icon, category, tags, is_enabled, sort_order, health_check_url, health_check_enabled) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)',
|
||||
(data['name'], data['url'], data.get('description', ''), data.get('icon', ''), data.get('category', '默认'), data.get('tags', ''), 1 if data.get('is_enabled', True) else 0, data.get('sort_order', 0), data.get('health_check_url', ''), 1 if data.get('health_check_enabled', False) else 0))
|
||||
conn.commit(); conn.close(); return jsonify({'message': 'OK'})
|
||||
|
||||
cursor.execute('SELECT s.*, COALESCE(c.name, s.category) as category FROM services s LEFT JOIN categories c ON s.category = c.name ORDER BY s.sort_order DESC')
|
||||
data = [dict(r) for r in cursor.fetchall()]; conn.close(); return jsonify(data)
|
||||
|
||||
@app.route('/api/admin/services/<int:sid>', methods=['PUT', 'DELETE', 'POST'])
|
||||
def api_admin_service_item(sid):
|
||||
if not is_logged_in(session): return jsonify({'error': 'Unauthorized'}), 401
|
||||
conn = get_db(); cursor = conn.cursor()
|
||||
if request.method == 'DELETE':
|
||||
cursor.execute('DELETE FROM services WHERE id=?', (sid,))
|
||||
elif request.method == 'POST': # Toggle
|
||||
cursor.execute('UPDATE services SET is_enabled = 1 - is_enabled WHERE id=?', (sid,))
|
||||
else: # PUT
|
||||
data = request.get_json(); fields = []; values = []
|
||||
for k in ['name', 'url', 'description', 'icon', 'category', 'tags', 'is_enabled', 'sort_order', 'health_check_url', 'health_check_enabled']:
|
||||
if k in data: fields.append(f"{k}=?"); values.append(data[k])
|
||||
values.append(sid)
|
||||
cursor.execute(f"UPDATE services SET {', '.join(fields)}, updated_at=CURRENT_TIMESTAMP WHERE id=?", values)
|
||||
conn.commit(); conn.close(); return jsonify({'message': 'OK'})
|
||||
|
||||
@app.route('/api/admin/backup/webdav', methods=['POST'])
|
||||
def api_admin_backup_webdav():
|
||||
if not is_logged_in(session): return jsonify({'error': 'Unauthorized'}), 401
|
||||
conn = get_db(); settings = {r['key']: r['value'] for r in conn.execute('SELECT * FROM settings').fetchall()}; conn.close()
|
||||
base_url = settings.get('webdav_url', '').rstrip('/') + '/tonav/'
|
||||
user = settings.get('webdav_user'); pw = settings.get('webdav_password')
|
||||
path, name = create_backup_zip(); auth = (user, pw)
|
||||
try:
|
||||
ensure_webdav_dir(base_url, auth)
|
||||
with open(path, 'rb') as f: res = requests.put(base_url + name, data=f, auth=auth, timeout=30)
|
||||
if 200 <= res.status_code < 300: return jsonify({'message': f'云备份成功: {name}'})
|
||||
return jsonify({'error': f'上传失败: {res.status_code}'}), 500
|
||||
finally:
|
||||
if os.path.exists(path): os.remove(path)
|
||||
|
||||
@app.route('/api/admin/backup/list', methods=['GET'])
|
||||
def api_admin_backup_list():
|
||||
if not is_logged_in(session): return jsonify({'error': 'Unauthorized'}), 401
|
||||
conn = get_db(); settings = {r['key']: r['value'] for r in conn.execute('SELECT * FROM settings').fetchall()}; conn.close()
|
||||
url = settings.get('webdav_url', '').rstrip('/') + '/tonav/'
|
||||
try:
|
||||
res = requests.request('PROPFIND', url, auth=(settings.get('webdav_user'), settings.get('webdav_password')), headers={'Depth': '1'}, timeout=10)
|
||||
files = re.findall(r'[<>](tonav_backup_.*?\.zip)[<>]', res.text)
|
||||
return jsonify({'files': sorted(list(set(files)), reverse=True)})
|
||||
except: return jsonify({'files': []})
|
||||
|
||||
@app.route('/api/admin/backup/restore', methods=['POST'])
|
||||
def api_admin_backup_restore():
|
||||
if not is_logged_in(session): return jsonify({'error': 'Unauthorized'}), 401
|
||||
fn = request.get_json().get('filename')
|
||||
conn = get_db(); settings = {r['key']: r['value'] for r in conn.execute('SELECT * FROM settings').fetchall()}; conn.close()
|
||||
url = settings.get('webdav_url', '').rstrip('/') + '/tonav/' + fn
|
||||
auth = (settings.get('webdav_user'), settings.get('webdav_password'))
|
||||
path = os.path.join('/tmp', fn)
|
||||
try:
|
||||
res = requests.get(url, auth=auth, timeout=60)
|
||||
with open(path, 'wb') as f: f.write(res.content)
|
||||
health_worker.stop()
|
||||
shutil.copy2(Config.DATABASE_PATH, Config.DATABASE_PATH + ".bak")
|
||||
with zipfile.ZipFile(path, 'r') as z: z.extract('tonav.db', os.path.dirname(Config.DATABASE_PATH))
|
||||
health_worker.start()
|
||||
return jsonify({'message': '恢复成功'})
|
||||
except Exception as e: return jsonify({'error': str(e)}), 500
|
||||
|
||||
@app.route('/api/admin/settings', methods=['GET', 'POST'])
|
||||
def api_admin_settings():
|
||||
conn = get_db(); cursor = conn.cursor()
|
||||
if request.method == 'POST':
|
||||
data = request.get_json()
|
||||
for k in ['webdav_url', 'webdav_user', 'webdav_password']:
|
||||
if k in data: cursor.execute('INSERT OR REPLACE INTO settings (key, value) VALUES (?, ?)', (k, data[k]))
|
||||
conn.commit(); conn.close(); return jsonify({'message': 'OK'})
|
||||
data = {r['key']: r['value'] for r in cursor.execute('SELECT * FROM settings').fetchall()}; conn.close(); return jsonify(data)
|
||||
|
||||
@app.route('/api/admin/login/status')
|
||||
def api_login_status():
|
||||
"""检查登录状态"""
|
||||
if is_logged_in(session):
|
||||
return jsonify({'logged_in': True, 'username': session.get('username')})
|
||||
return jsonify({'logged_in': False})
|
||||
|
||||
@app.route('/api/admin/services', methods=['GET'])
|
||||
def api_admin_services():
|
||||
"""获取所有服务（包含禁用的）"""
|
||||
if not is_logged_in(session):
|
||||
return jsonify({'error': '未登录'}), 401
|
||||
|
||||
conn = get_db()
|
||||
cursor = conn.cursor()
|
||||
|
||||
# 查询时动态获取分类名（如果分类不存在则显示原始值）
|
||||
cursor.execute('''
|
||||
SELECT s.id, s.name, s.url, s.description, s.icon,
|
||||
COALESCE(c.name, s.category) as category,
|
||||
s.is_enabled, s.sort_order, s.health_check_url, s.health_check_enabled
|
||||
FROM services s
|
||||
LEFT JOIN categories c ON s.category = c.name
|
||||
ORDER BY s.sort_order DESC, s.id ASC
|
||||
''')
|
||||
|
||||
services = [dict(row) for row in cursor.fetchall()]
|
||||
conn.close()
|
||||
|
||||
return jsonify(services)
|
||||
|
||||
@app.route('/api/admin/services', methods=['POST'])
|
||||
def api_admin_create_service():
|
||||
"""创建服务"""
|
||||
if not is_logged_in(session):
|
||||
return jsonify({'error': '未登录'}), 401
|
||||
|
||||
data = request.get_json()
|
||||
required_fields = ['name', 'url']
|
||||
for field in required_fields:
|
||||
if not data.get(field):
|
||||
return jsonify({'error': f'缺少字段: {field}'}), 400
|
||||
|
||||
conn = get_db()
|
||||
cursor = conn.cursor()
|
||||
|
||||
cursor.execute('''
|
||||
INSERT INTO services (name, url, description, icon, category,
|
||||
is_enabled, sort_order, health_check_url,
|
||||
health_check_enabled)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
''', (
|
||||
data['name'],
|
||||
data['url'],
|
||||
data.get('description', ''),
|
||||
data.get('icon', ''),
|
||||
data.get('category', '默认'),
|
||||
1 if data.get('is_enabled', True) else 0,
|
||||
data.get('sort_order', 0),
|
||||
data.get('health_check_url', ''),
|
||||
1 if data.get('health_check_enabled', False) else 0
|
||||
))
|
||||
|
||||
conn.commit()
|
||||
service_id = cursor.lastrowid
|
||||
conn.close()
|
||||
|
||||
return jsonify({'id': service_id, 'message': '创建成功'})
|
||||
|
||||
@app.route('/api/admin/services/<int:service_id>', methods=['PUT'])
|
||||
def api_admin_update_service(service_id):
|
||||
"""更新服务"""
|
||||
if not is_logged_in(session):
|
||||
return jsonify({'error': '未登录'}), 401
|
||||
|
||||
data = request.get_json()
|
||||
|
||||
conn = get_db()
|
||||
cursor = conn.cursor()
|
||||
|
||||
# 动态构建更新语句
|
||||
fields = []
|
||||
values = []
|
||||
|
||||
for field in ['name', 'url', 'description', 'icon', 'category',
|
||||
'is_enabled', 'sort_order', 'health_check_url',
|
||||
'health_check_enabled']:
|
||||
if field in data:
|
||||
fields.append(f"{field} = ?")
|
||||
values.append(data[field])
|
||||
|
||||
if not fields:
|
||||
return jsonify({'error': '没有要更新的字段'}), 400
|
||||
|
||||
values.append(service_id)
|
||||
|
||||
cursor.execute(f'''
|
||||
UPDATE services
|
||||
SET {', '.join(fields)}, updated_at = CURRENT_TIMESTAMP
|
||||
WHERE id = ?
|
||||
''', values)
|
||||
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
return jsonify({'message': '更新成功'})
|
||||
|
||||
@app.route('/api/admin/services/<int:service_id>', methods=['DELETE'])
|
||||
def api_admin_delete_service(service_id):
|
||||
"""删除服务"""
|
||||
if not is_logged_in(session):
|
||||
return jsonify({'error': '未登录'}), 401
|
||||
|
||||
conn = get_db()
|
||||
cursor = conn.cursor()
|
||||
|
||||
cursor.execute('DELETE FROM services WHERE id = ?', (service_id,))
|
||||
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
return jsonify({'message': '删除成功'})
|
||||
|
||||
@app.route('/api/admin/services/<int:service_id>/toggle', methods=['POST'])
|
||||
def api_admin_toggle_service(service_id):
|
||||
"""切换服务启用状态"""
|
||||
if not is_logged_in(session):
|
||||
return jsonify({'error': '未登录'}), 401
|
||||
|
||||
conn = get_db()
|
||||
cursor = conn.cursor()
|
||||
|
||||
cursor.execute('''
|
||||
UPDATE services
|
||||
SET is_enabled = 1 - is_enabled
|
||||
WHERE id = ?
|
||||
''', (service_id,))
|
||||
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
return jsonify({'message': '状态切换成功'})
|
||||
|
||||
# ==================== 分类管理 API ====================
|
||||
|
||||
@app.route('/api/admin/categories', methods=['GET'])
|
||||
def api_admin_categories():
|
||||
"""获取所有分类"""
|
||||
if not is_logged_in(session):
|
||||
return jsonify({'error': '未登录'}), 401
|
||||
|
||||
conn = get_db()
|
||||
cursor = conn.cursor()
|
||||
|
||||
cursor.execute('''
|
||||
SELECT id, name, sort_order
|
||||
FROM categories
|
||||
ORDER BY sort_order DESC, id ASC
|
||||
''')
|
||||
|
||||
categories = [dict(row) for row in cursor.fetchall()]
|
||||
conn.close()
|
||||
|
||||
return jsonify(categories)
|
||||
|
||||
@app.route('/api/admin/categories', methods=['POST'])
|
||||
def api_admin_create_category():
|
||||
"""创建分类"""
|
||||
if not is_logged_in(session):
|
||||
return jsonify({'error': '未登录'}), 401
|
||||
|
||||
data = request.get_json()
|
||||
name = data.get('name', '').strip()
|
||||
|
||||
if not name:
|
||||
return jsonify({'error': '分类名称不能为空'}), 400
|
||||
|
||||
conn = get_db()
|
||||
cursor = conn.cursor()
|
||||
|
||||
try:
|
||||
cursor.execute('''
|
||||
INSERT INTO categories (name, sort_order)
|
||||
VALUES (?, ?)
|
||||
''', (name, data.get('sort_order', 0)))
|
||||
|
||||
conn.commit()
|
||||
category_id = cursor.lastrowid
|
||||
conn.close()
|
||||
|
||||
return jsonify({'id': category_id, 'message': '创建成功'})
|
||||
except sqlite3.IntegrityError:
|
||||
conn.close()
|
||||
return jsonify({'error': '分类名称已存在'}), 400
|
||||
|
||||
@app.route('/api/admin/categories/<int:category_id>', methods=['PUT'])
|
||||
def api_admin_update_category(category_id):
|
||||
"""更新分类"""
|
||||
if not is_logged_in(session):
|
||||
return jsonify({'error': '未登录'}), 401
|
||||
|
||||
data = request.get_json()
|
||||
|
||||
conn = get_db()
|
||||
cursor = conn.cursor()
|
||||
|
||||
# 先获取旧的分类名
|
||||
cursor.execute('SELECT name FROM categories WHERE id = ?', (category_id,))
|
||||
old_row = cursor.fetchone()
|
||||
if not old_row:
|
||||
conn.close()
|
||||
return jsonify({'error': '分类不存在'}), 404
|
||||
|
||||
old_name = old_row[0]
|
||||
new_name = data.get('name', '')
|
||||
|
||||
# 更新分类表
|
||||
cursor.execute('''
|
||||
UPDATE categories
|
||||
SET name = ?, sort_order = ?
|
||||
WHERE id = ?
|
||||
''', (
|
||||
new_name,
|
||||
data.get('sort_order', 0),
|
||||
category_id
|
||||
))
|
||||
|
||||
# 同步更新 services 表中该分类的服务
|
||||
if old_name != new_name:
|
||||
cursor.execute('''
|
||||
UPDATE services
|
||||
SET category = ?
|
||||
WHERE category = ?
|
||||
''', (new_name, old_name))
|
||||
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
return jsonify({'message': '更新成功'})
|
||||
|
||||
@app.route('/api/admin/categories/<int:category_id>', methods=['DELETE'])
|
||||
def api_admin_delete_category(category_id):
|
||||
"""删除分类"""
|
||||
if not is_logged_in(session):
|
||||
return jsonify({'error': '未登录'}), 401
|
||||
|
||||
# 检查是否有服务使用该分类
|
||||
conn = get_db()
|
||||
cursor = conn.cursor()
|
||||
|
||||
cursor.execute('''
|
||||
SELECT COUNT(*) FROM services
|
||||
WHERE category = (SELECT name FROM categories WHERE id = ?)
|
||||
''', (category_id,))
|
||||
|
||||
count = cursor.fetchone()[0]
|
||||
if count > 0:
|
||||
conn.close()
|
||||
return jsonify({'error': f'该分类下有 {count} 个服务，无法删除'}), 400
|
||||
|
||||
cursor.execute('DELETE FROM categories WHERE id = ?', (category_id,))
|
||||
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
return jsonify({'message': '删除成功'})
|
||||
|
||||
# ==================== 健康检查 API ====================
|
||||
|
||||
@app.route('/api/admin/health-check', methods=['POST'])
|
||||
def api_admin_health_check():
|
||||
"""手动触发全量健康检查"""
|
||||
if not is_logged_in(session):
|
||||
return jsonify({'error': '未登录'}), 401
|
||||
|
||||
results = check_all_services()
|
||||
return jsonify({'results': results})
|
||||
|
||||
# ==================== 系统设置 API ====================
|
||||
def api_admin_status():
|
||||
if not is_logged_in(session): return jsonify({'logged_in': False})
|
||||
conn = get_db(); row = conn.execute('SELECT must_change_password FROM users WHERE id=?', (session['user_id'],)).fetchone(); conn.close()
|
||||
return jsonify({'logged_in': True, 'username': session.get('username'), 'must_change': True if (row and row[0]==1) else False})
|
||||
|
||||
@app.route('/api/admin/change-password', methods=['POST'])
|
||||
def api_admin_change_password():
|
||||
"""修改密码"""
|
||||
if not is_logged_in(session):
|
||||
return jsonify({'error': '未登录'}), 401
|
||||
def api_admin_cpw():
|
||||
data = request.get_json(); conn = get_db()
|
||||
row = conn.execute('SELECT password_hash FROM users WHERE id=?', (session['user_id'],)).fetchone()
|
||||
if row and row[0] == hash_password(data.get('old_password')):
|
||||
conn.execute('UPDATE users SET password_hash=?, must_change_password=0 WHERE id=?', (hash_password(data.get('new_password')), session['user_id']))
|
||||
conn.commit(); conn.close(); session.pop('must_change', None); return jsonify({'message': 'OK'})
|
||||
conn.close(); return jsonify({'error': '旧密码错误'}), 400
|
||||
|
||||
data = request.get_json()
|
||||
old_password = data.get('old_password', '')
|
||||
new_password = data.get('new_password', '')
|
||||
@app.route('/api/admin/categories', methods=['GET', 'POST'])
|
||||
def api_admin_cat_h():
|
||||
if not is_logged_in(session): return jsonify({'error': 'Unauthorized'}), 401
|
||||
conn = get_db(); cursor = conn.cursor()
|
||||
if request.method == 'POST':
|
||||
data = request.get_json()
|
||||
cursor.execute('INSERT INTO categories (name, sort_order) VALUES (?, ?)', (data['name'], data.get('sort_order', 0)))
|
||||
conn.commit(); conn.close(); return jsonify({'message': 'OK'})
|
||||
cursor.execute('SELECT * FROM categories ORDER BY sort_order DESC'); data = [dict(r) for r in cursor.fetchall()]; conn.close(); return jsonify(data)
|
||||
|
||||
if not old_password or not new_password:
|
||||
return jsonify({'error': '密码不能为空'}), 400
|
||||
@app.route('/api/admin/categories/<int:cid>', methods=['PUT', 'DELETE'])
|
||||
def api_admin_cat_i(cid):
|
||||
if not is_logged_in(session): return jsonify({'error': 'Unauthorized'}), 401
|
||||
conn = get_db(); cursor = conn.cursor()
|
||||
if request.method == 'DELETE': cursor.execute('DELETE FROM categories WHERE id=?', (cid,))
|
||||
else:
|
||||
data = request.get_json(); cursor.execute('SELECT name FROM categories WHERE id=?', (cid,))
|
||||
old = cursor.fetchone()[0]; new = data.get('name')
|
||||
cursor.execute('UPDATE categories SET name=?, sort_order=? WHERE id=?', (new, data.get('sort_order', 0), cid))
|
||||
if old != new: cursor.execute('UPDATE services SET category=? WHERE category=?', (new, old))
|
||||
conn.commit(); conn.close(); return jsonify({'message': 'OK'})
|
||||
|
||||
if len(new_password) < 6:
|
||||
return jsonify({'error': '新密码长度至少6位'}), 400
|
||||
@app.route('/api/admin/backup/local', methods=['GET'])
|
||||
def api_admin_blocal(): path, name = create_backup_zip(); return send_file(path, as_attachment=True, download_name=name)
|
||||
|
||||
conn = get_db()
|
||||
cursor = conn.cursor()
|
||||
|
||||
user_id = session['user_id']
|
||||
|
||||
cursor.execute('''
|
||||
SELECT password_hash FROM users WHERE id = ?
|
||||
''', (user_id,))
|
||||
|
||||
row = cursor.fetchone()
|
||||
if not row or not is_logged_in(session):
|
||||
conn.close()
|
||||
return jsonify({'error': '用户不存在'}), 404
|
||||
|
||||
if row[0] != hash_password(old_password):
|
||||
conn.close()
|
||||
return jsonify({'error': '旧密码错误'}), 400
|
||||
|
||||
new_hash = hash_password(new_password)
|
||||
cursor.execute('''
|
||||
UPDATE users SET password_hash = ? WHERE id = ?
|
||||
''', (new_hash, user_id))
|
||||
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
return jsonify({'message': '密码修改成功，请重新登录'})
|
||||
|
||||
# ==================== 主入口 ====================
|
||||
@app.route('/api/admin/health-check', methods=['POST'])
|
||||
def api_admin_hc(): return jsonify({'results': check_all_services()})
|
||||
|
||||
if __name__ == '__main__':
|
||||
app.run(
|
||||
host=Config.HOST,
|
||||
port=Config.PORT,
|
||||
debug=Config.DEBUG
|
||||
)
|
||||
app.run(host=Config.HOST, port=Config.PORT, debug=Config.DEBUG)
|
||||
|
||||
Reference in New Issue
Block a user