1 Commits
master ... dev

Author SHA1 Message Date
503a9780b2 refactor: 拆分admin路由为模块化结构,端口改为16013
- 将 routes/admin.py (672行) 拆分为 routes/admin/ 包
- 按功能模块拆分:auth, dashboard, articles, categories, tags, settings, stats, upload, authors
- 端口从 16012 改为 16013
- 初始化数据库和管理员账户
2026-06-25 11:21:36 +08:00
12 changed files with 683 additions and 674 deletions

View File

@@ -1,673 +0,0 @@
"""后台管理路由"""
from flask import Blueprint, render_template, redirect, url_for, request, flash, session, g, jsonify, send_from_directory
from functools import wraps
import os
import json
from app.services import ArticleService, CategoryService, TagService, AdminService, SiteConfigService, VisitService, UploadService, AuthorService, ReferenceService, AttachmentService
admin_bp = Blueprint('admin', __name__)
def login_required(f):
"""登录验证装饰器"""
@wraps(f)
def decorated_function(*args, **kwargs):
if 'admin_id' not in session:
flash('请先登录', 'warning')
return redirect(url_for('admin.login'))
return f(*args, **kwargs)
return decorated_function
# ==================== 登录登出 ====================
@admin_bp.route('/login', methods=['GET', 'POST'])
def login():
"""登录页面"""
if request.method == 'POST':
username = request.form.get('username')
password = request.form.get('password')
admin = AdminService.authenticate(username, password)
if admin:
session['admin_id'] = admin.id
session['admin_username'] = admin.username
flash('登录成功', 'success')
return redirect(url_for('admin.dashboard'))
else:
flash('用户名或密码错误', 'danger')
return render_template('admin/login.html')
@admin_bp.route('/logout')
def logout():
"""登出"""
session.clear()
flash('已退出登录', 'info')
return redirect(url_for('admin.login'))
# ==================== 仪表盘 ====================
@admin_bp.route('/')
@admin_bp.route('/dashboard')
@login_required
def dashboard():
"""仪表盘"""
# 获取统计数据
articles = ArticleService.get_all_list(per_page=5)
categories = CategoryService.get_all()
tags = TagService.get_all()
stats = {
'total_articles': len(articles.items),
'published_articles': sum(1 for a in articles.items if a.is_published),
'total_categories': len(categories),
'total_tags': len(tags),
'total_views': sum(a.view_count for a in articles.items)
}
return render_template('admin/dashboard.html', stats=stats, recent_articles=articles)
# ==================== 文章管理 ====================
@admin_bp.route('/articles')
@login_required
def articles():
"""文章列表"""
page = request.args.get('page', 1, type=int)
is_published = request.args.get('status')
if is_published == 'published':
is_published_filter = True
elif is_published == 'draft':
is_published_filter = False
else:
is_published_filter = None
articles = ArticleService.get_all_list(
page=page,
per_page=20,
is_published=is_published_filter
)
categories = CategoryService.get_all()
tags = TagService.get_all()
return render_template('admin/articles.html',
articles=articles,
categories=categories,
tags=tags,
current_status=is_published)
@admin_bp.route('/articles/create', methods=['GET', 'POST'])
@login_required
def article_create():
"""创建文章"""
if request.method == 'POST':
title = request.form.get('title')
content = request.form.get('content')
summary = request.form.get('summary')
author_id = request.form.get('author_id', type=int)
category_id = request.form.get('category_id', type=int)
cover_image = request.form.get('cover_image')
is_published = request.form.get('is_published') == 'on'
tag_ids = request.form.getlist('tags')
tag_ids = [int(t) for t in tag_ids if t]
# 处理参考来源
references_json = request.form.get('references', '[]')
references_data = json.loads(references_json) if references_json else []
article = ArticleService.create(
title=title,
content=content,
summary=summary,
author_id=author_id,
category_id=category_id,
tag_ids=tag_ids,
cover_image=cover_image,
is_published=is_published
)
# 添加参考来源
for ref in references_data:
if ref.get('title'):
ReferenceService.add(article.id, ref['title'], ref.get('url'), ref.get('description'))
flash('文章创建成功', 'success')
return redirect(url_for('admin.article_edit', article_id=article.id))
categories = CategoryService.get_all()
tags = TagService.get_all()
authors = AuthorService.get_all()
return render_template('admin/article_form.html',
article=None,
categories=categories,
tags=tags,
authors=authors,
references=[],
attachments=[])
@admin_bp.route('/articles/<int:article_id>/edit', methods=['GET', 'POST'])
@login_required
def article_edit(article_id):
"""编辑文章"""
article = ArticleService.get_by_id(article_id)
if not article:
flash('文章不存在', 'danger')
return redirect(url_for('admin.articles'))
if request.method == 'POST':
title = request.form.get('title')
content = request.form.get('content')
summary = request.form.get('summary')
author_id = request.form.get('author_id', type=int)
category_id = request.form.get('category_id', type=int)
cover_image = request.form.get('cover_image')
is_published = request.form.get('is_published') == 'on'
tag_ids = request.form.getlist('tags')
tag_ids = [int(t) for t in tag_ids if t]
# 处理参考来源
references_json = request.form.get('references', '[]')
references_data = json.loads(references_json) if references_json else []
ArticleService.update(
article_id,
title=title,
content=content,
summary=summary,
author_id=author_id,
category_id=category_id,
cover_image=cover_image,
is_published=is_published,
tag_ids=tag_ids
)
# 更新参考来源
ReferenceService.update_article_references(article_id, references_data)
flash('文章更新成功', 'success')
return redirect(url_for('admin.article_edit', article_id=article_id))
categories = CategoryService.get_all()
tags = TagService.get_all()
authors = AuthorService.get_all()
references = ReferenceService.get_by_article(article_id)
attachments = AttachmentService.get_by_article(article_id)
return render_template('admin/article_form.html',
article=article,
categories=categories,
tags=tags,
authors=authors,
references=references,
attachments=attachments)
@admin_bp.route('/articles/<int:article_id>/delete', methods=['POST'])
@login_required
def article_delete(article_id):
"""删除文章"""
if ArticleService.delete(article_id):
flash('文章已删除', 'success')
else:
flash('删除失败', 'danger')
return redirect(url_for('admin.articles'))
@admin_bp.route('/articles/<int:article_id>/toggle-publish', methods=['POST'])
@login_required
def article_toggle_publish(article_id):
"""切换文章发布状态"""
article = ArticleService.get_by_id(article_id)
if article:
ArticleService.update(article_id, is_published=not article.is_published)
status = '发布' if not article.is_published else '取消发布'
flash(f'文章已{status}', 'success')
else:
flash('文章不存在', 'danger')
return redirect(url_for('admin.articles'))
# ==================== 分类管理 ====================
@admin_bp.route('/categories')
@login_required
def categories():
"""分类列表"""
categories = CategoryService.get_all()
return render_template('admin/categories.html', categories=categories)
@admin_bp.route('/categories/create', methods=['GET', 'POST'])
@login_required
def category_create():
"""创建分类"""
if request.method == 'POST':
name = request.form.get('name')
description = request.form.get('description')
sort_order = request.form.get('sort_order', 0, type=int)
CategoryService.create(name=name, description=description, sort_order=sort_order)
flash('分类创建成功', 'success')
return redirect(url_for('admin.categories'))
return render_template('admin/category_form.html', category=None)
@admin_bp.route('/categories/<int:category_id>/edit', methods=['GET', 'POST'])
@login_required
def category_edit(category_id):
"""编辑分类"""
category = CategoryService.get_by_id(category_id)
if not category:
flash('分类不存在', 'danger')
return redirect(url_for('admin.categories'))
if request.method == 'POST':
name = request.form.get('name')
description = request.form.get('description')
sort_order = request.form.get('sort_order', 0, type=int)
CategoryService.update(category_id, name=name, description=description, sort_order=sort_order)
flash('分类更新成功', 'success')
return redirect(url_for('admin.categories'))
return render_template('admin/category_form.html', category=category)
@admin_bp.route('/categories/<int:category_id>/delete', methods=['POST'])
@login_required
def category_delete(category_id):
"""删除分类"""
if CategoryService.delete(category_id):
flash('分类已删除', 'success')
else:
flash('删除失败', 'danger')
return redirect(url_for('admin.categories'))
# ==================== 标签管理 ====================
@admin_bp.route('/tags')
@login_required
def tags():
"""标签列表"""
tags = TagService.get_all()
return render_template('admin/tags.html', tags=tags)
@admin_bp.route('/tags/create', methods=['GET', 'POST'])
@login_required
def tag_create():
"""创建标签"""
if request.method == 'POST':
name = request.form.get('name')
color = request.form.get('color', '#007bff')
TagService.create(name=name, color=color)
flash('标签创建成功', 'success')
return redirect(url_for('admin.tags'))
return render_template('admin/tag_form.html', tag=None)
@admin_bp.route('/tags/<int:tag_id>/edit', methods=['GET', 'POST'])
@login_required
def tag_edit(tag_id):
"""编辑标签"""
tag = TagService.get_by_id(tag_id)
if not tag:
flash('标签不存在', 'danger')
return redirect(url_for('admin.tags'))
if request.method == 'POST':
name = request.form.get('name')
color = request.form.get('color')
TagService.update(tag_id, name=name, color=color)
flash('标签更新成功', 'success')
return redirect(url_for('admin.tags'))
return render_template('admin/tag_form.html', tag=tag)
@admin_bp.route('/tags/<int:tag_id>/delete', methods=['POST'])
@login_required
def tag_delete(tag_id):
"""删除标签"""
if TagService.delete(tag_id):
flash('标签已删除', 'success')
else:
flash('删除失败', 'danger')
return redirect(url_for('admin.tags'))
# ==================== 网站设置 ====================
@admin_bp.route('/settings', methods=['GET', 'POST'])
@login_required
def settings():
"""网站设置"""
if request.method == 'POST':
# 处理图片上传
promo_image_1 = request.form.get('promo_image_1', '')
promo_image_2 = request.form.get('promo_image_2', '')
promo_image_3 = request.form.get('promo_image_3', '')
# 上传图片1
if 'promo_image_1_file' in request.files:
file = request.files['promo_image_1_file']
if file and file.filename:
image, error = UploadService.upload_image(file)
if image:
promo_image_1 = UploadService.get_image_url(image.filename)
# 上传图片2
if 'promo_image_2_file' in request.files:
file = request.files['promo_image_2_file']
if file and file.filename:
image, error = UploadService.upload_image(file)
if image:
promo_image_2 = UploadService.get_image_url(image.filename)
# 上传图片3
if 'promo_image_3_file' in request.files:
file = request.files['promo_image_3_file']
if file and file.filename:
image, error = UploadService.upload_image(file)
if image:
promo_image_3 = UploadService.get_image_url(image.filename)
# 更新设置
SiteConfigService.update(
site_name=request.form.get('site_name', '技术博客'),
site_subtitle=request.form.get('site_subtitle', ''),
site_logo=request.form.get('site_logo', ''),
site_keywords=request.form.get('site_keywords', ''),
site_description=request.form.get('site_description', ''),
footer_text=request.form.get('footer_text', ''),
footer_copyright=request.form.get('footer_copyright', ''),
footer_technology=request.form.get('footer_technology', ''),
promo_image_1=promo_image_1,
promo_image_1_title=request.form.get('promo_image_1_title', ''),
promo_image_1_link=request.form.get('promo_image_1_link', ''),
promo_image_2=promo_image_2,
promo_image_2_title=request.form.get('promo_image_2_title', ''),
promo_image_2_link=request.form.get('promo_image_2_link', ''),
promo_image_3=promo_image_3,
promo_image_3_title=request.form.get('promo_image_3_title', ''),
contact_email=request.form.get('contact_email', ''),
contact_github=request.form.get('contact_github', ''),
articles_per_page=request.form.get('articles_per_page', 10, type=int),
)
flash('网站设置已更新', 'success')
return redirect(url_for('admin.settings'))
config = SiteConfigService.get_config()
return render_template('admin/settings.html', config=config)
# ==================== 访问统计 ====================
@admin_bp.route('/stats')
@login_required
def stats():
"""访问统计概览"""
stats_summary = VisitService.get_stats_summary()
top_ips = VisitService.get_top_ips(10)
top_articles = VisitService.get_top_articles(7, 10)
top_paths = VisitService.get_top_paths(20)
return render_template('admin/stats.html',
stats=stats_summary,
top_ips=top_ips,
top_articles=top_articles,
top_paths=top_paths)
@admin_bp.route('/stats/logs')
@login_required
def stats_logs():
"""访问日志列表"""
page = request.args.get('page', 1, type=int)
path_filter = request.args.get('path', '')
logs = VisitService.get_visit_list(page=page, per_page=50, path_filter=path_filter)
return render_template('admin/stats_logs.html',
logs=logs,
path_filter=path_filter)
@admin_bp.route('/stats/charts')
@login_required
def stats_charts():
"""访问统计图表"""
stats_summary = VisitService.get_stats_summary()
# 准备图表数据
daily_data = []
for stat in stats_summary['daily_stats']:
daily_data.append({
'date': stat.date.strftime('%Y-%m-%d'),
'visits': stat.total_visits,
'unique_ips': stat.unique_ips,
'article_views': stat.article_views
})
return render_template('admin/stats_charts.html',
daily_data=daily_data,
stats=stats_summary)
@admin_bp.route('/stats/clear', methods=['POST'])
@login_required
def stats_clear():
"""清理旧访问日志"""
days = request.form.get('days', 90, type=int)
deleted = VisitService.clear_old_logs(days)
flash(f'已清理 {deleted} 条旧日志记录', 'success')
return redirect(url_for('admin.stats_logs'))
# ==================== 图片上传 ====================
@admin_bp.route('/upload/image', methods=['POST'])
@login_required
def upload_image():
"""上传图片(支持拖拽、粘贴、选择)"""
from flask import current_app
article_id = request.form.get('article_id', type=int)
uploader = session.get('admin_username', 'admin')
# 检查是否有文件上传
if 'file' not in request.files:
return jsonify({'success': False, 'message': '没有选择文件'}), 400
file = request.files['file']
if file.filename == '':
return jsonify({'success': False, 'message': '没有选择文件'}), 400
# 上传图片
image, error = UploadService.upload_image(file, article_id, uploader)
if error:
return jsonify({'success': False, 'message': error}), 400
# 返回图片URL
return jsonify({
'success': True,
'url': UploadService.get_image_url(image.filename),
'filename': image.filename
})
@admin_bp.route('/upload/paste', methods=['POST'])
@login_required
def upload_paste():
"""粘贴图片上传"""
from flask import current_app
import base64
from io import BytesIO
article_id = request.form.get('article_id', type=int)
uploader = session.get('admin_username', 'admin')
# 获取粘贴的图片数据
image_data = request.form.get('image_data')
if not image_data:
return jsonify({'success': False, 'message': '没有图片数据'}), 400
# 解码Base64图片
try:
# 移除data:image/xxx;base64,前缀
if 'base64,' in image_data:
image_data = image_data.split('base64,')[1]
image_bytes = base64.b64decode(image_data)
image_file = BytesIO(image_bytes)
# 确定图片格式
format_map = {'png': 'png', 'jpeg': 'jpg', 'jpg': 'jpg', 'gif': 'gif', 'webpo': 'webpo'}
# 从数据头识别格式(简化处理)
ext = 'png' # 默认png
# 生成文件名
from datetime import datetime
import uuid
filename = f'{datetime.now().strftime("%Y%m%d")}_{uuid.uuid4().hex[:8]}.{ext}'
# 保存文件
upload_folder = UploadService.get_upload_folder()
filepath = os.path.join(upload_folder, filename)
with open(filepath, 'wb') as f:
f.write(image_bytes)
# 记录到数据库
from app.models import UploadedImage
image = UploadedImage(
filename=filename,
filepath=filepath,
article_id=article_id,
uploader=uploader
)
db.session.add(image)
db.session.commit()
return jsonify({
'success': True,
'url': UploadService.get_image_url(filename),
'filename': filename
})
except Exception as e:
return jsonify({'success': False, 'message': str(e)}), 400
@admin_bp.route('/upload/attachment', methods=['POST'])
@login_required
def upload_attachment():
"""上传附件"""
article_id = request.form.get('article_id', type=int)
description = request.form.get('description', '')
if 'file' not in request.files:
return jsonify({'success': False, 'message': '没有选择文件'}), 400
file = request.files['file']
if file.filename == '':
return jsonify({'success': False, 'message': '没有选择文件'}), 400
attachment, error = UploadService.upload_attachment(file, article_id, description)
if error:
return jsonify({'success': False, 'message': error}), 400
return jsonify({
'success': True,
'attachment': {
'id': attachment.id,
'filename': attachment.filename,
'filesize': attachment.get_filesize_display(),
'url': f'/uploads/{attachment.filename}'
}
})
@admin_bp.route('/attachment/<int:attachment_id>/delete', methods=['POST'])
@login_required
def delete_attachment(attachment_id):
"""删除附件"""
if AttachmentService.delete_attachment(attachment_id):
return jsonify({'success': True})
return jsonify({'success': False, 'message': '删除失败'}), 400
# ==================== 作者管理 ====================
@admin_bp.route('/authors')
@login_required
def authors():
"""作者列表"""
authors = AuthorService.get_all()
return render_template('admin/authors.html', authors=authors)
@admin_bp.route('/authors/create', methods=['GET', 'POST'])
@login_required
def author_create():
"""创建作者"""
if request.method == 'POST':
name = request.form.get('name')
avatar = request.form.get('avatar')
bio = request.form.get('bio')
website = request.form.get('website')
AuthorService.create(name=name, avatar=avatar, bio=bio, website=website)
flash('作者创建成功', 'success')
return redirect(url_for('admin.authors'))
return render_template('admin/author_form.html', author=None)
@admin_bp.route('/authors/<int:author_id>/edit', methods=['GET', 'POST'])
@login_required
def author_edit(author_id):
"""编辑作者"""
author = AuthorService.get_by_id(author_id)
if not author:
flash('作者不存在', 'danger')
return redirect(url_for('admin.authors'))
if request.method == 'POST':
AuthorService.update(author_id,
name=request.form.get('name'),
avatar=request.form.get('avatar'),
bio=request.form.get('bio'),
website=request.form.get('website')
)
flash('作者更新成功', 'success')
return redirect(url_for('admin.authors'))
return render_template('admin/author_form.html', author=author)
@admin_bp.route('/authors/<int:author_id>/delete', methods=['POST'])
@login_required
def author_delete(author_id):
"""删除作者"""
if AuthorService.delete(author_id):
flash('作者已删除', 'success')
else:
flash('删除失败', 'danger')
return redirect(url_for('admin.authors'))
# ==================== 文件访问 ====================
# 注意:文件访问路由已在 app/__init__.py 中注册为全局路由
# 这里不需要重复注册

