xsl/ocean_project_manager

Fork 0

Files

T

xsl e4b59c5ee4 save code

2026-01-26 08:04:53 +08:00

56 KiB

Raw Blame History

海洋项目管理系统 - 后端实现计划

For Claude: REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.

Goal: 构建完整的后端API系统，包括认证、用户管理、项目管理和统计功能

Architecture: 基于FastAPI的异步后端服务，使用SQLAlchemy 2.0 ORM操作MySQL数据库，JWT Token认证，Pydantic v2进行数据验证。

Tech Stack:

FastAPI 0.104.1 - Web框架
SQLAlchemy 2.0.23 - ORM
aiomysql 0.2.0 - MySQL异步驱动
PyJWT 2.8.0 - JWT认证
bcrypt 4.1.1 - 密码加密
Pydantic 2.5.0 - 数据验证

Task 1: 创建项目基础配置文件

Files:

Create: backend/requirements.txt
Create: backend/config/__init__.py
Create: backend/config/settings.py
Create: backend/config/database.py

Step 1: 创建依赖文件

# backend/requirements.txt
fastapi==0.104.1
uvicorn[standard]==0.24.0
sqlalchemy==2.0.23
aiomysql==0.2.0
pydantic==2.5.0
pydantic-settings==2.1.0
pyjwt==2.8.0
bcrypt==4.1.1
python-multipart==0.0.6

Step 2: 创建配置类

# backend/config/settings.py
from pydantic_settings import BaseSettings
from functools import lru_cache


class Settings(BaseSettings):
    # 应用配置
    APP_NAME: str = "海洋项目管理系统"
    APP_VERSION: str = "1.0.0"
    DEBUG: bool = True

    # 数据库配置
    DB_HOST: str = "localhost"
    DB_PORT: int = 3306
    DB_USER: str = "root"
    DB_PASSWORD: str = "rootpassword"
    DB_NAME: str = "project_manager"
    DB_CHARSET: str = "utf8mb4"

    # JWT配置
    SECRET_KEY: str = "your-secret-key-change-this-in-production"
    ALGORITHM: str = "HS256"
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 1440  # 24小时

    # CORS配置
    CORS_ORIGINS: list[str] = ["http://localhost:3000", "http://127.0.0.1:3000"]

    class Config:
        env_file = ".env"
        case_sensitive = True


@lru_cache()
def get_settings():
    return Settings()

Step 3: 创建数据库配置

# backend/config/database.py
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from sqlalchemy.orm import declarative_base
from .settings import get_settings

settings = get_settings()

# 构建数据库连接URL
DATABASE_URL = (
    f"mysql+aiomysql://{settings.DB_USER}:{settings.DB_PASSWORD}"
    f"@{settings.DB_HOST}:{settings.DB_PORT}/{settings.DB_NAME}"
    f"?charset={settings.DB_CHARSET}"
)

# 创建异步引擎
engine = create_async_engine(DATABASE_URL, echo=settings.DEBUG, future=True)

# 创建会话工厂
AsyncSessionLocal = async_sessionmaker(
    engine, class_=AsyncSession, expire_on_commit=False
)

# 创建基类
Base = declarative_base()


# 依赖注入：获取数据库会话
async def get_db():
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()


# 初始化数据库表
async def init_db():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

Step 4: 创建配置包初始化文件

# backend/config/__init__.py
from .settings import get_settings, Settings
from .database import get_db, init_db, Base, engine, AsyncSessionLocal

__all__ = [
    "get_settings",
    "Settings",
    "get_db",
    "init_db",
    "Base",
    "engine",
    "AsyncSessionLocal",
]

Step 5: 安装依赖

Run: cd backend && pip install -r requirements.txt -r requirements-test.txt Expected: 所有依赖安装成功

Step 6: 提交

git add backend/requirements.txt backend/config/
git commit -m "[backend] feat: 添加项目基础配置"

Task 2: 创建工具函数模块

Files:

Create: backend/src/utils/__init__.py
Create: backend/src/utils/password.py
Create: backend/src/utils/jwt.py

Step 1: 创建密码工具

# backend/src/utils/password.py
import bcrypt


def hash_password(password: str) -> str:
    """哈希密码"""
    salt = bcrypt.gensalt(rounds=12)
    hashed = bcrypt.hashpw(password.encode("utf-8"), salt)
    return hashed.decode("utf-8")


def verify_password(password: str, hashed_password: str) -> bool:
    """验证密码"""
    return bcrypt.checkpw(
        password.encode("utf-8"), hashed_password.encode("utf-8")
    )

Step 2: 创建JWT工具

# backend/src/utils/jwt.py
from datetime import datetime, timedelta
from typing import Optional
from jose import JWTError, jwt
from config.settings import get_settings

settings = get_settings()


def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
    """创建访问令牌"""
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(
            minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES
        )
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(
        to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM
    )
    return encoded_jwt


def decode_access_token(token: str) -> Optional[dict]:
    """解码访问令牌"""
    try:
        payload = jwt.decode(
            token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM]
        )
        return payload
    except JWTError:
        return None

Step 3: 创建工具包初始化文件

# backend/src/utils/__init__.py
from .password import hash_password, verify_password
from .jwt import create_access_token, decode_access_token

__all__ = [
    "hash_password",
    "verify_password",
    "create_access_token",
    "decode_access_token",
]

Step 4: 测试密码工具

Run: cd backend && python -c "from src.utils.password import hash_password; print(hash_password('test123'))" Expected: 输出哈希后的密码

Step 5: 提交

git add backend/src/utils/
git commit -m "[backend] feat: 添加密码和JWT工具函数"

Task 3: 创建数据库模型

Files:

Create: backend/src/models/__init__.py
Create: backend/src/models/user.py
Create: backend/src/models/project.py

Step 1: 创建用户模型

# backend/src/models/user.py
from sqlalchemy import Column, Integer, String, Boolean, DateTime, Enum
from sqlalchemy.sql import func
from config.database import Base


class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, autoincrement=True, comment="用户ID")
    username = Column(String(50), unique=True, nullable=False, comment="用户名")
    password_hash = Column(String(255), nullable=False, comment="密码哈希")
    real_name = Column(String(100), nullable=False, comment="真实姓名")
    department = Column(String(50), nullable=False, comment="部门")
    role = Column(Enum("admin", "market", "other"), nullable=False, comment="角色")
    email = Column(String(100), unique=True, comment="邮箱")
    phone = Column(String(20), comment="电话")
    is_active = Column(Boolean, default=True, comment="是否激活")
    created_at = Column(
        DateTime,
        server_default=func.now(),
        comment="创建时间",
    )
    updated_at = Column(
        DateTime,
        server_default=func.now(),
        onupdate=func.now(),
        comment="更新时间",
    )

