hz4th_coder
e689fed71e
功能特性: - 前端文章展示 (Markdown渲染、代码高亮) - 后台管理系统 (文章、分类、标签、作者管理) - 图片上传 (点击、拖拽、粘贴) - 网站设置 (Logo、底部宣传图片上传) - 访问统计 (IP记录、趋势图表) - 参考来源、附件支持 - 单端口部署 (16012) 技术栈: - Flask 3.0 + SQLAlchemy - SQLite数据库 - marked.js + highlight.js (Markdown渲染) - Chart.js (统计图表)
137 lines
4.0 KiB
Python
137 lines
4.0 KiB
Python
"""Flask应用工厂"""
|
|
from flask import Flask, g, request, send_from_directory
|
|
from flask_sqlalchemy import SQLAlchemy
|
|
from flask_migrate import Migrate
|
|
import os
|
|
|
|
db = SQLAlchemy()
|
|
migrate = Migrate()
|
|
|
|
|
|
def create_app(config_name='default'):
|
|
"""创建Flask应用实例"""
|
|
from app.config import config_map
|
|
|
|
app = Flask(__name__)
|
|
app.config.from_object(config_map[config_name])
|
|
|
|
# 初始化扩展
|
|
db.init_app(app)
|
|
migrate.init_app(app, db)
|
|
|
|
# 确保数据目录和上传目录存在
|
|
data_dir = os.path.join(app.root_path, '..', 'data')
|
|
uploads_dir = os.path.join(app.root_path, '..', 'uploads')
|
|
os.makedirs(data_dir, exist_ok=True)
|
|
os.makedirs(uploads_dir, exist_ok=True)
|
|
|
|
# 配置上传目录
|
|
app.config['UPLOAD_FOLDER'] = uploads_dir
|
|
|
|
# 注册蓝图
|
|
from app.routes.frontend import frontend_bp
|
|
from app.routes.admin import admin_bp
|
|
|
|
app.register_blueprint(frontend_bp)
|
|
app.register_blueprint(admin_bp, url_prefix='/admin')
|
|
|
|
# 注册上传文件路由（在蓝图之前，确保全局可访问）
|
|
@app.route('/uploads/<path:filename>')
|
|
def serve_upload(filename):
|
|
"""提供上传文件访问"""
|
|
return send_from_directory(uploads_dir, filename)
|
|
|
|
# 注册模板过滤器
|
|
register_filters(app)
|
|
|
|
# 注册命令
|
|
register_commands(app)
|
|
|
|
# 注册请求钩子（访问统计和网站配置）
|
|
register_hooks(app)
|
|
|
|
return app
|
|
|
|
|
|
def register_filters(app):
|
|
"""注册自定义模板过滤器"""
|
|
from datetime import datetime
|
|
|
|
@app.template_filter('datetime_format')
|
|
def datetime_format(value, format='%Y-%m-%d %H:%M'):
|
|
"""格式化日期时间"""
|
|
if value is None:
|
|
return ''
|
|
return value.strftime(format)
|
|
|
|
@app.template_filter('truncate_content')
|
|
def truncate_content(value, length=200):
|
|
"""截断文章内容用于预览"""
|
|
if len(value) <= length:
|
|
return value
|
|
return value[:length] + '...'
|
|
|
|
|
|
def register_commands(app):
|
|
"""注册自定义命令"""
|
|
import click
|
|
|
|
@app.cli.command('init-db')
|
|
def init_db():
|
|
"""初始化数据库"""
|
|
db.create_all()
|
|
click.echo('数据库初始化完成。')
|
|
|
|
@app.cli.command('create-admin')
|
|
@click.argument('username')
|
|
@click.argument('password')
|
|
def create_admin(username, password):
|
|
"""创建管理员账户"""
|
|
from app.models.admin import Admin
|
|
from werkzeug.security import generate_password_hash
|
|
|
|
admin = Admin.query.filter_by(username=username).first()
|
|
if admin:
|
|
click.echo(f'管理员 {username} 已存在！')
|
|
return
|
|
|
|
admin = Admin(
|
|
username=username,
|
|
password_hash=generate_password_hash(password)
|
|
)
|
|
db.session.add(admin)
|
|
db.session.commit()
|
|
click.echo(f'管理员 {username} 创建成功！')
|
|
|
|
|
|
def register_hooks(app):
|
|
"""注册请求钩子"""
|
|
from app.services import SiteConfigService, VisitService
|
|
|
|
@app.before_request
|
|
def before_request():
|
|
"""请求前处理：注入网站配置"""
|
|
# 获取网站配置并注入到g对象
|
|
g.site_config = SiteConfigService.get_dict()
|
|
|
|
@app.after_request
|
|
def after_request(response):
|
|
"""请求后处理：记录访问日志"""
|
|
# 只记录前端请求，不记录静态资源和后台
|
|
if request.path.startswith('/static') or request.path.startswith('/admin'):
|
|
return response
|
|
|
|
# 只记录成功的请求
|
|
if response.status_code == 200:
|
|
# 检查是否是文章详情页
|
|
article_id = None
|
|
if hasattr(g, 'current_article_id'):
|
|
article_id = g.current_article_id
|
|
|
|
# 记录访问
|
|
try:
|
|
VisitService.log_visit(request, article_id)
|
|
except Exception:
|
|
pass # 忽略记录错误，不影响正常响应
|
|
|
|
return response |