From 5cef0b94fd4c75568f453ead1cabcf82d71bcec9 Mon Sep 17 00:00:00 2001 From: Jowe <123822645+Selei1983@users.noreply.github.com> Date: Mon, 23 Mar 2026 22:50:03 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=B7=BB=E5=8A=A0=20API=20Key=20?= =?UTF-8?q?=E5=8F=91=E5=B8=83=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 ApiKey 模型，支持 API 密钥认证 - 添加 require_api_key 装饰器实现 API 认证 - 实现 5 个站点管理 API 接口 (GET/POST/PUT/DELETE) - 添加后台 API Key 管理界面 - 添加数据库迁移脚本 Co-Authored-By: Claude Sonnet 4.6 --- app.py | 275 ++++++++++++++++++++++++++++- migrations/add_api_keys_table.py | 27 +++ models.py | 40 +++++ templates/admin/apikey/create.html | 15 ++ templates/admin/apikey/list.html | 17 ++ 5 files changed, 373 insertions(+), 1 deletion(-) create mode 100644 migrations/add_api_keys_table.py create mode 100644 templates/admin/apikey/create.html create mode 100644 templates/admin/apikey/list.html diff --git a/app.py b/app.py index 84f811c..f9631d8 100644 --- a/app.py +++ b/app.py @@ -8,9 +8,10 @@ from flask import Flask, render_template, redirect, url_for, request, flash, jso from flask_login import LoginManager, login_user, logout_user, login_required, current_user from flask_admin import Admin, AdminIndexView, expose from flask_admin.contrib.sqla import ModelView +from wtforms import fields from datetime import datetime, timedelta from config import config -from models import db, Site, Tag, Admin as AdminModel, News, site_tags, PromptTemplate, User, Folder, Collection +from models import db, Site, Tag, Admin as AdminModel, News, site_tags, PromptTemplate, User, Folder, Collection, ApiKey from utils.website_fetcher import WebsiteFetcher from utils.tag_generator import TagGenerator from utils.news_searcher import NewsSearcher @@ -477,6 +478,228 @@ def create_app(config_name='default'): }) + # ========== API Key 认证路由 ========== + def require_api_key(permissions=None): + """API Key 认证装饰器 + + Args: + permissions: 必需的权限列表，如 ['site:read', 'site:write'] + """ + def decorator(f): + from functools import wraps + + @wraps(f) + def decorated_function(*args, **kwargs): + # 从请求头获取 API Key + api_key = request.headers.get('X-API-Key') + + if not api_key: + return jsonify({'success': False, 'message': '缺少 API Key'}), 401 + + # 查找 API Key + key_obj = ApiKey.query.filter_by(key=api_key).first() + + if not key_obj: + return jsonify({'success': False, 'message': '无效的 API Key'}), 401 + + if not key_obj.is_active: + return jsonify({'success': False, 'message': 'API Key 已禁用'}), 401 + + # 检查权限 + if permissions: + for perm in permissions: + if not key_obj.has_permission(perm): + return jsonify({'success': False, 'message': f'权限不足，需要 {perm} 权限'}), 403 + + # 更新使用记录 + key_obj.last_used = datetime.now() + key_obj.usage_count = (key_obj.usage_count or 0) + 1 + db.session.commit() + + # 将 key_obj 传递给被装饰的函数 + kwargs['api_key_obj'] = key_obj + return f(*args, **kwargs) + + return decorated_function + return decorator + + @app.route('/api/key/sites', methods=['GET']) + @require_api_key(['site:read']) + def list_sites_via_api(**kwargs): + """通过 API Key 列出所有网站""" + try: + sites = Site.query.order_by(Site.created_at.desc()).all() + return jsonify({ + 'success': True, + 'sites': [site.to_dict() for site in sites] + }) + except Exception as e: + return jsonify({'success': False, 'message': str(e)}), 500 + + @app.route('/api/key/sites', methods=['POST']) + @require_api_key(['site:write']) + def create_site_via_api(**kwargs): + """通过 API Key 创建网站""" + try: + data = request.get_json() or {} + + # 必填字段验证 + name = data.get('name', '').strip() + url = data.get('url', '').strip() + + if not name: + return jsonify({'success': False, 'message': '请提供网站名称'}), 400 + + if not url: + return jsonify({'success': False, 'message': '请提供网站 URL'}), 400 + + # 创建网站 + site = Site() + site.name = name + site.url = url + site.slug = data.get('slug', '').strip() or None + site.logo = data.get('logo', '').strip() or None + site.short_desc = data.get('short_desc', '').strip() or None + site.description = data.get('description', '').strip() or None + site.features = data.get('features', '').strip() or None + site.news_keywords = data.get('news_keywords', '').strip() or None + site.is_active = data.get('is_active', True) + site.is_recommended = data.get('is_recommended', False) + site.sort_order = data.get('sort_order', 0) + + # 处理标签 + tags_data = data.get('tags', []) + if tags_data: + for tag_name in tags_data: + if isinstance(tag_name, str): + tag_name = tag_name.strip() + if tag_name: + # 查找或创建标签 + tag = Tag.query.filter_by(name=tag_name).first() + if not tag: + # 创建新标签（简化处理，slug 从 name 生成） + import re + from pypinyin import lazy_pinyin + tag_slug = ''.join(lazy_pinyin(tag_name)).lower() + tag_slug = re.sub(r'[^\w\s-]', '', tag_slug) + tag_slug = re.sub(r'[-\s]+', '-', tag_slug).strip('-')[:50] + tag = Tag(name=tag_name, slug=tag_slug) + db.session.add(tag) + if tag not in site.tags: + site.tags.append(tag) + + db.session.add(site) + db.session.commit() + + return jsonify({ + 'success': True, + 'site': site.to_dict() + }), 201 + + except Exception as e: + db.session.rollback() + return jsonify({'success': False, 'message': str(e)}), 500 + + @app.route('/api/key/sites/', methods=['GET']) + @require_api_key(['site:read']) + def get_site_via_api(code, **kwargs): + """通过 API Key 获取单个网站信息""" + site = Site.query.filter_by(code=code).first() + if not site: + return jsonify({'success': False, 'message': '网站不存在'}), 404 + + return jsonify({ + 'success': True, + 'site': site.to_dict() + }) + + @app.route('/api/key/sites/', methods=['PUT']) + @require_api_key(['site:write']) + def update_site_via_api(code, **kwargs): + """通过 API Key 更新网站""" + site = Site.query.filter_by(code=code).first() + if not site: + return jsonify({'success': False, 'message': '网站不存在'}), 404 + + try: + data = request.get_json() or {} + + # 更新字段 + if 'name' in data: + site.name = data['name'].strip() + if 'url' in data: + site.url = data['url'].strip() + if 'slug' in data: + site.slug = data['slug'].strip() if data['slug'] else None + if 'logo' in data: + site.logo = data['logo'].strip() if data['logo'] else None + if 'short_desc' in data: + site.short_desc = data['short_desc'].strip() if data['short_desc'] else None + if 'description' in data: + site.description = data['description'].strip() if data['description'] else None + if 'features' in data: + site.features = data['features'].strip() if data['features'] else None + if 'news_keywords' in data: + site.news_keywords = data['news_keywords'].strip() if data['news_keywords'] else None + if 'is_active' in data: + site.is_active = data['is_active'] + if 'is_recommended' in data: + site.is_recommended = data['is_recommended'] + if 'sort_order' in data: + site.sort_order = data['sort_order'] + + # 处理标签 + if 'tags' in data: + site.tags = [] + tags_data = data['tags'] + if tags_data: + import re + from pypinyin import lazy_pinyin + for tag_name in tags_data: + if isinstance(tag_name, str): + tag_name = tag_name.strip() + if tag_name: + tag = Tag.query.filter_by(name=tag_name).first() + if not tag: + tag_slug = ''.join(lazy_pinyin(tag_name)).lower() + tag_slug = re.sub(r'[^\w\s-]', '', tag_slug) + tag_slug = re.sub(r'[-\s]+', '-', tag_slug).strip('-')[:50] + tag = Tag(name=tag_name, slug=tag_slug) + db.session.add(tag) + if tag not in site.tags: + site.tags.append(tag) + + db.session.commit() + + return jsonify({ + 'success': True, + 'site': site.to_dict() + }) + + except Exception as e: + db.session.rollback() + return jsonify({'success': False, 'message': str(e)}), 500 + + @app.route('/api/key/sites/', methods=['DELETE']) + @require_api_key(['site:write']) + def delete_site_via_api(code, **kwargs): + """通过 API Key 删除网站""" + site = Site.query.filter_by(code=code).first() + if not site: + return jsonify({'success': False, 'message': '网站不存在'}), 404 + + try: + db.session.delete(site) + db.session.commit() + return jsonify({ + 'success': True, + 'message': '网站已删除' + }) + except Exception as e: + db.session.rollback() + return jsonify({'success': False, 'message': str(e)}), 500 + + # ========== 社媒营销路由 (v2.5新增) ========== @app.route('/api/generate-social-share', methods=['POST']) def generate_social_share(): @@ -2857,6 +3080,55 @@ Sitemap: {}sitemap.xml return self.render('admin/index.html', stats=stats, recent_sites=recent_sites) + # API Key 管理视图 + class ApiKeyAdmin(SecureModelView): + can_edit = True + can_delete = True + can_create = True + can_view_details = False + can_set_page_size = True + page_size = 20 + + column_display_actions = True + + column_list = ['id', 'name', 'key', 'permissions', 'is_active', 'usage_count', 'last_used', 'created_at'] + column_default_sort = ('created_at', True) + column_searchable_list = ['name', 'key'] + column_filters = ['is_active'] + column_labels = { + 'id': 'ID', + 'name': '密钥名称', + 'key': 'API Key', + 'permissions': '权限', + 'is_active': '是否启用', + 'usage_count': '使用次数', + 'last_used': '最后使用', + 'created_at': '创建时间' + } + + # 表单配置 + form_columns = ['name', 'permissions', 'is_active'] + + form_extra_fields = { + 'new_key': fields.StringField('新密钥（留空则自动生成）') + } + + def on_model_change(self, form, model, is_created): + """创建时自动生成 API Key""" + if is_created: + # 生成 64 位随机密钥 + model.key = secrets.token_urlsafe(32) # 生成约 43 字符，改用 hex + import secrets as sec + model.key = sec.token_hex(32) # 64 位十六进制字符串 + + def _form_validate(self, form): + """表单验证""" + return super()._form_validate(form) + + # 自定义列表模板，隐藏 key 字段但在创建时显示 + list_template = 'admin/apikey/list.html' + create_template = 'admin/apikey/create.html' + # 网站管理视图 class SiteAdmin(SecureModelView): # 自定义模板 @@ -3154,6 +3426,7 @@ Sitemap: {}sitemap.xml ) admin.add_view(SiteAdmin(Site, db.session, name='网站管理')) + admin.add_view(ApiKeyAdmin(ApiKey, db.session, name='API密钥', endpoint='api_keys')) admin.add_view(TagAdmin(Tag, db.session, name='标签管理')) admin.add_view(NewsAdmin(News, db.session, name='新闻管理')) admin.add_view(PromptAdmin(PromptTemplate, db.session, name='Prompt管理')) diff --git a/migrations/add_api_keys_table.py b/migrations/add_api_keys_table.py new file mode 100644 index 0000000..e880397 --- /dev/null +++ b/migrations/add_api_keys_table.py @@ -0,0 +1,27 @@ +""" +API Keys 表迁移脚本 + +执行方式: python migrations/add_api_keys_table.py +""" +import os +import sys + +# 添加项目根目录到路径 +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from app import create_app, db +from models import ApiKey + + +def create_api_keys_table(): + """创建 api_keys 表""" + app = create_app('development') + + with app.app_context(): + # 直接创建表 + db.create_all() + print('api_keys 表创建成功') + + +if __name__ == '__main__': + create_api_keys_table() \ No newline at end of file diff --git a/models.py b/models.py index 0fccf5d..868a434 100644 --- a/models.py +++ b/models.py @@ -155,6 +155,46 @@ class Admin(UserMixin, db.Model): def __repr__(self): return f'' + +class ApiKey(db.Model): + """API密钥模型""" + __tablename__ = 'api_keys' + + id = db.Column(db.Integer, primary_key=True) + key = db.Column(db.String(64), unique=True, nullable=False, comment='API密钥') + name = db.Column(db.String(50), nullable=False, comment='密钥名称') + permissions = db.Column(db.String(200), default='site:read,site:write', comment='权限列表') + is_active = db.Column(db.Boolean, default=True, comment='是否启用') + created_at = db.Column(db.DateTime, default=datetime.now, comment='创建时间') + last_used = db.Column(db.DateTime, comment='最后使用时间') + usage_count = db.Column(db.Integer, default=0, comment='使用次数') + + def __repr__(self): + return f'' + + def has_permission(self, permission): + """检查是否拥有指定权限""" + if not self.is_active: + return False + perms = self.permissions.split(',') if self.permissions else [] + return permission.strip() in [p.strip() for p in perms] + + def to_dict(self, include_key=False): + """转换为字典""" + result = { + 'id': self.id, + 'name': self.name, + 'permissions': self.permissions, + 'is_active': self.is_active, + 'created_at': self.created_at.strftime('%Y-%m-%d %H:%M:%S') if self.created_at else None, + 'last_used': self.last_used.strftime('%Y-%m-%d %H:%M:%S') if self.last_used else None, + 'usage_count': self.usage_count + } + if include_key: + result['key'] = self.key + return result + + class PromptTemplate(db.Model): """AI提示词模板模型""" __tablename__ = 'prompt_templates' diff --git a/templates/admin/apikey/create.html b/templates/admin/apikey/create.html new file mode 100644 index 0000000..6165810 --- /dev/null +++ b/templates/admin/apikey/create.html @@ -0,0 +1,15 @@ +{% extends 'admin/model/create.html' %} + +{% block tail %} +{{ super() }} + +{% endblock %} \ No newline at end of file diff --git a/templates/admin/apikey/list.html b/templates/admin/apikey/list.html new file mode 100644 index 0000000..933de83 --- /dev/null +++ b/templates/admin/apikey/list.html @@ -0,0 +1,17 @@ +{% extends 'admin/model/list.html' %} + +{% block tail %} +{{ super() }} + +{% endblock %} \ No newline at end of file