feat(B01): 프로젝트 카드 요약 함수 · API — 단계 · 썸네일 선 · 담당 · 멈춘 날 · 연장 · 측점 · 구조물 (PLAN 547-1)

- B01_Dashboard_Summary.project_summaries(ids) — 쿼리 다섯 번(개수 무관) · 넣은 차례 · 없는 값은 칸 뺌
- 썸네일 = 최신 노선 GeoJSON 중심선 100 점 안쪽 · 0~1 칸(가로세로 비 유지 · y 아래로)
- GET /api/dashboard/projects/summary?ids= — 보이는 프로젝트만(시스템 관리자 전부 · 회사 · 내 것)
- 실측 21 개 0.14 s

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

B01_Dashboard/B01_Dashboard_Router_Home.py

@@ -0,0 +1,47 @@
+"""B01 대시보드 홈 API (PLAN 547) — 프로젝트 카드 요약 · 역할별 홈."""
+
+from typing import Any
+
+import aiomysql
+from fastapi import APIRouter, Depends, Query
+
+from common_util.common_util_auth import verify_session
+from config.config_db import get_db_pool
+
+from .B01_Dashboard_Summary import project_summaries
+
+router = APIRouter(prefix="/api/dashboard", tags=["B01_Dashboard"])
+
+SUMMARY_LIMIT = 200
+
+
+async def visible_project_ids(session: dict[str, Any]) -> list[str]:
+    """보이는 프로젝트(최근 수정 순) — 시스템 관리자 = 전부 · 회사 = 회사 전체 · 없으면 내 것."""
+    where, args = "p.deleted_at IS NULL", ()
+    if session["role"] != "SYSTEM_ADMIN":
+        if session.get("company_id"):
+            where, args = where + " AND p.company_id = %s", (int(session["company_id"]),)
+        else:
+            where, args = where + " AND p.user_id = %s", (int(session["user_id"]),)
+    pool = get_db_pool()
+    async with pool.acquire() as connection, connection.cursor(aiomysql.DictCursor) as cursor:
+        await cursor.execute(
+            f"SELECT p.id FROM projects p WHERE {where} "
+            "ORDER BY p.updated_at DESC, p.created_at DESC",
+            args,
+        )
+        return [str(r["id"]) for r in await cursor.fetchall()]
+
+
+@router.get("/projects/summary")
+async def projects_summary(
+    ids: str | None = Query(None, max_length=8000, description="쉼표로 이은 프로젝트 id"),
+    session: dict[str, Any] = Depends(verify_session),
+):
+    """카드 요약 — ids 없으면 보이는 프로젝트 전부(최근 수정 순) · 안 보이는 id 는 빠짐."""
+    visible = await visible_project_ids(session)
+    if ids:
+        allowed = set(visible)
+        wanted = dict.fromkeys(pid.strip() for pid in ids.split(","))
+        visible = [pid for pid in wanted if pid in allowed]
+    return {"status": "success", "projects": await project_summaries(visible[:SUMMARY_LIMIT])}

B01_Dashboard/B01_Dashboard_Summary.py