View File

@@ -0,0 +1,20 @@
"""后台管理路由模块"""
from flask import Blueprint, session, flash, redirect, url_for
from functools import wraps
admin_bp = Blueprint('admin', __name__)
def login_required(f):
"""登录验证装饰器"""
@wraps(f)
def decorated_function(*args, **kwargs):
if 'admin_id' not in session:
flash('请先登录', 'warning')
return redirect(url_for('admin.login'))
return f(*args, **kwargs)
return decorated_function
# 导入子模块以注册路由
from app.routes.admin import auth, dashboard, articles, categories, tags, settings, stats, upload, authors

View File

@@ -0,0 +1,166 @@
"""文章管理路由"""
from flask import render_template, redirect, url_for, request, flash
import json
from app.routes.admin import admin_bp, login_required
from app.services import ArticleService, CategoryService, TagService, AuthorService, ReferenceService, AttachmentService
@admin_bp.route('/articles')
@login_required
def articles():
"""文章列表"""
page = request.args.get('page', 1, type=int)
is_published = request.args.get('status')
if is_published == 'published':
is_published_filter = True
elif is_published == 'draft':
is_published_filter = False
else:
is_published_filter = None
articles = ArticleService.get_all_list(
page=page,
per_page=20,
is_published=is_published_filter
)
categories = CategoryService.get_all()
tags = TagService.get_all()
return render_template('admin/articles.html',
articles=articles,
categories=categories,
tags=tags,
current_status=is_published)
@admin_bp.route('/articles/create', methods=['GET', 'POST'])
@login_required
def article_create():
"""创建文章"""
if request.method == 'POST':
title = request.form.get('title')
content = request.form.get('content')
summary = request.form.get('summary')
author_id = request.form.get('author_id', type=int)
category_id = request.form.get('category_id', type=int)
cover_image = request.form.get('cover_image')
is_published = request.form.get('is_published') == 'on'
tag_ids = request.form.getlist('tags')
tag_ids = [int(t) for t in tag_ids if t]
# 处理参考来源
references_json = request.form.get('references', '[]')
references_data = json.loads(references_json) if references_json else []
article = ArticleService.create(
title=title,
content=content,
summary=summary,
author_id=author_id,
category_id=category_id,
tag_ids=tag_ids,
cover_image=cover_image,
is_published=is_published
)
# 添加参考来源
for ref in references_data:
if ref.get('title'):
ReferenceService.add(article.id, ref['title'], ref.get('url'), ref.get('description'))
flash('文章创建成功', 'success')
return redirect(url_for('admin.article_edit', article_id=article.id))
categories = CategoryService.get_all()
tags = TagService.get_all()
authors = AuthorService.get_all()
return render_template('admin/article_form.html',
article=None,
categories=categories,
tags=tags,
authors=authors,
references=[],
attachments=[])
@admin_bp.route('/articles/<int:article_id>/edit', methods=['GET', 'POST'])
@login_required
def article_edit(article_id):
"""编辑文章"""
article = ArticleService.get_by_id(article_id)
if not article:
flash('文章不存在', 'danger')
return redirect(url_for('admin.articles'))
if request.method == 'POST':
title = request.form.get('title')
content = request.form.get('content')
summary = request.form.get('summary')
author_id = request.form.get('author_id', type=int)
category_id = request.form.get('category_id', type=int)
cover_image = request.form.get('cover_image')
is_published = request.form.get('is_published') == 'on'
tag_ids = request.form.getlist('tags')
tag_ids = [int(t) for t in tag_ids if t]
# 处理参考来源
references_json = request.form.get('references', '[]')
references_data = json.loads(references_json) if references_json else []
ArticleService.update(
article_id,
title=title,
content=content,
summary=summary,
author_id=author_id,
category_id=category_id,
cover_image=cover_image,
is_published=is_published,
tag_ids=tag_ids
)
# 更新参考来源
ReferenceService.update_article_references(article_id, references_data)
flash('文章更新成功', 'success')
return redirect(url_for('admin.article_edit', article_id=article_id))
categories = CategoryService.get_all()
tags = TagService.get_all()
authors = AuthorService.get_all()
references = ReferenceService.get_by_article(article_id)
attachments = AttachmentService.get_by_article(article_id)
return render_template('admin/article_form.html',
article=article,
categories=categories,
tags=tags,
authors=authors,
references=references,
attachments=attachments)
@admin_bp.route('/articles/<int:article_id>/delete', methods=['POST'])
@login_required
def article_delete(article_id):
"""删除文章"""
if ArticleService.delete(article_id):
flash('文章已删除', 'success')
else:
flash('删除失败', 'danger')
return redirect(url_for('admin.articles'))
@admin_bp.route('/articles/<int:article_id>/toggle-publish', methods=['POST'])
@login_required
def article_toggle_publish(article_id):
"""切换文章发布状态"""
article = ArticleService.get_by_id(article_id)
if article:
ArticleService.update(article_id, is_published=not article.is_published)
status = '发布' if not article.is_published else '取消发布'
flash(f'文章已{status}', 'success')
else:
flash('文章不存在', 'danger')
return redirect(url_for('admin.articles'))

