Initial commit: ToNav Personal Navigation Page
- Flask + SQLite 个人导航页系统 - 前台导航页（分类Tab、卡片展示） - 管理后台（服务管理、分类管理、健康检测） - 响应式设计 - Systemd 服务配置
This commit is contained in:
OpenClaw Agent
436
app.py
Normal file
436
app.py
Normal file
@@ -0,0 +1,436 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""ToNav - 个人导航页系统"""
|
||||
|
||||
from flask import Flask, render_template, request, jsonify, session, redirect, url_for
|
||||
import sqlite3
|
||||
import json
|
||||
import os
|
||||
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 应用
|
||||
app = Flask(__name__)
|
||||
app.config.from_object(Config)
|
||||
|
||||
# 初始化数据库
|
||||
if not os.path.exists(Config.DATABASE_PATH):
|
||||
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.route('/')
|
||||
def index():
|
||||
"""前台导航页"""
|
||||
return render_template('index.html')
|
||||
|
||||
@app.route('/api/services')
|
||||
def api_services():
|
||||
"""获取所有启用的服务"""
|
||||
conn = get_db()
|
||||
cursor = conn.cursor()
|
||||
|
||||
cursor.execute('''
|
||||
SELECT id, name, url, description, icon, category, sort_order,
|
||||
health_check_enabled
|
||||
FROM services
|
||||
WHERE is_enabled = 1
|
||||
ORDER BY sort_order DESC, id ASC
|
||||
''')
|
||||
|
||||
services = [dict(row) for row in cursor.fetchall()]
|
||||
conn.close()
|
||||
|
||||
return jsonify(services)
|
||||
|
||||
@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)
|
||||
|
||||
# ==================== 管理后台 ====================
|
||||
|
||||
@app.route('/admin')
|
||||
def admin_dashboard():
|
||||
"""管理后台首页"""
|
||||
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 user:
|
||||
session['user_id'] = user['id']
|
||||
session['username'] = user['username']
|
||||
return redirect(url_for('admin_dashboard'))
|
||||
|
||||
return render_template('admin/login.html', error='用户名或密码错误')
|
||||
|
||||
@app.route('/admin/logout')
|
||||
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'))
|
||||
|
||||
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'))
|
||||
|
||||
return render_template('admin/categories.html')
|
||||
|
||||
# ==================== 后台 API ====================
|
||||
|
||||
@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 id, name, url, description, icon, category, is_enabled,
|
||||
sort_order, health_check_url, health_check_enabled
|
||||
FROM services
|
||||
ORDER BY sort_order DESC, 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('''
|
||||
UPDATE categories
|
||||
SET name = ?, sort_order = ?
|
||||
WHERE id = ?
|
||||
''', (
|
||||
data.get('name', ''),
|
||||
data.get('sort_order', 0),
|
||||
category_id
|
||||
))
|
||||
|
||||
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 ====================
|
||||
|
||||
@app.route('/api/admin/change-password', methods=['POST'])
|
||||
def api_admin_change_password():
|
||||
"""修改密码"""
|
||||
if not is_logged_in(session):
|
||||
return jsonify({'error': '未登录'}), 401
|
||||
|
||||
data = request.get_json()
|
||||
old_password = data.get('old_password', '')
|
||||
new_password = data.get('new_password', '')
|
||||
|
||||
if not old_password or not new_password:
|
||||
return jsonify({'error': '密码不能为空'}), 400
|
||||
|
||||
if len(new_password) < 6:
|
||||
return jsonify({'error': '新密码长度至少6位'}), 400
|
||||
|
||||
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': '密码修改成功，请重新登录'})
|
||||
|
||||
# ==================== 主入口 ====================
|
||||
|
||||
if __name__ == '__main__':
|
||||
app.run(
|
||||
host=Config.HOST,
|
||||
port=Config.PORT,
|
||||
debug=Config.DEBUG
|
||||
)
|
||||
Reference in New Issue
Block a user