eomsangdonandClaude Opus 5
03a193af70
B04_PreProcess_Router.py 718줄 -> 493줄. - B04_PreProcess_Router_Status.py(221줄) — 분석 상태 조회·모델 프리뷰 엔드포인트. prefix 없이 만들고 본체가 include_router 로 실어 붙임(본체 prefix 를 그대로 상속). - B04_PreProcess_Router_Progress.py(47줄) — progress.json 읽기·쓰기 헬퍼. 본체와 상태 라우터가 함께 써야 해 순환 임포트를 피하려고 별도 모듈로 둠. 본체는 write_surface_progress 를 계속 재노출(B03_FileInput_Service_WF1 이 참조). 검증: ruff check/format 통과. 라우터 경로 9개가 분리 전과 동일(중복 prefix 없음 확인). 백엔드 재시작(8001) 후 openapi 에 surface 경로 12개 그대로, 브라우저에서 /surface/status 200(진행률 100·모델 3), /surface/models 200, /surface/models/415/preview 200 model/gltf-binary 7,946,756바이트. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
48 lines
1.8 KiB
Python
48 lines
1.8 KiB
Python
"""B04 지표면 분석 진행률 파일 입출력.
|
|
|
|
`B04_PreProcess_Router` 가 700줄을 넘겨 분리한 조각이다(2026-09-04). 라우터 본체와
|
|
상태 조회 라우터가 **같은 함수**를 써야 해서 여기에 둔다 — 순환 임포트를 피하려고
|
|
양쪽이 이 모듈을 바라본다. 동작·경로·값은 옮기기 전 그대로다.
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from common_util.common_util_json import atomic_write_json
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# 분석 진행률 파일: B04 산출 폴더 아래에 원자적으로 기록/조회한다.
|
|
PROGRESS_FILE_RELATIVE = ("B04_PreProcess", "processed", "progress.json")
|
|
|
|
|
|
def _progress_file_path(project_root: Path) -> Path:
|
|
return project_root.joinpath(*PROGRESS_FILE_RELATIVE)
|
|
|
|
|
|
def write_surface_progress(project_root: Path, percent: int, stage: str, message: str) -> None:
|
|
"""WF1 분석 진행률을 progress.json에 원자적으로 기록한다 (실패해도 분석은 계속)."""
|
|
try:
|
|
path = _progress_file_path(project_root)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
atomic_write_json(
|
|
path,
|
|
{"progress_percent": percent, "current_stage": stage, "message": message},
|
|
)
|
|
except OSError:
|
|
logger.warning("WF1 진행률 기록 실패: %s", project_root, exc_info=True)
|
|
|
|
|
|
def read_surface_progress(project_root: Path) -> dict[str, Any] | None:
|
|
"""progress.json을 읽어 반환한다. 없거나 손상 시 None."""
|
|
path = _progress_file_path(project_root)
|
|
if not path.is_file():
|
|
return None
|
|
try:
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
return data if isinstance(data, dict) else None
|
|
except (OSError, ValueError):
|
|
return None
|