31
app/routes/admin/auth.py Normal file
View File

@@ -0,0 +1,31 @@
"""认证相关路由:登录/登出"""
from flask import render_template, redirect, url_for, request, flash, session
from app.routes.admin import admin_bp
from app.services import AdminService
@admin_bp.route('/login', methods=['GET', 'POST'])
def login():
"""登录页面"""
if request.method == 'POST':
username = request.form.get('username')
password = request.form.get('password')
admin = AdminService.authenticate(username, password)
if admin:
session['admin_id'] = admin.id
session['admin_username'] = admin.username
flash('登录成功', 'success')
return redirect(url_for('admin.dashboard'))
else:
flash('用户名或密码错误', 'danger')
return render_template('admin/login.html')
@admin_bp.route('/logout')
def logout():
"""登出"""
session.clear()
flash('已退出登录', 'info')
return redirect(url_for('admin.login'))

View File

@@ -0,0 +1,62 @@
"""作者管理路由"""
from flask import render_template, redirect, url_for, request, flash
from app.routes.admin import admin_bp, login_required
from app.services import AuthorService
@admin_bp.route('/authors')
@login_required
def authors():
"""作者列表"""
authors = AuthorService.get_all()
return render_template('admin/authors.html', authors=authors)
@admin_bp.route('/authors/create', methods=['GET', 'POST'])
@login_required
def author_create():
"""创建作者"""
if request.method == 'POST':
name = request.form.get('name')
avatar = request.form.get('avatar')
bio = request.form.get('bio')
website = request.form.get('website')
AuthorService.create(name=name, avatar=avatar, bio=bio, website=website)
flash('作者创建成功', 'success')
return redirect(url_for('admin.authors'))
return render_template('admin/author_form.html', author=None)
@admin_bp.route('/authors/<int:author_id>/edit', methods=['GET', 'POST'])
@login_required
def author_edit(author_id):
"""编辑作者"""
author = AuthorService.get_by_id(author_id)
if not author:
flash('作者不存在', 'danger')
return redirect(url_for('admin.authors'))
if request.method == 'POST':
AuthorService.update(author_id,
name=request.form.get('name'),
avatar=request.form.get('avatar'),
bio=request.form.get('bio'),
website=request.form.get('website')
)
flash('作者更新成功', 'success')
return redirect(url_for('admin.authors'))
return render_template('admin/author_form.html', author=author)
@admin_bp.route('/authors/<int:author_id>/delete', methods=['POST'])
@login_required
def author_delete(author_id):
"""删除作者"""
if AuthorService.delete(author_id):
flash('作者已删除', 'success')
else:
flash('删除失败', 'danger')
return redirect(url_for('admin.authors'))

