eomsangdonandClaude Opus 5.5
ff8a8bad08
- DB 020(추가만): projects 종류 · 사업연도 끝 · 설계속도 · 원본 칸 · plan_route_years · plan_route_assignees · 회사 「산림청」 행 - is_forest_email · is_forest_service · /api/auth/session 에 is_forest_service - require_project_access 한 곳 — 지정 회사 = GET · HEAD 만 · 그 밖 403 - POST /api/d01/project(산림청만 · PLAN · 설계속도 기본 = resolve_design_speed) · GET /api/dashboard/plan-routes/assigned - GET /api/e01/source-candidates · POST /api/e01/project(B06 완료만 · STABILITY · 원본 칸 · 자료 복사는 528-4) - 종류별 단계 시드(PLAN 3 · STABILITY 4) · storage 허용 D0x · E0x · B01 목록에 project_kind - 가입 인증 — 메일 *.go.kr = 회사 산림청 · ACTIVE · 화면 회사 칸 「산림청」 잠금 표시 - 지킴 시험 test_528_project_groups.py Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
88 lines
3.2 KiB
Python
88 lines
3.2 KiB
Python
"""E01 안정성평가 생성 API (요구 345 · 347 · 요구서 4-1).
|
|
|
|
대상 = 같은 회사 임도(ROAD) 중 B06 횡단 단계(stage 3) 완료. 자료 복사는 528-4 — 지금은 행만.
|
|
"""
|
|
|
|
from typing import Any
|
|
|
|
import aiomysql
|
|
from fastapi import APIRouter, Depends, HTTPException, Request
|
|
from pydantic import BaseModel, Field
|
|
|
|
from common_util.common_util_auth import require_company
|
|
from common_util.common_util_project_kind import create_kind_project
|
|
from config.config_db import get_db_pool
|
|
|
|
router = APIRouter(prefix="/api/e01", tags=["E01_StabRegister"])
|
|
|
|
_SOURCE_SQL = """
|
|
SELECT p.id AS project_id, p.name, p.region, p.road_type, p.project_year,
|
|
DATE_FORMAT(p.updated_at, '%%Y-%%m-%%dT%%H:%%i:%%s') AS updated_at,
|
|
s.state = 'COMPLETE' AS section_complete
|
|
FROM projects p
|
|
LEFT JOIN project_workflow_stages s ON s.project_id = p.id AND s.stage_no = 3
|
|
WHERE p.company_id = %s AND p.project_kind = 'ROAD' AND p.deleted_at IS NULL
|
|
"""
|
|
|
|
|
|
class CreateStabRequest(BaseModel):
|
|
project_name: str | None = Field(default=None, max_length=255)
|
|
source_project_id: str = Field(..., min_length=36, max_length=36)
|
|
|
|
|
|
async def _sources(company_id: Any, project_id: str | None = None) -> list[dict[str, Any]]:
|
|
sql, args = _SOURCE_SQL, [company_id]
|
|
if project_id:
|
|
sql, args = sql + " AND p.id = %s", [company_id, project_id]
|
|
async with (
|
|
get_db_pool().acquire() as connection,
|
|
connection.cursor(aiomysql.DictCursor) as cursor,
|
|
):
|
|
await cursor.execute(sql + " ORDER BY p.updated_at DESC", tuple(args))
|
|
return list(await cursor.fetchall())
|
|
|
|
|
|
@router.get("/source-candidates")
|
|
async def source_candidates(
|
|
session: dict[str, Any] = Depends(require_company),
|
|
) -> dict[str, Any]:
|
|
"""B06 까지 끝난 같은 회사 임도 프로젝트."""
|
|
rows = await _sources(session.get("company_id"))
|
|
projects = [
|
|
{k: v for k, v in r.items() if k != "section_complete"}
|
|
for r in rows
|
|
if r["section_complete"]
|
|
]
|
|
return {"status": "success", "projects": projects}
|
|
|
|
|
|
@router.post("/project")
|
|
async def post_stab_project(
|
|
request: Request,
|
|
payload: CreateStabRequest,
|
|
session: dict[str, Any] = Depends(require_company),
|
|
) -> dict[str, str]:
|
|
"""안정성평가 프로젝트 생성 — 원본 B06 미완료면 400."""
|
|
company_id = session.get("company_id")
|
|
rows = await _sources(company_id, payload.source_project_id) if company_id else []
|
|
if not rows:
|
|
raise HTTPException(status_code=404, detail="같은 회사 임도 프로젝트가 아닙니다.")
|
|
source = rows[0]
|
|
if not source["section_complete"]:
|
|
raise HTTPException(
|
|
status_code=400, detail="원본 임도의 횡단 설계(B06)가 끝나지 않았습니다."
|
|
)
|
|
name = (payload.project_name or "").strip() or f"{source['name']} 안정성평가"
|
|
project_id = await create_kind_project(
|
|
request=request,
|
|
user_id=int(session["user_id"]),
|
|
company_id=int(company_id),
|
|
kind="STABILITY",
|
|
name=name,
|
|
region=source["region"],
|
|
road_type=source["road_type"],
|
|
project_year=source["project_year"],
|
|
source_project_id=source["project_id"],
|
|
)
|
|
return {"project_id": project_id}
|