133 lines
5.0 KiB
Python
133 lines
5.0 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, Query
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
from sqlalchemy import select, func, desc
|
||
from typing import Optional
|
||
from datetime import datetime
|
||
|
||
from app.database import get_db
|
||
from app.models.quality import QualityRecord
|
||
from app.schemas.quality import QualityRecordCreate, QualityRecordUpdate, QualityRecordOut
|
||
from app.schemas.common import Response, PageResponse
|
||
from app.services.auth_service import get_current_user
|
||
from app.services.prediction import QualityPredictionModel
|
||
|
||
router = APIRouter()
|
||
|
||
|
||
@router.get("/summary", response_model=Response[dict])
|
||
async def quality_summary(
|
||
db: AsyncSession = Depends(get_db),
|
||
_=Depends(get_current_user),
|
||
):
|
||
"""合格率、平均评分、班次分组统计"""
|
||
total_q = await db.execute(select(func.count()).select_from(QualityRecord))
|
||
passed_q = await db.execute(
|
||
select(func.count()).select_from(QualityRecord).where(QualityRecord.is_passed == True)
|
||
)
|
||
avg_pi_q = await db.execute(select(func.avg(QualityRecord.pi_score)).select_from(QualityRecord))
|
||
avg_suf_q = await db.execute(select(func.avg(QualityRecord.surface_score)).select_from(QualityRecord))
|
||
|
||
# Grade distribution
|
||
grade_counts: dict = {}
|
||
for grade in ["A1", "A2", "B1", "B2", "C"]:
|
||
cnt_q = await db.execute(
|
||
select(func.count()).select_from(QualityRecord).where(QualityRecord.overall_grade == grade)
|
||
)
|
||
grade_counts[grade] = cnt_q.scalar() or 0
|
||
|
||
total = total_q.scalar() or 0
|
||
passed = passed_q.scalar() or 0
|
||
pass_rate = round(passed / total * 100, 1) if total > 0 else 0.0
|
||
|
||
return Response.ok({
|
||
"total": total,
|
||
"passed": passed,
|
||
"pass_rate": pass_rate,
|
||
"avg_pi_score": round(avg_pi_q.scalar() or 0, 1),
|
||
"avg_surface_score": round(avg_suf_q.scalar() or 0, 1),
|
||
"grade_distribution": grade_counts,
|
||
})
|
||
|
||
|
||
@router.get("/", response_model=Response[PageResponse[QualityRecordOut]])
|
||
async def list_quality(
|
||
page: int = 1,
|
||
page_size: int = 20,
|
||
coil_no: Optional[str] = None,
|
||
overall_grade: Optional[str] = None,
|
||
start_date: Optional[datetime] = None,
|
||
end_date: Optional[datetime] = None,
|
||
db: AsyncSession = Depends(get_db),
|
||
_=Depends(get_current_user),
|
||
):
|
||
query = select(QualityRecord).order_by(desc(QualityRecord.created_at))
|
||
if coil_no:
|
||
query = query.where(QualityRecord.coil_no.ilike(f"%{coil_no}%"))
|
||
if overall_grade:
|
||
query = query.where(QualityRecord.overall_grade == overall_grade)
|
||
if start_date:
|
||
query = query.where(QualityRecord.created_at >= start_date)
|
||
if end_date:
|
||
query = query.where(QualityRecord.created_at <= end_date)
|
||
|
||
total = (await db.execute(select(func.count()).select_from(query.subquery()))).scalar()
|
||
result = await db.execute(query.offset((page - 1) * page_size).limit(page_size))
|
||
items = [QualityRecordOut.model_validate(r) for r in result.scalars()]
|
||
return Response.ok(PageResponse(total=total, page=page, page_size=page_size, items=items))
|
||
|
||
|
||
@router.post("/", response_model=Response[QualityRecordOut])
|
||
async def create_quality(
|
||
body: QualityRecordCreate,
|
||
db: AsyncSession = Depends(get_db),
|
||
_=Depends(get_current_user),
|
||
):
|
||
"""
|
||
创建质量记录。若未提供 pi_score / surface_score / overall_grade，
|
||
则尝试用 QualityPredictionModel 自动填充（需要thickness_actual和相关参数）。
|
||
"""
|
||
data = body.model_dump()
|
||
|
||
if (
|
||
data.get("pi_score") is None
|
||
and data.get("thickness_actual") is not None
|
||
):
|
||
# Use default avg values for auto-prediction when detailed params are absent
|
||
avg_speed = 100.0 # m/min default
|
||
acid_conc_avg = 160.0 # g/L default
|
||
acid_temp_avg = 75.0 # °C default
|
||
try:
|
||
pred = QualityPredictionModel(
|
||
thickness=data["thickness_actual"],
|
||
avg_speed=avg_speed,
|
||
acid_conc_avg=acid_conc_avg,
|
||
acid_temp_avg=acid_temp_avg,
|
||
).calculate()
|
||
data["pi_score"] = pred["pi_score"]
|
||
data["surface_score"] = pred["surface_score"]
|
||
data["overall_grade"] = pred["overall_grade"]
|
||
except Exception:
|
||
pass # best-effort, do not block creation
|
||
|
||
record = QualityRecord(**data)
|
||
db.add(record)
|
||
await db.flush()
|
||
return Response.ok(QualityRecordOut.model_validate(record))
|
||
|
||
|
||
@router.put("/{quality_id}", response_model=Response[QualityRecordOut])
|
||
async def update_quality(
|
||
quality_id: int,
|
||
body: QualityRecordUpdate,
|
||
db: AsyncSession = Depends(get_db),
|
||
_=Depends(get_current_user),
|
||
):
|
||
result = await db.execute(select(QualityRecord).where(QualityRecord.id == quality_id))
|
||
record = result.scalar_one_or_none()
|
||
if not record:
|
||
raise HTTPException(status_code=404, detail="质量记录不存在")
|
||
for k, v in body.model_dump(exclude_none=True).items():
|
||
setattr(record, k, v)
|
||
await db.flush()
|
||
return Response.ok(QualityRecordOut.model_validate(record))
|