Step 2: 创建项目模型

# backend/src/models/project.py
from sqlalchemy import Column, Integer, String, Text, Date, Numeric, DateTime, Enum
from sqlalchemy.sql import func
from config.database import Base


class Project(Base):
    __tablename__ = "projects"

    id = Column(Integer, primary_key=True, autoincrement=True, comment="项目ID")

    # 基础信息
    project_no = Column(String(50), unique=True, nullable=False, comment="合同编号")
    power_contract_no = Column(String(100), comment="供电局项目合同编号")
    name = Column(String(200), nullable=False, comment="项目名称")
    subitem_count = Column(Integer, default=0, comment="子项个数")
    subitem_code = Column(String(50), comment="子项编码")
    total_investment = Column(Numeric(15, 2), comment="项目总投资（万元）")
    contract_amount = Column(Numeric(15, 2), comment="中标合同金额（万元）")
    warranty_ratio = Column(Numeric(5, 2), comment="质保金比例")
    settlement_amount = Column(Numeric(15, 2), comment="结算金额（万元）")
    total_cost_estimated = Column(Numeric(15, 2), comment="总成本测算")
    voltage_level = Column(String(50), comment="工程电压等级")
    engineering_type = Column(String(50), comment="工程类别")
    owner_unit = Column(String(200), comment="业主单位")
    owner_contact = Column(String(200), comment="业主联系人及电话")
    bidding_type = Column(String(50), comment="中标形式")
    signing_date = Column(Date, comment="签订日期")
    start_date = Column(Date, comment="开工日期")
    planned_end_date = Column(Date, comment="计划竣工日期")
    actual_end_date = Column(Date, comment="实际竣工日期")

    # 质保信息
    warranty_amount = Column(Numeric(15, 2), default=0, comment="质保金（万元）")
    warranty_expiry_date = Column(Date, comment="质保期截止日")
    actual_warranty_refund_date = Column(Date, comment="实际退质保金日期")

    # 项目信息
    project_department = Column(String(100), comment="所属项目部")
    project_leader = Column(String(200), comment="项目负责人及电话")
    payment_method = Column(Text, comment="工程款拨付方式")

    # 成本控制
    total_cost_control = Column(Numeric(15, 2), comment="总体成本（控制）")
    is_adjusted = Column(Enum("是", "否"), default="否", comment="是否调整")
    labor_cost_control = Column(Numeric(15, 2), comment="人工成本（控制）")
    labor_cost_planned = Column(Numeric(15, 2), comment="农民工工资（计划）")
    labor_cost_paid = Column(Numeric(15, 2), comment="农民工工资（实付）")
    material_cost_control = Column(Numeric(15, 2), comment="乙供材料费（控制）")
    material_cost_payable = Column(Numeric(15, 2), comment="应付材料费")
    material_cost_actual = Column(Numeric(15, 2), comment="实际发生材料费")
    material_cost_paid = Column(Numeric(15, 2), comment="实际支付材料费")
    other_cost_control = Column(Numeric(15, 2), comment="其他费用（控制）")
    other_cost_payable = Column(Numeric(15, 2), comment="应付其他费")
    other_cost_actual = Column(Numeric(15, 2), comment="实际其他费用")

    # 财务信息
    tax_amount = Column(Numeric(15, 2), comment="税金")
    profit = Column(Numeric(15, 2), comment="利润（万元）")
    actual_profit = Column(Numeric(15, 2), comment="实际利润（万元）")
    cost_settlement_amount = Column(Numeric(15, 2), comment="成本结算金额（万元）")
    cumulative_progress = Column(Numeric(5, 2), comment="累计进度")
    receivable_amount = Column(Numeric(15, 2), comment="应收款（完成进度款）")
    invoice_amount = Column(Numeric(15, 2), comment="开票金额（万元）")
    actual_receipt_amount = Column(Numeric(15, 2), comment="实际收款金额（万元）")
    receipt_completion_rate = Column(Numeric(5, 2), comment="实际收款完成率")
    payable_amount = Column(Numeric(15, 2), comment="应付款金额（万元）")
    actual_payment_amount = Column(Numeric(15, 2), comment="实际付款金额（万元）")
    unpaid_amount = Column(Numeric(15, 2), comment="未收款（万元）")
    payment_completion_rate = Column(Numeric(5, 2), comment="实际付款完成率")

    # 民工工资
    labor_debt_amount = Column(Numeric(15, 2), comment="民工工资清欠金额（万元）")

    # 结算信息
    settlement_cost_amount = Column(Numeric(15, 2), comment="结算后成本测算金额")
    settlement_labor_cost = Column(Numeric(15, 2), comment="结算人工费")
    settlement_material_cost = Column(Numeric(15, 2), comment="结算材料费")
    settlement_other_cost = Column(Numeric(15, 2), comment="结算其他费")

    # 统计信息
    due_settlement_count = Column(Integer, comment="到期应结算项目个数")
    unsettlement_count = Column(Integer, comment="到期未完成结算个数")

    # 其他
    problems = Column(Text, comment="存在的问题")
    suggestions = Column(Text, comment="建议措施")
    remarks = Column(Text, comment="备注")

    # 系统字段
    created_by = Column(Integer, nullable=False, comment="创建人ID")
    created_at = Column(
        DateTime,
        server_default=func.now(),
        comment="创建时间",
    )
    updated_at = Column(
        DateTime,
        server_default=func.now(),
        onupdate=func.now(),
        comment="更新时间",
    )

Step 3: 创建模型包初始化文件

# backend/src/models/__init__.py
from .user import User
from .project import Project

__all__ = ["User", "Project"]

Step 4: 提交

git add backend/src/models/
git commit -m "[backend] feat: 添加用户和项目数据模型"

Task 4: 创建Pydantic Schema模型

Files:

Create: backend/src/schemas/__init__.py
Create: backend/src/schemas/user.py
Create: backend/src/schemas/project.py
Create: backend/src/schemas/auth.py

Step 1: 创建用户Schema

# backend/src/schemas/user.py
from pydantic import BaseModel, EmailStr, Field
from typing import Optional
from datetime import datetime


