feat(528): 신규 프로젝트 그룹 공통 백엔드 — 종류 칸 · 산림청 판정 · D01 · E01 생성 API · 공유 GET 만 · 가입 go.kr = 산림청

- 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>

A06_Login/A06_Login_Router.py

@@ -15,6 +15,7 @@ from common_util.common_util_auth import (
     hash_device_token,
     hash_password,
     hash_user_agent,
+    is_forest_service,
     set_device_token_cookie,
     set_session_cookie,
     verify_password,
@@ -173,6 +174,7 @@ async def current_session(session: dict = Depends(verify_session)):
             "role": session["role"],
             "company_id": session["company_id"],
             "is_master": session["is_master"],
+            "is_forest_service": is_forest_service(session),
         },
     }
 

A07_Register/A07_Register_Router.py

@@ -12,6 +12,7 @@ from common_util.common_util_auth import (
     hash_device_token,
     hash_password,
     hash_user_agent,
+    is_forest_email,
     set_device_token_cookie,
     verify_password,
 )
@@ -84,17 +85,20 @@ async def verify_registration(payload: RegisterVerifyRequest, request: Request):
     ):
         raise HTTPException(status_code=400, detail="인증 코드가 유효하지 않습니다.")
     await consume_otp(otp["id"])
-    await complete_registration(user["id"], bool(user["is_master"]))
+    # 메일 *.go.kr = 회사 「산림청」 고정 — 화면이 보낸 회사 값은 받지 않음(요구 347-8).
+    account_status = await complete_registration(
+        user["id"], bool(user["is_master"]), forest=is_forest_email(user["email"])
+    )
     # 가입 인증한 브라우저를 신뢰 등록하여 첫 로그인 시 OTP를 생략한다.
     agent = request.headers.get("user-agent", "unknown")[:1000]
     device_token = generate_device_token()
     await trust_device(user["id"], hash_device_token(device_token), hash_user_agent(agent))
     # 회원가입 시점에는 회사가 없으므로(NO_COMPANY) 마스터 알림은 발송하지 않는다.
-    # 회사 생성/연결은 로그인 후 B01_Dashboard에서 진행한다.
+    # 회사 생성/연결은 로그인 후 B01_Dashboard에서 진행한다(산림청 메일은 위에서 붙음).
     response = JSONResponse(
         {
             "status": "success",
-            "account_status": "NO_COMPANY",
+            "account_status": account_status,
         }
     )
     set_device_token_cookie(response, device_token)

A07_Register/A07_Register_UI_Auth_Page.ts

@@ -13,6 +13,7 @@ import { requestRegistration, verifyRegistration } from "./A07_Register_Api_Fetc
 import "./A07_Register_UI_Style.css";
 
 const L = (key: keyof typeof ui_locales): string => ui_locales[key][currentLanguageIndex];
+const FOREST_COMPANY_NAME = "산림청";
 
 interface AgreementField {
   root: HTMLElement;
@@ -80,6 +81,14 @@ export function renderA07Register(root: HTMLElement): void {
     type: "text",
   });
   const email = createInputField({ label: L("A07_Register_Field_Email"), type: "email" });