@@ -0,0 +1,172 @@
+"""프로젝트 카드 요약 (PLAN 547-1) — 홈 · 목록 · 상세 카드가 같이 씀.
+
+단계 = 대시보드 표 단계 막대와 같은 값(`workflow_state.current_stage` · 종류별 단계 목록).
+썸네일 = 최신 노선 경로 GeoJSON(B05 정본) 중심선을 100 점 안쪽으로 줄여 0~1 칸에 맞춤
+(가로세로 비 유지 · 가운데 · y 는 화면 방향 = 아래로 커짐). 값이 없으면 칸을 뺌.
+쿼리 = 프로젝트 · 단계 · 참여자 · 노선 · 측점 다섯 번(개수와 무관) + 노선 · 구조물 파일 읽기.
+"""
+
+from __future__ import annotations
+
+import json
+from datetime import datetime
+from pathlib import Path
+from typing import Any
+
+import aiomysql
+
+from config.config_db import get_db_pool
+from M02_MasterTemplete.M02_Template_Migrate import project_root
+
+from .B01_Dashboard_Repository import get_workflow_states_for_projects
+
+THUMB_POINTS = 100
+
+# 단계 이름 = 화면 단계 막대 글(ui_template_locale_common · _de) 한국어 · 모르는 종류 = 임도
+STAGE_NAMES: dict[str, tuple[str, ...]] = {
+    "ROAD": ("파일입력", "전처리", "노선설계", "횡단설계", "상세설계", "수량산출", "원가계산"),
+    "PLAN": ("기준 지정", "노선 분석", "확정 · 내보내기"),
+    "STABILITY": ("지반 조건", "2D 해석", "3D 해석", "보고서", "암반 해석", "광역 위험"),
+}
+
+
+def stage_names(kind: str | None) -> tuple[str, ...]:
+    return STAGE_NAMES.get(kind or "ROAD", STAGE_NAMES["ROAD"])
+
+
+def thumb_line(points: list[list[float]], limit: int = THUMB_POINTS) -> list[list[float]] | None:
+    """중심선 → limit 점 안쪽 · 0~1 칸(가로세로 비 유지 · y 아래로). 두 점 못 되면 None."""
+    if len(points) < 2:
+        return None
+    if len(points) > limit:
+        last = len(points) - 1
+        points = [points[round(i * last / (limit - 1))] for i in range(limit)]
+    xs = [p[0] for p in points]
+    ys = [p[1] for p in points]
+    dx, dy = max(xs) - min(xs), max(ys) - min(ys)
+    span = max(dx, dy) or 1.0
+    ox, oy = (1 - dx / span) / 2, (1 - dy / span) / 2
+    return [
+        [round(ox + (x - min(xs)) / span, 4), round(1 - oy - (y - min(ys)) / span, 4)]
+        for x, y in zip(xs, ys, strict=True)
+    ]
+
+
+def _route_points(root: Path | None, route_data_path: str | None) -> list[list[float]]:
+    from B05_Profile.B05_Profile_Engine_Sections import _load_route_polyline
+
+    if root is None or not route_data_path:
+        return []
+    try:
+        return [[x, y] for x, y, _ in _load_route_polyline(root, route_data_path)]
+    except (OSError, ValueError):
+        return []
+
+
+def _structure_count(root: Path | None) -> int | None:
+    from B05_Profile.B05_Profile_Structures_Repository import structures_file_path
+
+    if root is None:
+        return None
+    try:
+        with open(structures_file_path(str(root)), encoding="utf-8") as handle:
+            items = json.load(handle).get("structures")
+    except (OSError, ValueError, AttributeError):
+        return None
+    return len(items) if isinstance(items, list) else None
+
+
+def _marks(values: list[Any]) -> str:
+    return ", ".join(["%s"] * len(values))
+
+
+async def project_summaries(project_ids: list[str]) -> list[dict[str, Any]]:
+    """프로젝트 여럿 → 카드 요약(넣은 차례 · 지운 · 없는 프로젝트는 빠짐)."""
+    ids = list(dict.fromkeys(str(pid) for pid in project_ids))
+    if not ids:
+        return []
+    pool = get_db_pool()
+    async with pool.acquire() as connection, connection.cursor(aiomysql.DictCursor) as cursor:
+        await cursor.execute(
+            f"""SELECT p.id, p.name, p.project_kind, p.region, p.updated_at, p.storage_path,
+                       p.company_id, p.user_id, u.name AS owner_name
+                FROM projects p LEFT JOIN users u ON u.id = p.user_id
+                WHERE p.id IN ({_marks(ids)}) AND p.deleted_at IS NULL""",
+            tuple(ids),
+        )
+        rows = {str(r["id"]): r for r in await cursor.fetchall()}
+        states = await get_workflow_states_for_projects(cursor, list(rows))
+        await cursor.execute(
+            f"""SELECT m.project_id, u.name FROM project_members m
+                JOIN users u ON u.id = m.user_id AND u.deleted_at IS NULL
+                WHERE m.project_id IN ({_marks(ids)}) ORDER BY u.name""",
+            tuple(ids),
+        )
+        members: dict[str, list[str]] = {}
+        for r in await cursor.fetchall():
+            members.setdefault(str(r["project_id"]), []).append(r["name"])
+        # 최신 노선 = B05 get_latest_route 와 같은 차례(computed_at · id 내림)
+        await cursor.execute(
+            f"""SELECT id, project_id, total_length_m, route_data_path FROM routes
+                WHERE project_id IN ({_marks(ids)})
+                ORDER BY computed_at DESC, id DESC""",
+            tuple(ids),
+        )
+        routes: dict[str, dict[str, Any]] = {}
+        for r in await cursor.fetchall():
+            routes.setdefault(str(r["project_id"]), r)
+        stations: dict[int, int] = {}
+        route_ids = [int(r["id"]) for r in routes.values()]
+        if route_ids:
+            await cursor.execute(
+                f"""SELECT route_id, COUNT(*) AS n FROM cross_sections
+                    WHERE route_id IN ({_marks(route_ids)}) GROUP BY route_id""",
+                tuple(route_ids),
+            )
+            stations = {int(r["route_id"]): int(r["n"]) for r in await cursor.fetchall()}
+
+    now = datetime.now()
+    result = []
+    for pid in ids:
+        row = rows.get(pid)
+        if row is None:
+            continue
+        state = states.get(pid, {"current_stage": 0, "stages": []})
+        names = stage_names(row.get("project_kind"))
+        index = min(int(state["current_stage"]), len(names) - 1)
+        stages = state["stages"]
+        root = project_root(row.get("storage_path"))
+        route = routes.get(pid)
+        metrics: dict[str, Any] = {}
+        if route and route.get("total_length_m") is not None:
+            metrics["length_m"] = round(float(route["total_length_m"]), 1)
+        if route and stations.get(int(route["id"])):
+            metrics["stations"] = stations[int(route["id"])]
+        structures = _structure_count(root)
+        if structures is not None:
+            metrics["structures"] = structures
+        owners = [row["owner_name"]] if row.get("owner_name") else []
+        owners += [name for name in members.get(pid, []) if name and name not in owners]
+        updated = row.get("updated_at")
+        result.append(
+            {
+                "id": pid,
+                "name": row["name"],
+                "kind": row.get("project_kind") or "ROAD",
+                "region": row.get("region"),
+                "company_id": row.get("company_id"),
+                "stage_index": index,
+                "stage_total": len(names),
+                "stage_name": names[index],
+                "done": bool(stages) and all(s["state"] == "COMPLETE" for s in stages),
+                "updated_at": updated.isoformat() if updated else None,
+                "stale_days": (now - updated).days if updated else None,
+                "owner_user_id": row.get("user_id"),
+                "owner_names": owners,
+                "thumb": thumb_line(
+                    _route_points(root, route.get("route_data_path")) if route else []
+                ),
+                "metrics": metrics,
+            }
+        )
+    return result