# 用户基础信息
class UserBase(BaseModel):
    username: str = Field(..., min_length=3, max_length=50)
    real_name: str = Field(..., max_length=100)
    department: str = Field(..., max_length=50)
    role: str = Field(..., pattern="^(admin|market|other)$")
    email: Optional[EmailStr] = None
    phone: Optional[str] = Field(None, max_length=20)


# 创建用户请求
class UserCreate(UserBase):
    password: str = Field(..., min_length=6)


# 更新用户请求
class UserUpdate(BaseModel):
    real_name: Optional[str] = Field(None, max_length=100)
    email: Optional[EmailStr] = None
    phone: Optional[str] = Field(None, max_length=20)
    is_active: Optional[bool] = None


# 重置密码请求
class PasswordReset(BaseModel):
    new_password: str = Field(..., min_length=6)


# 用户响应
class UserResponse(UserBase):
    id: int
    is_active: bool
    created_at: datetime
    updated_at: Optional[datetime]

    class Config:
        from_attributes = True


# 用户列表响应
class UserListResponse(BaseModel):
    items: list[UserResponse]
    total: int
    page: int
    page_size: int

Step 2: 创建认证Schema

# backend/src/schemas/auth.py
from pydantic import BaseModel, Field
from typing import Optional
from .user import UserResponse


# 登录请求
class LoginRequest(BaseModel):
    username: str = Field(..., min_length=3)
    password: str = Field(..., min_length=6)


# 登录响应
class LoginResponse(BaseModel):
    token: str
    token_type: str = "bearer"
    user: UserResponse

Step 3: 创建项目Schema

# backend/src/schemas/project.py
from pydantic import BaseModel, Field
from typing import Optional
from datetime import date, datetime
from decimal import Decimal


# 项目基础信息
class ProjectBase(BaseModel):
    project_no: str = Field(..., max_length=50)
    name: str = Field(..., max_length=200)
    engineering_type: str = Field(..., max_length=50)
    contract_amount: Optional[Decimal] = Field(None, ge=0)


# 创建项目请求
class ProjectCreate(ProjectBase):
    power_contract_no: Optional[str] = Field(None, max_length=100)
    subitem_count: Optional[int] = Field(None, ge=0)
    subitem_code: Optional[str] = Field(None, max_length=50)
    total_investment: Optional[Decimal] = Field(None, ge=0)
    warranty_ratio: Optional[Decimal] = None
    settlement_amount: Optional[Decimal] = None
    total_cost_estimated: Optional[Decimal] = None
    voltage_level: Optional[str] = Field(None, max_length=50)
    owner_unit: Optional[str] = Field(None, max_length=200)
    owner_contact: Optional[str] = Field(None, max_length=200)
    bidding_type: Optional[str] = Field(None, max_length=50)
    signing_date: Optional[date] = None
    start_date: Optional[date] = None
    planned_end_date: Optional[date] = None
    project_department: Optional[str] = Field(None, max_length=100)
    project_leader: Optional[str] = Field(None, max_length=200)
    payment_method: Optional[str] = None
    total_cost_control: Optional[Decimal] = None
    labor_cost_control: Optional[Decimal] = None
    material_cost_control: Optional[Decimal] = None
    other_cost_control: Optional[Decimal] = None
    remarks: Optional[str] = None


# 更新项目请求
class ProjectUpdate(BaseModel):
    # 基础信息
    name: Optional[str] = Field(None, max_length=200)
    engineering_type: Optional[str] = Field(None, max_length=50)
    contract_amount: Optional[Decimal] = Field(None, ge=0)

    # 财务信息
    total_investment: Optional[Decimal] = None
    settlement_amount: Optional[Decimal] = None
    actual_receipt_amount: Optional[Decimal] = None
    actual_payment_amount: Optional[Decimal] = None
    invoice_amount: Optional[Decimal] = None
    payable_amount: Optional[Decimal] = None
    cumulative_progress: Optional[Decimal] = None

    # 成本信息
    labor_cost_paid: Optional[Decimal] = None
    material_cost_actual: Optional[Decimal] = None
    material_cost_paid: Optional[Decimal] = None
    other_cost_actual: Optional[Decimal] = None
    tax_amount: Optional[Decimal] = None
    profit: Optional[Decimal] = None

    # 其他
    remarks: Optional[str] = None
    problems: Optional[str] = None
    suggestions: Optional[str] = None


# 项目响应
class ProjectResponse(BaseModel):
    id: int
    project_no: str
    power_contract_no: Optional[str]
    name: str
    subitem_count: Optional[int]
    subitem_code: Optional[str]
    total_investment: Optional[Decimal]
    contract_amount: Optional[Decimal]
    warranty_ratio: Optional[Decimal]
    settlement_amount: Optional[Decimal]
    total_cost_estimated: Optional[Decimal]
    voltage_level: Optional[str]
    engineering_type: Optional[str]
    owner_unit: Optional[str]
    owner_contact: Optional[str]
    bidding_type: Optional[str]
    signing_date: Optional[date]
    start_date: Optional[date]
    planned_end_date: Optional[date]
    actual_end_date: Optional[date]
    warranty_amount: Optional[Decimal]
    warranty_expiry_date: Optional[date]
    actual_warranty_refund_date: Optional[date]
    project_department: Optional[str]
    project_leader: Optional[str]
    payment_method: Optional[str]
    total_cost_control: Optional[Decimal]
    is_adjusted: Optional[str]
    labor_cost_control: Optional[Decimal]
    labor_cost_planned: Optional[Decimal]
    labor_cost_paid: Optional[Decimal]
    material_cost_control: Optional[Decimal]
    material_cost_payable: Optional[Decimal]
    material_cost_actual: Optional[Decimal]
    material_cost_paid: Optional[Decimal]
    other_cost_control: Optional[Decimal]
    other_cost_payable: Optional[Decimal]
    other_cost_actual: Optional[Decimal]
    tax_amount: Optional[Decimal]
    profit: Optional[Decimal]
    actual_profit: Optional[Decimal]
    cost_settlement_amount: Optional[Decimal]
    cumulative_progress: Optional[Decimal]
    receivable_amount: Optional[Decimal]
    invoice_amount: Optional[Decimal]
    actual_receipt_amount: Optional[Decimal]
    receipt_completion_rate: Optional[Decimal]
    payable_amount: Optional[Decimal]
    actual_payment_amount: Optional[Decimal]
    unpaid_amount: Optional[Decimal]
    payment_completion_rate: Optional[Decimal]
    labor_debt_amount: Optional[Decimal]
    settlement_cost_amount: Optional[Decimal]
    settlement_labor_cost: Optional[Decimal]
    settlement_material_cost: Optional[Decimal]
    settlement_other_cost: Optional[Decimal]
    due_settlement_count: Optional[int]
    unsettlement_count: Optional[int]
    problems: Optional[str]
    suggestions: Optional[str]
    remarks: Optional[str]
    created_by: int
    created_by_name: Optional[str]
    created_at: datetime
    updated_at: Optional[datetime]

    class Config:
        from_attributes = True


