duxingchen/SCGL
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: initial commit and ignore qwen files

Browse Source
This commit is contained in:
dxc
2026-04-30 10:06:32 +08:00
commit def4f7d71f
55 changed files with 5252 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

280
backend/app/routers/project_stats.py Normal file
View File

@ -0,0 +1,280 @@
"""项目统计聚合接口
路由路径:
- GET /api/pms/projects/summary - 获取所有项目汇总
- GET /api/pms/projects/{id}/stats - 获取单个项目详情
注意: /summary 路由必须写在 /{project_id}/stats 之前，
否则 FastAPI 会将 "summary" 误匹配为 {project_id} 参数。
"""
from fastapi import APIRouter, Depends, Query, HTTPException
from sqlalchemy.orm import Session
from datetime import date
from typing import Optional
from app.database import get_db_pms
from app.models.production import (
PmsProject,
PmsWorkOrder,
PmsMaterialApproval,
ProjectStatus,
WorkOrderStatus,
ApprovalStatus,
)
from app.schemas.project import ProjectStats, ProjectSummary, ProjectCreate, ProjectResponse
router = APIRouter(prefix="/api/pms/projects", tags=["项目管理"])
# ============================================================
# 路由定义顺序很重要！更具体的路径必须写在前面！
# ============================================================
@router.post("", response_model=ProjectResponse, status_code=201)
async def create_project(
project_data: ProjectCreate,
db: Session = Depends(get_db_pms)
):
"""
创建新项目
请求体:
- project_no: 项目编号（必填，唯一）
- project_name: 项目名称（必填）
- start_date: 计划开始日期（可选）
- end_date: 计划结束日期（可选）
返回创建后的项目信息，包含自动生成的 id 和 created_at
"""
# 检查项目编号是否重复
existing = db.query(PmsProject).filter(
PmsProject.project_no == project_data.project_no
).first()
if existing:
raise HTTPException(
status_code=409,
detail=f"项目编号 '{project_data.project_no}' 已存在"
)
# 创建项目实例
project = PmsProject(
project_no=project_data.project_no,
name=project_data.name,
start_date=project_data.start_date,
end_date=project_data.end_date,
status=project_data.status,
)
db.add(project)
db.commit()
db.refresh(project)
return project
def calculate_project_stats(project: PmsProject, work_orders: list, approvals: list) -> ProjectStats:
"""计算单个项目的统计信息"""
today = date.today()
# 统计工单状态
total_work_orders = len(work_orders)
completed_work_orders = sum(1 for wo in work_orders if wo.status == WorkOrderStatus.COMPLETED)
in_progress_work_orders = sum(1 for wo in work_orders if wo.status == WorkOrderStatus.IN_PROGRESS)
pending_work_orders = sum(1 for wo in work_orders if wo.status == WorkOrderStatus.PENDING)
# 计算进度百分比
progress_percentage = (
round(completed_work_orders / total_work_orders * 100, 1)
if total_work_orders > 0 else 0.0
)
# 统计审批状态
total_approvals = len(approvals)
pending_approvals = sum(1 for a in approvals if a.status == ApprovalStatus.PENDING)
approved_approvals = sum(1 for a in approvals if a.status == ApprovalStatus.APPROVED)
rejected_approvals = sum(1 for a in approvals if a.status == ApprovalStatus.REJECTED)
# 物料到位率 = 已批准 / 总申请
material_ready_rate = (
round(approved_approvals / total_approvals * 100, 1)
if total_approvals > 0 else 100.0
)
# 延期预警计算
is_overdue = False
overdue_days: int | None = None
if project.end_date and project.status != ProjectStatus.COMPLETED:
if today > project.end_date:
is_overdue = True
overdue_days = (today - project.end_date).days
return ProjectStats(
project_id=project.id,
project_no=project.project_no,
project_name=project.name,
status=project.status,
start_date=project.start_date,
end_date=project.end_date,
total_work_orders=total_work_orders,
completed_work_orders=completed_work_orders,
in_progress_work_orders=in_progress_work_orders,
pending_work_orders=pending_work_orders,
progress_percentage=progress_percentage,
total_approvals=total_approvals,
pending_approvals=pending_approvals,
approved_approvals=approved_approvals,
rejected_approvals=rejected_approvals,
material_ready_rate=material_ready_rate,
is_overdue=is_overdue,
overdue_days=overdue_days,
created_at=project.created_at,
)
# ============================================================
# 路由定义顺序很重要！更具体的路径必须写在前面！
# ============================================================
@router.get("/summary", response_model=ProjectSummary)
async def get_project_summary(
status: Optional[ProjectStatus] = Query(None, description="项目状态筛选"),
db: Session = Depends(get_db_pms)
):
"""
获取项目总览汇总数据
包含所有项目的统计信息：
- 项目总数、活跃项目数、已完成项目数、延期项目数
- 全局工单统计和进度
- 物料到位率汇总
- 每个项目的详细统计
"""
print("Endpoint /summary reached!") # 调试日志
# 查询所有项目
query = db.query(PmsProject)
if status:
query = query.filter(PmsProject.status == status)
projects = query.order_by(PmsProject.created_at.desc()).all()
# 无项目时返回空数据
if not projects:
return ProjectSummary(
total_projects=0,
active_projects=0,
completed_projects=0,
overdue_projects=0,
total_work_orders=0,
completed_work_orders=0,
overall_progress=0.0,
total_pending_approvals=0,
overall_material_ready_rate=100.0,
projects=[],
)
# 获取所有项目ID
project_ids = [p.id for p in projects]
# 查询所有相关工单
all_work_orders = (
db.query(PmsWorkOrder)
.filter(PmsWorkOrder.project_id.in_(project_ids))
.all()
)
all_work_order_ids = [wo.id for wo in all_work_orders]
# 查询所有相关审批
all_approvals = (
db.query(PmsMaterialApproval)
.filter(PmsMaterialApproval.work_order_id.in_(all_work_order_ids))
.all()
)
# 按项目分组工单
work_orders_by_project: dict[int, list] = {}
for wo in all_work_orders:
work_orders_by_project.setdefault(wo.project_id, []).append(wo)
# 按工单分组审批（需要通过工单找到项目）
approvals_by_project: dict[int, list] = {}
work_order_to_project: dict[int, int] = {wo.id: wo.project_id for wo in all_work_orders}
for ap in all_approvals:
project_id = work_order_to_project.get(ap.work_order_id)
if project_id:
approvals_by_project.setdefault(project_id, []).append(ap)
# 计算每个项目的统计
project_stats_list: list[ProjectStats] = []
total_work_orders = 0
completed_work_orders = 0
total_pending_approvals = 0
overdue_projects = 0
for project in projects:
wo_list = work_orders_by_project.get(project.id, [])
ap_list = approvals_by_project.get(project.id, [])
stats = calculate_project_stats(project, wo_list, ap_list)
project_stats_list.append(stats)
total_work_orders += stats.total_work_orders
completed_work_orders += stats.completed_work_orders
total_pending_approvals += stats.pending_approvals
if stats.is_overdue:
overdue_projects += 1
# 计算全局进度
overall_progress = (
round(completed_work_orders / total_work_orders * 100, 1)
if total_work_orders > 0 else 0.0
)
# 计算全局物料到位率
total_approvals_count = sum(s.total_approvals for s in project_stats_list)
total_approved = sum(s.approved_approvals for s in project_stats_list)
overall_material_ready_rate = (
round(total_approved / total_approvals_count * 100, 1)
if total_approvals_count > 0 else 100.0
)
return ProjectSummary(
total_projects=len(projects),
active_projects=sum(1 for p in projects if p.status == ProjectStatus.ACTIVE),
completed_projects=sum(1 for p in projects if p.status == ProjectStatus.COMPLETED),
overdue_projects=overdue_projects,
total_work_orders=total_work_orders,
completed_work_orders=completed_work_orders,
overall_progress=overall_progress,
total_pending_approvals=total_pending_approvals,
overall_material_ready_rate=overall_material_ready_rate,
projects=project_stats_list,
)
@router.get("/{project_id}/stats", response_model=ProjectStats)
async def get_project_stats(
project_id: int,
db: Session = Depends(get_db_pms)
):
"""获取单个项目的详细统计信息"""
print(f"Endpoint /{project_id}/stats reached!") # 调试日志
project = db.query(PmsProject).filter(PmsProject.id == project_id).first()
if not project:
raise HTTPException(status_code=404, detail=f"Project ID={project_id} not found")
# 查询该项目的工单
work_orders = (
db.query(PmsWorkOrder)
.filter(PmsWorkOrder.project_id == project_id)
.all()
)
work_order_ids = [wo.id for wo in work_orders]
# 查询该项目的审批
approvals = (
db.query(PmsMaterialApproval)
.filter(PmsMaterialApproval.work_order_id.in_(work_order_ids))
.all()
)
return calculate_project_stats(project, work_orders, approvals)