292 lines
10 KiB
Python
292 lines
10 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
导航网站后端 API (Flask版本)
|
|
"""
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
from datetime import datetime
|
|
|
|
# 添加当前目录到 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')
|
|
|
|
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'
|
|
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
|
|
)
|
|
|
|
# ==================== 分类 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'])
|
|
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'])
|
|
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'])
|
|
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'])
|
|
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'])
|
|
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'])
|
|
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'})
|
|
|
|
if __name__ == '__main__':
|
|
app.run(host='0.0.0.0', port=5003, debug=False)
|