# 项目列表响应
class ProjectListResponse(BaseModel):
    items: list[ProjectResponse]
    total: int
    page: int
    page_size: int

Step 4: 创建Schema包初始化文件

# backend/src/schemas/__init__.py
from .auth import LoginRequest, LoginResponse
from .user import (
    UserCreate,
    UserUpdate,
    UserResponse,
    UserListResponse,
    PasswordReset,
)
from .project import (
    ProjectCreate,
    ProjectUpdate,
    ProjectResponse,
    ProjectListResponse,
)

__all__ = [
    "LoginRequest",
    "LoginResponse",
    "UserCreate",
    "UserUpdate",
    "UserResponse",
    "UserListResponse",
    "PasswordReset",
    "ProjectCreate",
    "ProjectUpdate",
    "ProjectResponse",
    "ProjectListResponse",
]

Step 5: 提交

git add backend/src/schemas/
git commit -m "[backend] feat: 添加Pydantic Schema模型"

Task 5: 创建认证中间件和依赖注入

Files:

Create: backend/src/middleware/__init__.py
Create: backend/src/middleware/auth.py
Create: backend/src/dependencies.py

Step 1: 创建认证中间件

# backend/src/middleware/auth.py
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from config.database import get_db
from src.models.user import User
from src.utils.jwt import decode_access_token

security = HTTPBearer()


async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security),
    db: AsyncSession = Depends(get_db),
) -> User:
    """获取当前用户"""
    token = credentials.credentials

    # 解码token
    payload = decode_access_token(token)
    if payload is None:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Token无效或过期",
            headers={"WWW-Authenticate": "Bearer"},
        )

    user_id = payload.get("sub")
    if user_id is None:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Token无效",
        )

    # 查询用户
    result = await db.execute(select(User).where(User.id == int(user_id)))
    user = result.scalar_one_or_none()

    if user is None:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="用户不存在",
        )

    if not user.is_active:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="用户已被禁用",
        )

    return user


async def get_current_active_user(
    current_user: User = Depends(get_current_user),
) -> User:
    """获取当前激活用户"""
    if not current_user.is_active:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="用户未激活",
        )
    return current_user


async def require_admin(
    current_user: User = Depends(get_current_active_user),
) -> User:
    """要求管理员权限"""
    if current_user.role != "admin":
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="需要管理员权限",
        )
    return current_user


async def require_market_or_admin(
    current_user: User = Depends(get_current_active_user),
) -> User:
    """要求市场部或管理员权限"""
    if current_user.role not in ["admin", "market"]:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="需要市场部或管理员权限",
        )
    return current_user

Step 2: 创建依赖注入模块

# backend/src/dependencies.py
from fastapi import Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from config.database import get_db
from src.models.user import User
from src.models.project import Project
from .middleware.auth import get_current_user


async def get_project_or_404(
    project_id: int,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
) -> Project:
    """获取项目或返回404"""
    result = await db.execute(select(Project).where(Project.id == project_id))
    project = result.scalar_one_or_none()

    if project is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="项目不存在",
        )

    return project


async def get_project_if_owner_or_admin(
    project_id: int,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
) -> Project:
    """获取项目（仅限创建者或管理员）"""
    project = await get_project_or_404(project_id, current_user, db)

    if current_user.role != "admin" and project.created_by != current_user.id:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="无权访问此项目",
        )

    return project


async def get_user_or_404(
    user_id: int,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
) -> User:
    """获取用户或返回404"""
    result = await db.execute(select(User).where(User.id == user_id))
    user = result.scalar_one_or_none()

    if user is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="用户不存在",
        )

    return user

Step 3: 创建中间件包初始化文件

# backend/src/middleware/__init__.py
from .auth import (
    get_current_user,
    get_current_active_user,
    require_admin,
    require_market_or_admin,
)

__all__ = [
    "get_current_user",
    "get_current_active_user",
    "require_admin",
    "require_market_or_admin",
]

Step 4: 提交

git add backend/src/middleware/ backend/src/dependencies.py
git commit -m "[backend] feat: 添加认证中间件和依赖注入"

Task 6: 创建认证路由

Files:

Create: backend/src/routes/__init__.py
Create: backend/src/routes/auth.py

Step 1: 创建认证路由

# backend/src/routes/auth.py
from fastapi import APIRouter, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from config.database import get_db
from src.models.user import User
from src.schemas.auth import LoginRequest, LoginResponse
from src.schemas.user import UserResponse
from src.utils.password import verify_password
from src.utils.jwt import create_access_token
from src.middleware.auth import get_current_user
from datetime import timedelta

router = APIRouter(prefix="/auth", tags=["认证"])


@router.post("/login", response_model=dict)
async def login(
    login_data: LoginRequest,
    db: AsyncSession = Depends(get_db),
):
    """用户登录"""
    # 查询用户
    result = await db.execute(
        select(User).where(User.username == login_data.username)
    )
    user = result.scalar_one_or_none()

    # 验证用户和密码
    if not user or not verify_password(login_data.password, user.password_hash):
        return {
            "success": False,
            "message": "用户名或密码错误",
            "data": None,
            "error_code": "1002",
        }

    if not user.is_active:
        return {
            "success": False,
            "message": "用户未激活",
            "data": None,
            "error_code": "1002",
        }

    # 生成token
    access_token = create_access_token(
        data={"sub": str(user.id), "username": user.username}
    )

    return {
        "success": True,
        "message": "登录成功",
        "data": {
            "token": access_token,
            "token_type": "bearer",
            "user": UserResponse.model_validate(user),
        },
        "error_code": None,
    }


@router.get("/me", response_model=dict)
async def get_current_user_info(
    current_user: User = Depends(get_current_user),
):
    """获取当前用户信息"""
    return {
        "success": True,
        "message": "获取成功",
        "data": UserResponse.model_validate(current_user),
        "error_code": None,
    }


