eomsangdonandClaude Opus 5.5
80d363c3aa
- D04 화면 — 연도 표(연장 · 시점 · 종점 · 누가 · 확정 · 지정 업체) · 업체 = 등록 회사 찾기 또는 이름 + 메일 · 캐시 d04-assignees · [저장] PUT · [확정] 연도 고름 → finalize(잠금판) · [내보내기] zip - PUT plan/assignees 에 years — 업체를 다 뺀 연도도 맞춤 - B03 노선 칸 옆 단추 · 머리 띠(새 파일에서 끼움 · 기존 동작 그대로) — 연결 = existing 길 · 옛 결과 지울 때 확인 창 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
308 lines
13 KiB
Python
308 lines
13 KiB
Python
"""D 예정노선 확정 · 업체 지정 API (요구 345 · PLAN 531-1 · 531-2).
|
|
|
|
- `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(산림청 = 전 연도 + 전체 · 공유 = 자기 연도)
|
|
|
|
지정 · 확정은 산림청(시스템 관리자 또는 메일 *.go.kr)만. 지정받은 회사는 공용 접근 판정이
|
|
GET 만 통과시킴. 계산 자리 ③ 서버 단독 — 내보내기 파일은 저장된 `route_json` 으로만 만듦.
|
|
"""
|
|
|
|
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 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,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
router = APIRouter(prefix="/api/projects", tags=["D04 Plan Confirm"])
|
|
|
|
|
|
class AssigneeIn(BaseModel):
|
|
year: int
|
|
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["year"], "c", row["company_id"])
|
|
return (row["year"], "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`)만 맞춰 바꿈 — (지울 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()
|
|
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 {
|
|
"year": item.year,
|
|
"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 {"year": item.year, "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
|
|
|
|
|
|
@router.get("/{project_id}/plan/assignees", response_model=None)
|
|
async def get_plan_assignees(project_id: UUID) -> dict[str, Any]:
|
|
await _plan_project(project_id)
|
|
return {"status": "success", "assignees": await repo.list_assignees(str(project_id))}
|
|
|
|
|
|
@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)
|
|
known = {row["year"] for row in await repo.list_year_routes(str(project_id))}
|
|
years = {a.year for a in request.assignees} | set(request.years or [])
|
|
unknown = sorted(years - known)
|
|
if unknown:
|
|
raise HTTPException(status_code=400, detail=f"예정노선 연도가 아닙니다: {unknown}")
|
|
wanted = [await _resolve_assignee(item) for item in request.assignees]
|
|
existing = await repo.list_assignees(str(project_id))
|
|
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 repo.list_assignees(str(project_id))}
|
|
|
|
|
|
def year_routes_from_rows(rows: list[dict[str, Any]]) -> list[YearRoute]:
|
|
"""분석이 저장된 연도 → 내보내기 입력 (D03 `route_json["planned"]` = 프로젝트 좌표 꺾은선)."""
|
|
routes = []
|
|
for row in rows:
|
|
body = row.get("route_json")
|
|
if not body or not body.get("planned"):
|
|
continue
|
|
routes.append(
|
|
YearRoute(
|
|
year=int(row["year"]),
|
|
points=[(p[0], p[1]) for p in body["planned"]],
|
|
crs_epsg=int(row["epsg"]),
|
|
start_station_m=body.get("start_chainage_m"),
|
|
end_station_m=body.get("end_chainage_m"),
|
|
)
|
|
)
|
|
return routes
|
|
|
|
|
|
def _result(assignee: dict[str, Any], state: str) -> dict[str, Any]:
|
|
return {
|
|
"year": assignee["year"],
|
|
"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)
|
|
routes = year_routes_from_rows(await repo.list_year_routes(str(project_id)))
|
|
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 '없음'}"
|
|
)
|
|
await repo.mark_confirmed(str(project_id), 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 repo.list_assignees(str(project_id)):
|
|
year = assignee["year"]
|
|
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
|
|
zip_name, zip_bytes = company_zip(project["name"], year, bundle.by_year[year])
|
|
subject, body = build_completion_mail(
|
|
project_name=project["name"],
|
|
region=project["region"],
|
|
year=year,
|
|
length_m=bundle.lengths_m[year],
|
|
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
|
|
|
|
|
|
@router.get("/{project_id}/plan/summary", response_model=None)
|
|
async def get_plan_summary(project_id: UUID) -> dict[str, Any]:
|
|
project = await _plan_project(project_id)
|
|
rows = await repo.list_year_routes(str(project_id))
|
|
return {
|
|
"status": "success",
|
|
"name": project["name"],
|
|
"years": [year_summary(row) for row in rows],
|
|
"assignees": await repo.list_assignees(str(project_id)),
|
|
}
|
|
|
|
|
|
def export_zip(bundle: Any, years: list[int], with_all: bool) -> bytes:
|
|
"""연도마다 `<연도>/` · 전체는 `전체/` — 좌표표 이름이 연도마다 같아 폴더로 가름."""
|
|
buffer = io.BytesIO()
|
|
with zipfile.ZipFile(buffer, "w", zipfile.ZIP_DEFLATED) as archive:
|
|
for year in years:
|
|
for name, data in bundle.by_year[year].items():
|
|
archive.writestr(f"{year}/{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)
|
|
rows = await repo.list_year_routes(str(project_id))
|
|
routes = year_routes_from_rows(rows)
|
|
if not routes:
|
|
raise HTTPException(status_code=409, detail="분석 [저장]이 된 연도가 없습니다.")
|
|
forest = is_forest_service(session)
|
|
if forest:
|
|
years = [r.year for r in routes]
|
|
else:
|
|
confirmed = {int(r["year"]) for r in rows if r.get("confirmed_at")}
|
|
mine = {
|
|
a["year"]
|
|
for a in await repo.list_assignees(str(project_id))
|
|
if a["company_id"] is not None and a["company_id"] == session.get("company_id")
|
|
}
|
|
years = sorted({r.year for r in routes} & confirmed & mine)
|
|
if not years:
|
|
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, years, 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)}"},
|
|
)
|