feat(B01): 홈 API — 역할별 진행 중 카드 · 처리할 일 · 회사 · 산림청 · 시스템 요약 (PLAN 547-2)

- GET /api/dashboard/home — role(USER · ADMIN · FOREST · SYSTEM_ADMIN) · 진행 중 카드(최근 수정 순 · 썸네일 앞 8 개) - 처리할 일 = 새로 받은 · 업데이트 있는 예정노선 · 가입 요청 수 · 7 일 안 만료 보관 파일 · 보관 묶음 수 - 회사 관리자 = 회사 이름 · 팀원 수 · 산림청 = 예정노선 연도 상태별 수 · 업체 지정 대기 · 시스템 = 회사 · 사용자 수 · 서버 지금 값 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-26 18:07:54 +09:00
parent 8b1b286e04
commit bc1c8b44e8
1 changed files with 163 additions and 1 deletions
@@ -1,18 +1,23 @@
 """B01 대시보드 홈 API (PLAN 547) — 프로젝트 카드 요약 · 역할별 홈."""

+from datetime import datetime, timedelta
 from typing import Any

 import aiomysql
 from fastapi import APIRouter, Depends, Query

-from common_util.common_util_auth import verify_session
+from common_util.common_util_auth import is_forest_email, verify_session
 from config.config_db import get_db_pool

+from .B01_Dashboard_Repository import get_system_resources
 from .B01_Dashboard_Summary import project_summaries

 router = APIRouter(prefix="/api/dashboard", tags=["B01_Dashboard"])

 SUMMARY_LIMIT = 200
+# 홈 카드 썸네일은 앞 8 개만(9 넘음 꼴은 그림 대신 숫자 · 그래프)
+HOME_THUMBS = 8
+EXPIRE_SOON_DAYS = 7


 async def visible_project_ids(session: dict[str, Any]) -> list[str]:
@@ -23,6 +28,10 @@ async def visible_project_ids(session: dict[str, Any]) -> list[str]:
            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"]),)
+    return await _ids(where, args)
+
+
+async def _ids(where: str, args: tuple) -> list[str]:
    pool = get_db_pool()
    async with pool.acquire() as connection, connection.cursor(aiomysql.DictCursor) as cursor:
        await cursor.execute(
@@ -45,3 +54,156 @@ async def projects_summary(
        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])}
