import os from flask import Flask, render_template, redirect, url_for, request, flash, jsonify from flask_login import LoginManager, login_user, logout_user, login_required, current_user from flask_admin import Admin, AdminIndexView from flask_admin.contrib.sqla import ModelView from datetime import datetime from config import config from models import db, Site, Tag, Admin as AdminModel from utils.website_fetcher import WebsiteFetcher def create_app(config_name='default'): """应用工厂函数""" app = Flask(__name__) # 加载配置 app.config.from_object(config[config_name]) # 初始化数据库 db.init_app(app) # 初始化登录管理 login_manager = LoginManager() login_manager.init_app(app) login_manager.login_view = 'admin_login' login_manager.login_message = '请先登录' @login_manager.user_loader def load_user(user_id): return AdminModel.query.get(int(user_id)) # ========== 前台路由 ========== @app.route('/') def index(): """首页""" # 获取所有启用的标签 tags = Tag.query.order_by(Tag.sort_order.desc(), Tag.id).all() # 获取筛选的标签 tag_slug = request.args.get('tag') selected_tag = None if tag_slug: selected_tag = Tag.query.filter_by(slug=tag_slug).first() if selected_tag: sites = Site.query.filter( Site.is_active == True, Site.tags.contains(selected_tag) ).order_by(Site.sort_order.desc(), Site.id.desc()).all() else: sites = [] else: # 获取所有启用的网站 sites = Site.query.filter_by(is_active=True).order_by( Site.sort_order.desc(), Site.id.desc() ).all() return render_template('index.html', sites=sites, tags=tags, selected_tag=selected_tag) @app.route('/site/') def site_detail(slug): """网站详情页""" site = Site.query.filter_by(slug=slug, is_active=True).first_or_404() # 增加浏览次数 site.view_count += 1 db.session.commit() return render_template('detail.html', site=site) # ========== 后台登录路由 ========== @app.route('/admin/login', methods=['GET', 'POST']) def admin_login(): """管理员登录""" if current_user.is_authenticated: return redirect(url_for('admin.index')) if request.method == 'POST': username = request.form.get('username') password = request.form.get('password') admin = AdminModel.query.filter_by(username=username).first() if admin and admin.check_password(password) and admin.is_active: login_user(admin) admin.last_login = datetime.now() db.session.commit() return redirect(url_for('admin.index')) else: flash('用户名或密码错误', 'error') return render_template('admin_login.html') @app.route('/admin/logout') @login_required def admin_logout(): """管理员登出""" logout_user() return redirect(url_for('index')) # ========== API路由 ========== @app.route('/api/fetch-website-info', methods=['POST']) @login_required def fetch_website_info(): """抓取网站信息API""" try: data = request.get_json() url = data.get('url', '').strip() if not url: return jsonify({ 'success': False, 'message': '请提供网站URL' }), 400 # 创建抓取器 fetcher = WebsiteFetcher(timeout=15) # 抓取网站信息 info = fetcher.fetch_website_info(url) if not info: return jsonify({ 'success': False, 'message': '无法获取网站信息,请检查URL是否正确或手动填写' }) # 下载Logo(如果有) logo_path = None if info.get('logo_url'): logo_path = fetcher.download_logo(info['logo_url']) return jsonify({ 'success': True, 'data': { 'name': info.get('name', ''), 'description': info.get('description', ''), 'logo': logo_path or info.get('logo_url', '') } }) except Exception as e: return jsonify({ 'success': False, 'message': f'抓取失败: {str(e)}' }), 500 # ========== Flask-Admin 配置 ========== class SecureModelView(ModelView): """需要登录的模型视图""" def is_accessible(self): return current_user.is_authenticated def inaccessible_callback(self, name, **kwargs): return redirect(url_for('admin_login')) class SecureAdminIndexView(AdminIndexView): """需要登录的管理首页""" def is_accessible(self): return current_user.is_authenticated def inaccessible_callback(self, name, **kwargs): return redirect(url_for('admin_login')) # 网站管理视图 class SiteAdmin(SecureModelView): # 自定义模板 create_template = 'admin/site/create.html' edit_template = 'admin/site/edit.html' column_list = ['id', 'name', 'url', 'slug', 'is_active', 'view_count', 'created_at'] column_searchable_list = ['name', 'url', 'description'] column_filters = ['is_active', 'tags'] column_labels = { 'id': 'ID', 'name': '网站名称', 'url': 'URL', 'slug': 'URL别名', 'logo': 'Logo', 'short_desc': '简短描述', 'description': '详细介绍', 'features': '主要功能', 'is_active': '是否启用', 'view_count': '浏览次数', 'sort_order': '排序权重', 'tags': '标签', 'created_at': '创建时间', 'updated_at': '更新时间' } form_columns = ['name', 'url', 'slug', 'logo', 'short_desc', 'description', 'features', 'tags', 'is_active', 'sort_order'] # 标签管理视图 class TagAdmin(SecureModelView): column_list = ['id', 'name', 'slug', 'description', 'sort_order'] column_searchable_list = ['name', 'description'] column_labels = { 'id': 'ID', 'name': '标签名称', 'slug': 'URL别名', 'description': '标签描述', 'icon': '图标', 'sort_order': '排序权重', 'created_at': '创建时间' } form_columns = ['name', 'slug', 'description', 'icon', 'sort_order'] # 管理员视图 class AdminAdmin(SecureModelView): column_list = ['id', 'username', 'email', 'is_active', 'last_login', 'created_at'] column_searchable_list = ['username', 'email'] column_filters = ['is_active'] column_labels = { 'id': 'ID', 'username': '用户名', 'email': '邮箱', 'is_active': '是否启用', 'created_at': '创建时间', 'last_login': '最后登录' } form_columns = ['username', 'email', 'is_active'] def on_model_change(self, form, model, is_created): # 如果是新建管理员,设置默认密码 if is_created: model.set_password('admin123') # 默认密码 # 初始化 Flask-Admin admin = Admin( app, name='ZJPB 焦提示词 - 后台管理', template_mode='bootstrap4', index_view=SecureAdminIndexView(), base_template='admin/custom_base.html' ) admin.add_view(SiteAdmin(Site, db.session, name='网站管理')) admin.add_view(TagAdmin(Tag, db.session, name='标签管理')) admin.add_view(AdminAdmin(AdminModel, db.session, name='管理员', endpoint='admin_users')) return app if __name__ == '__main__': app = create_app(os.getenv('FLASK_ENV', 'development')) app.run(host='0.0.0.0', port=5000, debug=True)