+  // 메일 *.go.kr = 회사 「산림청」 고정(요구 347-8) — 서버도 같은 규칙으로 붙임.
+  const company = createInputField({ label: L("A07_Register_Field_Company"), type: "text" });
+  company.input.value = FOREST_COMPANY_NAME;
+  company.input.readOnly = true;
+  company.root.hidden = true;
+  email.input.addEventListener("input", () => {
+    company.root.hidden = !email.input.value.trim().toLowerCase().endsWith(".go.kr");
+  });
   const phone = createInputField({ label: L("A07_Register_Field_Phone"), type: "tel" });
   const password = createInputField({ label: L("A07_Register_Field_Password"), type: "password" });
   const passwordConfirm = createInputField({
@@ -132,6 +141,7 @@ export function renderA07Register(root: HTMLElement): void {
       name.root,
       position.root,
       email.root,
+      company.root,
       phone.root,
       password.root,
       passwordConfirm.root,

B01_Dashboard/B01_Dashboard_Repository.py

@@ -193,7 +193,7 @@ async def list_user_projects(user_id: int) -> list[dict[str, Any]]:
     pool = get_db_pool()
     async with pool.acquire() as connection, connection.cursor(aiomysql.DictCursor) as cursor:
         await cursor.execute(
-            """SELECT id, company_id, name, region, road_type, project_year,
+            """SELECT id, company_id, name, region, road_type, project_kind, project_year,
                       estimated_length_m, route_start_m, route_end_m,
                       memo, status, updated_at, created_at,
                       client_org, project_number, work_amount, design_date,
@@ -210,7 +210,8 @@ async def list_company_projects(company_id: int) -> list[dict[str, Any]]:
     pool = get_db_pool()
     async with pool.acquire() as connection, connection.cursor(aiomysql.DictCursor) as cursor:
         await cursor.execute(
-            """SELECT p.id, p.company_id, p.name, p.region, p.road_type, p.project_year,
+            """SELECT p.id, p.company_id, p.name, p.region, p.road_type, p.project_kind,
+                      p.project_year,
                       p.estimated_length_m, p.route_start_m, p.route_end_m,
                       p.memo, p.status, p.updated_at, p.created_at,
                       p.client_org, p.project_number, p.work_amount, p.design_date,
@@ -229,7 +230,8 @@ async def list_all_projects() -> list[dict[str, Any]]:
     pool = get_db_pool()
     async with pool.acquire() as connection, connection.cursor(aiomysql.DictCursor) as cursor:
         await cursor.execute(
-            """SELECT p.id, p.company_id, p.name, p.region, p.road_type, p.project_year,
+            """SELECT p.id, p.company_id, p.name, p.region, p.road_type, p.project_kind,
+                      p.project_year,
                       p.estimated_length_m, p.route_start_m, p.route_end_m,
                       p.memo, p.status, p.updated_at, p.created_at,
                       p.client_org, p.project_number, p.work_amount, p.design_date,

D01_PlanRegister/D01_PlanRegister_Router.py

@@ -0,0 +1,87 @@
+"""D01 예정노선 등록 API · 대시보드 「받은 예정노선」 (요구 345 · 요구서 3-1 · 2-2)."""
+
+from typing import Any
+
+import aiomysql
+from fastapi import APIRouter, Depends, HTTPException, Request
+from pydantic import BaseModel, Field
+
+from B05_Profile.B05_Profile_Engine_Grade import resolve_design_speed
+from B05_Profile.B05_Profile_PageSettings import grade_class_of
+from common_util.common_util_auth import is_forest_service, 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/d01", tags=["D01_PlanRegister"])
+dashboard_router = APIRouter(prefix="/api/dashboard", tags=["D01_PlanRegister"])
+
+
+class CreatePlanRequest(BaseModel):
+    project_name: str = Field(..., min_length=1, max_length=255)
+    project_year: int = Field(..., ge=2000, le=2100)
+    project_year_end: int | None = Field(default=None, ge=2000, le=2100)
+    region: str = Field(..., min_length=1, max_length=100)
+    road_type: str = Field(..., pattern="^(main|fire|work)$")
+    design_speed_kph: int | None = None
+    start_chainage_m: float | None = Field(default=None, ge=0)
+    end_chainage_m: float | None = Field(default=None, ge=0)
+
+
+@router.post("/project")
+async def post_plan_project(
+    request: Request,
+    payload: CreatePlanRequest,
+    session: dict[str, Any] = Depends(require_company),
+) -> dict[str, str]:
+    """예정노선 프로젝트 생성 — 산림청(시스템관리자 · *.go.kr)만."""
+    if not is_forest_service(session):
+        raise HTTPException(status_code=403, detail="예정노선은 산림청만 만들 수 있습니다.")
+    if session.get("company_id") is None:
+        raise HTTPException(status_code=403, detail="회사 연결이 필요합니다.")
+    year_end = payload.project_year_end or payload.project_year
+    if year_end < payload.project_year:
+        raise HTTPException(status_code=400, detail="사업연도 끝이 시작보다 앞입니다.")
+    start_m, end_m = payload.start_chainage_m, payload.end_chainage_m
+    if start_m is not None and end_m is not None and start_m >= end_m:
+        raise HTTPException(
+            status_code=400, detail="노선 시작 누가거리는 종료 누가거리보다 작아야 합니다."
+        )
+    speed = resolve_design_speed(grade_class_of(payload.road_type, {}), payload.design_speed_kph)
+    project_id = await create_kind_project(
+        request=request,
+        user_id=int(session["user_id"]),
+        company_id=int(session["company_id"]),
+        kind="PLAN",
+        name=payload.project_name.strip(),
+        region=payload.region.strip(),
+        road_type=payload.road_type,
+        project_year=payload.project_year,
+        project_year_end=year_end,
+        design_speed_kph=speed,
+        route_start_m=start_m,
+        route_end_m=end_m,
+    )
+    return {"project_id": project_id}
+
+
+@dashboard_router.get("/plan-routes/assigned")
+async def assigned_plan_routes(
+    session: dict[str, Any] = Depends(require_company),
+) -> dict[str, Any]:
+    """우리 회사에 지정된 예정노선 연도 — 보기 · 연결 상태."""
+    async with (
+        get_db_pool().acquire() as connection,
+        connection.cursor(aiomysql.DictCursor) as cursor,
+    ):
+        await cursor.execute(
+            """SELECT a.project_id, p.name AS project_name, a.year, a.linked_project_id
+               FROM plan_route_assignees a
+               JOIN projects p ON p.id = a.project_id AND p.deleted_at IS NULL
+               WHERE a.company_id = %s
+               ORDER BY p.name, a.year""",
+            (session.get("company_id"),),
+        )
+        rows = await cursor.fetchall()
+    # ponytail: update_available 자리만 — 지문 비교는 531-3.
+    routes = [{**r, "update_available": False} for r in rows]
+    return {"status": "success", "routes": routes}

E01_StabRegister/E01_StabRegister_Router.py

@@ -0,0 +1,87 @@
+"""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}

common_util/common_util_auth.py

@@ -148,6 +148,16 @@ async def get_optional_session(request: Request) -> dict[str, Any] | None:
         return None
 
 
+def is_forest_email(email: str | None) -> bool:
+    """메일이 *.go.kr(대소문자 무시) — 가입 때 회사 「산림청」 고정."""
+    return bool(email) and email.strip().lower().endswith(".go.kr")
+
+
+def is_forest_service(session: dict[str, Any]) -> bool:
+    """산림청 = 시스템관리자 또는 메일 *.go.kr — 예정노선 생성 · 고침 권한."""
+    return session.get("role") == "SYSTEM_ADMIN" or is_forest_email(session.get("email"))
+
+
 async def require_master(session: dict[str, Any] = Depends(verify_session)) -> dict[str, Any]:
     if not session["is_master"] and session["role"] != "SYSTEM_ADMIN":
         raise HTTPException(status_code=403, detail="마스터 권한이 필요합니다.")
@@ -196,6 +206,19 @@ async def require_project_access(
         )
         row = await cursor.fetchone()
 
+        if row and row[0] != session["company_id"]:
+            # 예정노선을 지정받은 회사 = 보기만(요구 345 · GET 만 통과).
+            await cursor.execute(
+                "SELECT 1 FROM plan_route_assignees"
+                " WHERE project_id = %s AND company_id = %s LIMIT 1",
+                (str(project_id), session["company_id"]),
+            )
+            shared = await cursor.fetchone()
+            if shared:
+                if request.method in ("GET", "HEAD"):
+                    return session
+                raise HTTPException(status_code=403, detail="보기만 허용된 예정노선입니다.")
+
     if not row:
         raise HTTPException(status_code=404, detail="프로젝트를 찾을 수 없습니다.")
     if row[0] != session["company_id"]:

common_util/common_util_auth_repository.py

@@ -148,15 +148,32 @@ async def consume_otp(otp_id: int) -> None:
         await connection.commit()
 
 
-async def complete_registration(user_id: int, is_master: bool) -> None:
+async def complete_registration(user_id: int, is_master: bool, forest: bool = False) -> str:
+    """메일 인증 끝 — 계정 상태를 돌려줌. forest(*.go.kr) = 회사 「산림청」 고정 · ACTIVE."""
     pool = get_db_pool()
     async with pool.acquire() as connection, connection.cursor() as cursor:
-        await cursor.execute(
-            """UPDATE users SET status = 'NO_COMPANY', last_email_verified_at = CURRENT_TIMESTAMP
-               WHERE id = %s""",
-            (user_id,),
-        )
+        company = None
+        if forest:
+            await cursor.execute(
+                "SELECT id FROM companies WHERE name = '산림청' AND deleted_at IS NULL"
+            )
+            company = await cursor.fetchone()
+        if company:
+            await cursor.execute(
+                """UPDATE users SET status = 'ACTIVE', company_id = %s, role = 'USER',
+                          last_email_verified_at = CURRENT_TIMESTAMP
+                   WHERE id = %s""",
+                (company[0], user_id),
+            )
+        else:
+            await cursor.execute(
+                """UPDATE users SET status = 'NO_COMPANY',
+                          last_email_verified_at = CURRENT_TIMESTAMP
+                   WHERE id = %s""",
+                (user_id,),
+            )
         await connection.commit()
+    return "ACTIVE" if company else "NO_COMPANY"
 
 
 async def company_master_email(company_id: int) -> str | None:

common_util/common_util_project_kind.py

@@ -0,0 +1,102 @@
+"""종류별 프로젝트 행 만들기 — D 예정노선 · E 안정성평가(요구 345).
+
+임도설계(ROAD)는 B02 그대로. 저장소 경로 · 초기 폴더는 B02 함수를 그대로 씀.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Any
+from uuid import uuid4
+
+from B02_ProjRegister.B02_ProjRegister_Repository import (
+    _build_project_storage,
+    _initialize_project_storage,
+)
+from common_util.common_util_audit import record_audit
+from common_util.common_util_storage import get_project_stage_path
+from common_util.common_util_workflow_state import initialize_project_stages
+from config.config_db import get_db_pool
+
+KIND_STAGE_DIRS = {
+    "PLAN": ("D02_PlanBase", "D03_PlanAnalysis", "D04_PlanConfirm"),
+    "STABILITY": ("E02_StabGround", "E03_StabSection", "E04_Stab3D", "E05_StabReport"),
+}
+
+
+async def create_kind_project(
+    *,
+    request: Any | None,
+    user_id: int,
+    company_id: int,
+    kind: str,
+    name: str,
+    region: str | None,
+    road_type: str | None,
+    project_year: int | None,
+    project_year_end: int | None = None,
+    design_speed_kph: int | None = None,
+    route_start_m: float | None = None,
+    route_end_m: float | None = None,
+    source_project_id: str | None = None,
+) -> str:
+    """projects 행 + 참여자 + 종류별 단계 + 저장소 폴더 → 새 project_id."""
+    project_id = str(uuid4())
+    storage_path, project_root = _build_project_storage(company_id, user_id, project_id)
+    now = datetime.utcnow()
+
+    async with get_db_pool().acquire() as connection, connection.cursor() as cursor:
+        try:
+            await connection.begin()
+            await cursor.execute(
+                """
+                INSERT INTO projects (
+                    id, user_id, company_id, name, region, road_type, project_kind,
+                    project_year, project_year_end, design_speed_kph,
+                    route_start_m, route_end_m, status, crs_epsg, storage_path,
+                    source_project_id, source_copied_at, created_at, updated_at
+                )
+                VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, 'NEW', 5178, %s,
+                        %s, %s, %s, %s)
+                """,
+                (
+                    project_id,
+                    user_id,
+                    company_id,
+                    name,
+                    region,
+                    road_type,
+                    kind,
+                    project_year,
+                    project_year_end,
+                    design_speed_kph,
+                    route_start_m,
+                    route_end_m,
+                    storage_path,
+                    source_project_id,
+                    now if source_project_id else None,
+                    now,
+                    now,
+                ),
+            )
+            await cursor.execute(
+                "INSERT IGNORE INTO project_members (project_id, user_id) VALUES (%s, %s)",
+                (project_id, user_id),
+            )
+            await initialize_project_stages(cursor, project_id, kind)
+            await record_audit(
+                cursor,
+                actor_id=user_id,
+                action="PROJECT_CREATE",
+                resource_type="project",
+                resource_ref=project_id,
+                request=request,
+            )
+            _initialize_project_storage(project_root, project_id)
+            for stage_dir in KIND_STAGE_DIRS[kind]:
+                get_project_stage_path(str(project_root), stage_dir)
+            await connection.commit()
+        except Exception:
+            await connection.rollback()
+            raise
+    return project_id

common_util/common_util_workflow_state.py

@@ -20,9 +20,19 @@ STAGE_KEYS = [
 ]
 
 
-async def initialize_project_stages(cursor: aiomysql.DictCursor, project_id: str) -> None:
-    """프로젝트 생성 시 7개의 단계를 NOT_STARTED 상태로 시드한다."""
-    for stage_no, stage_key in enumerate(STAGE_KEYS):
+# 종류별 단계(요구 345) — ROAD = B03~B09 · PLAN = D02~D04 · STABILITY = E02~E05.
+STAGE_KEYS_BY_KIND = {
+    "ROAD": STAGE_KEYS,
+    "PLAN": ["PLAN_BASE", "PLAN_ANALYSIS", "PLAN_CONFIRM"],
+    "STABILITY": ["STAB_GROUND", "STAB_SECTION", "STAB_3D", "STAB_REPORT"],
+}
+
+
+async def initialize_project_stages(
+    cursor: aiomysql.DictCursor, project_id: str, kind: str = "ROAD"
+) -> None:
+    """프로젝트 생성 시 종류별 단계를 NOT_STARTED 상태로 시드한다."""
+    for stage_no, stage_key in enumerate(STAGE_KEYS_BY_KIND[kind]):
         await cursor.execute(
             """
             INSERT INTO project_workflow_stages (

config/config_system.py

@@ -211,6 +211,15 @@ PROJECT_STORAGE_STAGE_DIRS = frozenset(
         "B07_DesignDetail",
         "B08_Quantity",
         "B09_Estimation",
+        "D01_PlanRegister",
+        "D02_PlanBase",
+        "D03_PlanAnalysis",
+        "D04_PlanConfirm",
+        "E01_StabRegister",
+        "E02_StabGround",
+        "E03_StabSection",
+        "E04_Stab3D",
+        "E05_StabReport",
     }
 )
 

db_management/020_project_groups.sql

@@ -0,0 +1,59 @@
+-- 020_project_groups.sql
+-- 신규 프로젝트 그룹 — 종류 · 예정노선 연도 구간 · 지정 업체 · 산림청 회사 (요구 345 · 347-8)
+--
+-- 추가만 함 — 기존 행 = ROAD(임도설계) · 기존 칸 · 동작 그대로.
+-- 예정노선 시작 · 종료 누가거리는 기존 route_start_m · route_end_m 을 씀.
+
+USE aislo_db;
+
+ALTER TABLE projects
+  ADD COLUMN IF NOT EXISTS project_kind VARCHAR(20) NOT NULL DEFAULT 'ROAD'
+    COMMENT 'ROAD 임도설계 · PLAN 예정노선 · EROSION 사방설계 · STABILITY 안정성평가' AFTER road_type,
+  ADD COLUMN IF NOT EXISTS project_year_end INT NULL
+    COMMENT '사업연도 범위 끝(예정노선) — 시작은 project_year' AFTER project_year,
+  ADD COLUMN IF NOT EXISTS design_speed_kph INT NULL
+    COMMENT '설계속도(km/h · 예정노선 등록값)' AFTER project_year_end,
+  ADD COLUMN IF NOT EXISTS source_project_id CHAR(36) NULL
+    COMMENT '원본 프로젝트(E = 임도 · 임도 = 예정노선)',
+  ADD COLUMN IF NOT EXISTS source_copied_at DATETIME NULL COMMENT '원본 복사 시각',
+  ADD COLUMN IF NOT EXISTS source_fingerprint CHAR(40) NULL COMMENT '복사 때 원본 지문(sha1)';
+
+CREATE TABLE IF NOT EXISTS plan_route_years (
+  project_id CHAR(36) NOT NULL COMMENT 'projects.id (PLAN)',
+  year INT NOT NULL,
+  start_x DOUBLE NULL,
+  start_y DOUBLE NULL,
+  end_x DOUBLE NULL,
+  end_y DOUBLE NULL,
+  epsg INT NULL,
+  route_json JSON NULL COMMENT '분석 결과 노선',
+  fingerprint CHAR(40) NULL COMMENT '노선 내용 지문(sha1)',
+  confirmed_at DATETIME NULL,
+  updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+  PRIMARY KEY (project_id, year),
+  CONSTRAINT fk_plan_route_years_project FOREIGN KEY (project_id)
+    REFERENCES projects (id) ON DELETE CASCADE
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='예정노선 연도 구간';
+
+CREATE TABLE IF NOT EXISTS plan_route_assignees (
+  id INT AUTO_INCREMENT PRIMARY KEY,
+  project_id CHAR(36) NOT NULL COMMENT 'projects.id (PLAN)',
+  year INT NOT NULL,
+  company_id INT NULL COMMENT '등록 회사 — NULL 이면 수기 업체',
+  company_name VARCHAR(255) NULL,
+  email VARCHAR(255) NULL,
+  notified_at DATETIME NULL,
+  linked_project_id CHAR(36) NULL COMMENT '업체가 연결한 임도 프로젝트',
+  created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  KEY idx_plan_route_assignees_project (project_id, year),
+  KEY idx_plan_route_assignees_company (company_id),
+  CONSTRAINT fk_plan_route_assignees_project FOREIGN KEY (project_id)
+    REFERENCES projects (id) ON DELETE CASCADE,
+  CONSTRAINT fk_plan_route_assignees_company FOREIGN KEY (company_id)
+    REFERENCES companies (id) ON DELETE SET NULL,
+  CONSTRAINT fk_plan_route_assignees_linked FOREIGN KEY (linked_project_id)
+    REFERENCES projects (id) ON DELETE SET NULL
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='예정노선 연도별 지정 업체';
+
+-- 산림청 회사 — 메일 *.go.kr 가입자가 자동으로 붙음.
+INSERT IGNORE INTO companies (name, company_code, status) VALUES ('산림청', 'FOREST', 'ACTIVE');

main.py

@@ -72,6 +72,9 @@ from B08_Quantity.B08_Quantity_Router_Summary import router as b08_summary_route
 from B08_Quantity.B08_Quantity_Router_Material import router as b08_material_router
 from B08_Quantity.B08_Quantity_Router_Milestone import router as b08_milestone_router
 from B09_Estimation.B09_Estimation_Router import router as b09_estimation_router
+from D01_PlanRegister.D01_PlanRegister_Router import dashboard_router as d01_dashboard_router
+from D01_PlanRegister.D01_PlanRegister_Router import router as d01_plan_register_router
+from E01_StabRegister.E01_StabRegister_Router import router as e01_stab_register_router
 from M01_MasterData.M01_MasterData_Router import router as m01_master_data_router
 from M02_MasterTemplete.M02_Labor_Router import router as m02_labor_router
 from M02_MasterTemplete.M02_MasterTemplete_Router import router as m02_master_template_router
@@ -449,6 +452,10 @@ app.include_router(a09_security_router)
 app.include_router(b01_dashboard_router)
 app.include_router(b02_proj_register_router)
 protected = [Depends(verify_session)]
+# 신규 그룹 등록(요구 345) — 권한은 길 안에서(산림청 · 회사).
+app.include_router(d01_plan_register_router, dependencies=protected)
+app.include_router(d01_dashboard_router, dependencies=protected)
+app.include_router(e01_stab_register_router, dependencies=protected)
 # 프로젝트 경로가 붙는 라우터는 그 프로젝트가 내 회사 것인지도 함께 본다
 # (2026-09-01 — 지금까지 id만 알면 남의 회사 자료에 닿았다).
 protected_with_company = [

resources/tester/test_528_project_groups.py

@@ -0,0 +1,258 @@
+"""528-1 · 3 · 5 신규 프로젝트 그룹 공통(요구 345 · 347-8).
+
+- 산림청 판정 — *.go.kr(대소문자 · 하위 도메인) · 시스템관리자 · 아닌 메일
+- D01 생성 — 일반 사용자 403 · 산림청 = PLAN · 설계속도 기본 · 연도 끝 기본
+- E01 생성 — 원본 B06 미완료 400 · 남의 것 404 · 완료 = STABILITY + 원본 칸
+- 공유 판정 — 지정 회사 GET 200 · PUT 403 · 미지정 403 · 같은 회사 그대로
+- 가입 — go.kr = 회사 「산림청」 · ACTIVE · 화면이 보낸 회사 값 무시 · 다른 메일 = NO_COMPANY
+- 종류별 단계 시드 · storage 허용 목록
+"""
+
+import asyncio
+
+import pytest
+from fastapi import HTTPException
+
+from A07_Register import A07_Register_Router as reg_router
+from A07_Register.A07_Register_Schema import RegisterRequest
+from common_util import common_util_auth as auth
+from common_util import common_util_auth_repository as auth_repo
+from common_util.common_util_workflow_state import initialize_project_stages
+from config.config_system import PROJECT_STORAGE_STAGE_DIRS
+from D01_PlanRegister import D01_PlanRegister_Router as d01
+from E01_StabRegister import E01_StabRegister_Router as e01
+
+
+class FakeCursor:
+    """SQL 조각 → 돌려줄 한 줄. 실행한 SQL · 인자를 쌓음."""
+
+    def __init__(self, answers, log):
+        self.answers, self.log, self.last = answers, log, None
+
+    async def execute(self, sql, args=None):
+        self.log.append((" ".join(sql.split()), args))
+        self.last = next((v for k, v in self.answers.items() if k in sql), None)
+
+    async def fetchone(self):
+        return self.last
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *a):
+        return False
+
+
+class FakePool:
+    def __init__(self, answers=None):
+        self.answers, self.log = answers or {}, []
+
+    def acquire(self):
+        return self
+
+    def cursor(self, *a):
+        return FakeCursor(self.answers, self.log)
+
+    async def commit(self):
+        self.log.append(("commit", None))
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *a):
+        return False
+
+
+class Req:
+    def __init__(self, method="GET", project_id="p1"):
+        self.method, self.path_params = method, {"project_id": project_id}
+
+
+USER = {"user_id": 9, "role": "USER", "company_id": 1, "email": "a@gmail.com"}
+FOREST = {"user_id": 8, "role": "USER", "company_id": 4, "email": "Kim@Forest.GO.KR"}
+ADMIN = {"user_id": 1, "role": "SYSTEM_ADMIN", "company_id": 1, "email": "x@aislo.kr"}
+
+
+@pytest.mark.parametrize(
+    "email,expected",
+    [
+        ("a@forest.go.kr", True),
+        ("A@KFS.GO.KR", True),
+        ("a@nifos.sub.forest.go.kr", True),
+        ("a@gmail.com", False),
+        ("a@notgo.kr", False),
+        ("a@go.kr.evil.com", False),
+        (None, False),
+    ],
+)
+def test_forest_email(email, expected):
+    assert auth.is_forest_email(email) is expected
+
+
+def test_forest_service():
+    assert auth.is_forest_service(FOREST)
+    assert auth.is_forest_service(ADMIN)
+    assert not auth.is_forest_service(USER)
+
+
+def _plan(**kw):
+    body = {"project_name": "예정", "project_year": 2026, "region": "홍천", "road_type": "fire"}
+    return d01.CreatePlanRequest(**{**body, **kw})
+
+
+def test_d01_plan_forbidden_for_user():
+    with pytest.raises(HTTPException) as e:
+        asyncio.run(d01.post_plan_project(None, _plan(), USER))
+    assert e.value.status_code == 403
+
+
+def test_d01_plan_created(monkeypatch):
+    seen = {}
+
+    async def fake_create(**kw):
+        seen.update(kw)
+        return "new-id"
+
+    monkeypatch.setattr(d01, "create_kind_project", fake_create)
+    out = asyncio.run(d01.post_plan_project(None, _plan(start_chainage_m=0), FOREST))
+    assert out == {"project_id": "new-id"}
+    assert seen["kind"] == "PLAN" and seen["company_id"] == 4
+    assert seen["design_speed_kph"] == 20 and seen["project_year_end"] == 2026
+    asyncio.run(d01.post_plan_project(None, _plan(design_speed_kph=40), ADMIN))
+    assert seen["design_speed_kph"] == 40
+    asyncio.run(d01.post_plan_project(None, _plan(road_type="work", design_speed_kph=40), ADMIN))
+    assert seen["design_speed_kph"] == 20  # 작업임도 = 20 만
+    with pytest.raises(HTTPException) as e:
+        asyncio.run(d01.post_plan_project(None, _plan(project_year_end=2025), FOREST))
+    assert e.value.status_code == 400
+
+
+def _e01(monkeypatch, rows):
+    seen = {}
+
+    async def fake_sources(company_id, project_id=None):
+        return rows
+
+    async def fake_create(**kw):
+        seen.update(kw)
+        return "stab-id"
+
+    monkeypatch.setattr(e01, "_sources", fake_sources)
+    monkeypatch.setattr(e01, "create_kind_project", fake_create)
+    payload = e01.CreateStabRequest(source_project_id="s" * 36)
+    return seen, lambda: asyncio.run(e01.post_stab_project(None, payload, USER))
+
+
+def test_e01_requires_b06(monkeypatch):
+    src = {"project_id": "s" * 36, "name": "임도", "region": "r", "road_type": "fire"}
+    _, call = _e01(monkeypatch, [{**src, "project_year": 2026, "section_complete": 0}])
+    with pytest.raises(HTTPException) as e:
+        call()
+    assert e.value.status_code == 400
+    _, call = _e01(monkeypatch, [])
+    with pytest.raises(HTTPException) as e:
+        call()
+    assert e.value.status_code == 404
+    seen, call = _e01(monkeypatch, [{**src, "project_year": 2026, "section_complete": 1}])
+    assert call() == {"project_id": "stab-id"}
+    assert seen["kind"] == "STABILITY" and seen["source_project_id"] == "s" * 36
+    assert seen["name"] == "임도 안정성평가"
+
+
+def _access(monkeypatch, method, assigned, owner_company=2):
+    pool = FakePool(
+        {
+            "FROM projects": (owner_company,),
+            "plan_route_assignees": (1,) if assigned else None,
+        }
+    )
+    monkeypatch.setattr(auth, "get_db_pool", lambda: pool)
+    return asyncio.run(auth.require_project_access(Req(method), USER))
+
+
+def test_shared_get_only(monkeypatch):
+    assert _access(monkeypatch, "GET", True) is USER
+    for method in ("PUT", "POST", "DELETE"):
+        with pytest.raises(HTTPException) as e:
+            _access(monkeypatch, method, True)
+        assert e.value.status_code == 403
+    with pytest.raises(HTTPException) as e:
+        _access(monkeypatch, "GET", False)
+    assert e.value.status_code == 403
+    assert _access(monkeypatch, "PUT", False, owner_company=1) is USER
+
+
+def test_register_forest_company(monkeypatch):
+    pool = FakePool({"FROM companies": (4,)})
+    monkeypatch.setattr(auth_repo, "get_db_pool", lambda: pool)
+    assert asyncio.run(auth_repo.complete_registration(5, False, forest=True)) == "ACTIVE"
+    update = [a for s, a in pool.log if s.startswith("UPDATE users")]
+    assert update == [(4, 5)]
+    pool = FakePool({"FROM companies": (4,)})
+    monkeypatch.setattr(auth_repo, "get_db_pool", lambda: pool)
+    assert asyncio.run(auth_repo.complete_registration(5, False)) == "NO_COMPANY"
+    assert not any("companies" in s for s, _ in pool.log)
+
+
+def test_register_verify_forces_forest(monkeypatch):
+    seen = {}
+
+    async def fake_user(email):
+        return {"id": 5, "email": email, "status": "PENDING_EMAIL", "is_master": 0}
+
+    async def fake_otp(user_id, kind):
+        from datetime import datetime, timedelta
+
+        return {"id": 1, "expires_at": datetime.utcnow() + timedelta(minutes=5), "otp_hash": "h"}
+
+    async def fake_complete(user_id, is_master, forest=False):
+        seen["forest"] = forest
+        return "ACTIVE" if forest else "NO_COMPANY"
+
+    async def noop(*a, **k):
+        return None
+
+    monkeypatch.setattr(reg_router, "get_user_by_email", fake_user)
+    monkeypatch.setattr(reg_router, "get_active_otp", fake_otp)
+    monkeypatch.setattr(reg_router, "verify_password", lambda a, b: True)
+    monkeypatch.setattr(reg_router, "consume_otp", noop)
+    monkeypatch.setattr(reg_router, "trust_device", noop)
+    monkeypatch.setattr(reg_router, "complete_registration", fake_complete)
+
+    class R:
+        headers = {}
+
+    body = reg_router.RegisterVerifyRequest(email="Lee@KFS.go.kr", otp_code="123456")
+    resp = asyncio.run(reg_router.verify_registration(body, R()))
+    assert seen["forest"] is True and b'"ACTIVE"' in resp.body
+    body = reg_router.RegisterVerifyRequest(email="lee@gmail.com", otp_code="123456")
+    resp = asyncio.run(reg_router.verify_registration(body, R()))
+    assert seen["forest"] is False and b'"NO_COMPANY"' in resp.body
+
+
+def test_register_ignores_sent_company():
+    data = RegisterRequest(
+        email="a@kfs.go.kr",
+        password="12345678",
+        password_confirm="12345678",
+        name="이",
+        terms_version="1.0",
+        terms_agreed=True,
+        privacy_agreed=True,
+        company_name="다른회사",
+    ).model_dump()
+    assert "company_name" not in data
+
+
+@pytest.mark.parametrize("kind,count", [("ROAD", 7), ("PLAN", 3), ("STABILITY", 4)])
+def test_stage_seed_by_kind(kind, count):
+    log = []
+    asyncio.run(initialize_project_stages(FakeCursor({}, log), "p1", kind))
+    assert len(log) == count
+    if kind == "PLAN":
+        assert [a[2] for _, a in log] == ["PLAN_BASE", "PLAN_ANALYSIS", "PLAN_CONFIRM"]
+
+
+def test_storage_dirs_allowed():
+    for name in ("D02_PlanBase", "D04_PlanConfirm", "E02_StabGround", "E05_StabReport"):
+        assert name in PROJECT_STORAGE_STAGE_DIRS