View File

@@ -0,0 +1,60 @@
"""分类管理路由"""
from flask import render_template, redirect, url_for, request, flash
from app.routes.admin import admin_bp, login_required
from app.services import CategoryService
@admin_bp.route('/categories')
@login_required
def categories():
"""分类列表"""
categories = CategoryService.get_all()
return render_template('admin/categories.html', categories=categories)
@admin_bp.route('/categories/create', methods=['GET', 'POST'])
@login_required
def category_create():
"""创建分类"""
if request.method == 'POST':
name = request.form.get('name')
description = request.form.get('description')
sort_order = request.form.get('sort_order', 0, type=int)
CategoryService.create(name=name, description=description, sort_order=sort_order)
flash('分类创建成功', 'success')
return redirect(url_for('admin.categories'))
return render_template('admin/category_form.html', category=None)
@admin_bp.route('/categories/<int:category_id>/edit', methods=['GET', 'POST'])
@login_required
def category_edit(category_id):
"""编辑分类"""
category = CategoryService.get_by_id(category_id)
if not category:
flash('分类不存在', 'danger')
return redirect(url_for('admin.categories'))
if request.method == 'POST':
name = request.form.get('name')
description = request.form.get('description')
sort_order = request.form.get('sort_order', 0, type=int)
CategoryService.update(category_id, name=name, description=description, sort_order=sort_order)
flash('分类更新成功', 'success')
return redirect(url_for('admin.categories'))
return render_template('admin/category_form.html', category=category)
@admin_bp.route('/categories/<int:category_id>/delete', methods=['POST'])
@login_required
def category_delete(category_id):
"""删除分类"""
if CategoryService.delete(category_id):
flash('分类已删除', 'success')
else:
flash('删除失败', 'danger')
return redirect(url_for('admin.categories'))