@router.post("/logout", response_model=dict)
async def logout(current_user: User = Depends(get_current_user)):
    """用户登出"""
    return {
        "success": True,
        "message": "登出成功",
        "data": None,
        "error_code": None,
    }

Step 2: 创建路由包初始化文件

# backend/src/routes/__init__.py
from .auth import router as auth_router
from .users import router as users_router
from .projects import router as projects_router

__all__ = [
    "auth_router",
    "users_router",
    "projects_router",
]

Step 3: 提交

git add backend/src/routes/auth.py
git commit -m "[backend] feat: 添加认证路由"

Task 7: 创建用户管理路由

Files:

Create: backend/src/routes/users.py

Step 1: 创建用户管理路由

# backend/src/routes/users.py
from fastapi import APIRouter, Depends, Query
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, func, or_
from config.database import get_db
from src.models.user import User
from src.schemas.user import (
    UserCreate,
    UserUpdate,
    UserResponse,
    UserListResponse,
    PasswordReset,
)
from src.utils.password import hash_password
from src.middleware.auth import require_admin
from typing import Optional

router = APIRouter(prefix="/users", tags=["用户管理"])


@router.get("", response_model=dict)
async def get_users(
    page: int = Query(1, ge=1),
    page_size: int = Query(10, ge=1, le=100),
    department: Optional[str] = None,
    role: Optional[str] = None,
    keyword: Optional[str] = None,
    current_user: User = Depends(require_admin),
    db: AsyncSession = Depends(get_db),
):
    """获取用户列表"""
    # 构建查询
    query = select(User)

    # 筛选条件
    if department:
        query = query.where(User.department == department)
    if role:
        query = query.where(User.role == role)
    if keyword:
        query = query.where(
            or_(
                User.username.contains(keyword),
                User.real_name.contains(keyword),
                User.email.contains(keyword),
            )
        )

    # 总数
    total_query = select(func.count()).select_from(query.subquery())
    total_result = await db.execute(total_query)
    total = total_result.scalar()

    # 分页
    query = query.offset((page - 1) * page_size).limit(page_size)
    result = await db.execute(query)
    users = result.scalars().all()

    return {
        "success": True,
        "message": "获取成功",
        "data": {
            "items": [UserResponse.model_validate(u) for u in users],
            "total": total,
            "page": page,
            "page_size": page_size,
        },
        "error_code": None,
    }


@router.post("", response_model=dict)
async def create_user(
    user_data: UserCreate,
    current_user: User = Depends(require_admin),
    db: AsyncSession = Depends(get_db),
):
    """创建用户"""
    # 检查用户名是否已存在
    result = await db.execute(
        select(User).where(User.username == user_data.username)
    )
    if result.scalar_one_or_none():
        return {
            "success": False,
            "message": "用户名已存在",
            "data": None,
            "error_code": "2002",
        }

    # 检查邮箱是否已存在
    if user_data.email:
        result = await db.execute(
            select(User).where(User.email == user_data.email)
        )
        if result.scalar_one_or_none():
            return {
                "success": False,
                "message": "邮箱已存在",
                "data": None,
                "error_code": "2002",
            }

    # 创建用户
    user = User(
        username=user_data.username,
        password_hash=hash_password(user_data.password),
        real_name=user_data.real_name,
        department=user_data.department,
        role=user_data.role,
        email=user_data.email,
        phone=user_data.phone,
    )
    db.add(user)
    await db.commit()
    await db.refresh(user)

    return {
        "success": True,
        "message": "用户创建成功",
        "data": UserResponse.model_validate(user),
        "error_code": None,
    }


@router.get("/{user_id}", response_model=dict)
async def get_user(
    user_id: int,
    current_user: User = Depends(require_admin),
    db: AsyncSession = Depends(get_db),
):
    """获取用户详情"""
    result = await db.execute(select(User).where(User.id == user_id))
    user = result.scalar_one_or_none()

    if not user:
        return {
            "success": False,
            "message": "用户不存在",
            "data": None,
            "error_code": "2001",
        }

    return {
        "success": True,
        "message": "获取成功",
        "data": UserResponse.model_validate(user),
        "error_code": None,
    }


@router.put("/{user_id}", response_model=dict)
async def update_user(
    user_id: int,
    user_data: UserUpdate,
    current_user: User = Depends(require_admin),
    db: AsyncSession = Depends(get_db),
):
    """更新用户"""
    result = await db.execute(select(User).where(User.id == user_id))
    user = result.scalar_one_or_none()

    if not user:
        return {
            "success": False,
            "message": "用户不存在",
            "data": None,
            "error_code": "2001",
        }

    # 更新字段
    update_data = user_data.model_dump(exclude_unset=True)
    for field, value in update_data.items():
        setattr(user, field, value)

    await db.commit()
    await db.refresh(user)

    return {
        "success": True,
        "message": "用户更新成功",
        "data": UserResponse.model_validate(user),
        "error_code": None,
    }


@router.delete("/{user_id}", response_model=dict)
async def delete_user(
    user_id: int,
    current_user: User = Depends(require_admin),
    db: AsyncSession = Depends(get_db),
):
    """删除用户"""
    result = await db.execute(select(User).where(User.id == user_id))
    user = result.scalar_one_or_none()

    if not user:
        return {
            "success": False,
            "message": "用户不存在",
            "data": None,
            "error_code": "2001",
        }

    await db.delete(user)
    await db.commit()

    return {
        "success": True,
        "message": "用户删除成功",
        "data": None,
        "error_code": None,
    }


@router.post("/{user_id}/reset-password", response_model=dict)
async def reset_password(
    user_id: int,
    password_data: PasswordReset,
    current_user: User = Depends(require_admin),
    db: AsyncSession = Depends(get_db),
):
    """重置用户密码"""
    result = await db.execute(select(User).where(User.id == user_id))
    user = result.scalar_one_or_none()

    if not user:
        return {
            "success": False,
            "message": "用户不存在",
            "data": None,
            "error_code": "2001",
        }

    user.password_hash = hash_password(password_data.new_password)
    await db.commit()

    return {
        "success": True,
        "message": "密码重置成功",
        "data": None,
        "error_code": None,
    }

Step 2: 提交

git add backend/src/routes/users.py
git commit -m "[backend] feat: 添加用户管理路由"

Task 8: 创建项目管理路由

Files:

Create: backend/src/routes/projects.py

Step 1: 创建项目管理路由