main.py

@@ -28,6 +28,7 @@ from A07_Register.A07_Register_Router import router as a07_register_router
 from A08_Support.A08_Support_Router import router as a08_support_router
 from A09_Security.A09_Security_Router import router as a09_security_router
 from B01_Dashboard.B01_Dashboard_Router import router as b01_dashboard_router
+from B01_Dashboard.B01_Dashboard_Router_Home import router as b01_home_router
 from B02_ProjRegister.B02_ProjRegister_Router import router as b02_proj_register_router
 from B03_FileInput.B03_FileInput_Repository import ProjectNotFoundError
 from B03_FileInput.B03_FileInput_Router import router as b03_file_input_router
@@ -464,6 +465,7 @@ app.include_router(a07_register_router)
 app.include_router(a08_support_router)
 app.include_router(a09_security_router)
 app.include_router(b01_dashboard_router)
+app.include_router(b01_home_router)
 app.include_router(b02_proj_register_router)
 protected = [Depends(verify_session)]
 # 신규 그룹 등록(요구 345) — 권한은 길 안에서(산림청 · 회사).

resources/tester/test_547_1_project_summary.py

@@ -0,0 +1,143 @@
+"""547-1 프로젝트 카드 요약 — 썸네일 줄임 · 단계 · 없는 값 빼기 · 보이는 프로젝트만."""
+
+import asyncio
+import json
+from datetime import datetime, timedelta
+
+import B01_Dashboard.B01_Dashboard_Router_Home as home
+import B01_Dashboard.B01_Dashboard_Summary as summary
+
+ROUTE = "B05_Profile/route/r.geojson"
+
+
+def test_thumb_line_fits_unit_box_and_caps_points():
+    line = [[1000.0 + i, 2000.0 + i * 0.5] for i in range(500)]
+    thumb = summary.thumb_line(line)
+    assert len(thumb) == 100
+    assert thumb[0] == [0.0, 0.75] and thumb[-1] == [1.0, 0.25]  # y 아래로 · 세로 가운데
+    assert all(0 <= x <= 1 and 0 <= y <= 1 for x, y in thumb)
+    assert summary.thumb_line([[0.0, 0.0]]) is None
+
+
+def test_stage_names_follow_step_bar():
+    assert len(summary.stage_names("ROAD")) == 7
+    assert summary.stage_names("PLAN")[0] == "기준 지정"
+    assert summary.stage_names("EROSION") == summary.stage_names("ROAD")
+
+
+class _Cursor:
+    def __init__(self, db):
+        self.db, self.rows = db, []
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *_):
+        return False
+
+    async def execute(self, sql, args=()):
+        self.db["queries"] += 1
+        key = next(k for k in self.db["answers"] if k in sql)
+        self.rows = self.db["answers"][key]
+
+    async def fetchall(self):
+        return self.rows
+
+
+class _Pool:
+    def __init__(self, db):
+        self.db = db
+
+    def acquire(self):
+        pool = self
+
+        class _Conn:
+            async def __aenter__(self):
+                return self
+
+            async def __aexit__(self, *_):
+                return False
+
+            def cursor(self, *_):
+                return _Cursor(pool.db)
+
+        return _Conn()
+
+
+def test_summaries_in_batch(monkeypatch, tmp_path):
+    now = datetime.now()
+    route_dir = tmp_path / "B05_Profile" / "route"
+    route_dir.mkdir(parents=True)
+    coords = [[0, 0, 1], [10, 0, 1], [10, 5, 1]]
+    (route_dir / "r.geojson").write_text(json.dumps({"geometry": {"coordinates": coords}}))
+    (route_dir / "structures.json").write_text(json.dumps({"structures": [{}, {}]}))
+    db = {
+        "queries": 0,
+        "answers": {
+            "FROM projects p LEFT JOIN users": [
+                {
+                    "id": "a",
+                    "name": "가",
+                    "project_kind": "ROAD",
+                    "region": "강원",
+                    "updated_at": now - timedelta(days=20),
+                    "storage_path": "storage/a",
+                    "company_id": 1,
+                    "user_id": 7,
+                    "owner_name": "엄",
+                },
+                {
+                    "id": "b",
+                    "name": "나",
+                    "project_kind": "PLAN",
+                    "region": None,
+                    "updated_at": now,
+                    "storage_path": None,
+                    "company_id": 1,
+                    "user_id": 7,
+                    "owner_name": "엄",
+                },
+            ],
+            "project_workflow_stages": [
+                {"project_id": "a", "stage_no": n, "stage_key": "", "state": s}
+                | {"progress_percent": 0, "params": None, "message": None}
+                | {"started_at": None, "completed_at": None}
+                for n, s in enumerate(["COMPLETE", "COMPLETE", "IN_PROGRESS"] + ["NOT"] * 4)
+            ],
+            "project_members": [{"project_id": "a", "name": "박"}],
+            "FROM routes": [
+                {"id": 9, "project_id": "a", "total_length_m": 15.04, "route_data_path": ROUTE},
+                {"id": 5, "project_id": "a", "total_length_m": 1.0, "route_data_path": "old"},
+            ],
+            "cross_sections": [{"route_id": 9, "n": 3}],
+        },
+    }
+    monkeypatch.setattr(summary, "get_db_pool", lambda: _Pool(db))
+    monkeypatch.setattr(summary, "project_root", lambda p: tmp_path if p else None)
+    out = asyncio.run(summary.project_summaries(["b", "a", "zz"]))
+    assert [s["id"] for s in out] == ["b", "a"]  # 넣은 차례 · 없는 id 빠짐
+    b, a = out
+    assert (a["stage_index"], a["stage_total"], a["stage_name"]) == (2, 7, "노선설계")
+    assert a["stale_days"] == 20 and a["owner_names"] == ["엄", "박"]
+    assert a["metrics"] == {"length_m": 15.0, "stations": 3, "structures": 2}
+    assert len(a["thumb"]) == 3 and a["done"] is False
+    assert b["thumb"] is None and b["metrics"] == {} and b["stage_name"] == "기준 지정"
+    assert db["queries"] == 5
+
+
+def test_summary_api_drops_invisible_ids(monkeypatch):
+    seen = {}
+
+    async def fake_visible(session):
+        return ["a", "b"]
+
+    async def fake_summaries(ids):
+        seen["ids"] = ids
+        return []
+
+    monkeypatch.setattr(home, "visible_project_ids", fake_visible)
+    monkeypatch.setattr(home, "project_summaries", fake_summaries)
+    asyncio.run(home.projects_summary(ids="b, x,a,b", session={}))
+    assert seen["ids"] == ["b", "a"]
+    asyncio.run(home.projects_summary(ids=None, session={}))
+    assert seen["ids"] == ["a", "b"]