View File

@@ -0,0 +1,24 @@
"""仪表盘路由"""
from flask import render_template
from app.routes.admin import admin_bp, login_required
from app.services import ArticleService, CategoryService, TagService
@admin_bp.route('/')
@admin_bp.route('/dashboard')
@login_required
def dashboard():
"""仪表盘"""
articles = ArticleService.get_all_list(per_page=5)
categories = CategoryService.get_all()
tags = TagService.get_all()
stats = {
'total_articles': len(articles.items),
'published_articles': sum(1 for a in articles.items if a.is_published),
'total_categories': len(categories),
'total_tags': len(tags),
'total_views': sum(a.view_count for a in articles.items)
}
return render_template('admin/dashboard.html', stats=stats, recent_articles=articles)

View File

@@ -0,0 +1,67 @@
"""网站设置路由"""
from flask import render_template, redirect, url_for, request, flash
from app.routes.admin import admin_bp, login_required
from app.services import SiteConfigService, UploadService
@admin_bp.route('/settings', methods=['GET', 'POST'])
@login_required
def settings():
"""网站设置"""
if request.method == 'POST':
# 处理图片上传
promo_image_1 = request.form.get('promo_image_1', '')
promo_image_2 = request.form.get('promo_image_2', '')
promo_image_3 = request.form.get('promo_image_3', '')
# 上传图片1
if 'promo_image_1_file' in request.files:
file = request.files['promo_image_1_file']
if file and file.filename:
image, error = UploadService.upload_image(file)
if image:
promo_image_1 = UploadService.get_image_url(image.filename)
# 上传图片2
if 'promo_image_2_file' in request.files:
file = request.files['promo_image_2_file']
if file and file.filename:
image, error = UploadService.upload_image(file)
if image:
promo_image_2 = UploadService.get_image_url(image.filename)
# 上传图片3
if 'promo_image_3_file' in request.files:
file = request.files['promo_image_3_file']
if file and file.filename:
image, error = UploadService.upload_image(file)
if image:
promo_image_3 = UploadService.get_image_url(image.filename)
# 更新设置
SiteConfigService.update(
site_name=request.form.get('site_name', '技术博客'),
site_subtitle=request.form.get('site_subtitle', ''),
site_logo=request.form.get('site_logo', ''),
site_keywords=request.form.get('site_keywords', ''),
site_description=request.form.get('site_description', ''),
footer_text=request.form.get('footer_text', ''),
footer_copyright=request.form.get('footer_copyright', ''),
footer_technology=request.form.get('footer_technology', ''),
promo_image_1=promo_image_1,
promo_image_1_title=request.form.get('promo_image_1_title', ''),
promo_image_1_link=request.form.get('promo_image_1_link', ''),
promo_image_2=promo_image_2,
promo_image_2_title=request.form.get('promo_image_2_title', ''),
promo_image_2_link=request.form.get('promo_image_2_link', ''),
promo_image_3=promo_image_3,
promo_image_3_title=request.form.get('promo_image_3_title', ''),
contact_email=request.form.get('contact_email', ''),
contact_github=request.form.get('contact_github', ''),
articles_per_page=request.form.get('articles_per_page', 10, type=int),
)
flash('网站设置已更新', 'success')
return redirect(url_for('admin.settings'))
config = SiteConfigService.get_config()
return render_template('admin/settings.html', config=config)