# backend/src/routes/projects.py
from fastapi import APIRouter, Depends, Query
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, func, and_
from config.database import get_db
from src.models.user import User
from src.models.project import Project
from src.schemas.project import (
    ProjectCreate,
    ProjectUpdate,
    ProjectResponse,
    ProjectListResponse,
)
from src.middleware.auth import (
    get_current_user,
    require_market_or_admin,
)
from src.dependencies import get_project_or_404
from typing import Optional
from datetime import date
from decimal import Decimal

router = APIRouter(prefix="/projects", tags=["项目管理"])


@router.get("", response_model=dict)
async def get_projects(
    page: int = Query(1, ge=1),
    page_size: int = Query(10, ge=1, le=100),
    project_no: Optional[str] = None,
    engineering_type: Optional[str] = None,
    project_department: Optional[str] = None,
    signing_date_start: Optional[date] = None,
    signing_date_end: Optional[date] = None,
    contract_amount_min: Optional[Decimal] = None,
    contract_amount_max: Optional[Decimal] = None,
    keyword: Optional[str] = None,
    sort_by: Optional[str] = None,
    sort_order: Optional[str] = Query(None, pattern="^(asc|desc)$"),
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """获取项目列表"""
    # 构建查询
    query = select(Project)

    # 筛选条件
    conditions = []
    if project_no:
        conditions.append(Project.project_no == project_no)
    if engineering_type:
        conditions.append(Project.engineering_type == engineering_type)
    if project_department:
        conditions.append(Project.project_department == project_department)
    if signing_date_start:
        conditions.append(Project.signing_date >= signing_date_start)
    if signing_date_end:
        conditions.append(Project.signing_date <= signing_date_end)
    if contract_amount_min:
        conditions.append(Project.contract_amount >= contract_amount_min)
    if contract_amount_max:
        conditions.append(Project.contract_amount <= contract_amount_max)
    if keyword:
        conditions.append(
            (Project.name.contains(keyword)) |
            (Project.owner_unit.contains(keyword))
        )

    if conditions:
        query = query.where(and_(*conditions))

    # 排序
    if sort_by:
        sort_column = getattr(Project, sort_by, None)
        if sort_column:
            order = desc if sort_order == "desc" else asc
            query = query.order_by(order(sort_column))
    else:
        query = query.order_by(Project.created_at.desc())

    # 总数
    total_query = select(func.count()).select_from(query.subquery())
    total_result = await db.execute(total_query)
    total = total_result.scalar()

    # 分页
    query = query.offset((page - 1) * page_size).limit(page_size)
    result = await db.execute(query)
    projects = result.scalars().all()

    # 获取创建人姓名
    project_data = []
    for p in projects:
        creator_result = await db.execute(
            select(User.real_name).where(User.id == p.created_by)
        )
        creator_name = creator_result.scalar_one_or_none()
        p_dict = ProjectResponse.model_validate(p).model_dump()
        p_dict["created_by_name"] = creator_name
        project_data.append(p_dict)

    return {
        "success": True,
        "message": "获取成功",
        "data": {
            "items": project_data,
            "total": total,
            "page": page,
            "page_size": page_size,
        },
        "error_code": None,
    }


@router.get("/{project_id}", response_model=dict)
async def get_project(
    project: Project = Depends(get_project_or_404),
    current_user: User = Depends(get_current_user),
):
    """获取项目详情"""
    from sqlalchemy import select
    result = await current_user.execute(
        select(User.real_name).where(User.id == project.created_by)
    )
    creator_name = result.scalar_one_or_none()

    project_dict = ProjectResponse.model_validate(project).model_dump()
    project_dict["created_by_name"] = creator_name

    return {
        "success": True,
        "message": "获取成功",
        "data": project_dict,
        "error_code": None,
    }


@router.post("", response_model=dict)
async def create_project(
    project_data: ProjectCreate,
    current_user: User = Depends(require_market_or_admin),
    db: AsyncSession = Depends(get_db),
):
    """创建项目"""
    # 检查项目编号是否已存在
    result = await db.execute(
        select(Project).where(Project.project_no == project_data.project_no)
    )
    if result.scalar_one_or_none():
        return {
            "success": False,
            "message": "项目编号已存在",
            "data": None,
            "error_code": "2002",
        }

    # 创建项目
    project = Project(
        **project_data.model_dump(),
        created_by=current_user.id,
    )
    db.add(project)
    await db.commit()
    await db.refresh(project)

    return {
        "success": True,
        "message": "项目创建成功",
        "data": ProjectResponse.model_validate(project),
        "error_code": None,
    }


@router.put("/{project_id}", response_model=dict)
async def update_project(
    project_id: int,
    project_data: ProjectUpdate,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """更新项目"""
    # 获取项目
    result = await db.execute(select(Project).where(Project.id == project_id))
    project = result.scalar_one_or_none()

    if not project:
        return {
            "success": False,
            "message": "项目不存在",
            "data": None,
            "error_code": "2001",
        }

    # 权限检查
    if current_user.role == "market" and project.created_by != current_user.id:
        return {
            "success": False,
            "message": "无权修改此项目",
            "data": None,
            "error_code": "3001",
        }

    # 更新字段
    update_data = project_data.model_dump(exclude_unset=True)
    for field, value in update_data.items():
        setattr(project, field, value)

    await db.commit()
    await db.refresh(project)

    return {
        "success": True,
        "message": "项目更新成功",
        "data": ProjectResponse.model_validate(project),
        "error_code": None,
    }


@router.delete("/{project_id}", response_model=dict)
async def delete_project(
    project_id: int,
    current_user: User = Depends(require_market_or_admin),
    db: AsyncSession = Depends(get_db),
):
    """删除项目"""
    result = await db.execute(select(Project).where(Project.id == project_id))
    project = result.scalar_one_or_none()

    if not project:
        return {
            "success": False,
            "message": "项目不存在",
            "data": None,
            "error_code": "2001",
        }

    # 权限检查
    if current_user.role == "market" and project.created_by != current_user.id:
        return {
            "success": False,
            "message": "无权删除此项目",
            "data": None,
            "error_code": "3001",
        }

    await db.delete(project)
    await db.commit()

    return {
        "success": True,
        "message": "项目删除成功",
        "data": None,
        "error_code": None,
    }


@router.get("/statistics/basic", response_model=dict)
async def get_statistics(
    engineering_type: Optional[str] = None,
    signing_date_start: Optional[date] = None,
    signing_date_end: Optional[date] = None,
    contract_amount_min: Optional[Decimal] = None,
    contract_amount_max: Optional[Decimal] = None,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """获取基础统计"""
    # 构建查询
    query = select(Project)

    conditions = []
    if engineering_type:
        conditions.append(Project.engineering_type == engineering_type)
    if signing_date_start:
        conditions.append(Project.signing_date >= signing_date_start)
    if signing_date_end:
        conditions.append(Project.signing_date <= signing_date_end)
    if contract_amount_min:
        conditions.append(Project.contract_amount >= contract_amount_min)
    if contract_amount_max:
        conditions.append(Project.contract_amount <= contract_amount_max)

    if conditions:
        query = query.where(and_(*conditions))

    result = await db.execute(query)
    projects = result.scalars().all()

    # 统计
    total_count = len(projects)
    total_contract_amount = sum(
        p.contract_amount or Decimal(0) for p in projects
    )
    total_receipt_amount = sum(
        p.actual_receipt_amount or Decimal(0) for p in projects
    )
    total_payment_amount = sum(
        p.actual_payment_amount or Decimal(0) for p in projects
    )

    avg_receipt_rate = 0
    if total_contract_amount > 0:
        avg_receipt_rate = (
            (total_receipt_amount / total_contract_amount) * 100
        )

    avg_payment_rate = 0
    total_payable = sum(p.payable_amount or Decimal(0) for p in projects)
    if total_payable > 0:
        avg_payment_rate = (
            (total_payment_amount / total_payable) * 100
        )

    avg_progress = 0
    if total_count > 0:
        avg_progress = sum(
            p.cumulative_progress or Decimal(0) for p in projects
        ) / total_count

    return {
        "success": True,
        "message": "统计成功",
        "data": {
            "total_count": total_count,
            "total_contract_amount": float(total_contract_amount),
            "total_receipt_amount": float(total_receipt_amount),
            "total_payment_amount": float(total_payment_amount),
            "avg_receipt_completion_rate": float(avg_receipt_rate),
            "avg_payment_completion_rate": float(avg_payment_rate),
            "avg_cumulative_progress": float(avg_progress),
        },
        "error_code": None,
    }


@router.get("/statistics/group", response_model=dict)
async def get_group_statistics(
    group_by: str = Query(..., pattern="^(engineering_type|project_department)$"),
    signing_date_start: Optional[date] = None,
    signing_date_end: Optional[date] = None,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """获取分组统计"""
    # 构建查询
    query = select(Project)

    if signing_date_start:
        query = query.where(Project.signing_date >= signing_date_start)
    if signing_date_end:
        query = query.where(Project.signing_date <= signing_date_end)

    result = await db.execute(query)
    projects = result.scalars().all()

    # 分组统计
    groups = {}
    for p in projects:
        group_value = getattr(p, group_by)
        if group_value not in groups:
            groups[group_value] = {
                "count": 0,
                "total_contract_amount": Decimal(0),
                "total_receipt_amount": Decimal(0),
                "total_payment_amount": Decimal(0),
            }
        groups[group_value]["count"] += 1
        groups[group_value]["total_contract_amount"] += p.contract_amount or Decimal(0)
        groups[group_value]["total_receipt_amount"] += p.actual_receipt_amount or Decimal(0)
        groups[group_value]["total_payment_amount"] += p.actual_payment_amount or Decimal(0)

    # 转换为列表
    result_list = []
    for key, value in groups.items():
        if group_by == "engineering_type":
            item = {"engineering_type": key}
        else:
            item = {"project_department": key}
        item.update({
            "count": value["count"],
            "total_contract_amount": float(value["total_contract_amount"]),
            "total_receipt_amount": float(value["total_receipt_amount"]),
            "total_payment_amount": float(value["total_payment_amount"]),
        })
        result_list.append(item)

    return {
        "success": True,
        "message": "统计成功",
        "data": result_list,
        "error_code": None,
    }


@router.get("/statistics/timeline", response_model=dict)
async def get_timeline_statistics(
    time_field: str = Query(..., pattern="^(signing_date|start_date|planned_end_date)$"),
    group_by: str = Query("month", pattern="^(day|month|year)$"),
    engineering_type: Optional[str] = None,
    start_date: Optional[date] = None,
    end_date: Optional[date] = None,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """获取时间维度统计"""
    # 构建查询
    query = select(Project)

    if engineering_type:
        query = query.where(Project.engineering_type == engineering_type)
    if start_date:
        date_column = getattr(Project, time_field)
        query = query.where(date_column >= start_date)
    if end_date:
        date_column = getattr(Project, time_field)
        query = query.where(date_column <= end_date)

    result = await db.execute(query)
    projects = result.scalars().all()

    # 时间分组
    groups = {}
    for p in projects:
        date_value = getattr(p, time_field)
        if not date_value:
            continue

        if group_by == "day":
            key = date_value.strftime("%Y-%m-%d")
        elif group_by == "month":
            key = date_value.strftime("%Y-%m")
        else:  # year
            key = str(date_value.year)

        if key not in groups:
            groups[key] = {"count": 0, "total_contract_amount": Decimal(0)}
        groups[key]["count"] += 1
        groups[key]["total_contract_amount"] += p.contract_amount or Decimal(0)

    # 转换为列表并排序
    result_list = []
    for key in sorted(groups.keys()):
        value = groups[key]
        if group_by == "day":
            item = {"day": key}
        elif group_by == "month":
            item = {"month": key}
        else:
            item = {"year": key}
        item.update({
            "count": value["count"],
            "total_contract_amount": float(value["total_contract_amount"]),
        })
        result_list.append(item)

    return {
        "success": True,
        "message": "统计成功",
        "data": result_list,
        "error_code": None,
    }

Step 2: 提交

git add backend/src/routes/projects.py
git commit -m "[backend] feat: 添加项目管理路由"

Task 9: 创建FastAPI主应用

Files:

Create: backend/main.py

Step 1: 创建主应用

# backend/main.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from config.settings import get_settings
from config.database import init_db
from src.routes import auth_router, users_router, projects_router

settings = get_settings()

# 创建FastAPI应用
app = FastAPI(
    title=settings.APP_NAME,
    version=settings.APP_VERSION,
    description="海洋项目管理系统后端API",
    docs_url="/docs",
    redoc_url="/redoc",
)

# 配置CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.CORS_ORIGINS,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 注册路由
app.include_router(auth_router, prefix="/api/v1")
app.include_router(users_router, prefix="/api/v1")
app.include_router(projects_router, prefix="/api/v1")


@app.on_event("startup")
async def startup_event():
    """应用启动事件"""
    await init_db()


@app.get("/")
async def root():
    """根路径"""
    return {
        "app": settings.APP_NAME,
        "version": settings.APP_VERSION,
        "status": "running",
    }


@app.get("/health")
async def health():
    """健康检查"""
    return {"status": "healthy"}


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(
        "main:app",
        host="0.0.0.0",
        port=5000,
        reload=settings.DEBUG,
    )

Step 2: 提交

git add backend/main.py
git commit -m "[backend] feat: 创建FastAPI主应用"

Task 10: 创建环境配置文件

Files:

Create: backend/.env.example
Create: backend/.env

Step 1: 创建环境变量示例文件

# backend/.env.example
# 应用配置
APP_NAME=海洋项目管理系统
APP_VERSION=1.0.0
DEBUG=True

# 数据库配置
DB_HOST=localhost
DB_PORT=3306
DB_USER=root
DB_PASSWORD=rootpassword
DB_NAME=project_manager
DB_CHARSET=utf8mb4

# JWT配置
SECRET_KEY=your-secret-key-change-this-in-production
ALGORITHM=HS256
ACCESS_TOKEN_EXPIRE_MINUTES=1440

# CORS配置
CORS_ORIGINS=http://localhost:3000,http://127.0.0.1:3000

Step 2: 创建环境变量文件

# backend/.env
APP_NAME=海洋项目管理系统
APP_VERSION=1.0.0
DEBUG=True

DB_HOST=localhost
DB_PORT=3306
DB_USER=root
DB_PASSWORD=rootpassword
DB_NAME=project_manager
DB_CHARSET=utf8mb4

SECRET_KEY=ocean-project-manager-secret-key-2026
ALGORITHM=HS256
ACCESS_TOKEN_EXPIRE_MINUTES=1440

CORS_ORIGINS=http://localhost:3000,http://127.0.0.1:3000

Step 3: 创建.gitignore

# backend/.gitignore
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
env/
venv/
ENV/
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
*.egg-info/
.installed.cfg
*.egg
.env
.pytest_cache/
.coverage
htmlcov/
*.cover
.hypothesis/
.mypy_cache/
.dmypy.json
dmypy.json

Step 4: 提交

git add backend/.env.example backend/.env backend/.gitignore
git commit -m "[backend] feat: 添加环境配置文件"

Task 11: 初始化数据库并创建测试用户

Step 1: 启动MySQL服务

Run: sudo systemctl start mysql (如果MySQL未运行) Expected: MySQL服务启动成功

Step 2: 创建数据库

Run: mysql -u root -p -e "CREATE DATABASE IF NOT EXISTS project_manager DEFAULT CHARACTER SET utf8mb4 DEFAULT COLLATE utf8mb4_unicode_ci;" Expected: 数据库创建成功

Step 3: 执行初始化脚本

Run: mysql -u root -p project_manager < backend/config/init-database.sql Expected: 表结构和初始数据创建成功

Step 4: 验证数据库

Run: mysql -u root -p project_manager -e "SHOW TABLES;" Expected: 显示users和projects表

Task 12: 运行测试并修复问题

Step 1: 运行所有测试

Run: cd backend && pytest tests/ -v Expected: 测试运行

Step 2: 查看测试结果

Run: cd backend && pytest tests/ --cov=src --cov-report=term-missing Expected: 生成覆盖率报告

Step 3: 修复失败的测试

根据测试输出，修复代码中的问题

Step 4: 重新运行测试

Run: cd backend && pytest tests/ -v Expected: 所有测试通过

Step 5: 提交

git add backend/
git commit -m "[backend] test: 修复测试问题，所有测试通过"

Task 13: 启动应用并验证API

Step 1: 启动FastAPI应用

Run: cd backend && python main.py Expected: 应用启动在 http://0.0.0.0:5000

Step 2: 访问Swagger文档

Run: 在浏览器中打开 http://localhost:5000/docs Expected: 显示API文档

Step 3: 测试登录接口

Run: curl -X POST http://localhost:5000/api/v1/auth/login -H "Content-Type: application/json" -d '{"username":"admin","password":"admin123"}' Expected: 返回token和用户信息

Step 4: 测试获取用户信息

Run: 使用返回的token

curl http://localhost:5000/api/v1/auth/me -H "Authorization: Bearer <token>"

Expected: 返回用户信息

Step 5: 提交

git commit --allow-empty -m "[backend] feat: 后端开发完成，所有功能正常"

Task 14: 更新文档

Files:

Modify: backend/README.md

Step 1: 更新README

# 海洋项目管理系统 - 后端

## 快速开始

### 1. 安装依赖

```bash
pip install -r requirements.txt -r requirements-test.txt

2. 配置环境变量

复制 .env.example 为 .env 并修改配置

cp .env.example .env

3. 初始化数据库

mysql -u root -p < config/init-database.sql

4. 启动应用

python main.py

应用将在 http://localhost:5000 启动

5. 访问API文档

Swagger UI: http://localhost:5000/docs ReDoc: http://localhost:5000/redoc

运行测试

pytest tests/ -v

生成覆盖率报告:

pytest tests/ --cov=src --cov-report=html

API文档

详细的API文档请参考: docs/api.md

开发规范

详见: WORKSTANDARDS.md


**Step 2: 提交**

```bash
git add backend/README.md
git commit -m "[backend] docs: 更新README文档"

验证检查清单

在完成所有任务后，验证以下内容：

所有依赖安装成功
数据库表创建成功
应用可以正常启动
Swagger文档可以访问
登录接口测试通过
所有测试用例通过（51个测试）
测试覆盖率 > 80%
文档更新完成

56 KiB Raw Blame History

海洋项目管理系统 - 后端实现计划

Task 1: 创建项目基础配置文件

Task 2: 创建工具函数模块

Task 3: 创建数据库模型

Task 4: 创建Pydantic Schema模型

Task 5: 创建认证中间件和依赖注入

Task 6: 创建认证路由

Task 7: 创建用户管理路由

Task 8: 创建项目管理路由

Task 9: 创建FastAPI主应用

Task 10: 创建环境配置文件

Task 11: 初始化数据库并创建测试用户

Task 12: 运行测试并修复问题

Task 13: 启动应用并验证API

Task 14: 更新文档

2. 配置环境变量

3. 初始化数据库

4. 启动应用

5. 访问API文档

运行测试

API文档

开发规范

验证检查清单

56 KiB

Raw Blame History