init: TEALab 2025 实验室网站项目

从百度网盘 /apps/bypy/dofile/ 下载的完整项目结构,包含:
- dofile/: Flask 后端(端口 5000)+ 静态前端(端口 6002)
- officefile/: 项目相关文档
- .gitignore: Python venv / 日志 / 缓存

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
2026-06-14 01:07:19 +08:00
commit a3eb5c7b2d
39 changed files with 7110 additions and 0 deletions
+88
View File
@@ -0,0 +1,88 @@
# DAEL Lab Backend API
Flask后端API服务器,提供用户认证、内容管理和数据API功能。
## 安装和设置
### 1. 安装依赖
```bash
cd backend
pip install -r requirements.txt
```
### 2. 配置环境变量
复制 `.env.example``.env` 并修改配置:
```bash
cp .env.example .env
```
### 3. 初始化数据库
```bash
python init_data.py
```
这将创建数据库表和默认账户:
- 管理员:`admin` / `admin123`
- 测试用户:`user` / `user123`
### 4. 启动服务器
```bash
python app.py
```
服务器将在 `http://localhost:5000` 启动。
## API端点
### 认证
- `POST /api/auth/register` - 用户注册
- `POST /api/auth/login` - 用户登录
- `POST /api/auth/logout` - 用户登出
- `GET /api/auth/check` - 检查认证状态
- `GET /api/auth/me` - 获取当前用户信息
### 内容管理
- `GET /api/cms/articles` - 获取文章列表
- `GET /api/cms/articles/<id>` - 获取单篇文章
- `POST /api/cms/articles` - 创建文章(管理员)
- `PUT /api/cms/articles/<id>` - 更新文章(管理员)
- `DELETE /api/cms/articles/<id>` - 删除文章(管理员)
- `GET /api/cms/resources` - 获取资源列表
- `POST /api/cms/resources` - 上传资源(管理员)
- `GET /api/cms/team` - 获取团队成员列表
- `POST /api/cms/team` - 创建团队成员(管理员)
### 其他API
- `GET /api/download/<resource_id>` - 下载资源文件
- `GET /api/stats` - 获取统计数据(需登录)
- `GET /api/protected-content` - 获取受保护内容(需登录)
## 数据库模型
- **User**: 用户模型
- **Article**: 文章/新闻模型
- **Resource**: 资源文件模型
- **TeamMember**: 团队成员模型
## 开发说明
- 数据库文件存储在 `../data/dael.db`
- 上传文件存储在 `../uploads/`
- 使用SQLite作为开发数据库,生产环境建议使用PostgreSQL
+76
View File
@@ -0,0 +1,76 @@
from flask import Flask, request, jsonify
from flask_cors import CORS
from flask_login import LoginManager
from models import db, User
import os
from dotenv import load_dotenv
load_dotenv()
app = Flask(__name__)
app.config['SECRET_KEY'] = os.getenv('SECRET_KEY', 'dev-secret-key-change-in-production')
# 获取项目根目录(backend的父目录)
base_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
data_dir = os.path.join(base_dir, 'data')
db_path = os.path.join(data_dir, 'dael.db')
# 确保data目录存在
os.makedirs(data_dir, exist_ok=True)
# 配置数据库URI(使用绝对路径)
app.config['SQLALCHEMY_DATABASE_URI'] = os.getenv('DATABASE_URL', f'sqlite:///{db_path}')
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['UPLOAD_FOLDER'] = os.path.join(base_dir, 'uploads')
app.config['MAX_CONTENT_LENGTH'] = 100 * 1024 * 1024 # 100MB max file size
# 确保上传目录存在
os.makedirs(app.config['UPLOAD_FOLDER'], exist_ok=True)
db.init_app(app)
# 配置CORS以支持跨域请求和cookies
CORS(app,
origins=['http://localhost:6002', 'http://127.0.0.1:6002', 'http://8.152.100.83:6002'],
supports_credentials=True,
allow_headers=['Content-Type', 'Authorization'])
# Flask-Login配置
login_manager = LoginManager()
login_manager.init_app(app)
login_manager.login_view = 'login'
@login_manager.user_loader
def load_user(user_id):
return User.query.get(int(user_id))
# 导入路由
from routes import auth, cms, api, meetings, research
app.register_blueprint(auth.bp, url_prefix='/api/auth')
app.register_blueprint(cms.bp, url_prefix='/api/cms')
app.register_blueprint(api.bp, url_prefix='/api')
app.register_blueprint(meetings.bp, url_prefix='/api/meetings')
app.register_blueprint(research.bp, url_prefix='/api/research')
@app.route('/api/health', methods=['GET'])
def health_check():
"""健康检查"""
return jsonify({'status': 'ok', 'message': 'DAEL Lab API is running'})
def init_db():
"""初始化数据库"""
with app.app_context():
db.create_all()
# 创建默认管理员账户(如果不存在)
if not User.query.filter_by(username='admin').first():
admin = User(username='admin', email='admin@dael.edu.cn', is_admin=True)
admin.set_password('admin123') # 默认密码,生产环境应更改
db.session.add(admin)
db.session.commit()
print("Default admin user created: admin/admin123")
if __name__ == '__main__':
init_db()
app.run(debug=True, host='0.0.0.0', port=5000)
+126
View File
@@ -0,0 +1,126 @@
#!/usr/bin/env python3
"""初始化数据库并添加示例数据"""
import os
import sys
# 确保可以导入app
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from app import app, db
from models import User, Article, Resource, TeamMember
from datetime import datetime
def init_data():
# 确保data目录存在
base_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
data_dir = os.path.join(base_dir, 'data')
os.makedirs(data_dir, exist_ok=True)
with app.app_context():
# 创建数据库表
db.create_all()
# 创建默认管理员
if not User.query.filter_by(username='admin').first():
admin = User(username='admin', email='admin@dael.edu.cn', is_admin=True)
admin.set_password('admin123')
db.session.add(admin)
print("Created admin user: admin/admin123")
# 创建示例用户
if not User.query.filter_by(username='user').first():
user = User(username='user', email='user@dael.edu.cn', is_admin=False)
user.set_password('user123')
db.session.add(user)
print("Created test user: user/user123")
# 创建示例文章(受保护)
if Article.query.count() == 0:
articles = [
Article(
title='实验室内部研究进展报告',
content='这是实验室的内部研究进展报告,包含详细的研究数据和成果分析...',
summary='2024年度实验室内部研究进展总结',
category='news',
is_protected=True,
author_id=1,
published=True
),
Article(
title='内部技术文档:生成式生态学算法详解',
content='本文档详细介绍了生成式生态学算法的实现原理和应用案例...',
summary='内部技术文档,仅供实验室成员查阅',
category='publication',
is_protected=True,
author_id=1,
published=True
)
]
for article in articles:
db.session.add(article)
print("Created sample protected articles")
# 创建示例资源(受保护)
if Resource.query.count() == 0:
resources = [
Resource(
name='内部数据集 v2.0',
description='实验室内部使用的完整数据集,包含所有研究数据',
file_path='/uploads/internal_dataset.zip',
file_size=1024 * 1024 * 500, # 500MB
file_type='zip',
category='dataset',
is_protected=True
),
Resource(
name='内部研究报告模板',
description='实验室内部研究报告的标准模板',
file_path='/uploads/report_template.docx',
file_size=1024 * 50, # 50KB
file_type='docx',
category='document',
is_protected=True
)
]
for resource in resources:
db.session.add(resource)
print("Created sample protected resources")
# 创建团队成员
if TeamMember.query.count() == 0:
members = [
TeamMember(
name='Dr. Lin Chen',
title='Principal Investigator / 实验室主任',
title_en='Principal Investigator',
bio='博士毕业于MIT Media Lab,专注于计算设计与人机交互领域的研究。',
bio_en='PhD from MIT Media Lab, focusing on computational design and HCI.',
role='professor',
email='lchen@dael.edu.cn',
order=1
),
TeamMember(
name='Sarah Wu',
title='Lead Ecologist / 生态学顾问',
title_en='Lead Ecologist',
bio='城市生态学专家,研究重点为城市微气候与生物多样性修复。',
bio_en='Urban ecology expert, focusing on microclimate and biodiversity restoration.',
role='faculty',
email='swu@dael.edu.cn',
order=2
)
]
for member in members:
db.session.add(member)
print("Created sample team members")
db.session.commit()
print("\nDatabase initialized successfully!")
print("\nDefault accounts:")
print(" Admin: admin / admin123")
print(" User: user / user123")
if __name__ == '__main__':
init_data()
+24
View File
@@ -0,0 +1,24 @@
#!/bin/bash
# 初始化数据库脚本
cd "$(dirname "$0")"
# 检查虚拟环境
if [ -d ".venv" ]; then
source .venv/bin/activate
python init_data.py
else
echo "Virtual environment not found. Please run: python3 -m venv .venv"
echo "Then install dependencies: pip install -r requirements.txt"
exit 1
fi
+342
View File
@@ -0,0 +1,342 @@
from flask_sqlalchemy import SQLAlchemy
from flask_login import UserMixin
from werkzeug.security import generate_password_hash, check_password_hash
from datetime import datetime
db = SQLAlchemy()
class User(UserMixin, db.Model):
"""用户模型"""
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(80), unique=True, nullable=False)
email = db.Column(db.String(120), unique=True, nullable=False)
password_hash = db.Column(db.String(255), nullable=False)
is_admin = db.Column(db.Boolean, default=False)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
def set_password(self, password):
self.password_hash = generate_password_hash(password)
def check_password(self, password):
return check_password_hash(self.password_hash, password)
def to_dict(self):
return {
'id': self.id,
'username': self.username,
'email': self.email,
'is_admin': self.is_admin,
'created_at': self.created_at.isoformat()
}
class Article(db.Model):
"""文章/新闻模型"""
__tablename__ = 'articles'
id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.String(200), nullable=False)
content = db.Column(db.Text, nullable=False)
summary = db.Column(db.Text)
category = db.Column(db.String(50), default='news') # news, publication, etc.
is_protected = db.Column(db.Boolean, default=False) # 是否需要登录查看
author_id = db.Column(db.Integer, db.ForeignKey('users.id'))
created_at = db.Column(db.DateTime, default=datetime.utcnow)
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
published = db.Column(db.Boolean, default=True)
author = db.relationship('User', backref='articles')
def to_dict(self):
return {
'id': self.id,
'title': self.title,
'content': self.content,
'summary': self.summary,
'category': self.category,
'is_protected': self.is_protected,
'author_id': self.author_id,
'created_at': self.created_at.isoformat(),
'updated_at': self.updated_at.isoformat(),
'published': self.published
}
class Resource(db.Model):
"""资源文件模型"""
__tablename__ = 'resources'
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(200), nullable=False)
description = db.Column(db.Text)
file_path = db.Column(db.String(500), nullable=False)
file_size = db.Column(db.Integer) # 文件大小(字节)
file_type = db.Column(db.String(50)) # 文件类型
category = db.Column(db.String(50)) # software, dataset, document, etc.
is_protected = db.Column(db.Boolean, default=False) # 是否需要登录下载
download_count = db.Column(db.Integer, default=0)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
def to_dict(self):
return {
'id': self.id,
'name': self.name,
'description': self.description,
'file_path': self.file_path,
'file_size': self.file_size,
'file_type': self.file_type,
'category': self.category,
'is_protected': self.is_protected,
'download_count': self.download_count,
'created_at': self.created_at.isoformat(),
'updated_at': self.updated_at.isoformat()
}
class TeamMember(db.Model):
"""团队成员模型"""
__tablename__ = 'team_members'
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(100), nullable=False)
title = db.Column(db.String(200))
title_en = db.Column(db.String(200))
bio = db.Column(db.Text)
bio_en = db.Column(db.Text)
role = db.Column(db.String(50)) # professor, student, alumni
email = db.Column(db.String(120))
website = db.Column(db.String(500))
avatar_path = db.Column(db.String(500))
order = db.Column(db.Integer, default=0) # 排序
created_at = db.Column(db.DateTime, default=datetime.utcnow)
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
def to_dict(self):
return {
'id': self.id,
'name': self.name,
'title': self.title,
'title_en': self.title_en,
'bio': self.bio,
'bio_en': self.bio_en,
'role': self.role,
'email': self.email,
'website': self.website,
'avatar_path': self.avatar_path,
'order': self.order,
'created_at': self.created_at.isoformat(),
'updated_at': self.updated_at.isoformat()
}
class Meeting(db.Model):
"""组会模型"""
__tablename__ = 'meetings'
id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.String(200), nullable=False)
description = db.Column(db.Text)
date = db.Column(db.Date, nullable=False)
time = db.Column(db.Time, nullable=False)
location = db.Column(db.String(200))
organizer_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
status = db.Column(db.String(50), default='planned') # planned, ongoing, completed, cancelled
meeting_type = db.Column(db.String(50), default='weekly') # weekly, monthly, special
meeting_notes = db.Column(db.Text)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
organizer = db.relationship('User', backref='organized_meetings')
registrations = db.relationship('MeetingRegistration', backref='meeting', cascade='all, delete-orphan')
files = db.relationship('MeetingFile', backref='meeting', cascade='all, delete-orphan')
def to_dict(self):
return {
'id': self.id,
'title': self.title,
'description': self.description,
'date': self.date.isoformat() if self.date else None,
'time': self.time.strftime('%H:%M') if self.time else None,
'location': self.location,
'organizer_id': self.organizer_id,
'organizer_name': self.organizer.username if self.organizer else None,
'status': self.status,
'meeting_type': self.meeting_type,
'meeting_notes': self.meeting_notes,
'created_at': self.created_at.isoformat(),
'updated_at': self.updated_at.isoformat(),
'attendee_count': len(self.registrations) if self.registrations else 0
}
class MeetingRegistration(db.Model):
"""组会报名模型"""
__tablename__ = 'meeting_registrations'
id = db.Column(db.Integer, primary_key=True)
meeting_id = db.Column(db.Integer, db.ForeignKey('meetings.id'), nullable=False)
user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
registered_at = db.Column(db.DateTime, default=datetime.utcnow)
status = db.Column(db.String(50), default='registered') # registered, checked_in, absent
user = db.relationship('User', backref='meeting_registrations')
def to_dict(self):
return {
'id': self.id,
'meeting_id': self.meeting_id,
'user_id': self.user_id,
'username': self.user.username if self.user else None,
'registered_at': self.registered_at.isoformat(),
'status': self.status
}
class MeetingFile(db.Model):
"""组会文件模型"""
__tablename__ = 'meeting_files'
id = db.Column(db.Integer, primary_key=True)
meeting_id = db.Column(db.Integer, db.ForeignKey('meetings.id'), nullable=False)
name = db.Column(db.String(200), nullable=False)
file_path = db.Column(db.String(500), nullable=False)
file_size = db.Column(db.Integer)
file_type = db.Column(db.String(50))
uploader_id = db.Column(db.Integer, db.ForeignKey('users.id'))
uploaded_at = db.Column(db.DateTime, default=datetime.utcnow)
uploader = db.relationship('User', backref='uploaded_meeting_files')
def to_dict(self):
return {
'id': self.id,
'meeting_id': self.meeting_id,
'name': self.name,
'file_path': self.file_path,
'file_size': self.file_size,
'file_type': self.file_type,
'uploader_id': self.uploader_id,
'uploaded_at': self.uploaded_at.isoformat()
}
class ResearchProject(db.Model):
"""研究项目模型"""
__tablename__ = 'research_projects'
id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.String(200), nullable=False)
description = db.Column(db.Text)
start_date = db.Column(db.Date)
end_date = db.Column(db.Date)
owner_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
status = db.Column(db.String(50), default='active') # active, completed, paused, cancelled
progress = db.Column(db.Integer, default=0) # 0-100
tags = db.Column(db.String(500)) # 逗号分隔的标签
created_at = db.Column(db.DateTime, default=datetime.utcnow)
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
owner = db.relationship('User', backref='research_projects')
notes = db.relationship('ResearchNote', backref='project', cascade='all, delete-orphan')
tasks = db.relationship('ResearchTask', backref='project', cascade='all, delete-orphan')
files = db.relationship('ResearchFile', backref='project', cascade='all, delete-orphan')
def to_dict(self):
return {
'id': self.id,
'title': self.title,
'description': self.description,
'start_date': self.start_date.isoformat() if self.start_date else None,
'end_date': self.end_date.isoformat() if self.end_date else None,
'owner_id': self.owner_id,
'owner_name': self.owner.username if self.owner else None,
'status': self.status,
'progress': self.progress,
'tags': self.tags.split(',') if self.tags else [],
'created_at': self.created_at.isoformat(),
'updated_at': self.updated_at.isoformat()
}
class ResearchNote(db.Model):
"""研究笔记模型"""
__tablename__ = 'research_notes'
id = db.Column(db.Integer, primary_key=True)
project_id = db.Column(db.Integer, db.ForeignKey('research_projects.id'), nullable=False)
title = db.Column(db.String(200), nullable=False)
content = db.Column(db.Text, nullable=False)
author_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
note_type = db.Column(db.String(50), default='note') # note, experiment, literature, idea
created_at = db.Column(db.DateTime, default=datetime.utcnow)
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
author = db.relationship('User', backref='research_notes')
def to_dict(self):
return {
'id': self.id,
'project_id': self.project_id,
'title': self.title,
'content': self.content,
'author_id': self.author_id,
'author_name': self.author.username if self.author else None,
'note_type': self.note_type,
'created_at': self.created_at.isoformat(),
'updated_at': self.updated_at.isoformat()
}
class ResearchTask(db.Model):
"""研究任务模型"""
__tablename__ = 'research_tasks'
id = db.Column(db.Integer, primary_key=True)
project_id = db.Column(db.Integer, db.ForeignKey('research_projects.id'), nullable=False)
title = db.Column(db.String(200), nullable=False)
description = db.Column(db.Text)
assignee_id = db.Column(db.Integer, db.ForeignKey('users.id'))
due_date = db.Column(db.Date)
status = db.Column(db.String(50), default='todo') # todo, in_progress, completed, cancelled
priority = db.Column(db.String(50), default='medium') # high, medium, low
created_at = db.Column(db.DateTime, default=datetime.utcnow)
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
assignee = db.relationship('User', backref='assigned_tasks')
def to_dict(self):
return {
'id': self.id,
'project_id': self.project_id,
'title': self.title,
'description': self.description,
'assignee_id': self.assignee_id,
'assignee_name': self.assignee.username if self.assignee else None,
'due_date': self.due_date.isoformat() if self.due_date else None,
'status': self.status,
'priority': self.priority,
'created_at': self.created_at.isoformat(),
'updated_at': self.updated_at.isoformat()
}
class ResearchFile(db.Model):
"""研究文件模型"""
__tablename__ = 'research_files'
id = db.Column(db.Integer, primary_key=True)
project_id = db.Column(db.Integer, db.ForeignKey('research_projects.id'), nullable=False)
name = db.Column(db.String(200), nullable=False)
file_path = db.Column(db.String(500), nullable=False)
file_size = db.Column(db.Integer)
file_type = db.Column(db.String(50))
uploader_id = db.Column(db.Integer, db.ForeignKey('users.id'))
uploaded_at = db.Column(db.DateTime, default=datetime.utcnow)
uploader = db.relationship('User', backref='uploaded_research_files')
def to_dict(self):
return {
'id': self.id,
'project_id': self.project_id,
'name': self.name,
'file_path': self.file_path,
'file_size': self.file_size,
'file_type': self.file_type,
'uploader_id': self.uploader_id,
'uploaded_at': self.uploaded_at.isoformat()
}
+17
View File
@@ -0,0 +1,17 @@
Flask==3.0.0
Flask-SQLAlchemy==3.1.1
Flask-Login==0.6.3
Flask-CORS==4.0.0
Werkzeug==3.0.1
bcrypt==4.1.2
python-dotenv==1.0.0
+11
View File
@@ -0,0 +1,11 @@
# Routes package
+59
View File
@@ -0,0 +1,59 @@
from flask import Blueprint, request, jsonify, send_from_directory
from flask_login import login_required, current_user
import sys
import os
# 添加父目录到路径
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from models import db, Resource
bp = Blueprint('api', __name__)
@bp.route('/download/<int:resource_id>', methods=['GET'])
def download_resource(resource_id):
"""下载资源文件"""
resource = Resource.query.get_or_404(resource_id)
# 检查是否需要登录
if resource.is_protected and not current_user.is_authenticated:
return jsonify({'error': 'Authentication required'}), 401
# 增加下载计数
resource.download_count += 1
db.session.commit()
# 返回文件
uploads_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), 'uploads')
filename = os.path.basename(resource.file_path)
return send_from_directory(uploads_dir, filename, as_attachment=True)
@bp.route('/stats', methods=['GET'])
@login_required
def get_stats():
"""获取统计数据(需要登录)"""
from models import Article, Resource, TeamMember
stats = {
'articles': Article.query.count(),
'resources': Resource.query.count(),
'team_members': TeamMember.query.count(),
'total_downloads': db.session.query(db.func.sum(Resource.download_count)).scalar() or 0
}
return jsonify({'stats': stats}), 200
@bp.route('/protected-content', methods=['GET'])
@login_required
def get_protected_content():
"""获取受保护内容(需要登录)"""
from models import Article, Resource
protected_articles = Article.query.filter_by(is_protected=True, published=True).all()
protected_resources = Resource.query.filter_by(is_protected=True).all()
return jsonify({
'articles': [article.to_dict() for article in protected_articles],
'resources': [resource.to_dict() for resource in protected_resources]
}), 200
+87
View File
@@ -0,0 +1,87 @@
from flask import Blueprint, request, jsonify
from flask_login import login_user, logout_user, login_required, current_user
import sys
import os
# 添加父目录到路径
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from models import db, User
from werkzeug.security import check_password_hash
bp = Blueprint('auth', __name__)
@bp.route('/register', methods=['POST'])
def register():
"""用户注册"""
data = request.get_json()
if not data or not data.get('username') or not data.get('email') or not data.get('password'):
return jsonify({'error': 'Missing required fields'}), 400
# 检查用户名是否已存在
if User.query.filter_by(username=data['username']).first():
return jsonify({'error': 'Username already exists'}), 400
# 检查邮箱是否已存在
if User.query.filter_by(email=data['email']).first():
return jsonify({'error': 'Email already exists'}), 400
# 创建新用户
user = User(
username=data['username'],
email=data['email'],
is_admin=data.get('is_admin', False)
)
user.set_password(data['password'])
db.session.add(user)
db.session.commit()
return jsonify({
'message': 'User registered successfully',
'user': user.to_dict()
}), 201
@bp.route('/login', methods=['POST'])
def login():
"""用户登录"""
data = request.get_json()
if not data or not data.get('username') or not data.get('password'):
return jsonify({'error': 'Username and password required'}), 400
user = User.query.filter_by(username=data['username']).first()
if user and user.check_password(data['password']):
login_user(user, remember=data.get('remember', False))
return jsonify({
'message': 'Login successful',
'user': user.to_dict()
}), 200
else:
return jsonify({'error': 'Invalid username or password'}), 401
@bp.route('/logout', methods=['POST'])
@login_required
def logout():
"""用户登出"""
logout_user()
return jsonify({'message': 'Logout successful'}), 200
@bp.route('/me', methods=['GET'])
@login_required
def get_current_user():
"""获取当前登录用户信息"""
return jsonify({'user': current_user.to_dict()}), 200
@bp.route('/check', methods=['GET'])
def check_auth():
"""检查认证状态"""
if current_user.is_authenticated:
return jsonify({
'authenticated': True,
'user': current_user.to_dict()
}), 200
else:
return jsonify({'authenticated': False}), 200
+227
View File
@@ -0,0 +1,227 @@
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
+256
View File
@@ -0,0 +1,256 @@
from flask import Blueprint, request, jsonify, send_from_directory, current_app
from flask_login import login_required, current_user
from werkzeug.utils import secure_filename
import sys
import os
from datetime import datetime, date, time
# 添加父目录到路径
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from models import db, Meeting, MeetingRegistration, MeetingFile, User
bp = Blueprint('meetings', __name__)
def allowed_file(filename, allowed_extensions):
"""检查文件扩展名是否允许"""
return '.' in filename and filename.rsplit('.', 1)[1].lower() in allowed_extensions
@bp.route('/meetings', methods=['GET'])
@login_required
def get_meetings():
"""获取组会列表(实验室成员可查看所有)"""
status = request.args.get('status')
meeting_type = request.args.get('type')
query = Meeting.query
if status:
query = query.filter_by(status=status)
if meeting_type:
query = query.filter_by(meeting_type=meeting_type)
meetings = query.order_by(Meeting.date.desc(), Meeting.time.desc()).all()
return jsonify({
'meetings': [meeting.to_dict() for meeting in meetings]
}), 200
@bp.route('/meetings/<int:meeting_id>', methods=['GET'])
@login_required
def get_meeting(meeting_id):
"""获取单个组会详情"""
meeting = Meeting.query.get_or_404(meeting_id)
# 获取参会人员列表
registrations = MeetingRegistration.query.filter_by(meeting_id=meeting_id).all()
attendees = [reg.to_dict() for reg in registrations]
# 获取文件列表
files = MeetingFile.query.filter_by(meeting_id=meeting_id).all()
file_list = [f.to_dict() for f in files]
result = meeting.to_dict()
result['attendees'] = attendees
result['files'] = file_list
return jsonify({'meeting': result}), 200
@bp.route('/meetings', methods=['POST'])
@login_required
def create_meeting():
"""创建组会(需要登录)"""
data = request.get_json()
# 解析日期和时间
meeting_date = datetime.strptime(data['date'], '%Y-%m-%d').date() if data.get('date') else None
meeting_time = datetime.strptime(data['time'], '%H:%M').time() if data.get('time') else None
meeting = Meeting(
title=data.get('title'),
description=data.get('description'),
date=meeting_date,
time=meeting_time,
location=data.get('location'),
organizer_id=current_user.id,
status=data.get('status', 'planned'),
meeting_type=data.get('meeting_type', 'weekly')
)
db.session.add(meeting)
db.session.commit()
return jsonify({
'message': 'Meeting created successfully',
'meeting': meeting.to_dict()
}), 201
@bp.route('/meetings/<int:meeting_id>', methods=['PUT'])
@login_required
def update_meeting(meeting_id):
"""更新组会(只有创建者可以编辑)"""
meeting = Meeting.query.get_or_404(meeting_id)
# 检查权限:只有创建者可以编辑
if meeting.organizer_id != current_user.id:
return jsonify({'error': 'Only the organizer can edit this meeting'}), 403
data = request.get_json()
if 'title' in data:
meeting.title = data['title']
if 'description' in data:
meeting.description = data['description']
if 'date' in data:
meeting.date = datetime.strptime(data['date'], '%Y-%m-%d').date()
if 'time' in data:
meeting.time = datetime.strptime(data['time'], '%H:%M').time()
if 'location' in data:
meeting.location = data['location']
if 'status' in data:
meeting.status = data['status']
if 'meeting_type' in data:
meeting.meeting_type = data['meeting_type']
if 'meeting_notes' in data:
meeting.meeting_notes = data['meeting_notes']
meeting.updated_at = datetime.utcnow()
db.session.commit()
return jsonify({
'message': 'Meeting updated successfully',
'meeting': meeting.to_dict()
}), 200
@bp.route('/meetings/<int:meeting_id>', methods=['DELETE'])
@login_required
def delete_meeting(meeting_id):
"""删除组会(只有创建者可以删除)"""
meeting = Meeting.query.get_or_404(meeting_id)
# 检查权限
if meeting.organizer_id != current_user.id:
return jsonify({'error': 'Only the organizer can delete this meeting'}), 403
db.session.delete(meeting)
db.session.commit()
return jsonify({'message': 'Meeting deleted successfully'}), 200
@bp.route('/meetings/<int:meeting_id>/register', methods=['POST'])
@login_required
def register_meeting(meeting_id):
"""报名参加组会"""
meeting = Meeting.query.get_or_404(meeting_id)
# 检查是否已报名
existing = MeetingRegistration.query.filter_by(
meeting_id=meeting_id,
user_id=current_user.id
).first()
if existing:
return jsonify({'error': 'Already registered'}), 400
registration = MeetingRegistration(
meeting_id=meeting_id,
user_id=current_user.id,
status='registered'
)
db.session.add(registration)
db.session.commit()
return jsonify({
'message': 'Registered successfully',
'registration': registration.to_dict()
}), 201
@bp.route('/meetings/<int:meeting_id>/checkin', methods=['POST'])
@login_required
def checkin_meeting(meeting_id):
"""签到组会"""
registration = MeetingRegistration.query.filter_by(
meeting_id=meeting_id,
user_id=current_user.id
).first_or_404()
registration.status = 'checked_in'
db.session.commit()
return jsonify({
'message': 'Checked in successfully',
'registration': registration.to_dict()
}), 200
@bp.route('/meetings/<int:meeting_id>/files', methods=['POST'])
@login_required
def upload_meeting_file(meeting_id):
"""上传组会文件(只有创建者可以上传)"""
meeting = Meeting.query.get_or_404(meeting_id)
# 检查权限
if meeting.organizer_id != current_user.id:
return jsonify({'error': 'Only the organizer can upload files'}), 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', 'doc', 'docx', 'ppt', 'pptx', 'xls', 'xlsx', 'zip', 'rar'}):
filename = secure_filename(file.filename)
uploads_dir = current_app.config['UPLOAD_FOLDER']
meeting_dir = os.path.join(uploads_dir, 'meetings', str(meeting_id))
os.makedirs(meeting_dir, exist_ok=True)
filepath = os.path.join(meeting_dir, filename)
file.save(filepath)
meeting_file = MeetingFile(
meeting_id=meeting_id,
name=request.form.get('name', filename),
file_path=f'/uploads/meetings/{meeting_id}/{filename}',
file_size=os.path.getsize(filepath),
file_type=filename.rsplit('.', 1)[1].lower(),
uploader_id=current_user.id
)
db.session.add(meeting_file)
db.session.commit()
return jsonify({
'message': 'File uploaded successfully',
'file': meeting_file.to_dict()
}), 201
return jsonify({'error': 'Invalid file type'}), 400
@bp.route('/meetings/<int:meeting_id>/files/<int:file_id>', methods=['GET'])
@login_required
def download_meeting_file(meeting_id, file_id):
"""下载组会文件"""
meeting_file = MeetingFile.query.filter_by(
meeting_id=meeting_id,
id=file_id
).first_or_404()
uploads_dir = current_app.config['UPLOAD_FOLDER']
file_path = os.path.join(uploads_dir, 'meetings', str(meeting_id), os.path.basename(meeting_file.file_path))
return send_from_directory(
os.path.dirname(file_path),
os.path.basename(file_path),
as_attachment=True
)
+410
View File
@@ -0,0 +1,410 @@
from flask import Blueprint, request, jsonify, send_from_directory, current_app
from flask_login import login_required, current_user
from werkzeug.utils import secure_filename
import sys
import os
from datetime import datetime, date
# 添加父目录到路径
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from models import db, ResearchProject, ResearchNote, ResearchTask, ResearchFile, User
bp = Blueprint('research', __name__)
def allowed_file(filename, allowed_extensions):
"""检查文件扩展名是否允许"""
return '.' in filename and filename.rsplit('.', 1)[1].lower() in allowed_extensions
# 研究项目管理
@bp.route('/projects', methods=['GET'])
@login_required
def get_projects():
"""获取研究项目列表(实验室成员可查看所有)"""
status = request.args.get('status')
owner_id = request.args.get('owner_id') # 可选:筛选特定用户的项目
query = ResearchProject.query
if status:
query = query.filter_by(status=status)
if owner_id:
query = query.filter_by(owner_id=owner_id)
projects = query.order_by(ResearchProject.created_at.desc()).all()
return jsonify({
'projects': [project.to_dict() for project in projects]
}), 200
@bp.route('/projects/<int:project_id>', methods=['GET'])
@login_required
def get_project(project_id):
"""获取单个研究项目详情"""
project = ResearchProject.query.get_or_404(project_id)
# 获取笔记、任务、文件
notes = ResearchNote.query.filter_by(project_id=project_id).order_by(ResearchNote.created_at.desc()).all()
tasks = ResearchTask.query.filter_by(project_id=project_id).order_by(ResearchTask.created_at.desc()).all()
files = ResearchFile.query.filter_by(project_id=project_id).order_by(ResearchFile.uploaded_at.desc()).all()
result = project.to_dict()
result['notes'] = [note.to_dict() for note in notes]
result['tasks'] = [task.to_dict() for task in tasks]
result['files'] = [f.to_dict() for f in files]
return jsonify({'project': result}), 200
@bp.route('/projects', methods=['POST'])
@login_required
def create_project():
"""创建研究项目(需要登录)"""
data = request.get_json()
start_date = datetime.strptime(data['start_date'], '%Y-%m-%d').date() if data.get('start_date') else None
end_date = datetime.strptime(data['end_date'], '%Y-%m-%d').date() if data.get('end_date') else None
tags_str = ','.join(data.get('tags', [])) if isinstance(data.get('tags'), list) else data.get('tags', '')
project = ResearchProject(
title=data.get('title'),
description=data.get('description'),
start_date=start_date,
end_date=end_date,
owner_id=current_user.id,
status=data.get('status', 'active'),
progress=data.get('progress', 0),
tags=tags_str
)
db.session.add(project)
db.session.commit()
return jsonify({
'message': 'Project created successfully',
'project': project.to_dict()
}), 201
@bp.route('/projects/<int:project_id>', methods=['PUT'])
@login_required
def update_project(project_id):
"""更新研究项目(只有创建者可以编辑)"""
project = ResearchProject.query.get_or_404(project_id)
# 检查权限
if project.owner_id != current_user.id:
return jsonify({'error': 'Only the owner can edit this project'}), 403
data = request.get_json()
if 'title' in data:
project.title = data['title']
if 'description' in data:
project.description = data['description']
if 'start_date' in data:
project.start_date = datetime.strptime(data['start_date'], '%Y-%m-%d').date() if data['start_date'] else None
if 'end_date' in data:
project.end_date = datetime.strptime(data['end_date'], '%Y-%m-%d').date() if data['end_date'] else None
if 'status' in data:
project.status = data['status']
if 'progress' in data:
project.progress = data['progress']
if 'tags' in data:
tags_str = ','.join(data['tags']) if isinstance(data['tags'], list) else data['tags']
project.tags = tags_str
project.updated_at = datetime.utcnow()
db.session.commit()
return jsonify({
'message': 'Project updated successfully',
'project': project.to_dict()
}), 200
@bp.route('/projects/<int:project_id>', methods=['DELETE'])
@login_required
def delete_project(project_id):
"""删除研究项目(只有创建者可以删除)"""
project = ResearchProject.query.get_or_404(project_id)
# 检查权限
if project.owner_id != current_user.id:
return jsonify({'error': 'Only the owner can delete this project'}), 403
db.session.delete(project)
db.session.commit()
return jsonify({'message': 'Project deleted successfully'}), 200
# 研究笔记管理
@bp.route('/projects/<int:project_id>/notes', methods=['GET'])
@login_required
def get_notes(project_id):
"""获取项目笔记列表"""
note_type = request.args.get('type')
query = ResearchNote.query.filter_by(project_id=project_id)
if note_type:
query = query.filter_by(note_type=note_type)
notes = query.order_by(ResearchNote.created_at.desc()).all()
return jsonify({
'notes': [note.to_dict() for note in notes]
}), 200
@bp.route('/projects/<int:project_id>/notes', methods=['POST'])
@login_required
def create_note(project_id):
"""创建研究笔记(需要登录)"""
project = ResearchProject.query.get_or_404(project_id)
data = request.get_json()
note = ResearchNote(
project_id=project_id,
title=data.get('title'),
content=data.get('content'),
author_id=current_user.id,
note_type=data.get('note_type', 'note')
)
db.session.add(note)
db.session.commit()
return jsonify({
'message': 'Note created successfully',
'note': note.to_dict()
}), 201
@bp.route('/notes/<int:note_id>', methods=['PUT'])
@login_required
def update_note(note_id):
"""更新研究笔记(只有创建者可以编辑)"""
note = ResearchNote.query.get_or_404(note_id)
# 检查权限
if note.author_id != current_user.id:
return jsonify({'error': 'Only the author can edit this note'}), 403
data = request.get_json()
if 'title' in data:
note.title = data['title']
if 'content' in data:
note.content = data['content']
if 'note_type' in data:
note.note_type = data['note_type']
note.updated_at = datetime.utcnow()
db.session.commit()
return jsonify({
'message': 'Note updated successfully',
'note': note.to_dict()
}), 200
@bp.route('/notes/<int:note_id>', methods=['DELETE'])
@login_required
def delete_note(note_id):
"""删除研究笔记(只有创建者可以删除)"""
note = ResearchNote.query.get_or_404(note_id)
# 检查权限
if note.author_id != current_user.id:
return jsonify({'error': 'Only the author can delete this note'}), 403
db.session.delete(note)
db.session.commit()
return jsonify({'message': 'Note deleted successfully'}), 200
# 研究任务管理
@bp.route('/projects/<int:project_id>/tasks', methods=['GET'])
@login_required
def get_tasks(project_id):
"""获取项目任务列表"""
status = request.args.get('status')
assignee_id = request.args.get('assignee_id')
query = ResearchTask.query.filter_by(project_id=project_id)
if status:
query = query.filter_by(status=status)
if assignee_id:
query = query.filter_by(assignee_id=assignee_id)
tasks = query.order_by(ResearchTask.due_date, ResearchTask.priority).all()
return jsonify({
'tasks': [task.to_dict() for task in tasks]
}), 200
@bp.route('/projects/<int:project_id>/tasks', methods=['POST'])
@login_required
def create_task(project_id):
"""创建研究任务(项目创建者可以创建)"""
project = ResearchProject.query.get_or_404(project_id)
# 只有项目创建者可以创建任务
if project.owner_id != current_user.id:
return jsonify({'error': 'Only the project owner can create tasks'}), 403
data = request.get_json()
due_date = datetime.strptime(data['due_date'], '%Y-%m-%d').date() if data.get('due_date') else None
task = ResearchTask(
project_id=project_id,
title=data.get('title'),
description=data.get('description'),
assignee_id=data.get('assignee_id'),
due_date=due_date,
status=data.get('status', 'todo'),
priority=data.get('priority', 'medium')
)
db.session.add(task)
db.session.commit()
return jsonify({
'message': 'Task created successfully',
'task': task.to_dict()
}), 201
@bp.route('/tasks/<int:task_id>', methods=['PUT'])
@login_required
def update_task(task_id):
"""更新研究任务(项目创建者或任务分配者可以编辑)"""
task = ResearchTask.query.get_or_404(task_id)
project = ResearchProject.query.get(task.project_id)
# 检查权限:项目创建者或任务分配者
if project.owner_id != current_user.id and task.assignee_id != current_user.id:
return jsonify({'error': 'Only the project owner or assignee can edit this task'}), 403
data = request.get_json()
if 'title' in data:
task.title = data['title']
if 'description' in data:
task.description = data['description']
if 'assignee_id' in data:
task.assignee_id = data['assignee_id']
if 'due_date' in data:
task.due_date = datetime.strptime(data['due_date'], '%Y-%m-%d').date() if data['due_date'] else None
if 'status' in data:
task.status = data['status']
if 'priority' in data:
task.priority = data['priority']
task.updated_at = datetime.utcnow()
db.session.commit()
return jsonify({
'message': 'Task updated successfully',
'task': task.to_dict()
}), 200
@bp.route('/tasks/<int:task_id>', methods=['DELETE'])
@login_required
def delete_task(task_id):
"""删除研究任务(只有项目创建者可以删除)"""
task = ResearchTask.query.get_or_404(task_id)
project = ResearchProject.query.get(task.project_id)
# 检查权限
if project.owner_id != current_user.id:
return jsonify({'error': 'Only the project owner can delete this task'}), 403
db.session.delete(task)
db.session.commit()
return jsonify({'message': 'Task deleted successfully'}), 200
# 研究文件管理
@bp.route('/projects/<int:project_id>/files', methods=['POST'])
@login_required
def upload_research_file(project_id):
"""上传研究文件(需要登录)"""
project = ResearchProject.query.get_or_404(project_id)
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', 'doc', 'docx', 'ppt', 'pptx', 'xls', 'xlsx', 'zip', 'rar', 'txt', 'csv', 'json', 'py', 'ipynb'}):
filename = secure_filename(file.filename)
uploads_dir = current_app.config['UPLOAD_FOLDER']
project_dir = os.path.join(uploads_dir, 'research', str(project_id))
os.makedirs(project_dir, exist_ok=True)
filepath = os.path.join(project_dir, filename)
file.save(filepath)
research_file = ResearchFile(
project_id=project_id,
name=request.form.get('name', filename),
file_path=f'/uploads/research/{project_id}/{filename}',
file_size=os.path.getsize(filepath),
file_type=filename.rsplit('.', 1)[1].lower(),
uploader_id=current_user.id
)
db.session.add(research_file)
db.session.commit()
return jsonify({
'message': 'File uploaded successfully',
'file': research_file.to_dict()
}), 201
return jsonify({'error': 'Invalid file type'}), 400
@bp.route('/projects/<int:project_id>/files/<int:file_id>', methods=['GET'])
@login_required
def download_research_file(project_id, file_id):
"""下载研究文件"""
research_file = ResearchFile.query.filter_by(
project_id=project_id,
id=file_id
).first_or_404()
uploads_dir = current_app.config['UPLOAD_FOLDER']
file_path = os.path.join(uploads_dir, 'research', str(project_id), os.path.basename(research_file.file_path))
return send_from_directory(
os.path.dirname(file_path),
os.path.basename(file_path),
as_attachment=True
)
@bp.route('/projects/<int:project_id>/files/<int:file_id>', methods=['DELETE'])
@login_required
def delete_research_file(project_id, file_id):
"""删除研究文件(只有上传者可以删除)"""
research_file = ResearchFile.query.filter_by(
project_id=project_id,
id=file_id
).first_or_404()
# 检查权限
if research_file.uploader_id != current_user.id:
return jsonify({'error': 'Only the uploader can delete this file'}), 403
db.session.delete(research_file)
db.session.commit()
return jsonify({'message': 'File deleted successfully'}), 200