65
app/routes/admin/stats.py Normal file
View File

@@ -0,0 +1,65 @@
"""访问统计路由"""
from flask import render_template, redirect, url_for, request, flash
from app.routes.admin import admin_bp, login_required
from app.services import VisitService
@admin_bp.route('/stats')
@login_required
def stats():
"""访问统计概览"""
stats_summary = VisitService.get_stats_summary()
top_ips = VisitService.get_top_ips(10)
top_articles = VisitService.get_top_articles(7, 10)
top_paths = VisitService.get_top_paths(20)
return render_template('admin/stats.html',
stats=stats_summary,
top_ips=top_ips,
top_articles=top_articles,
top_paths=top_paths)
@admin_bp.route('/stats/logs')
@login_required
def stats_logs():
"""访问日志列表"""
page = request.args.get('page', 1, type=int)
path_filter = request.args.get('path', '')
logs = VisitService.get_visit_list(page=page, per_page=50, path_filter=path_filter)
return render_template('admin/stats_logs.html',
logs=logs,
path_filter=path_filter)
@admin_bp.route('/stats/charts')
@login_required
def stats_charts():
"""访问统计图表"""
stats_summary = VisitService.get_stats_summary()
# 准备图表数据
daily_data = []
for stat in stats_summary['daily_stats']:
daily_data.append({
'date': stat.date.strftime('%Y-%m-%d'),
'visits': stat.total_visits,
'unique_ips': stat.unique_ips,
'article_views': stat.article_views
})
return render_template('admin/stats_charts.html',
daily_data=daily_data,
stats=stats_summary)
@admin_bp.route('/stats/clear', methods=['POST'])
@login_required
def stats_clear():
"""清理旧访问日志"""
days = request.form.get('days', 90, type=int)
deleted = VisitService.clear_old_logs(days)
flash(f'已清理 {deleted} 条旧日志记录', 'success')
return redirect(url_for('admin.stats_logs'))

