492 lines
17 KiB
Python
492 lines
17 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
导航网站后端 API (Flask版本)
|
|
包含用户认证功能
|
|
"""
|
|
import os
|
|
import sys
|
|
import jwt
|
|
import hashlib
|
|
from pathlib import Path
|
|
from datetime import datetime, timedelta
|
|
from functools import wraps
|
|
|
|
# 添加当前目录到 path
|
|
BASE_DIR = Path(__file__).resolve().parent
|
|
sys.path.insert(0, str(BASE_DIR))
|
|
|
|
from flask import Flask, request, jsonify, make_response
|
|
import pymysql
|
|
from pymysql.cursors import DictCursor
|
|
|
|
# 数据库配置
|
|
DB_HOST = os.environ.get('DB_HOST', '192.168.8.160')
|
|
DB_PORT = int(os.environ.get('DB_PORT', '3306'))
|
|
DB_USER = os.environ.get('DB_USER', 'dbuser')
|
|
DB_PASS = os.environ.get('DB_PASS', 'Tata@1234')
|
|
DB_NAME = os.environ.get('DB_NAME', 'favorites_nav')
|
|
|
|
# JWT 配置
|
|
JWT_SECRET = os.environ.get('JWT_SECRET', 'easynavi-secret-key-change-in-production')
|
|
JWT_ALGORITHM = 'HS256'
|
|
JWT_EXPIRATION_HOURS = 24 * 30 # 30天过期
|
|
|
|
app = Flask(__name__)
|
|
|
|
# 简单 CORS 中间件
|
|
@app.after_request
|
|
def add_cors_headers(response):
|
|
response.headers['Access-Control-Allow-Origin'] = '*'
|
|
response.headers['Access-Control-Allow-Methods'] = 'GET, POST, PUT, DELETE, OPTIONS'
|
|
response.headers['Access-Control-Allow-Headers'] = 'Content-Type, Authorization'
|
|
return response
|
|
|
|
def get_db_connection():
|
|
"""获取数据库连接"""
|
|
return pymysql.connect(
|
|
host=DB_HOST,
|
|
port=DB_PORT,
|
|
user=DB_USER,
|
|
password=DB_PASS,
|
|
database=DB_NAME,
|
|
charset='utf8mb4',
|
|
cursorclass=DictCursor
|
|
)
|
|
|
|
# ==================== 密码加密 ====================
|
|
|
|
def hash_password(password):
|
|
"""哈希密码"""
|
|
return hashlib.sha256(password.encode()).hexdigest()
|
|
|
|
def verify_password(password, password_hash):
|
|
"""验证密码"""
|
|
return hashlib.sha256(password.encode()).hexdigest() == password_hash
|
|
|
|
# ==================== JWT 工具 ====================
|
|
|
|
def create_token(user_id, username):
|
|
"""创建 JWT token"""
|
|
payload = {
|
|
'user_id': user_id,
|
|
'username': username,
|
|
'exp': datetime.utcnow() + timedelta(hours=JWT_EXPIRATION_HOURS),
|
|
'iat': datetime.utcnow()
|
|
}
|
|
return jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM)
|
|
|
|
def decode_token(token):
|
|
"""解码 JWT token"""
|
|
try:
|
|
return jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGORITHM])
|
|
except jwt.ExpiredSignatureError:
|
|
return None
|
|
except jwt.InvalidTokenError:
|
|
return None
|
|
|
|
def token_required(f):
|
|
"""装饰器：验证 JWT token"""
|
|
@wraps(f)
|
|
def decorated(*args, **kwargs):
|
|
token = None
|
|
|
|
# 从 header 获取 token
|
|
auth_header = request.headers.get('Authorization')
|
|
if auth_header and auth_header.startswith('Bearer '):
|
|
token = auth_header.split(' ')[1]
|
|
|
|
if not token:
|
|
return jsonify({'error': '未登录', 'message': '请先登录'}), 401
|
|
|
|
payload = decode_token(token)
|
|
if not payload:
|
|
return jsonify({'error': '登录已过期', 'message': '请重新登录'}), 401
|
|
|
|
# 将用户信息传递给路由
|
|
request.user_id = payload.get('user_id')
|
|
request.username = payload.get('username')
|
|
|
|
return f(*args, **kwargs)
|
|
return decorated
|
|
|
|
# ==================== 用户认证 API ====================
|
|
|
|
@app.route('/api/auth/login', methods=['POST'])
|
|
def login():
|
|
"""用户登录"""
|
|
data = request.get_json()
|
|
username = data.get('username', '').strip()
|
|
password = data.get('password', '')
|
|
|
|
if not username or not password:
|
|
return jsonify({'error': '参数错误', 'message': '请输入用户名和密码'}), 400
|
|
|
|
conn = get_db_connection()
|
|
try:
|
|
with conn.cursor() as cursor:
|
|
# 查找用户
|
|
cursor.execute(
|
|
"SELECT id, username, password_hash FROM users WHERE username = %s",
|
|
(username,)
|
|
)
|
|
user = cursor.fetchone()
|
|
|
|
if not user:
|
|
return jsonify({'error': '登录失败', 'message': '用户名或密码错误'}), 401
|
|
|
|
# 验证密码
|
|
if not verify_password(password, user['password_hash']):
|
|
return jsonify({'error': '登录失败', 'message': '用户名或密码错误'}), 401
|
|
|
|
# 创建 token
|
|
token = create_token(user['id'], user['username'])
|
|
|
|
return jsonify({
|
|
'token': token,
|
|
'user_id': user['id'],
|
|
'username': user['username']
|
|
})
|
|
finally:
|
|
conn.close()
|
|
|
|
@app.route('/api/auth/register', methods=['POST'])
|
|
def register():
|
|
"""用户注册"""
|
|
data = request.get_json()
|
|
username = data.get('username', '').strip()
|
|
password = data.get('password', '')
|
|
email = data.get('email', '').strip()
|
|
|
|
if not username or not password:
|
|
return jsonify({'error': '参数错误', 'message': '请输入用户名和密码'}), 400
|
|
|
|
if len(username) < 3:
|
|
return jsonify({'error': '用户名太短', 'message': '用户名至少3个字符'}), 400
|
|
|
|
if len(password) < 6:
|
|
return jsonify({'error': '密码太短', 'message': '密码至少6个字符'}), 400
|
|
|
|
conn = get_db_connection()
|
|
try:
|
|
with conn.cursor() as cursor:
|
|
# 检查用户名是否已存在
|
|
cursor.execute("SELECT id FROM users WHERE username = %s", (username,))
|
|
if cursor.fetchone():
|
|
return jsonify({'error': '注册失败', 'message': '用户名已存在'}), 409
|
|
|
|
# 创建用户
|
|
password_hash = hash_password(password)
|
|
cursor.execute(
|
|
"INSERT INTO users (username, password_hash, email) VALUES (%s, %s, %s)",
|
|
(username, password_hash, email)
|
|
)
|
|
conn.commit()
|
|
user_id = cursor.lastrowid
|
|
|
|
# 创建 token
|
|
token = create_token(user_id, username)
|
|
|
|
return jsonify({
|
|
'token': token,
|
|
'user_id': user_id,
|
|
'username': username
|
|
}), 201
|
|
finally:
|
|
conn.close()
|
|
|
|
@app.route('/api/auth/profile', methods=['GET'])
|
|
@token_required
|
|
def get_profile():
|
|
"""获取当前用户信息"""
|
|
return jsonify({
|
|
'user_id': request.user_id,
|
|
'username': request.username
|
|
})
|
|
|
|
@app.route('/api/auth/logout', methods=['POST'])
|
|
@token_required
|
|
def logout():
|
|
"""退出登录（前端删除 token 即可）"""
|
|
return jsonify({'message': '已退出登录'})
|
|
|
|
# ==================== 分类 API ====================
|
|
|
|
@app.route('/api/categories', methods=['GET'])
|
|
def get_categories():
|
|
"""获取所有分类"""
|
|
conn = get_db_connection()
|
|
try:
|
|
with conn.cursor() as cursor:
|
|
cursor.execute("SELECT * FROM categories ORDER BY sort_order")
|
|
categories = cursor.fetchall()
|
|
# 转换日期时间
|
|
for cat in categories:
|
|
if cat.get('created_at'):
|
|
cat['created_at'] = cat['created_at'].isoformat()
|
|
return jsonify(categories)
|
|
finally:
|
|
conn.close()
|
|
|
|
@app.route('/api/categories', methods=['POST'])
|
|
@token_required
|
|
def create_category():
|
|
"""创建分类（需要登录）"""
|
|
data = request.get_json()
|
|
|
|
conn = get_db_connection()
|
|
try:
|
|
with conn.cursor() as cursor:
|
|
sql = "INSERT INTO categories (name, icon, color, sort_order) VALUES (%s, %s, %s, %s)"
|
|
cursor.execute(sql, (data['name'], data.get('icon', 'folder'), data.get('color', '#6366f1'), data.get('sort_order', 0)))
|
|
conn.commit()
|
|
category_id = cursor.lastrowid
|
|
|
|
cursor.execute("SELECT * FROM categories WHERE id = %s", (category_id,))
|
|
category = cursor.fetchone()
|
|
if category.get('created_at'):
|
|
category['created_at'] = category['created_at'].isoformat()
|
|
return jsonify(category)
|
|
finally:
|
|
conn.close()
|
|
|
|
@app.route('/api/categories/<int:category_id>', methods=['PUT'])
|
|
@token_required
|
|
def update_category(category_id):
|
|
"""更新分类（需要登录）"""
|
|
data = request.get_json()
|
|
|
|
conn = get_db_connection()
|
|
try:
|
|
with conn.cursor() as cursor:
|
|
sql = "UPDATE categories SET name=%s, icon=%s, color=%s, sort_order=%s WHERE id=%s"
|
|
cursor.execute(sql, (data['name'], data.get('icon', 'folder'), data.get('color', '#6366f1'), data.get('sort_order', 0), category_id))
|
|
conn.commit()
|
|
|
|
cursor.execute("SELECT * FROM categories WHERE id = %s", (category_id,))
|
|
category = cursor.fetchone()
|
|
if category.get('created_at'):
|
|
category['created_at'] = category['created_at'].isoformat()
|
|
return jsonify(category)
|
|
finally:
|
|
conn.close()
|
|
|
|
@app.route('/api/categories/<int:category_id>', methods=['DELETE'])
|
|
@token_required
|
|
def delete_category(category_id):
|
|
"""删除分类（需要登录）"""
|
|
conn = get_db_connection()
|
|
try:
|
|
with conn.cursor() as cursor:
|
|
cursor.execute("DELETE FROM categories WHERE id = %s", (category_id,))
|
|
conn.commit()
|
|
return jsonify({'message': '删除成功'})
|
|
finally:
|
|
conn.close()
|
|
|
|
# ==================== 网站 API ====================
|
|
|
|
@app.route('/api/websites', methods=['GET'])
|
|
def get_websites():
|
|
"""获取网站列表"""
|
|
category_id = request.args.get('category_id')
|
|
|
|
conn = get_db_connection()
|
|
try:
|
|
with conn.cursor() as cursor:
|
|
if category_id:
|
|
sql = "SELECT * FROM websites WHERE category_id = %s ORDER BY is_favorite DESC, click_count DESC"
|
|
cursor.execute(sql, (category_id,))
|
|
else:
|
|
sql = "SELECT * FROM websites ORDER BY is_favorite DESC, click_count DESC"
|
|
cursor.execute(sql)
|
|
|
|
websites = cursor.fetchall()
|
|
|
|
# 获取分类信息
|
|
cursor.execute("SELECT * FROM categories")
|
|
categories = {c['id']: c for c in cursor.fetchall()}
|
|
|
|
# 转换日期时间并添加分类信息
|
|
for site in websites:
|
|
if site.get('created_at') and hasattr(site['created_at'], 'isoformat'):
|
|
site['created_at'] = site['created_at'].isoformat()
|
|
if site.get('updated_at') and hasattr(site['updated_at'], 'isoformat'):
|
|
site['updated_at'] = site['updated_at'].isoformat()
|
|
|
|
cat_id = site.get('category_id')
|
|
if cat_id and cat_id in categories:
|
|
cat = categories[cat_id]
|
|
if cat.get('created_at') and hasattr(cat['created_at'], 'isoformat'):
|
|
cat['created_at'] = cat['created_at'].isoformat()
|
|
site['category'] = cat
|
|
else:
|
|
site['category'] = None
|
|
|
|
return jsonify(websites)
|
|
finally:
|
|
conn.close()
|
|
|
|
@app.route('/api/websites', methods=['POST'])
|
|
@token_required
|
|
def create_website():
|
|
"""创建网站（需要登录）"""
|
|
data = request.get_json()
|
|
|
|
conn = get_db_connection()
|
|
try:
|
|
with conn.cursor() as cursor:
|
|
sql = """INSERT INTO websites (name, url, icon, description, category_id, is_favorite)
|
|
VALUES (%s, %s, %s, %s, %s, %s)"""
|
|
cursor.execute(sql, (
|
|
data['name'],
|
|
data['url'],
|
|
data.get('icon', ''),
|
|
data.get('description', ''),
|
|
data.get('category_id') or None,
|
|
data.get('is_favorite', False)
|
|
))
|
|
conn.commit()
|
|
website_id = cursor.lastrowid
|
|
|
|
cursor.execute("SELECT * FROM websites WHERE id = %s", (website_id,))
|
|
site = cursor.fetchone()
|
|
|
|
if site.get('created_at'):
|
|
site['created_at'] = site['created_at'].isoformat()
|
|
if site.get('updated_at'):
|
|
site['updated_at'] = site['updated_at'].isoformat()
|
|
site['category'] = None
|
|
|
|
return jsonify(site)
|
|
finally:
|
|
conn.close()
|
|
|
|
@app.route('/api/websites/<int:website_id>', methods=['PUT'])
|
|
@token_required
|
|
def update_website(website_id):
|
|
"""更新网站（需要登录）"""
|
|
data = request.get_json()
|
|
|
|
conn = get_db_connection()
|
|
try:
|
|
with conn.cursor() as cursor:
|
|
fields = []
|
|
values = []
|
|
|
|
if 'name' in data:
|
|
fields.append('name = %s')
|
|
values.append(data['name'])
|
|
if 'url' in data:
|
|
fields.append('url = %s')
|
|
values.append(data['url'])
|
|
if 'icon' in data:
|
|
fields.append('icon = %s')
|
|
values.append(data['icon'])
|
|
if 'description' in data:
|
|
fields.append('description = %s')
|
|
values.append(data['description'])
|
|
if 'category_id' in data:
|
|
fields.append('category_id = %s')
|
|
values.append(data['category_id'] or None)
|
|
if 'is_favorite' in data:
|
|
fields.append('is_favorite = %s')
|
|
values.append(data['is_favorite'])
|
|
|
|
if fields:
|
|
values.append(website_id)
|
|
sql = f"UPDATE websites SET {', '.join(fields)} WHERE id = %s"
|
|
cursor.execute(sql, values)
|
|
conn.commit()
|
|
|
|
cursor.execute("SELECT * FROM websites WHERE id = %s", (website_id,))
|
|
site = cursor.fetchone()
|
|
|
|
if site.get('created_at'):
|
|
site['created_at'] = site['created_at'].isoformat()
|
|
if site.get('updated_at'):
|
|
site['updated_at'] = site['updated_at'].isoformat()
|
|
|
|
# 获取分类信息
|
|
if site.get('category_id'):
|
|
cursor.execute("SELECT * FROM categories WHERE id = %s", (site['category_id'],))
|
|
cat = cursor.fetchone()
|
|
if cat and cat.get('created_at'):
|
|
cat['created_at'] = cat['created_at'].isoformat()
|
|
site['category'] = cat
|
|
else:
|
|
site['category'] = None
|
|
|
|
return jsonify(site)
|
|
finally:
|
|
conn.close()
|
|
|
|
@app.route('/api/websites/<int:website_id>', methods=['DELETE'])
|
|
@token_required
|
|
def delete_website(website_id):
|
|
"""删除网站（需要登录）"""
|
|
conn = get_db_connection()
|
|
try:
|
|
with conn.cursor() as cursor:
|
|
cursor.execute("DELETE FROM websites WHERE id = %s", (website_id,))
|
|
conn.commit()
|
|
return jsonify({'message': '删除成功'})
|
|
finally:
|
|
conn.close()
|
|
|
|
@app.route('/api/websites/<int:website_id>/click', methods=['POST'])
|
|
def click_website(website_id):
|
|
"""点击网站（无需登录）"""
|
|
conn = get_db_connection()
|
|
try:
|
|
with conn.cursor() as cursor:
|
|
cursor.execute("UPDATE websites SET click_count = click_count + 1 WHERE id = %s", (website_id,))
|
|
conn.commit()
|
|
|
|
cursor.execute("SELECT click_count FROM websites WHERE id = %s", (website_id,))
|
|
result = cursor.fetchone()
|
|
return jsonify({'click_count': result['click_count'] if result else 0})
|
|
finally:
|
|
conn.close()
|
|
|
|
# ==================== 健康检查 ====================
|
|
|
|
@app.route('/api/health', methods=['GET'])
|
|
def health_check():
|
|
return jsonify({'status': 'ok', 'timestamp': datetime.utcnow().isoformat()})
|
|
|
|
@app.route('/', methods=['GET'])
|
|
def root():
|
|
return jsonify({'message': '导航网站 API', 'docs': '/api/health'})
|
|
|
|
# ==================== 启动时创建 users 表 ====================
|
|
|
|
def init_db():
|
|
"""初始化数据库，创建 users 表（如果不存在）"""
|
|
try:
|
|
conn = get_db_connection()
|
|
try:
|
|
with conn.cursor() as cursor:
|
|
cursor.execute("""
|
|
CREATE TABLE IF NOT EXISTS users (
|
|
id INT AUTO_INCREMENT PRIMARY KEY,
|
|
username VARCHAR(50) UNIQUE NOT NULL,
|
|
password_hash VARCHAR(64) NOT NULL,
|
|
email VARCHAR(100),
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
|
INDEX idx_username (username)
|
|
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
|
|
""")
|
|
conn.commit()
|
|
print("Users table initialized successfully")
|
|
finally:
|
|
conn.close()
|
|
except Exception as e:
|
|
print(f"Failed to initialize users table: {e}")
|
|
|
|
# 应用启动时初始化
|
|
with app.app_context():
|
|
init_db()
|
|
|
|
if __name__ == '__main__':
|
|
app.run(host='0.0.0.0', port=5003, debug=False)
|