pengxiao
a3eb5c7b2d
从百度网盘 /apps/bypy/dofile/ 下载的完整项目结构，包含： - dofile/: Flask 后端（端口 5000）+ 静态前端（端口 6002） - officefile/: 项目相关文档 - .gitignore: Python venv / 日志 / 缓存 Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
228 lines
7.2 KiB
Python
228 lines
7.2 KiB
Python
from flask import Blueprint, request, jsonify, send_file, current_app
|
|
from flask_login import login_required, current_user
|
|
from werkzeug.utils import secure_filename
|
|
import sys
|
|
import os
|
|
|
|
# 添加父目录到路径
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
from models import db, Article, Resource, TeamMember
|
|
from datetime import datetime
|
|
|
|
bp = Blueprint('cms', __name__)
|
|
|
|
def allowed_file(filename, allowed_extensions):
|
|
"""检查文件扩展名是否允许"""
|
|
return '.' in filename and filename.rsplit('.', 1)[1].lower() in allowed_extensions
|
|
|
|
# 文章管理
|
|
@bp.route('/articles', methods=['GET'])
|
|
def get_articles():
|
|
"""获取文章列表"""
|
|
category = request.args.get('category', 'news')
|
|
published_only = request.args.get('published_only', 'true').lower() == 'true'
|
|
|
|
query = Article.query.filter_by(category=category)
|
|
|
|
if published_only:
|
|
query = query.filter_by(published=True)
|
|
|
|
# 如果未登录，过滤受保护内容
|
|
if not current_user.is_authenticated:
|
|
query = query.filter_by(is_protected=False)
|
|
|
|
articles = query.order_by(Article.created_at.desc()).all()
|
|
|
|
return jsonify({
|
|
'articles': [article.to_dict() for article in articles]
|
|
}), 200
|
|
|
|
@bp.route('/articles/<int:article_id>', methods=['GET'])
|
|
def get_article(article_id):
|
|
"""获取单篇文章"""
|
|
article = Article.query.get_or_404(article_id)
|
|
|
|
# 检查是否需要登录
|
|
if article.is_protected and not current_user.is_authenticated:
|
|
return jsonify({'error': 'Authentication required'}), 401
|
|
|
|
if not article.published:
|
|
if not current_user.is_authenticated or not current_user.is_admin:
|
|
return jsonify({'error': 'Article not found'}), 404
|
|
|
|
return jsonify({'article': article.to_dict()}), 200
|
|
|
|
@bp.route('/articles', methods=['POST'])
|
|
@login_required
|
|
def create_article():
|
|
"""创建文章（需要管理员权限）"""
|
|
if not current_user.is_admin:
|
|
return jsonify({'error': 'Admin access required'}), 403
|
|
|
|
data = request.get_json()
|
|
|
|
article = Article(
|
|
title=data.get('title'),
|
|
content=data.get('content'),
|
|
summary=data.get('summary'),
|
|
category=data.get('category', 'news'),
|
|
is_protected=data.get('is_protected', False),
|
|
author_id=current_user.id,
|
|
published=data.get('published', True)
|
|
)
|
|
|
|
db.session.add(article)
|
|
db.session.commit()
|
|
|
|
return jsonify({
|
|
'message': 'Article created successfully',
|
|
'article': article.to_dict()
|
|
}), 201
|
|
|
|
@bp.route('/articles/<int:article_id>', methods=['PUT'])
|
|
@login_required
|
|
def update_article(article_id):
|
|
"""更新文章（需要管理员权限）"""
|
|
if not current_user.is_admin:
|
|
return jsonify({'error': 'Admin access required'}), 403
|
|
|
|
article = Article.query.get_or_404(article_id)
|
|
data = request.get_json()
|
|
|
|
article.title = data.get('title', article.title)
|
|
article.content = data.get('content', article.content)
|
|
article.summary = data.get('summary', article.summary)
|
|
article.category = data.get('category', article.category)
|
|
article.is_protected = data.get('is_protected', article.is_protected)
|
|
article.published = data.get('published', article.published)
|
|
article.updated_at = datetime.utcnow()
|
|
|
|
db.session.commit()
|
|
|
|
return jsonify({
|
|
'message': 'Article updated successfully',
|
|
'article': article.to_dict()
|
|
}), 200
|
|
|
|
@bp.route('/articles/<int:article_id>', methods=['DELETE'])
|
|
@login_required
|
|
def delete_article(article_id):
|
|
"""删除文章（需要管理员权限）"""
|
|
if not current_user.is_admin:
|
|
return jsonify({'error': 'Admin access required'}), 403
|
|
|
|
article = Article.query.get_or_404(article_id)
|
|
db.session.delete(article)
|
|
db.session.commit()
|
|
|
|
return jsonify({'message': 'Article deleted successfully'}), 200
|
|
|
|
# 资源管理
|
|
@bp.route('/resources', methods=['GET'])
|
|
def get_resources():
|
|
"""获取资源列表"""
|
|
category = request.args.get('category')
|
|
|
|
query = Resource.query
|
|
|
|
if category:
|
|
query = query.filter_by(category=category)
|
|
|
|
# 如果未登录，过滤受保护资源
|
|
if not current_user.is_authenticated:
|
|
query = query.filter_by(is_protected=False)
|
|
|
|
resources = query.order_by(Resource.created_at.desc()).all()
|
|
|
|
return jsonify({
|
|
'resources': [resource.to_dict() for resource in resources]
|
|
}), 200
|
|
|
|
@bp.route('/resources', methods=['POST'])
|
|
@login_required
|
|
def upload_resource():
|
|
"""上传资源文件（需要管理员权限）"""
|
|
if not current_user.is_admin:
|
|
return jsonify({'error': 'Admin access required'}), 403
|
|
|
|
if 'file' not in request.files:
|
|
return jsonify({'error': 'No file provided'}), 400
|
|
|
|
file = request.files['file']
|
|
if file.filename == '':
|
|
return jsonify({'error': 'No file selected'}), 400
|
|
|
|
if file and allowed_file(file.filename, {'pdf', 'zip', 'rar', 'doc', 'docx', 'xls', 'xlsx', 'ppt', 'pptx'}):
|
|
filename = secure_filename(file.filename)
|
|
uploads_dir = current_app.config['UPLOAD_FOLDER']
|
|
os.makedirs(uploads_dir, exist_ok=True)
|
|
filepath = os.path.join(uploads_dir, filename)
|
|
file.save(filepath)
|
|
|
|
resource = Resource(
|
|
name=request.form.get('name', filename),
|
|
description=request.form.get('description', ''),
|
|
file_path=f'/uploads/{filename}',
|
|
file_size=os.path.getsize(filepath),
|
|
file_type=filename.rsplit('.', 1)[1].lower(),
|
|
category=request.form.get('category', 'document'),
|
|
is_protected=request.form.get('is_protected', 'false').lower() == 'true'
|
|
)
|
|
|
|
db.session.add(resource)
|
|
db.session.commit()
|
|
|
|
return jsonify({
|
|
'message': 'Resource uploaded successfully',
|
|
'resource': resource.to_dict()
|
|
}), 201
|
|
|
|
return jsonify({'error': 'Invalid file type'}), 400
|
|
|
|
# 团队成员管理
|
|
@bp.route('/team', methods=['GET'])
|
|
def get_team():
|
|
"""获取团队成员列表"""
|
|
role = request.args.get('role')
|
|
|
|
query = TeamMember.query
|
|
|
|
if role:
|
|
query = query.filter_by(role=role)
|
|
|
|
members = query.order_by(TeamMember.order, TeamMember.created_at).all()
|
|
|
|
return jsonify({
|
|
'members': [member.to_dict() for member in members]
|
|
}), 200
|
|
|
|
@bp.route('/team', methods=['POST'])
|
|
@login_required
|
|
def create_team_member():
|
|
"""创建团队成员（需要管理员权限）"""
|
|
if not current_user.is_admin:
|
|
return jsonify({'error': 'Admin access required'}), 403
|
|
|
|
data = request.get_json()
|
|
|
|
member = TeamMember(
|
|
name=data.get('name'),
|
|
title=data.get('title'),
|
|
title_en=data.get('title_en'),
|
|
bio=data.get('bio'),
|
|
bio_en=data.get('bio_en'),
|
|
role=data.get('role', 'student'),
|
|
email=data.get('email'),
|
|
website=data.get('website'),
|
|
order=data.get('order', 0)
|
|
)
|
|
|
|
db.session.add(member)
|
|
db.session.commit()
|
|
|
|
return jsonify({
|
|
'message': 'Team member created successfully',
|
|
'member': member.to_dict()
|
|
}), 201
|
|
|