58
app/routes/admin/tags.py Normal file
View File

@@ -0,0 +1,58 @@
"""标签管理路由"""
from flask import render_template, redirect, url_for, request, flash
from app.routes.admin import admin_bp, login_required
from app.services import TagService
@admin_bp.route('/tags')
@login_required
def tags():
"""标签列表"""
tags = TagService.get_all()
return render_template('admin/tags.html', tags=tags)
@admin_bp.route('/tags/create', methods=['GET', 'POST'])
@login_required
def tag_create():
"""创建标签"""
if request.method == 'POST':
name = request.form.get('name')
color = request.form.get('color', '#007bff')
TagService.create(name=name, color=color)
flash('标签创建成功', 'success')
return redirect(url_for('admin.tags'))
return render_template('admin/tag_form.html', tag=None)
@admin_bp.route('/tags/<int:tag_id>/edit', methods=['GET', 'POST'])
@login_required
def tag_edit(tag_id):
"""编辑标签"""
tag = TagService.get_by_id(tag_id)
if not tag:
flash('标签不存在', 'danger')
return redirect(url_for('admin.tags'))
if request.method == 'POST':
name = request.form.get('name')
color = request.form.get('color')
TagService.update(tag_id, name=name, color=color)
flash('标签更新成功', 'success')
return redirect(url_for('admin.tags'))
return render_template('admin/tag_form.html', tag=tag)
@admin_bp.route('/tags/<int:tag_id>/delete', methods=['POST'])
@login_required
def tag_delete(tag_id):
"""删除标签"""
if TagService.delete(tag_id):
flash('标签已删除', 'success')
else:
flash('删除失败', 'danger')
return redirect(url_for('admin.tags'))

