"""B03 파일 입력 라우터 보조 — 필수 파일 판정·중복 업로드 판별·단계 기록·알림. 라우터 본체(`B03_FileInput_Router.py`)가 700줄을 넘어 떼어낸 조각이다(2026-09-04). 엔드포인트는 두지 않는다 — 순수 보조 함수와 파일 종류 상수만 둔다. """ import asyncio import json import logging from pathlib import Path from typing import Any from uuid import UUID import aiomysql from B03_FileInput.B03_FileInput_Email import ( send_file_upload_complete_email, ) from B03_FileInput.B03_FileInput_Repository import ( find_input_file_by_name, get_project_input_readiness, get_project_storage_relative_path, ) from B03_FileInput.B03_FileInput_Schema import ( ChunkSessionCreateRequest, ChunkSessionCreateResponse, UploadedFileResult, ) from common_util.common_util_initial_snapshot import clear_designing, discard_initial_snapshot from common_util.common_util_json import atomic_write_json from common_util.common_util_project_reset import purge_project_outputs from common_util.common_util_storage import resolve_stored_project_path from common_util.common_util_workflow_state import ( complete_stage, reset_stages_after_input_change, ) from config.config_db import get_db_pool logger = logging.getLogger(__name__) _ANALYSIS_RUNNING_MESSAGE = "이 프로젝트는 지금 분석 중입니다. 끝난 뒤에 새 자료를 올려 주세요." _REQUIRED_FILE_TYPES = frozenset({"prj", "tfw"}) _POINT_CLOUD_FILE_TYPES = frozenset({"las", "laz"}) # 계획노선은 CSV 또는 shapefile 중 하나면 된다 (2026-08-31 — 원청 정식 노선이 shapefile). _ROUTE_FILE_TYPES = frozenset({"csv", "shp"}) # shapefile은 이것들이 다 있어야 열린다. `.cpg`는 없으면 CP949로 읽으므로 필수가 아니다. # `route_prj`는 노선 세트 폴더에 있는 PRJ — 지형 PRJ(`prj`)와 따로 센다. _SHAPEFILE_REQUIRED_TYPES = frozenset({"shx", "dbf", "route_prj"}) def _total_chunks(size_bytes: int, chunk_size_bytes: int) -> int: return max(1, (size_bytes + chunk_size_bytes - 1) // chunk_size_bytes) def _is_point_cloud_result(result: UploadedFileResult) -> bool: """포인트클라우드 결과인지 — 임시 보관함 안내 메일 경로에서 쓴다. 프로젝트 업로드 경로는 더 이상 이 판정으로 메일을 보내지 않는다 ([[_send_upload_complete_notification]] 주석 참고). """ return result.file_type.lower() in _POINT_CLOUD_FILE_TYPES def upload_file_types(filenames: list[str]) -> set[str]: """업로드 요청의 파일명을 **완료 검사와 같은 기준**으로 유형화한다. 확장자만 세면 노선 세트의 `.prj`가 지형 PRJ로 오인돼, 요청 검사는 통과하고 저장 뒤 완료 검사에서 누락으로 갈린다(2026-09-06 실측 — LAS 없이 설계가 이 어긋남으로 막혔다). 노선 도형(`.shp`)과 basename이 같은 PRJ는 화면의 카드 배정과 같은 규칙으로 `route_prj`. """ route_stem = next( (Path(name).stem for name in filenames if Path(name).suffix.lower() == ".shp"), None ) types: set[str] = set() for name in filenames: suffix = Path(name).suffix.lower().lstrip(".") if suffix == "prj" and route_stem is not None and Path(name).stem == route_stem: suffix = "route_prj" types.add(suffix) return types def _missing_required_file_types(file_types: set[str], las_free: bool = False) -> list[str]: # LAS 없는 설계(도엽등고선 기반, 2026-08-30)는 지형 한 벌(포인트클라우드·지형 PRJ· # TFW)을 통째로 받지 않는다 — 화면도 그 카드들을 필수에서 뺀다. missing = [] if las_free else sorted(_REQUIRED_FILE_TYPES - file_types) if not file_types.intersection(_ROUTE_FILE_TYPES): missing.append("csv/shp") # shapefile로 왔으면 형제 파일이 다 있어야 노선을 읽는다. if "shp" in file_types: missing.extend(sorted(_SHAPEFILE_REQUIRED_TYPES - file_types)) if not las_free and not file_types.intersection(_POINT_CLOUD_FILE_TYPES): missing.append("las/laz") return missing def _require_complete_file_set(file_types: set[str], las_free: bool = False) -> None: missing = _missing_required_file_types(file_types, las_free) if missing: raise ValueError(f"B03 필수 입력 파일이 없습니다: {', '.join(missing)}") def _stored_fingerprint(metadata: Any) -> str | None: """입력 파일 메타데이터에 적어 둔 지문을 꺼낸다.""" if isinstance(metadata, str): try: metadata = json.loads(metadata) except (TypeError, ValueError): return None if not isinstance(metadata, dict): return None value = metadata.get("fingerprint") return str(value) if value else None async def _already_uploaded( connection: aiomysql.Connection, project_id: UUID, payload: ChunkSessionCreateRequest, ) -> ChunkSessionCreateResponse | None: """같은 이름으로 **같은 내용**이 이미 올라와 있으면 전송을 건너뛰라는 응답을 만든다. 1.7GB를 다 받은 뒤에 비교하면 아낄 게 없으므로, 세션을 만들기 전에 화면이 보내 준 지문으로 가린다. 지문이 없거나 다르면 그냥 올린다 — 애매하면 올리는 쪽이 안전하다. """ if not payload.fingerprint: return None existing = await find_input_file_by_name(connection, project_id, payload.original_filename) if not existing or _stored_fingerprint(existing.get("metadata")) != payload.fingerprint: return None logger.info( "B03 같은 파일 재업로드 — 전송 생략: project_id=%s file=%s", project_id, payload.original_filename, ) return ChunkSessionCreateResponse( project_id=str(project_id), upload_session_id="", original_filename=payload.original_filename, file_size_bytes=payload.size_bytes, chunk_size_bytes=payload.chunk_size_bytes, total_chunks=0, already_uploaded=True, ) async def _complete_file_input_if_ready( connection: aiomysql.Connection, project_id: UUID, las_free: bool = False, ) -> int: file_types, point_cloud_input_id, route_csv_input_id = await get_project_input_readiness( connection, project_id ) _require_complete_file_set(file_types, las_free) if point_cloud_input_id is None: if not las_free: raise ValueError("LAS 또는 LAZ 입력 파일을 찾을 수 없습니다.") if route_csv_input_id is None: raise ValueError("계획 노선 입력 파일(CSV 또는 shapefile)을 찾을 수 없습니다.") # 자료가 갈렸으니 옛 계산 결과(파일 + DB)를 지우고 진행 표시도 되돌린다. 남겨 두면 # 아직 다시 만들어지지 않은 뒷단계 화면이 옛 결과를 새 자료 것인 양 보여준다. stored_path = await get_project_storage_relative_path(connection, project_id) project_root = Path(resolve_stored_project_path(stored_path)) # 지형 파일 여러 장은 합쳐서 전처리한다 — 다른 사업지 파일이 섞이면 합친 범위가 # 통째로 어긋나므로 여기서 막는다(2026-09-06 사용자 지시). 머리글만 읽어 즉시 끝난다. from B04_PreProcess.B04_PreProcess_Engine_Structurize import merge_gap_error from B04_PreProcess.B04_PreProcess_Repository import list_project_point_cloud_paths terrain_paths = await list_project_point_cloud_paths(connection, project_id, project_root) gap_message = merge_gap_error(terrain_paths) if gap_message: raise ValueError(gap_message) clear_designing(project_root) discard_initial_snapshot(project_root) await purge_project_outputs(connection, str(project_id), project_root) async with connection.cursor(aiomysql.DictCursor) as cursor: await reset_stages_after_input_change(cursor, str(project_id)) await complete_stage(cursor, str(project_id), 0) # LAS가 있으면 LAS, 없으면(las_free) 계획노선 CSV가 WF1 분석 입력이다. return point_cloud_input_id if point_cloud_input_id is not None else int(route_csv_input_id) def _write_stage_metadata( stage_root: Path, project_id: UUID, results: list[UploadedFileResult], ) -> None: metadata_path = stage_root / "metadata.json" existing_files: list[dict[str, Any]] = [] if metadata_path.exists(): try: payload = json.loads(metadata_path.read_text(encoding="utf-8")) existing_files = list(payload.get("files") or []) except (OSError, TypeError, ValueError): logger.warning("B03 metadata.json을 읽지 못해 새로 작성합니다: %s", metadata_path) merged = { str(item.get("relative_path") or item.get("original_filename")): item for item in existing_files } for result in results: dumped = result.model_dump() merged[result.relative_path] = dumped atomic_write_json( metadata_path, {"project_id": str(project_id), "files": list(merged.values())}, ) # 실행 중인 백그라운드 작업의 강한 참조. 이벤트 루프는 작업을 약한 참조로만 들고 있어, # 여기서 붙잡지 않으면 GC가 대기 중인 작업을 통째로 회수해 WF1 분석이 조용히 사라진다 # (업로드는 성공·stage 0은 COMPLETE인데 stage 1은 NOT_STARTED로 남는 증상). _BACKGROUND_TASKS: set[asyncio.Task] = set() def _schedule_background_task(coro: Any, *, task_name: str) -> None: task = asyncio.create_task(coro, name=task_name) _BACKGROUND_TASKS.add(task) def _log_task_failure(completed: asyncio.Task) -> None: _BACKGROUND_TASKS.discard(completed) try: completed.result() except asyncio.CancelledError: logger.warning("백그라운드 작업 취소됨: %s", task_name) except Exception: logger.exception("백그라운드 작업 실패: %s", task_name) task.add_done_callback(_log_task_failure) logger.info("백그라운드 작업 시작: %s", task_name) async def _get_project_notification_info( connection: aiomysql.Connection, project_id: UUID, ) -> dict[str, Any] | None: async with connection.cursor(aiomysql.DictCursor) as cursor: await cursor.execute( """ SELECT p.id, p.name AS project_name, u.email AS user_email, u.name AS user_name FROM projects p JOIN users u ON u.id = p.user_id WHERE p.id = %s AND p.deleted_at IS NULL AND u.deleted_at IS NULL """, (str(project_id),), ) row = await cursor.fetchone() return dict(row) if row else None async def _update_project_status(project_id: UUID, status: str) -> None: pool = get_db_pool() async with pool.acquire() as connection, connection.cursor() as cursor: await cursor.execute( """ UPDATE projects SET status = %s, updated_at = NOW() WHERE id = %s AND deleted_at IS NULL """, (status, str(project_id)), ) await connection.commit() async def _send_upload_complete_notification( *, project_id: UUID, uploaded_file: UploadedFileResult, ) -> None: """저장만 끝났을 때 보내는 안내. 프로젝트 업로드 경로에서는 호출하지 않는다 — 그 흐름은 초기 설계까지 마친 뒤 통합 메일 한 통으로 알린다. 프로젝트 생성 전 임시 보관함 업로드에서 쓸 예정이라 지워두지 않았다(2026-08-08 사용자 지시). """ pool = get_db_pool() async with pool.acquire() as connection: project_info = await _get_project_notification_info(connection, project_id) if not project_info or not project_info.get("user_email"): logger.warning("업로드 완료 이메일 수신자 없음: project_id=%s", project_id) return await send_file_upload_complete_email( to_email=str(project_info["user_email"]), user_name=str(project_info.get("user_name") or "사용자"), project_name=str(project_info.get("project_name") or project_id), file_name=uploaded_file.original_filename, file_size_mb=uploaded_file.size_bytes / (1024 * 1024), metadata=uploaded_file.metadata, )