feat: 添加 API Key 发布功能

- 新增 ApiKey 模型，支持 API 密钥认证
- 添加 require_api_key 装饰器实现 API 认证
- 实现 5 个站点管理 API 接口 (GET/POST/PUT/DELETE)
- 添加后台 API Key 管理界面
- 添加数据库迁移脚本

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

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/<code>', 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/<code>', 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/<code>', 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管理'))

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()

models.py

@@ -155,6 +155,46 @@ class Admin(UserMixin, db.Model):
     def __repr__(self):
         return f'<Admin {self.username}>'
 
+
+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'<ApiKey {self.name}>'
+
+    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'

templates/admin/apikey/create.html

@@ -0,0 +1,15 @@
+{% extends 'admin/model/create.html' %}
+
+{% block tail %}
+{{ super() }}
+<script>
+document.addEventListener('DOMContentLoaded', function() {
+    // 创建成功后显示 API Key
+    {% if created_at and model and model.key %}
+    setTimeout(function() {
+        alert('API Key 已创建：\n\n{{ model.key }}\n\n请妥善保存，此密钥只会显示一次！');
+    }, 500);
+    {% endif %}
+});
+</script>
+{% endblock %}

templates/admin/apikey/list.html

@@ -0,0 +1,17 @@
+{% extends 'admin/model/list.html' %}
+
+{% block tail %}
+{{ super() }}
+<style>
+    .api-key-cell {
+        font-family: monospace;
+        font-size: 12px;
+        max-width: 200px;
+        overflow: hidden;
+        text-overflow: ellipsis;
+    }
+    .api-key-masked {
+        color: #6c757d;
+    }
+</style>
+{% endblock %}