129
app/routes/admin/upload.py Normal file
View File

@@ -0,0 +1,129 @@
"""文件上传路由:图片/附件"""
from flask import render_template, redirect, url_for, request, flash, session, jsonify
import os
from app.routes.admin import admin_bp, login_required
from app.services import UploadService, AttachmentService
@admin_bp.route('/upload/image', methods=['POST'])
@login_required
def upload_image():
"""上传图片(支持拖拽、粘贴、选择)"""
article_id = request.form.get('article_id', type=int)
uploader = session.get('admin_username', 'admin')
# 检查是否有文件上传
if 'file' not in request.files:
return jsonify({'success': False, 'message': '没有选择文件'}), 400
file = request.files['file']
if file.filename == '':
return jsonify({'success': False, 'message': '没有选择文件'}), 400
# 上传图片
image, error = UploadService.upload_image(file, article_id, uploader)
if error:
return jsonify({'success': False, 'message': error}), 400
# 返回图片URL
return jsonify({
'success': True,
'url': UploadService.get_image_url(image.filename),
'filename': image.filename
})
@admin_bp.route('/upload/paste', methods=['POST'])
@login_required
def upload_paste():
"""粘贴图片上传"""
import base64
from io import BytesIO
from datetime import datetime
import uuid
article_id = request.form.get('article_id', type=int)
uploader = session.get('admin_username', 'admin')
# 获取粘贴的图片数据
image_data = request.form.get('image_data')
if not image_data:
return jsonify({'success': False, 'message': '没有图片数据'}), 400
# 解码Base64图片
try:
# 移除data:image/xxx;base64,前缀
if 'base64,' in image_data:
image_data = image_data.split('base64,')[1]
image_bytes = base64.b64decode(image_data)
# 确定图片格式
ext = 'png' # 默认png
# 生成文件名
filename = f'{datetime.now().strftime("%Y%m%d")}_{uuid.uuid4().hex[:8]}.{ext}'
# 保存文件
upload_folder = UploadService.get_upload_folder()
filepath = os.path.join(upload_folder, filename)
with open(filepath, 'wb') as f:
f.write(image_bytes)
# 记录到数据库
from app.models import UploadedImage
from app import db
image = UploadedImage(
filename=filename,
filepath=filepath,
article_id=article_id,
uploader=uploader
)
db.session.add(image)
db.session.commit()
return jsonify({
'success': True,
'url': UploadService.get_image_url(filename),
'filename': filename
})
except Exception as e:
return jsonify({'success': False, 'message': str(e)}), 400
@admin_bp.route('/upload/attachment', methods=['POST'])
@login_required
def upload_attachment():
"""上传附件"""
article_id = request.form.get('article_id', type=int)
description = request.form.get('description', '')
if 'file' not in request.files:
return jsonify({'success': False, 'message': '没有选择文件'}), 400
file = request.files['file']
if file.filename == '':
return jsonify({'success': False, 'message': '没有选择文件'}), 400
attachment, error = UploadService.upload_attachment(file, article_id, description)
if error:
return jsonify({'success': False, 'message': error}), 400
return jsonify({
'success': True,
'attachment': {
'id': attachment.id,
'filename': attachment.filename,
'filesize': attachment.get_filesize_display(),
'url': f'/uploads/{attachment.filename}'
}
})
@admin_bp.route('/attachment/<int:attachment_id>/delete', methods=['POST'])
@login_required
def delete_attachment(attachment_id):
"""删除附件"""
if AttachmentService.delete_attachment(attachment_id):
return jsonify({'success': True})
return jsonify({'success': False, 'message': '删除失败'}), 400

2
run.py
View File

@@ -7,5 +7,5 @@ app = create_app(os.environ.get('FLASK_ENV', 'development'))
if __name__ == '__main__':
# 从环境变量获取端口默认16012
port = int(os.environ.get('PORT', 16012))
port = int(os.environ.get('PORT', 16013))
app.run(host='0.0.0.0', port=port)