eomsangdonandClaude Opus 5.5
54e08d0d65
- 저장 모듈 D03_PlanAnalysis_Store — 읽기 · 원자 쓰기 · 판 · 옛 DB 칸 읽을 때 옮김(지점 셋 꼴 · 연도 행 접기 · 옛 연도 노선) · plan_state(조각 · 연도 지문 · 확정 = 지문 같을 때만) - D03 [저장] · [초기화] · D02 지점 = 파일만 + 연도 상태 줄(지문 · 확정) · 프로젝트 잠금 · 파일 쓰기 실패 = DB 되돌림 · 빈 연도 줄 지움 - D04 요약 · 업체 지정 · 확정 메일 · 내보내기 · 대시보드 [연결] · 받은 예정노선 · 산림청 홈 · 개요 연결 = 파일에서 · 업체 줄 연도 칸 안 읽음(⑤) - 산림청 홈 「분석 전」 = 노선 안 그은 예정노선 프로젝트 수 · plan-source 지운 원본 거름 - 옛 연도 단위 지정 · 옛 연도 행 노선 길 걷음(DB 실측 0) · 연결 때 설계속도 씨앗 걷음(591-1 이음) - SQL 023 = 업체 줄 year NULL 허용(옛 칸은 590-3 에서 지움) - 시험: 새 지킴 590_1(옮김 · 원자 쓰기 · 연도 읽기 · 잠금) · 낡은 시험 고침 — 529_3 · 530_2 · 531_3 · 531_d04 · 547_2 · 563_2(DB 조각 · 연도 행 흉내 → 파일 흉내 · 옛 연도 지정 시험 · 설계속도 씨앗 시험 지움 = 그 길 걷음) Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
438 lines
18 KiB
Python
438 lines
18 KiB
Python
"""D 예정노선 확정 · 업체 지정 API (요구 345 · 391 · 452 · PLAN 531-1 · 531-2 · 563-2 · 590-1).
|
||
|
||
- `GET /api/projects/{id}/plan/assignees` — 조각별 지정 업체(지정 회사 = 자기 행만)
|
||
- `PUT /api/projects/{id}/plan/assignees` — 조각별 지정 업체 저장(등록 회사 또는 업체명 + 메일)
|
||
- `POST /api/projects/{id}/plan/finalize` — 연도 노선 확정 → 업체별 조각 zip → 메일 첨부 → 발송 시각
|
||
- `GET /api/projects/{id}/plan/summary` — D04 화면 연도 · 조각 요약(지정 회사 = 자기 조각 · 행만)
|
||
- `GET /api/projects/{id}/plan/export` — [내보내기] zip(산림청 = 전 조각 + 전체 · 공유 = 자기 조각)
|
||
|
||
단위 = 조각(설계 내용 파일 `D03_PlanAnalysis_Store` · 구역 × 갈래 · 요구 391). 확정은 연도 단위
|
||
(`plan_route_years.confirmed_at` · 지문이 파일과 같을 때만). 업체 줄 = 조각 번호 ·
|
||
연도는 파일 조각에서.
|
||
|
||
지정 · 확정은 산림청(시스템 관리자 또는 메일 *.go.kr)만. 지정받은 회사는 공용 접근 판정이
|
||
GET 만 통과시킴. 계산 자리 ③ 서버 단독 — 내보내기 파일은 파일 노선을 서버가 다시 센 것으로만.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import io
|
||
import logging
|
||
import zipfile
|
||
from typing import Any
|
||
from urllib.parse import quote
|
||
from uuid import UUID
|
||
|
||
from fastapi import APIRouter, Depends, HTTPException
|
||
from fastapi.responses import Response
|
||
from pydantic import BaseModel
|
||
|
||
from common_util.common_util_auth import is_forest_service, verify_session
|
||
from common_util.common_util_crs import tm_to_lonlat
|
||
from D03_PlanAnalysis import D03_PlanAnalysis_Store as store
|
||
from D04_PlanConfirm import D04_PlanConfirm_Repository as repo
|
||
from D04_PlanConfirm.D04_PlanConfirm_Email import build_completion_mail, send_completion_mail
|
||
from D04_PlanConfirm.D04_PlanConfirm_Export import (
|
||
MIME_ZIP,
|
||
YearRoute,
|
||
build_export,
|
||
company_zip,
|
||
safe_name,
|
||
station_label,
|
||
)
|
||
|
||
logger = logging.getLogger(__name__)
|
||
router = APIRouter(prefix="/api/projects", tags=["D04 Plan Confirm"])
|
||
|
||
|
||
class AssigneeIn(BaseModel):
|
||
year: int | None = None # 안 씀 — 연도 = 파일 조각 연도
|
||
piece_id: str
|
||
company_id: int | None = None # 등록 회사 — 없으면 수기 업체
|
||
company_name: str | None = None
|
||
email: str | None = None
|
||
|
||
|
||
class AssigneesRequest(BaseModel):
|
||
assignees: list[AssigneeIn]
|
||
years: list[int] | None = None # 맞출 연도(업체를 다 뺀 연도도) — 비우면 보낸 업체의 연도
|
||
|
||
|
||
class FinalizeRequest(BaseModel):
|
||
years: list[int] | None = None # 비우면 분석이 저장된 모든 연도
|
||
resend: bool = False # 이미 메일을 보낸 업체에도 다시 보냄
|
||
|
||
|
||
def _require_forest(session: dict[str, Any]) -> None:
|
||
if not is_forest_service(session):
|
||
raise HTTPException(
|
||
status_code=403, detail="산림청만 예정노선을 확정 · 지정할 수 있습니다."
|
||
)
|
||
|
||
|
||
def _key(row: dict[str, Any]) -> tuple:
|
||
"""같은 업체 판정 — 같은 조각에서 등록 회사 = id · 수기 = 메일(대소문자 무시)."""
|
||
if row.get("company_id") is not None:
|
||
return (row["piece_id"], "c", row["company_id"])
|
||
return (row["piece_id"], "e", (row.get("email") or "").strip().lower())
|
||
|
||
|
||
def plan_assignee_changes(
|
||
existing: list[dict[str, Any]], wanted: list[dict[str, Any]], years: set[int]
|
||
) -> tuple[list[int], list[dict], list[dict]]:
|
||
"""보낸 연도(`years` · 업체 줄 `year` = 파일 조각 연도)만 맞춰 바꿈 —
|
||
(지울 id, 고칠 행, 넣을 행).
|
||
|
||
같은 업체는 행을 살려 두어 발송 시각 · 연결한 프로젝트가 남는다.
|
||
"""
|
||
old = {_key(row): row for row in existing if row["year"] in years}
|
||
new = {_key(row): row for row in wanted}
|
||
delete_ids = [row["id"] for key, row in old.items() if key not in new]
|
||
updates = [{**row, "id": old[key]["id"]} for key, row in new.items() if key in old]
|
||
inserts = [row for key, row in new.items() if key not in old]
|
||
return delete_ids, updates, inserts
|
||
|
||
|
||
async def _resolve_assignee(item: AssigneeIn) -> dict[str, Any]:
|
||
"""등록 회사면 이름 · 대표 메일을 DB 에서, 수기면 이름 + 메일을 검사."""
|
||
email = (item.email or "").strip()
|
||
name = (item.company_name or "").strip()
|
||
unit = {"piece_id": item.piece_id}
|
||
if item.company_id is not None:
|
||
company = await repo.get_company_contact(item.company_id)
|
||
if not company:
|
||
raise HTTPException(
|
||
status_code=400, detail=f"등록 회사를 찾을 수 없습니다: {item.company_id}"
|
||
)
|
||
return {
|
||
**unit,
|
||
"company_id": item.company_id,
|
||
"company_name": company["name"],
|
||
"email": email or company.get("email") or None,
|
||
}
|
||
if not name or "@" not in email:
|
||
raise HTTPException(status_code=400, detail="업체명과 메일 주소가 필요합니다.")
|
||
return {**unit, "company_id": None, "company_name": name, "email": email}
|
||
|
||
|
||
async def _plan_project(project_id: UUID) -> dict[str, Any]:
|
||
project = await repo.get_plan_project(str(project_id))
|
||
if not project:
|
||
raise HTTPException(status_code=404, detail="프로젝트를 찾을 수 없습니다.")
|
||
if project["project_kind"] != "PLAN":
|
||
raise HTTPException(status_code=400, detail="예정노선 프로젝트가 아닙니다.")
|
||
return project
|
||
|
||
|
||
def route_units(pieces: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||
"""표 한 줄 단위 = 파일 조각(`store.plan_units` · 연도 차례)."""
|
||
return [
|
||
{
|
||
"piece_id": p["id"],
|
||
"line": p["line"],
|
||
"line_name": p.get("line_name"),
|
||
"year": int(p["year"]),
|
||
"epsg": int(p["epsg"]),
|
||
"planned": p["planned"],
|
||
"start_chainage_m": float(p["from_m"]),
|
||
"end_chainage_m": float(p["to_m"]),
|
||
"length_m": float(p["length_m"]),
|
||
"confirmed_at": p.get("confirmed_at"),
|
||
}
|
||
for p in pieces
|
||
]
|
||
|
||
|
||
def with_years(
|
||
assignees: list[dict[str, Any]], units: list[dict[str, Any]]
|
||
) -> list[dict[str, Any]]:
|
||
"""업체 줄에 연도 = 그 조각의 파일 연도(조각이 없으면 None) — 업체 줄 `year` 칸은 안 봄."""
|
||
years = {u["piece_id"]: u["year"] for u in units}
|
||
return [{**a, "year": years.get(a.get("piece_id"))} for a in assignees]
|
||
|
||
|
||
def unit_key(unit: dict[str, Any]) -> str | int:
|
||
"""내보내기 묶음 열쇠 — 조각 id."""
|
||
return unit["piece_id"]
|
||
|
||
|
||
def unit_label(unit: dict[str, Any]) -> str:
|
||
"""zip 폴더 · 첨부 이름 — 「2026_갈래1_p3」."""
|
||
line = (unit["line_name"] or unit["line"]).replace(" ", "")
|
||
return f"{unit['year']}_{line}_{unit['piece_id']}"
|
||
|
||
|
||
def unit_of(assignee: dict[str, Any], units: list[dict[str, Any]]) -> dict[str, Any] | None:
|
||
"""업체 행의 단위 = 그 조각."""
|
||
return next((u for u in units if u["piece_id"] == assignee.get("piece_id")), None)
|
||
|
||
|
||
def _saved(unit: dict[str, Any]) -> bool:
|
||
return len(unit["planned"]) >= 2
|
||
|
||
|
||
def visible_to(
|
||
session: dict[str, Any], units: list[dict[str, Any]], assignees: list[dict[str, Any]]
|
||
) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
|
||
"""산림청 = 전부 · 지정 회사 = 자기 회사가 지정된 단위 · 자기 업체 행만(남의 이름 · 메일 뺌)."""
|
||
if is_forest_service(session):
|
||
return units, assignees
|
||
company = session.get("company_id")
|
||
mine = [a for a in assignees if a["company_id"] is not None and a["company_id"] == company]
|
||
own = [unit_of(a, units) for a in mine]
|
||
return [u for u in units if any(u is o for o in own)], mine
|
||
|
||
|
||
async def _units(project_id: UUID) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
|
||
"""(연도 줄, 단위) — 파일에서 · 연도 줄 = {year, epsg, route_json, fingerprint,
|
||
confirmed_at}."""
|
||
state = await store.plan_state(project_id)
|
||
epsg = state["plan"]["epsg"]
|
||
rows = [
|
||
{"year": year, "epsg": epsg, "route_json": row["route"], **row}
|
||
for year, row in state["years"].items()
|
||
]
|
||
return rows, route_units(state["units"])
|
||
|
||
|
||
async def _assignees(project_id: UUID, units: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||
return with_years(await repo.list_assignees(str(project_id)), units)
|
||
|
||
|
||
@router.get("/{project_id}/plan/assignees", response_model=None)
|
||
async def get_plan_assignees(
|
||
project_id: UUID, session: dict = Depends(verify_session)
|
||
) -> dict[str, Any]:
|
||
await _plan_project(project_id)
|
||
_, units = await _units(project_id)
|
||
_, assignees = visible_to(session, [], await _assignees(project_id, units))
|
||
return {"status": "success", "assignees": assignees}
|
||
|
||
|
||
@router.put("/{project_id}/plan/assignees", response_model=None)
|
||
async def put_plan_assignees(
|
||
project_id: UUID, request: AssigneesRequest, session: dict = Depends(verify_session)
|
||
) -> dict[str, Any]:
|
||
"""보낸 연도의 지정 업체를 그대로 맞춤 — 연도를 안 보낸 곳은 건드리지 않음.
|
||
|
||
연도는 파일 조각에서 읽음(보낸 연도는 무시)."""
|
||
_require_forest(session)
|
||
await _plan_project(project_id)
|
||
_, units = await _units(project_id)
|
||
piece_years = {u["piece_id"]: u["year"] for u in units}
|
||
for item in request.assignees:
|
||
if item.piece_id not in piece_years:
|
||
raise HTTPException(status_code=400, detail=f"조각이 없습니다: {item.piece_id}")
|
||
years = {piece_years[a.piece_id] for a in request.assignees} | set(request.years or [])
|
||
unknown = sorted(years - set(piece_years.values()))
|
||
if unknown:
|
||
raise HTTPException(status_code=400, detail=f"예정노선 연도가 아닙니다: {unknown}")
|
||
wanted = [
|
||
{**await _resolve_assignee(item), "year": piece_years[item.piece_id]}
|
||
for item in request.assignees
|
||
]
|
||
existing = await _assignees(project_id, units)
|
||
delete_ids, updates, inserts = plan_assignee_changes(existing, wanted, years)
|
||
await repo.apply_assignee_changes(str(project_id), delete_ids, updates, inserts)
|
||
return {"status": "success", "assignees": await _assignees(project_id, units)}
|
||
|
||
|
||
def unit_routes(units: list[dict[str, Any]]) -> list[YearRoute]:
|
||
"""저장된 단위 → 내보내기 입력(`planned` = 프로젝트 좌표 꺾은선)."""
|
||
return [
|
||
YearRoute(
|
||
year=unit["year"],
|
||
points=[(p[0], p[1]) for p in unit["planned"]],
|
||
crs_epsg=int(unit["epsg"]),
|
||
start_station_m=unit["start_chainage_m"],
|
||
end_station_m=unit["end_chainage_m"],
|
||
piece=unit["piece_id"],
|
||
line=(unit["line_name"] or unit["line"] or "").replace(" ", "") or None,
|
||
)
|
||
for unit in units
|
||
if _saved(unit)
|
||
]
|
||
|
||
|
||
def _result(assignee: dict[str, Any], state: str) -> dict[str, Any]:
|
||
return {
|
||
"year": assignee["year"],
|
||
"piece_id": assignee.get("piece_id"),
|
||
"company_name": assignee["company_name"],
|
||
"email": assignee["email"],
|
||
"state": state,
|
||
}
|
||
|
||
|
||
@router.post("/{project_id}/plan/finalize", response_model=None)
|
||
async def finalize_plan(
|
||
project_id: UUID, request: FinalizeRequest, session: dict = Depends(verify_session)
|
||
) -> dict[str, Any]:
|
||
"""연도 노선 확정 → 지정 업체마다 그 조각 zip 을 첨부해 메일 → 보낸 곳에 발송 시각."""
|
||
_require_forest(session)
|
||
project = await _plan_project(project_id)
|
||
rows, units = await _units(project_id)
|
||
routes = unit_routes(units)
|
||
saved = {r.year for r in routes}
|
||
years = sorted(set(request.years) if request.years else saved)
|
||
missing = [y for y in years if y not in saved]
|
||
if not years or missing:
|
||
raise HTTPException(
|
||
status_code=409, detail=f"분석 [저장]이 안 된 연도입니다: {missing or '없음'}"
|
||
)
|
||
fingerprints = {r["year"]: r["fingerprint"] for r in rows}
|
||
await repo.mark_confirmed(str(project_id), {y: fingerprints[y] for y in years})
|
||
|
||
# 옛 누가거리가 연도마다 이어지므로 저장된 전 단위로 한 번에 만들고, 보낼 때만 고름.
|
||
first = float(project.get("route_start_m") or 0.0)
|
||
bundle = await asyncio.to_thread(
|
||
build_export, project["name"], project["region"] or "", routes, first
|
||
)
|
||
results = []
|
||
for assignee in await _assignees(project_id, units):
|
||
year = assignee["year"] # 파일 조각 연도(요구 452 ⑤)
|
||
if year not in years:
|
||
continue
|
||
if assignee["notified_at"] and not request.resend:
|
||
results.append(_result(assignee, "already_sent"))
|
||
continue
|
||
if not assignee["email"]:
|
||
results.append(_result(assignee, "no_email"))
|
||
continue
|
||
unit = unit_of(assignee, units)
|
||
if unit is None or unit_key(unit) not in bundle.by_year:
|
||
results.append(_result(assignee, "no_route"))
|
||
continue
|
||
key = unit_key(unit)
|
||
zip_name, zip_bytes = company_zip(project["name"], unit_label(unit), bundle.by_year[key])
|
||
subject, body = build_completion_mail(
|
||
project_name=project["name"],
|
||
region=project["region"],
|
||
year=year,
|
||
length_m=bundle.lengths_m[key],
|
||
recipient=assignee["company_name"] or assignee["email"],
|
||
attachment_names=[zip_name],
|
||
)
|
||
sent = await send_completion_mail(
|
||
to_email=assignee["email"],
|
||
subject=subject,
|
||
body=body,
|
||
attachments=[(zip_name, zip_bytes, MIME_ZIP)],
|
||
)
|
||
if sent:
|
||
await repo.mark_notified(assignee["id"])
|
||
results.append(_result(assignee, "sent" if sent else "failed"))
|
||
return {"status": "success", "confirmed_years": years, "notifications": results}
|
||
|
||
|
||
def _end_point(point: list[float], epsg: int) -> dict[str, float]:
|
||
lon, lat = tm_to_lonlat(float(point[0]), float(point[1]), epsg)
|
||
return {"x": float(point[0]), "y": float(point[1]), "lon": lon, "lat": lat}
|
||
|
||
|
||
def year_summary(row: dict[str, Any]) -> dict[str, Any]:
|
||
"""연도 한 줄 — 분석 [저장]이 없으면 saved False 만."""
|
||
body = row.get("route_json") or {}
|
||
planned = body.get("planned") or []
|
||
out = {"year": int(row["year"]), "saved": len(planned) >= 2, "confirmed_at": None}
|
||
if row.get("confirmed_at"):
|
||
out["confirmed_at"] = str(row["confirmed_at"])
|
||
if out["saved"]:
|
||
epsg = int(row["epsg"])
|
||
out.update(
|
||
length_m=body.get("length_m"),
|
||
start_chainage_m=body.get("start_chainage_m"),
|
||
end_chainage_m=body.get("end_chainage_m"),
|
||
mode=body.get("mode"),
|
||
start=_end_point(planned[0], epsg),
|
||
end=_end_point(planned[-1], epsg),
|
||
)
|
||
return out
|
||
|
||
|
||
def unit_summary(unit: dict[str, Any]) -> dict[str, Any]:
|
||
"""표 한 줄(조각) — label = 「본선 0+0~12+10」."""
|
||
out = {key: unit[key] for key in ("piece_id", "line", "line_name", "year", "length_m")}
|
||
out.update(saved=_saved(unit), confirmed_at=None, label=None)
|
||
if unit["confirmed_at"]:
|
||
out["confirmed_at"] = str(unit["confirmed_at"])
|
||
if out["saved"]:
|
||
epsg = int(unit["epsg"])
|
||
out.update(
|
||
start_chainage_m=unit["start_chainage_m"],
|
||
end_chainage_m=unit["end_chainage_m"],
|
||
start=_end_point(unit["planned"][0], epsg),
|
||
end=_end_point(unit["planned"][-1], epsg),
|
||
)
|
||
if unit["piece_id"]:
|
||
out["label"] = piece_label(
|
||
unit["line_name"] or unit["line"], unit["start_chainage_m"], unit["end_chainage_m"]
|
||
)
|
||
return out
|
||
|
||
|
||
def piece_label(line_name: str, from_m: float, to_m: float) -> str:
|
||
"""조각 이름 「본선 0+0~12+10」(측점 = 선마다 No.0)."""
|
||
return f"{line_name} {station_label(from_m)}~{station_label(to_m)}"
|
||
|
||
|
||
@router.get("/{project_id}/plan/summary", response_model=None)
|
||
async def get_plan_summary(
|
||
project_id: UUID, session: dict = Depends(verify_session)
|
||
) -> dict[str, Any]:
|
||
"""`years` = 연도 줄(확정 고르기) · `pieces` = 표 한 줄 단위(조각)."""
|
||
project = await _plan_project(project_id)
|
||
rows, units = await _units(project_id)
|
||
units, assignees = visible_to(session, units, await _assignees(project_id, units))
|
||
own_years = {u["year"] for u in units}
|
||
years = [year_summary(row) for row in rows if int(row["year"]) in own_years]
|
||
return {
|
||
"status": "success",
|
||
"name": project["name"],
|
||
"years": years,
|
||
"pieces": [unit_summary(u) for u in units],
|
||
"assignees": assignees,
|
||
}
|
||
|
||
|
||
def export_zip(bundle: Any, units: list[dict[str, Any]], with_all: bool) -> bytes:
|
||
"""단위마다 `<연도>_<선>_<조각>/` · 전체 `전체/` — 좌표표 이름이 같아 폴더로."""
|
||
buffer = io.BytesIO()
|
||
with zipfile.ZipFile(buffer, "w", zipfile.ZIP_DEFLATED) as archive:
|
||
for unit in units:
|
||
for name, data in bundle.by_year[unit_key(unit)].items():
|
||
archive.writestr(f"{unit_label(unit)}/{name}", data)
|
||
if with_all:
|
||
for name, data in bundle.all.items():
|
||
archive.writestr(f"전체/{name}", data)
|
||
return buffer.getvalue()
|
||
|
||
|
||
@router.get("/{project_id}/plan/export", response_model=None)
|
||
async def export_plan(project_id: UUID, session: dict = Depends(verify_session)) -> Response:
|
||
"""산림청 = 저장된 전 조각 + 전체 · 지정 회사 = 확정된 자기 조각만(전체 없음)."""
|
||
project = await _plan_project(project_id)
|
||
_, units = await _units(project_id)
|
||
routes = unit_routes(units)
|
||
if not routes:
|
||
raise HTTPException(status_code=409, detail="분석 [저장]이 된 연도가 없습니다.")
|
||
forest = is_forest_service(session)
|
||
picked = [u for u in units if _saved(u)]
|
||
if not forest:
|
||
own, _ = visible_to(session, picked, await _assignees(project_id, units))
|
||
picked = [u for u in own if u["confirmed_at"]]
|
||
if not picked:
|
||
raise HTTPException(status_code=403, detail="받을 수 있는 조각이 없습니다.")
|
||
first = float(project.get("route_start_m") or 0.0)
|
||
bundle = await asyncio.to_thread(
|
||
build_export, project["name"], project["region"] or "", routes, first
|
||
)
|
||
data = await asyncio.to_thread(export_zip, bundle, picked, forest)
|
||
name = f"{safe_name(project['name'])}_예정노선.zip"
|
||
return Response(
|
||
content=data,
|
||
media_type=MIME_ZIP,
|
||
headers={"Content-Disposition": f"attachment; filename*=UTF-8''{quote(name)}"},
|
||
)
|