+
+
+def home_role(session: dict[str, Any]) -> str:
+    """홈 꼴 역할 — SYSTEM_ADMIN · FOREST(*.go.kr) · ADMIN(회사 관리자 · 마스터) · USER."""
+    if session["role"] == "SYSTEM_ADMIN":
+        return "SYSTEM_ADMIN"
+    if is_forest_email(session.get("email")):
+        return "FOREST"
+    if session["role"] == "ADMIN" or session.get("is_master"):
+        return "ADMIN"
+    return "USER"
+
+
+async def home_project_ids(session: dict[str, Any], role: str) -> list[str]:
+    """홈 프로젝트 — 시스템 관리자 = 전부 · 회사 관리자 = 회사 · 그 밖 = 내가 만든 · 참여한 것."""
+    if role in ("SYSTEM_ADMIN", "ADMIN") or not session.get("company_id"):
+        return await visible_project_ids(session)
+    me = int(session["user_id"])
+    return await _ids(
+        "p.deleted_at IS NULL AND p.company_id = %s AND (p.user_id = %s OR EXISTS "
+        "(SELECT 1 FROM project_members m WHERE m.project_id = p.id AND m.user_id = %s))",
+        (int(session["company_id"]), me, me),
+    )
+
+
+async def _one(cursor: Any, sql: str, args: tuple = ()) -> dict[str, Any]:
+    await cursor.execute(sql, args)
+    return await cursor.fetchone() or {}
+
+
+async def _todos(cursor: Any, session: dict[str, Any], role: str) -> dict[str, Any]:
+    """처리할 일 · 보관함 — 받은 예정노선(새로 · 업데이트) · 가입 요청 · 곧 만료 보관 파일."""
+    from D01_PlanRegister.D01_PlanRegister_Router import assigned_plan_routes
+
+    company_id = session.get("company_id")
+    plans = (await assigned_plan_routes(session))["routes"] if company_id else []
+    pick = ("project_id", "project_name", "year", "linked_project_id")
+    todos: dict[str, Any] = {
+        "new_plans": [
+            {k: p[k] for k in pick}
+            for p in plans
+            if not p["linked_project_id"] and p["confirmed_at"]
+        ],
+        "updated_plans": [{k: p[k] for k in pick} for p in plans if p["update_available"]],
+    }
+    if role == "SYSTEM_ADMIN" or (role == "ADMIN" and company_id):
+        where = "" if role == "SYSTEM_ADMIN" else " AND company_id = %s"
+        row = await _one(
+            cursor,
+            f"SELECT COUNT(*) AS n FROM join_requests WHERE status = 'PENDING'{where}",
+            () if role == "SYSTEM_ADMIN" else (int(company_id),),
+        )
+        todos["join_requests"] = int(row.get("n") or 0)
+    await cursor.execute(
+        """SELECT id, name, status, expires_at FROM temp_upload_batches
+           WHERE user_id = %s AND status <> 'linked'
+             AND (expires_at IS NULL OR expires_at > NOW())
+           ORDER BY expires_at IS NULL, expires_at""",
+        (int(session["user_id"]),),
+    )
+    batches = list(await cursor.fetchall())
+    soon = datetime.now() + timedelta(days=EXPIRE_SOON_DAYS)
+    todos["expiring_temp"] = [
+        {
+            "batch_id": str(b["id"]),
+            "name": b["name"],
+            "expires_at": b["expires_at"].isoformat(),
+            "days_left": max(0, (b["expires_at"] - datetime.now()).days),
+        }
+        for b in batches
+        if b["expires_at"] and b["expires_at"] <= soon
+    ]
+    todos["temp_count"] = len(batches)
+    return todos
+
+
+async def _company(cursor: Any, company_id: int) -> dict[str, Any]:
+    row = await _one(
+        cursor,
+        """SELECT c.name, (SELECT COUNT(*) FROM users u
+                 WHERE u.company_id = c.id AND u.deleted_at IS NULL) AS member_count
+           FROM companies c WHERE c.id = %s""",
+        (company_id,),
+    )
+    return {"id": company_id, "name": row.get("name"), "member_count": row.get("member_count")}
+
+
+async def _forest(cursor: Any, company_id: int) -> dict[str, Any]:
+    """예정노선 연도 상태별 수(분석 전 · 분석됨 · 확정) · 업체 지정 대기(확정 · 업체 없음)."""
+    await cursor.execute(
+        """SELECT y.route_json IS NOT NULL AS analyzed, y.confirmed_at IS NOT NULL AS confirmed,
+                  EXISTS (SELECT 1 FROM plan_route_assignees a
+                          WHERE a.project_id = y.project_id AND a.year = y.year) AS assigned
+           FROM plan_route_years y
+           JOIN projects p ON p.id = y.project_id AND p.deleted_at IS NULL
+           WHERE p.company_id = %s""",
+        (company_id,),
+    )
+    status = {"draft": 0, "analyzed": 0, "confirmed": 0}
+    waiting = 0
+    for r in await cursor.fetchall():
+        key = "confirmed" if r["confirmed"] else "analyzed" if r["analyzed"] else "draft"
+        status[key] += 1
+        waiting += bool(r["confirmed"] and not r["assigned"])
+    return {"plan_status": status, "assign_waiting": waiting}
+
+
+async def _system(cursor: Any) -> dict[str, Any]:
+    row = await _one(
+        cursor,
+        """SELECT (SELECT COUNT(*) FROM companies WHERE deleted_at IS NULL) AS companies,
+                  (SELECT COUNT(*) FROM users WHERE deleted_at IS NULL) AS users""",
+    )
+    await cursor.execute("SELECT id, name FROM companies WHERE deleted_at IS NULL")
+    names = {int(r["id"]): r["name"] for r in await cursor.fetchall()}
+    current = (await get_system_resources(days=1))["current"]
+    server = {k: current.get(f"{k}_usage_percent") for k in ("cpu", "memory", "disk")}
+    return {
+        "companies": int(row.get("companies") or 0),
+        "users": int(row.get("users") or 0),
+        "company_names": names,
+        "server": server,
+    }
+
+
+@router.get("/home")
+async def dashboard_home(session: dict[str, Any] = Depends(verify_session)):
+    """역할별 홈 — 진행 중 프로젝트 카드(최근 수정 순) · 처리할 일 · 회사 · 산림청 · 시스템 요약.
+
+    단계 분포 · 팀원별 · 종류별 · 회사별 수는 화면이 `projects` 로 셈(같은 목록 한 벌).
+    """
+    role = home_role(session)
+    ids = await home_project_ids(session, role)
+    cards = [s for s in await project_summaries(ids[:SUMMARY_LIMIT]) if not s["done"]]
+    for card in cards[HOME_THUMBS:]:
+        card["thumb"] = None
+    company_id = session.get("company_id")
+    pool = get_db_pool()
+    async with pool.acquire() as connection, connection.cursor(aiomysql.DictCursor) as cursor:
+        body: dict[str, Any] = {
+            "status": "success",
+            "role": role,
+            "projects": cards,
+            "project_total": len(ids),
+            "todos": await _todos(cursor, session, role),
+        }
+        if role == "ADMIN" and company_id:
+            body["company"] = await _company(cursor, int(company_id))
+        if role == "FOREST" and company_id:
+            body["forest"] = await _forest(cursor, int(company_id))
+        if role == "SYSTEM_ADMIN":
+            body["system"] = await _system(cursor)
+    return body