eomsangdonandClaude Opus 5.5
070791854c
- 상세 화면(view=project) — 머리 · [⋯] · [작업 열기 — 지금 단계 ›] · 단계 진행(누름 = 그 단계) · 노선 지도(배경 타일 + 선 · 시점 · 종점) · 핵심 숫자 · 기본 정보 · 연결된 것 · 산출물 · 최근 변경(있는 칸만) - 목록 · 상세 숫자 · 썸네일 = 547-1 카드 요약(project_summaries) 한 벌 - 색 = 셸 변수(--b01-*) 이름에 맞춤 - 시험 test_548_project_overview — 칸반 나눔 · 산출물 · 내 것 · 단계 수 짝 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
328 lines
13 KiB
Python
328 lines
13 KiB
Python
"""B01 대시보드 프로젝트 목록 · 상세 API (요구 369~374 · PLAN 548).
|
|
|
|
목록 = 카드 · 표 · 칸반 한 벌(칸반 칸은 단계 상태로 저절로 · `board_column`).
|
|
상세 = 머리 · 단계 · 지도 선 · 핵심 숫자 · 기본 정보 · 연결 · 산출물 · 최근 변경(있는 것만).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import aiomysql
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
|
|
from common_util.common_util_auth import verify_session
|
|
from config.config_db import get_db_pool
|
|
|
|
from .B01_Dashboard_Repository import _project_rows
|
|
from .B01_Dashboard_Summary import project_summaries
|
|
|
|
router = APIRouter(prefix="/api/dashboard/overview", tags=["B01_Dashboard"])
|
|
|
|
#: 종류별 단계 수 — 화면 단계 막대(`group_workflow.ts` · `WORKFLOW_STEP_ROUTES`)와 같음
|
|
STAGE_TOTAL = {"ROAD": 7, "PLAN": 3, "STABILITY": 6}
|
|
#: 손대지 않은 날이 이만큼 넘으면 「오래 멈춤」
|
|
STALE_DAYS = 14
|
|
|
|
_COLUMNS = """p.id, p.user_id, p.company_id, p.name, p.region, p.road_type, p.project_kind,
|
|
p.project_year, p.project_year_end, p.design_speed_kph,
|
|
p.estimated_length_m, p.route_start_m, p.route_end_m,
|
|
p.memo, p.status, p.updated_at, p.created_at, p.crs_epsg,
|
|
p.client_org, p.project_number, p.work_amount, p.design_date,
|
|
p.pm_user_id, p.field_lead_user_id, p.designer_user_id,
|
|
p.logo_asset_id, p.signature_asset_id, p.source_project_id,
|
|
u.name AS owner_name, COALESCE(pm.name, u.name) AS manager_name"""
|
|
_FROM = """FROM projects p LEFT JOIN users u ON u.id = p.user_id
|
|
LEFT JOIN users pm ON pm.id = p.pm_user_id"""
|
|
|
|
|
|
def board_column(kind: str | None, stages: list[dict[str, Any]]) -> str:
|
|
"""칸반 칸 — start 시작 전 · design 설계 중 · review 검토 · confirm 확정 대기 · done 완료.
|
|
|
|
review = 앞 단계가 바뀌어 다시 볼 단계(STALE) · 실패(FAILED)가 있음.
|
|
confirm = 앞 단계를 다 마치고 마지막 단계만 남음.
|
|
"""
|
|
total = STAGE_TOTAL.get(kind or "ROAD", 0)
|
|
states = {int(s["stage_no"]): s["state"] for s in stages}
|
|
if total and states.get(total - 1) == "COMPLETE":
|
|
return "done"
|
|
if all(state == "NOT_STARTED" for state in states.values()):
|
|
return "start"
|
|
if any(state in ("STALE", "FAILED") for state in states.values()):
|
|
return "review"
|
|
done_head = 0
|
|
while states.get(done_head) == "COMPLETE":
|
|
done_head += 1
|
|
if total and done_head >= total - 1:
|
|
return "confirm"
|
|
return "design"
|
|
|
|
|
|
def idle_days(updated_at: datetime | None, now: datetime | None = None) -> int | None:
|
|
"""마지막 수정 뒤 지난 날 수."""
|
|
if updated_at is None:
|
|
return None
|
|
return ((now or datetime.now()) - updated_at).days
|
|
|
|
|
|
def is_mine(row: dict[str, Any], user_id: int) -> bool:
|
|
"""내 것 = 만든 사람 · 참여자 · 책임 · 현장 · 설계 담당."""
|
|
people = (row.get("user_id"), row.get("pm_user_id"), row.get("field_lead_user_id"))
|
|
return user_id in (*people, row.get("designer_user_id")) or user_id in row.get(
|
|
"member_user_ids", []
|
|
)
|
|
|
|
|
|
def _decorate(row: dict[str, Any]) -> dict[str, Any]:
|
|
stages = row.get("workflow_state", {}).get("stages", [])
|
|
kind = row.get("project_kind") or "ROAD"
|
|
days = idle_days(row.get("updated_at"))
|
|
return {
|
|
**row,
|
|
"project_kind": kind,
|
|
"stage_total": STAGE_TOTAL.get(kind, 0),
|
|
"board": board_column(kind, stages),
|
|
"idle_days": days,
|
|
"stalled": days is not None and days > STALE_DAYS,
|
|
}
|
|
|
|
|
|
def _scope_where(session: dict[str, Any], company_id: int | None) -> tuple[str, tuple[Any, ...]]:
|
|
"""볼 범위 — 시스템 관리자 = 고른 회사(없음 = 전체) · 회사 사람 = 자기 회사 · 그 밖 = 내 것."""
|
|
if session["role"] == "SYSTEM_ADMIN":
|
|
return ("p.company_id = %s", (company_id,)) if company_id else ("1 = 1", ())
|
|
own = session.get("company_id")
|
|
if company_id is not None and int(company_id) != int(own or 0):
|
|
raise HTTPException(status_code=403, detail="다른 회사의 자료는 볼 수 없습니다.")
|
|
if own:
|
|
return "p.company_id = %s", (int(own),)
|
|
return "p.user_id = %s", (int(session["user_id"]),)
|
|
|
|
|
|
@router.get("/projects")
|
|
async def overview_projects(
|
|
scope: str = Query("company", pattern="^(mine|company)$"),
|
|
company_id: int | None = Query(None, gt=0),
|
|
session: dict[str, Any] = Depends(verify_session),
|
|
):
|
|
where, args = _scope_where(session, company_id)
|
|
pool = get_db_pool()
|
|
async with pool.acquire() as connection, connection.cursor(aiomysql.DictCursor) as cursor:
|
|
await cursor.execute(
|
|
f"""SELECT {_COLUMNS} {_FROM}
|
|
WHERE {where} AND p.deleted_at IS NULL
|
|
ORDER BY p.updated_at DESC, p.created_at DESC""",
|
|
args,
|
|
)
|
|
rows = await _project_rows(cursor, await cursor.fetchall())
|
|
if scope == "mine":
|
|
rows = [r for r in rows if is_mine(r, int(session["user_id"]))]
|
|
summaries = {s["id"]: s for s in await project_summaries([str(r["id"]) for r in rows])}
|
|
return {
|
|
"status": "success",
|
|
"projects": [_with_summary(_decorate(r), summaries.get(str(r["id"]))) for r in rows],
|
|
}
|
|
|
|
|
|
def _with_summary(row: dict[str, Any], summary: dict[str, Any] | None) -> dict[str, Any]:
|
|
"""카드 요약(547-1) 의 썸네일 선 · 핵심 숫자를 붙임."""
|
|
summary = summary or {}
|
|
return {**row, "thumb": summary.get("thumb"), "metrics": summary.get("metrics", {})}
|
|
|
|
|
|
async def _visible_project(cursor: Any, session: dict[str, Any], project_id: str) -> dict:
|
|
await cursor.execute(
|
|
f"""SELECT {_COLUMNS}, p.storage_path,
|
|
fl.name AS field_lead_name, ds.name AS designer_name {_FROM}
|
|
LEFT JOIN users fl ON fl.id = p.field_lead_user_id
|
|
LEFT JOIN users ds ON ds.id = p.designer_user_id
|
|
WHERE p.id = %s AND p.deleted_at IS NULL""",
|
|
(project_id,),
|
|
)
|
|
row = await cursor.fetchone()
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail="프로젝트를 찾을 수 없습니다.")
|
|
if session["role"] != "SYSTEM_ADMIN":
|
|
own = session.get("company_id")
|
|
allowed = (
|
|
int(row["company_id"] or 0) == int(own)
|
|
if own
|
|
else int(row["user_id"]) == int(session["user_id"])
|
|
)
|
|
if not allowed:
|
|
raise HTTPException(status_code=403, detail="이 프로젝트를 볼 권한이 없습니다.")
|
|
return row
|
|
|
|
|
|
def _route_lonlat(root: Path, limit: int = 300) -> tuple[list[list[float]], float | None]:
|
|
"""지도 선 = 설계 노선(작업본 → 초기값 → 예상노선) [lat, lon] · 연장 — 못 읽으면 ([], None)."""
|
|
from pyproj import Transformer
|
|
|
|
from B04_PreProcess.B04_PreProcess_Engine_Extent import project_epsg_from_prj
|
|
from common_util.common_util_route_geometry import (
|
|
expected_route_csv_path,
|
|
planned_route_initial_path,
|
|
planned_route_working_path,
|
|
read_planned_route_csv,
|
|
)
|
|
|
|
for path in (
|
|
planned_route_working_path(root),
|
|
planned_route_initial_path(root),
|
|
expected_route_csv_path(root),
|
|
):
|
|
route = read_planned_route_csv(path) if path.is_file() else None
|
|
if route is None or len(route.vertices) < 2:
|
|
continue
|
|
picked = route.vertices[:: max(1, len(route.vertices) // limit)]
|
|
if picked[-1] is not route.vertices[-1]:
|
|
picked.append(route.vertices[-1])
|
|
try:
|
|
crs = route.crs_input or project_epsg_from_prj(root)
|
|
to_wgs = Transformer.from_crs(crs, "EPSG:4326", always_xy=True)
|
|
lonlat = [to_wgs.transform(v.x, v.y) for v in picked]
|
|
except Exception: # 좌표계 못 읽음 = 지도 없음 · 연장은 그대로
|
|
return [], route.vertices[-1].chainage_m
|
|
points = [[round(lat, 7), round(lon, 7)] for lon, lat in lonlat]
|
|
return points, route.vertices[-1].chainage_m
|
|
return [], None
|
|
|
|
|
|
def _numbers(metrics: dict[str, Any]) -> list[dict[str, Any]]:
|
|
"""핵심 숫자 — 카드 요약 값 가운데 있는 것만(연장 · 측점 · 구조물)."""
|
|
keys = (
|
|
("length_m", "length", "m"),
|
|
("stations", "stations", None),
|
|
("structures", "structures", None),
|
|
)
|
|
return [
|
|
{"key": key, "value": metrics[name], **({"unit": unit} if unit else {})}
|
|
for name, key, unit in keys
|
|
if metrics.get(name)
|
|
]
|
|
|
|
|
|
def project_outputs(kind: str, stages: list[dict[str, Any]]) -> list[dict[str, str]]:
|
|
"""산출물 — 인쇄 · 내려받기 화면(b09-print)에서 지금 받을 수 있는 것만(임도설계).
|
|
|
|
도면 = 횡단(3) 마침 · 수량 = 수량(5) 마침 · 내역 = 원가(6) 들어감.
|
|
"""
|
|
if kind != "ROAD":
|
|
return []
|
|
state = {int(s["stage_no"]): s["state"] for s in stages}
|
|
out = []
|
|
if state.get(3) == "COMPLETE":
|
|
out.append({"key": "drawings", "label": "도면 PDF"})
|
|
if state.get(5) == "COMPLETE":
|
|
out.append({"key": "quantity", "label": "수량 xlsx"})
|
|
if state.get(6, "NOT_STARTED") != "NOT_STARTED":
|
|
out.append({"key": "cost", "label": "내역 xlsx"})
|
|
return out
|
|
|
|
|
|
def _link(kind: str, row: dict[str, Any], note: str | None = None) -> dict[str, Any]:
|
|
return {
|
|
"kind": kind,
|
|
"project_id": row["id"],
|
|
"name": row["name"],
|
|
"project_kind": row.get("project_kind") or "PLAN",
|
|
"note": note,
|
|
}
|
|
|
|
|
|
async def _links(cursor: Any, row: dict[str, Any]) -> list[dict[str, Any]]:
|
|
"""연결된 것 — 원본 · 이 프로젝트에서 나온 것 · 연결한 예정노선(원본과 같으면 한 번만)."""
|
|
links: list[dict[str, Any]] = []
|
|
source = row.get("source_project_id")
|
|
if source:
|
|
await cursor.execute(
|
|
"SELECT id, name, project_kind FROM projects WHERE id = %s AND deleted_at IS NULL",
|
|
(source,),
|
|
)
|
|
links += [_link("source", r) for r in await cursor.fetchall()]
|
|
await cursor.execute(
|
|
"""SELECT id, name, project_kind FROM projects
|
|
WHERE source_project_id = %s AND deleted_at IS NULL ORDER BY created_at""",
|
|
(row["id"],),
|
|
)
|
|
links += [_link("derived", r) for r in await cursor.fetchall()]
|
|
await cursor.execute(
|
|
"""SELECT p.id, p.name, a.year FROM plan_route_assignees a
|
|
JOIN projects p ON p.id = a.project_id AND p.deleted_at IS NULL
|
|
WHERE a.linked_project_id = %s ORDER BY a.year""",
|
|
(row["id"],),
|
|
)
|
|
for r in await cursor.fetchall():
|
|
if r["id"] == source:
|
|
links[0]["note"] = f"{r['year']} 년"
|
|
else:
|
|
links.append(_link("plan", r, f"{r['year']} 년"))
|
|
return links
|
|
|
|
|
|
#: 최근 변경 — 감사 기록 중 사람 글로 보일 것
|
|
_AUDIT_ACTIONS = ("PROJECT_CREATE", "PROJECT_UPDATE")
|
|
|
|
|
|
async def _changes(cursor: Any, project_id: str, stages: list[dict[str, Any]]) -> list[dict]:
|
|
"""최근 변경 — 감사 기록(만듦 · 정보 고침) + 단계 마친 때 · 새것부터 8 개."""
|
|
await cursor.execute(
|
|
"""SELECT l.action, DATE_FORMAT(l.timestamp, '%%Y-%%m-%%d %%H:%%i:%%s') AS at,
|
|
u.name AS who
|
|
FROM system_audit_logs l LEFT JOIN users u ON u.id = l.user_id
|
|
WHERE l.resource_type = 'project' AND l.resource_ref = %s
|
|
AND l.action IN (%s, %s)
|
|
ORDER BY l.timestamp DESC LIMIT 8""",
|
|
(project_id, *_AUDIT_ACTIONS),
|
|
)
|
|
items: list[dict[str, Any]] = [dict(r) for r in await cursor.fetchall()]
|
|
items += [
|
|
{
|
|
"at": s["completed_at"].replace("T", " "),
|
|
"who": None,
|
|
"action": "STAGE_COMPLETE",
|
|
"stage_no": s["stage_no"],
|
|
}
|
|
for s in stages
|
|
if s.get("state") == "COMPLETE" and s.get("completed_at")
|
|
]
|
|
return sorted(items, key=lambda item: item["at"], reverse=True)[:8]
|
|
|
|
|
|
@router.get("/projects/{project_id}")
|
|
async def overview_project_detail(
|
|
project_id: str, session: dict[str, Any] = Depends(verify_session)
|
|
):
|
|
from common_util.common_util_storage import resolve_stored_project_path
|
|
|
|
pool = get_db_pool()
|
|
async with pool.acquire() as connection, connection.cursor(aiomysql.DictCursor) as cursor:
|
|
raw = await _visible_project(cursor, session, project_id)
|
|
row = _decorate((await _project_rows(cursor, [raw]))[0])
|
|
links = await _links(cursor, row)
|
|
changes = await _changes(cursor, project_id, row["workflow_state"]["stages"])
|
|
try:
|
|
root = Path(resolve_stored_project_path(raw["storage_path"]))
|
|
except (ValueError, TypeError, AttributeError):
|
|
root = None
|
|
line = _route_lonlat(root)[0] if root else []
|
|
summary = (await project_summaries([project_id]) or [{}])[0]
|
|
people = {
|
|
"pm": row["manager_name"] if row.get("pm_user_id") else None,
|
|
"field_lead": row.pop("field_lead_name", None),
|
|
"designer": row.pop("designer_name", None),
|
|
}
|
|
row.pop("storage_path", None)
|
|
return {
|
|
"status": "success",
|
|
"project": row,
|
|
"people": people,
|
|
"route": {"points": line} if line else None,
|
|
"numbers": _numbers(summary.get("metrics", {})),
|
|
"links": links,
|
|
"outputs": project_outputs(row["project_kind"], row["workflow_state"]["stages"]),
|
|
"changes": changes,
|
|
}
|