bluish
6ecd95ad5d
主要改动： - 更新依赖：使用 fastapi、uvicorn 替代 Flask - 数据库层：从 Flask-SQLAlchemy 迁移到纯 SQLAlchemy - 新增 database.py 管理数据库连接和会话 - 路由层：从 Blueprint 迁移到 APIRouter，添加 Pydantic 模型验证 - 应用层：使用 FastAPI 中间件替代 Flask 插件 - 启动方式：使用 uvicorn 替代 Flask 开发服务器 - 更新 Docker 配置以支持 FastAPI 优势： - 更高的性能和异步支持 - 自动生成 OpenAPI 文档 - 更好的类型安全和数据验证 - 所有 API 端点保持向后兼容 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
100 lines
3.7 KiB
Python
100 lines
3.7 KiB
Python
from sqlalchemy import Column, Integer, String, Text, DateTime, ForeignKey, func
|
|
from sqlalchemy.ext.declarative import declarative_base
|
|
from sqlalchemy.orm import relationship
|
|
from datetime import datetime
|
|
from werkzeug.security import generate_password_hash, check_password_hash
|
|
|
|
Base = declarative_base()
|
|
|
|
class User(Base):
|
|
"""用户模型"""
|
|
__tablename__ = 'users'
|
|
|
|
id = Column(Integer, primary_key=True)
|
|
username = Column(String(80), unique=True, nullable=False)
|
|
password_hash = Column(String(200), nullable=False)
|
|
created_at = Column(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,
|
|
'created_at': self.created_at.isoformat() if self.created_at else None
|
|
}
|
|
|
|
class Task(Base):
|
|
"""任务模型"""
|
|
__tablename__ = 'tasks'
|
|
|
|
id = Column(Integer, primary_key=True)
|
|
title = Column(String(200), nullable=False)
|
|
description = Column(Text)
|
|
polished_description = Column(Text) # AI润色后的描述
|
|
status = Column(String(20), default='pending') # pending, in_progress, completed
|
|
created_at = Column(DateTime, default=datetime.utcnow)
|
|
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
|
|
|
|
# 关联关系
|
|
time_records = relationship('TimeRecord', back_populates='task', cascade='all, delete-orphan')
|
|
|
|
def to_dict(self, db_session=None):
|
|
result = {
|
|
'id': self.id,
|
|
'title': self.title,
|
|
'description': self.description,
|
|
'polished_description': self.polished_description,
|
|
'status': self.status,
|
|
'created_at': self.created_at.isoformat() if self.created_at else None,
|
|
'updated_at': self.updated_at.isoformat() if self.updated_at else None,
|
|
}
|
|
if db_session:
|
|
result['total_time'] = self.get_total_time(db_session)
|
|
else:
|
|
result['total_time'] = 0
|
|
return result
|
|
|
|
def get_total_time(self, db_session):
|
|
"""获取任务总时长（秒）"""
|
|
total_seconds = db_session.query(func.sum(TimeRecord.duration)).filter(
|
|
TimeRecord.task_id == self.id
|
|
).scalar() or 0
|
|
return int(total_seconds)
|
|
|
|
class TimeRecord(Base):
|
|
"""时间记录模型"""
|
|
__tablename__ = 'time_records'
|
|
|
|
id = Column(Integer, primary_key=True)
|
|
task_id = Column(Integer, ForeignKey('tasks.id'), nullable=False)
|
|
start_time = Column(DateTime, nullable=False)
|
|
end_time = Column(DateTime)
|
|
duration = Column(Integer) # 时长（秒）
|
|
created_at = Column(DateTime, default=datetime.utcnow)
|
|
|
|
# 关联关系
|
|
task = relationship('Task', back_populates='time_records')
|
|
|
|
def to_dict(self):
|
|
return {
|
|
'id': self.id,
|
|
'task_id': self.task_id,
|
|
'start_time': self.start_time.isoformat() if self.start_time else None,
|
|
'end_time': self.end_time.isoformat() if self.end_time else None,
|
|
'duration': self.duration,
|
|
'created_at': self.created_at.isoformat() if self.created_at else None
|
|
}
|
|
|
|
def calculate_duration(self):
|
|
"""计算时长"""
|
|
if self.start_time and self.end_time:
|
|
self.duration = int((self.end_time - self.start_time).total_seconds())
|
|
return self.duration
|