diff --git a/B03_FileInput/B03_FileInput_Router.py b/B03_FileInput/B03_FileInput_Router.py index 222b0666..2867abd5 100644 --- a/B03_FileInput/B03_FileInput_Router.py +++ b/B03_FileInput/B03_FileInput_Router.py @@ -5,302 +5,59 @@ import json import logging from pathlib import Path from typing import Any -from uuid import UUID, uuid4 +from uuid import UUID import aiomysql from fastapi import APIRouter, Depends, File, Form, UploadFile from fastapi.responses import JSONResponse -from B03_FileInput.B03_FileInput_Email import ( - send_file_upload_complete_email, -) from B03_FileInput.B03_FileInput_Engine import ( - merge_upload_chunks, - remove_chunk_session, - resolve_chunk_session_dir, resolve_upload_destination, - save_upload_chunk, save_upload_stream, ) from B03_FileInput.B03_FileInput_Engine_Analyze import analyze_input_metadata from B03_FileInput.B03_FileInput_Repository import ( create_input_file, - create_upload_session, - find_input_file_by_name, get_project_input_readiness, get_project_storage_relative_path, - get_upload_session, - list_completed_chunk_indexes, list_incomplete_upload_sessions, list_project_input_files, - mark_upload_session_completed, - mark_upload_session_failed, - supersede_previous_input_files, - upsert_upload_chunk, ) +from B03_FileInput.B03_FileInput_Router_Chunks import router as chunk_router from B03_FileInput.B03_FileInput_Router_Errors import lookup_error_response +from B03_FileInput.B03_FileInput_Router_Helpers import ( + _REQUIRED_FILE_TYPES, + _complete_file_input_if_ready, + _missing_required_file_types, + _schedule_background_task, + _write_stage_metadata, +) from B03_FileInput.B03_FileInput_Schema import ( - ChunkSessionCreateRequest, - ChunkSessionCreateResponse, - ChunkUploadResponse, FileUploadDescriptor, FileUploadResponse, UploadedFileResult, - UploadFinalizeRequest, UploadOverviewFile, UploadOverviewResponse, UploadOverviewSession, - UploadStatusResponse, ) from B03_FileInput.B03_FileInput_Service_WF1 import trigger_wf1_analysis_and_email from common_util.common_util_auth import verify_session -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 import load_project_workflow from common_util.common_util_workflow_state import ( - complete_stage, get_workflow_state, - is_analysis_running, - reset_stages_after_input_change, ) from config.config_db import get_db_pool from config.config_system import ( - UPLOAD_CHUNK_SIZE_BYTES, UPLOAD_MAX_FILES, ) -_ANALYSIS_RUNNING_MESSAGE = "이 프로젝트는 지금 분석 중입니다. 끝난 뒤에 새 자료를 올려 주세요." - logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/projects", tags=["B03 File Input"]) -_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 _missing_required_file_types(file_types: set[str], las_free: bool = False) -> list[str]: - missing = 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)) - # LAS 없는 설계(도엽등고선 기반, 2026-08-30)는 LAS 필수를 면제한다. - 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)) - 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, - ) +# 청크 업로드 엔드포인트는 파일이 700줄을 넘어 떼어냈다(2026-09-04) — 경로는 그대로다. +router.include_router(chunk_router) @router.post("/{project_id}/files", response_model=FileUploadResponse) @@ -458,329 +215,6 @@ async def upload_project_files( await upload.close() -@router.post("/{project_id}/upload-sessions", response_model=ChunkSessionCreateResponse) -async def create_project_upload_session( - project_id: UUID, - payload: ChunkSessionCreateRequest, - session: dict[str, Any] = Depends(verify_session), -) -> ChunkSessionCreateResponse | JSONResponse: - """대용량 파일 청크 업로드 세션을 생성한다.""" - # LAS 없는 설계를 켠 상태면 포인트클라우드는 받지 않는다 — 큰 LAS는 이 경로로 - # 들어오므로 여기서 막지 않으면 `/files` 검사를 통째로 비켜 간다. - if payload.las_free and Path(payload.original_filename).suffix.lower() in {".las", ".laz"}: - return JSONResponse( - status_code=400, - content={ - "status": "error", - "message": "LAS 없이 설계를 켠 상태에서는 LAS/LAZ 파일을 올릴 수 없습니다.", - }, - ) - chunk_size_bytes = min(payload.chunk_size_bytes, UPLOAD_CHUNK_SIZE_BYTES) - total_chunks = _total_chunks(payload.size_bytes, chunk_size_bytes) - session_id = str(uuid4()) - point_cloud_input_id: int | None = None - skipped: ChunkSessionCreateResponse | None = None - - pool = get_db_pool() - try: - async with pool.acquire() as connection: - await get_project_storage_relative_path(connection, project_id) - # 분석이 도는 중이면 새 자료를 받지 않는다 — 받아 봐야 분석 2개가 같은 산출물 - # 경로에서 부딪힌다. 화면도 버튼을 잠그지만 새로고침으로 우회할 수 있어 여기서 막는다. - async with connection.cursor(aiomysql.DictCursor) as cursor: - if await is_analysis_running(cursor, str(project_id)): - return JSONResponse( - status_code=409, - content={"status": "error", "message": _ANALYSIS_RUNNING_MESSAGE}, - ) - skipped = await _already_uploaded(connection, project_id, payload) - if skipped is not None: - if payload.complete_upload: - await connection.begin() - try: - point_cloud_input_id = await _complete_file_input_if_ready( - connection, - project_id, - payload.las_free, - ) - await connection.commit() - except Exception: - await connection.rollback() - raise - else: - await create_upload_session( - connection, - session_id=session_id, - project_id=project_id, - original_filename=payload.original_filename, - file_size_bytes=payload.size_bytes, - chunk_size_bytes=chunk_size_bytes, - total_chunks=total_chunks, - ) - if skipped is not None: - if point_cloud_input_id is not None: - _schedule_background_task( - trigger_wf1_analysis_and_email( - project_id=project_id, - input_file_id=point_cloud_input_id, - user_role=str(session["role"]), - ), - task_name=f"b04-preprocess-auto-{project_id}", - ) - return skipped - return ChunkSessionCreateResponse( - project_id=str(project_id), - upload_session_id=session_id, - original_filename=payload.original_filename, - file_size_bytes=payload.size_bytes, - chunk_size_bytes=chunk_size_bytes, - total_chunks=total_chunks, - ) - except LookupError as exc: - return lookup_error_response(exc, logger, context="B03 업로드", project_id=project_id) - except (OSError, ValueError) as exc: - return JSONResponse(status_code=400, content={"status": "error", "message": str(exc)}) - except Exception: - logger.exception("B03 청크 세션 생성 실패: project_id=%s", project_id) - return JSONResponse( - status_code=500, - content={"status": "error", "message": "업로드 세션 생성 중 오류가 발생했습니다."}, - ) - - -@router.post("/{project_id}/chunks", response_model=ChunkUploadResponse) -async def upload_project_chunk( - project_id: UUID, - session_id: str = Form(...), - chunk_index: int = Form(...), - chunk_data: UploadFile = File(...), -) -> ChunkUploadResponse | JSONResponse: - """단일 파일 청크를 B03 임시 폴더에 저장하고 DB에 기록한다.""" - pool = get_db_pool() - try: - async with pool.acquire() as connection: - upload_session = await get_upload_session( - connection, - project_id=project_id, - session_id=session_id, - ) - if chunk_index < 0 or chunk_index >= int(upload_session["total_chunks"]): - return JSONResponse( - status_code=400, - content={"status": "error", "message": "청크 인덱스가 범위를 벗어났습니다."}, - ) - stored_path = await get_project_storage_relative_path(connection, project_id) - project_root = Path(resolve_stored_project_path(stored_path)) - session_dir = resolve_chunk_session_dir(project_root, session_id) - chunk_path, size_bytes, chunk_hash = await save_upload_chunk( - chunk_data, - session_dir, - chunk_index, - expected_max_bytes=int(upload_session["chunk_size_bytes"]), - ) - relative_chunk_path = chunk_path.relative_to(project_root).as_posix() - completed_chunks = await upsert_upload_chunk( - connection, - session_id=session_id, - chunk_index=chunk_index, - chunk_hash=chunk_hash, - size_bytes=size_bytes, - stored_at=relative_chunk_path, - ) - return ChunkUploadResponse( - upload_session_id=session_id, - chunk_index=chunk_index, - completed_chunks=completed_chunks, - total_chunks=int(upload_session["total_chunks"]), - chunk_hash=chunk_hash, - ) - except LookupError as exc: - return lookup_error_response(exc, logger, context="B03 업로드", project_id=project_id) - except (OSError, ValueError) as exc: - return JSONResponse(status_code=400, content={"status": "error", "message": str(exc)}) - except Exception: - logger.exception( - "B03 청크 업로드 실패: project_id=%s session_id=%s", - project_id, - session_id, - ) - return JSONResponse( - status_code=500, - content={"status": "error", "message": "청크 업로드 중 오류가 발생했습니다."}, - ) - finally: - await chunk_data.close() - - -@router.post("/{project_id}/finalize", response_model=FileUploadResponse) -async def finalize_project_upload( - project_id: UUID, - payload: UploadFinalizeRequest, - session: dict[str, Any] = Depends(verify_session), -) -> FileUploadResponse | JSONResponse: - """청크 업로드를 최종 병합하고 input_files 메타데이터를 기록한다.""" - pool = get_db_pool() - final_path: Path | None = None - point_cloud_input_id: int | None = None - try: - async with pool.acquire() as connection: - upload_session = await get_upload_session( - connection, - project_id=project_id, - session_id=payload.session_id, - ) - if int(upload_session["total_chunks"]) != payload.total_chunks: - return JSONResponse( - status_code=400, - content={"status": "error", "message": "세션 청크 개수가 일치하지 않습니다."}, - ) - completed_indexes = await list_completed_chunk_indexes( - connection, - session_id=payload.session_id, - ) - expected_indexes = list(range(payload.total_chunks)) - if completed_indexes != expected_indexes: - return JSONResponse( - status_code=400, - content={"status": "error", "message": "아직 업로드되지 않은 청크가 있습니다."}, - ) - - stored_path = await get_project_storage_relative_path(connection, project_id) - project_root = Path(resolve_stored_project_path(stored_path)) - descriptor = FileUploadDescriptor( - original_filename=str(upload_session["original_filename"]), - size_bytes=int(upload_session["file_size_bytes"]), - ) - final_path = merge_upload_chunks( - project_root, - descriptor, - payload.session_id, - payload.total_chunks, - ) - metadata = await asyncio.to_thread(analyze_input_metadata, final_path) - # 다음에 같은 파일이 올라오면 전송을 건너뛸 수 있도록 지문을 함께 남긴다. - fingerprint = payload.fingerprint or None - if fingerprint: - metadata = {**metadata, "fingerprint": fingerprint} - relative_path = final_path.relative_to(project_root).as_posix() - file_type = final_path.suffix.lower().lstrip(".") - crs_epsg = metadata.get("epsg") - - await connection.begin() - try: - input_file_id = await create_input_file( - connection, - project_id=project_id, - file_type=file_type, - original_filename=descriptor.original_filename, - relative_path=relative_path, - file_size_bytes=descriptor.size_bytes, - upload_by=None, - crs_epsg=int(crs_epsg) if crs_epsg is not None else None, - metadata=metadata, - ) - # 같은 이름의 옛 행은 내려 둔다 — 목록·분석이 최신 1건만 보게 한다. - await supersede_previous_input_files( - connection, - project_id, - descriptor.original_filename, - input_file_id, - ) - await mark_upload_session_completed(connection, session_id=payload.session_id) - if payload.complete_upload: - point_cloud_input_id = await _complete_file_input_if_ready( - connection, - project_id, - payload.las_free, - ) - await connection.commit() - except Exception: - await connection.rollback() - await mark_upload_session_failed(connection, session_id=payload.session_id) - raise - - remove_chunk_session(project_root, payload.session_id) - result = UploadedFileResult( - input_file_id=input_file_id, - original_filename=descriptor.original_filename, - file_type=file_type, - relative_path=relative_path, - size_bytes=descriptor.size_bytes, - metadata=metadata, - ) - stage_root = project_root / "B03_FileInput" - _write_stage_metadata(stage_root, project_id, [result]) - # 업로드 직후 안내 메일은 보내지 않는다 — 위 일반 업로드 경로와 같은 이유. - if point_cloud_input_id is not None: - _schedule_background_task( - trigger_wf1_analysis_and_email( - project_id=project_id, - input_file_id=point_cloud_input_id, - user_role=str(session["role"]), - ), - task_name=f"b04-preprocess-auto-{project_id}", - ) - return FileUploadResponse(project_id=str(project_id), files=[result]) - except LookupError as exc: - return lookup_error_response(exc, logger, context="B03 업로드", project_id=project_id) - except (OSError, ValueError) as exc: - if final_path is not None: - final_path.unlink(missing_ok=True) - return JSONResponse(status_code=400, content={"status": "error", "message": str(exc)}) - except Exception: - if final_path is not None: - final_path.unlink(missing_ok=True) - logger.exception("B03 청크 업로드 최종 병합 실패: project_id=%s", project_id) - return JSONResponse( - status_code=500, - content={"status": "error", "message": "업로드 최종 처리 중 오류가 발생했습니다."}, - ) - - -@router.get("/{project_id}/upload-status/{session_id}", response_model=UploadStatusResponse) -async def get_project_upload_status( - project_id: UUID, - session_id: str, -) -> UploadStatusResponse | JSONResponse: - """업로드 세션의 청크 완료 상태를 조회한다.""" - pool = get_db_pool() - try: - async with pool.acquire() as connection: - upload_session = await get_upload_session( - connection, - project_id=project_id, - session_id=session_id, - ) - completed_indexes = await list_completed_chunk_indexes( - connection, - session_id=session_id, - ) - return UploadStatusResponse( - upload_session_id=session_id, - upload_status=str(upload_session["status"]), - original_filename=str(upload_session["original_filename"]), - file_size_bytes=int(upload_session["file_size_bytes"]), - chunk_size_bytes=int(upload_session["chunk_size_bytes"]), - total_chunks=int(upload_session["total_chunks"]), - completed_chunks=len(completed_indexes), - completed_chunk_indexes=completed_indexes, - ) - except LookupError as exc: - return lookup_error_response(exc, logger, context="B03 업로드", project_id=project_id) - except (OSError, ValueError) as exc: - return JSONResponse(status_code=400, content={"status": "error", "message": str(exc)}) - except Exception: - logger.exception( - "B03 업로드 상태 조회 실패: project_id=%s session_id=%s", - project_id, - session_id, - ) - return JSONResponse( - status_code=500, - content={"status": "error", "message": "업로드 상태 조회 중 오류가 발생했습니다."}, - ) - - def _parse_metadata(raw: Any) -> dict[str, Any] | None: """DB에 JSON 문자열로 저장된 분석 메타데이터를 dict로 돌린다(깨지면 생략).""" if isinstance(raw, dict): diff --git a/B03_FileInput/B03_FileInput_Router_Chunks.py b/B03_FileInput/B03_FileInput_Router_Chunks.py new file mode 100644 index 00000000..0ded7774 --- /dev/null +++ b/B03_FileInput/B03_FileInput_Router_Chunks.py @@ -0,0 +1,389 @@ +"""B03 청크 업로드 엔드포인트 — 세션 생성·조각 전송·마무리·진행 조회. + +라우터 본체(`B03_FileInput_Router.py`)가 700줄을 넘어 떼어낸 조각이다(2026-09-04). +경로·태그는 본체와 같다 — 본체가 `include_router` 로 붙여 URL 이 그대로 유지된다. +""" + +import asyncio +import logging +from pathlib import Path +from typing import Any +from uuid import UUID, uuid4 + +import aiomysql +from fastapi import APIRouter, Depends, File, Form, UploadFile +from fastapi.responses import JSONResponse + +from B03_FileInput.B03_FileInput_Engine import ( + merge_upload_chunks, + remove_chunk_session, + resolve_chunk_session_dir, + save_upload_chunk, +) +from B03_FileInput.B03_FileInput_Engine_Analyze import analyze_input_metadata +from B03_FileInput.B03_FileInput_Repository import ( + create_input_file, + create_upload_session, + get_project_storage_relative_path, + get_upload_session, + list_completed_chunk_indexes, + mark_upload_session_completed, + mark_upload_session_failed, + supersede_previous_input_files, + upsert_upload_chunk, +) +from B03_FileInput.B03_FileInput_Router_Errors import lookup_error_response +from B03_FileInput.B03_FileInput_Router_Helpers import ( + _ANALYSIS_RUNNING_MESSAGE, + _already_uploaded, + _complete_file_input_if_ready, + _schedule_background_task, + _total_chunks, + _write_stage_metadata, +) +from B03_FileInput.B03_FileInput_Schema import ( + ChunkSessionCreateRequest, + ChunkSessionCreateResponse, + ChunkUploadResponse, + FileUploadDescriptor, + FileUploadResponse, + UploadedFileResult, + UploadFinalizeRequest, + UploadStatusResponse, +) +from B03_FileInput.B03_FileInput_Service_WF1 import trigger_wf1_analysis_and_email +from common_util.common_util_auth import verify_session +from common_util.common_util_storage import resolve_stored_project_path +from common_util.common_util_workflow_state import ( + is_analysis_running, +) +from config.config_db import get_db_pool +from config.config_system import ( + UPLOAD_CHUNK_SIZE_BYTES, +) + +logger = logging.getLogger(__name__) +router = APIRouter(tags=["B03 File Input"]) + + +@router.post("/{project_id}/upload-sessions", response_model=ChunkSessionCreateResponse) +async def create_project_upload_session( + project_id: UUID, + payload: ChunkSessionCreateRequest, + session: dict[str, Any] = Depends(verify_session), +) -> ChunkSessionCreateResponse | JSONResponse: + """대용량 파일 청크 업로드 세션을 생성한다.""" + # LAS 없는 설계를 켠 상태면 포인트클라우드는 받지 않는다 — 큰 LAS는 이 경로로 + # 들어오므로 여기서 막지 않으면 `/files` 검사를 통째로 비켜 간다. + if payload.las_free and Path(payload.original_filename).suffix.lower() in {".las", ".laz"}: + return JSONResponse( + status_code=400, + content={ + "status": "error", + "message": "LAS 없이 설계를 켠 상태에서는 LAS/LAZ 파일을 올릴 수 없습니다.", + }, + ) + chunk_size_bytes = min(payload.chunk_size_bytes, UPLOAD_CHUNK_SIZE_BYTES) + total_chunks = _total_chunks(payload.size_bytes, chunk_size_bytes) + session_id = str(uuid4()) + point_cloud_input_id: int | None = None + skipped: ChunkSessionCreateResponse | None = None + + pool = get_db_pool() + try: + async with pool.acquire() as connection: + await get_project_storage_relative_path(connection, project_id) + # 분석이 도는 중이면 새 자료를 받지 않는다 — 받아 봐야 분석 2개가 같은 산출물 + # 경로에서 부딪힌다. 화면도 버튼을 잠그지만 새로고침으로 우회할 수 있어 여기서 막는다. + async with connection.cursor(aiomysql.DictCursor) as cursor: + if await is_analysis_running(cursor, str(project_id)): + return JSONResponse( + status_code=409, + content={"status": "error", "message": _ANALYSIS_RUNNING_MESSAGE}, + ) + skipped = await _already_uploaded(connection, project_id, payload) + if skipped is not None: + if payload.complete_upload: + await connection.begin() + try: + point_cloud_input_id = await _complete_file_input_if_ready( + connection, + project_id, + payload.las_free, + ) + await connection.commit() + except Exception: + await connection.rollback() + raise + else: + await create_upload_session( + connection, + session_id=session_id, + project_id=project_id, + original_filename=payload.original_filename, + file_size_bytes=payload.size_bytes, + chunk_size_bytes=chunk_size_bytes, + total_chunks=total_chunks, + ) + if skipped is not None: + if point_cloud_input_id is not None: + _schedule_background_task( + trigger_wf1_analysis_and_email( + project_id=project_id, + input_file_id=point_cloud_input_id, + user_role=str(session["role"]), + ), + task_name=f"b04-preprocess-auto-{project_id}", + ) + return skipped + return ChunkSessionCreateResponse( + project_id=str(project_id), + upload_session_id=session_id, + original_filename=payload.original_filename, + file_size_bytes=payload.size_bytes, + chunk_size_bytes=chunk_size_bytes, + total_chunks=total_chunks, + ) + except LookupError as exc: + return lookup_error_response(exc, logger, context="B03 업로드", project_id=project_id) + except (OSError, ValueError) as exc: + return JSONResponse(status_code=400, content={"status": "error", "message": str(exc)}) + except Exception: + logger.exception("B03 청크 세션 생성 실패: project_id=%s", project_id) + return JSONResponse( + status_code=500, + content={"status": "error", "message": "업로드 세션 생성 중 오류가 발생했습니다."}, + ) + + +@router.post("/{project_id}/chunks", response_model=ChunkUploadResponse) +async def upload_project_chunk( + project_id: UUID, + session_id: str = Form(...), + chunk_index: int = Form(...), + chunk_data: UploadFile = File(...), +) -> ChunkUploadResponse | JSONResponse: + """단일 파일 청크를 B03 임시 폴더에 저장하고 DB에 기록한다.""" + pool = get_db_pool() + try: + async with pool.acquire() as connection: + upload_session = await get_upload_session( + connection, + project_id=project_id, + session_id=session_id, + ) + if chunk_index < 0 or chunk_index >= int(upload_session["total_chunks"]): + return JSONResponse( + status_code=400, + content={"status": "error", "message": "청크 인덱스가 범위를 벗어났습니다."}, + ) + stored_path = await get_project_storage_relative_path(connection, project_id) + project_root = Path(resolve_stored_project_path(stored_path)) + session_dir = resolve_chunk_session_dir(project_root, session_id) + chunk_path, size_bytes, chunk_hash = await save_upload_chunk( + chunk_data, + session_dir, + chunk_index, + expected_max_bytes=int(upload_session["chunk_size_bytes"]), + ) + relative_chunk_path = chunk_path.relative_to(project_root).as_posix() + completed_chunks = await upsert_upload_chunk( + connection, + session_id=session_id, + chunk_index=chunk_index, + chunk_hash=chunk_hash, + size_bytes=size_bytes, + stored_at=relative_chunk_path, + ) + return ChunkUploadResponse( + upload_session_id=session_id, + chunk_index=chunk_index, + completed_chunks=completed_chunks, + total_chunks=int(upload_session["total_chunks"]), + chunk_hash=chunk_hash, + ) + except LookupError as exc: + return lookup_error_response(exc, logger, context="B03 업로드", project_id=project_id) + except (OSError, ValueError) as exc: + return JSONResponse(status_code=400, content={"status": "error", "message": str(exc)}) + except Exception: + logger.exception( + "B03 청크 업로드 실패: project_id=%s session_id=%s", + project_id, + session_id, + ) + return JSONResponse( + status_code=500, + content={"status": "error", "message": "청크 업로드 중 오류가 발생했습니다."}, + ) + finally: + await chunk_data.close() + + +@router.post("/{project_id}/finalize", response_model=FileUploadResponse) +async def finalize_project_upload( + project_id: UUID, + payload: UploadFinalizeRequest, + session: dict[str, Any] = Depends(verify_session), +) -> FileUploadResponse | JSONResponse: + """청크 업로드를 최종 병합하고 input_files 메타데이터를 기록한다.""" + pool = get_db_pool() + final_path: Path | None = None + point_cloud_input_id: int | None = None + try: + async with pool.acquire() as connection: + upload_session = await get_upload_session( + connection, + project_id=project_id, + session_id=payload.session_id, + ) + if int(upload_session["total_chunks"]) != payload.total_chunks: + return JSONResponse( + status_code=400, + content={"status": "error", "message": "세션 청크 개수가 일치하지 않습니다."}, + ) + completed_indexes = await list_completed_chunk_indexes( + connection, + session_id=payload.session_id, + ) + expected_indexes = list(range(payload.total_chunks)) + if completed_indexes != expected_indexes: + return JSONResponse( + status_code=400, + content={"status": "error", "message": "아직 업로드되지 않은 청크가 있습니다."}, + ) + + stored_path = await get_project_storage_relative_path(connection, project_id) + project_root = Path(resolve_stored_project_path(stored_path)) + descriptor = FileUploadDescriptor( + original_filename=str(upload_session["original_filename"]), + size_bytes=int(upload_session["file_size_bytes"]), + ) + final_path = merge_upload_chunks( + project_root, + descriptor, + payload.session_id, + payload.total_chunks, + ) + metadata = await asyncio.to_thread(analyze_input_metadata, final_path) + # 다음에 같은 파일이 올라오면 전송을 건너뛸 수 있도록 지문을 함께 남긴다. + fingerprint = payload.fingerprint or None + if fingerprint: + metadata = {**metadata, "fingerprint": fingerprint} + relative_path = final_path.relative_to(project_root).as_posix() + file_type = final_path.suffix.lower().lstrip(".") + crs_epsg = metadata.get("epsg") + + await connection.begin() + try: + input_file_id = await create_input_file( + connection, + project_id=project_id, + file_type=file_type, + original_filename=descriptor.original_filename, + relative_path=relative_path, + file_size_bytes=descriptor.size_bytes, + upload_by=None, + crs_epsg=int(crs_epsg) if crs_epsg is not None else None, + metadata=metadata, + ) + # 같은 이름의 옛 행은 내려 둔다 — 목록·분석이 최신 1건만 보게 한다. + await supersede_previous_input_files( + connection, + project_id, + descriptor.original_filename, + input_file_id, + ) + await mark_upload_session_completed(connection, session_id=payload.session_id) + if payload.complete_upload: + point_cloud_input_id = await _complete_file_input_if_ready( + connection, + project_id, + payload.las_free, + ) + await connection.commit() + except Exception: + await connection.rollback() + await mark_upload_session_failed(connection, session_id=payload.session_id) + raise + + remove_chunk_session(project_root, payload.session_id) + result = UploadedFileResult( + input_file_id=input_file_id, + original_filename=descriptor.original_filename, + file_type=file_type, + relative_path=relative_path, + size_bytes=descriptor.size_bytes, + metadata=metadata, + ) + stage_root = project_root / "B03_FileInput" + _write_stage_metadata(stage_root, project_id, [result]) + # 업로드 직후 안내 메일은 보내지 않는다 — 위 일반 업로드 경로와 같은 이유. + if point_cloud_input_id is not None: + _schedule_background_task( + trigger_wf1_analysis_and_email( + project_id=project_id, + input_file_id=point_cloud_input_id, + user_role=str(session["role"]), + ), + task_name=f"b04-preprocess-auto-{project_id}", + ) + return FileUploadResponse(project_id=str(project_id), files=[result]) + except LookupError as exc: + return lookup_error_response(exc, logger, context="B03 업로드", project_id=project_id) + except (OSError, ValueError) as exc: + if final_path is not None: + final_path.unlink(missing_ok=True) + return JSONResponse(status_code=400, content={"status": "error", "message": str(exc)}) + except Exception: + if final_path is not None: + final_path.unlink(missing_ok=True) + logger.exception("B03 청크 업로드 최종 병합 실패: project_id=%s", project_id) + return JSONResponse( + status_code=500, + content={"status": "error", "message": "업로드 최종 처리 중 오류가 발생했습니다."}, + ) + + +@router.get("/{project_id}/upload-status/{session_id}", response_model=UploadStatusResponse) +async def get_project_upload_status( + project_id: UUID, + session_id: str, +) -> UploadStatusResponse | JSONResponse: + """업로드 세션의 청크 완료 상태를 조회한다.""" + pool = get_db_pool() + try: + async with pool.acquire() as connection: + upload_session = await get_upload_session( + connection, + project_id=project_id, + session_id=session_id, + ) + completed_indexes = await list_completed_chunk_indexes( + connection, + session_id=session_id, + ) + return UploadStatusResponse( + upload_session_id=session_id, + upload_status=str(upload_session["status"]), + original_filename=str(upload_session["original_filename"]), + file_size_bytes=int(upload_session["file_size_bytes"]), + chunk_size_bytes=int(upload_session["chunk_size_bytes"]), + total_chunks=int(upload_session["total_chunks"]), + completed_chunks=len(completed_indexes), + completed_chunk_indexes=completed_indexes, + ) + except LookupError as exc: + return lookup_error_response(exc, logger, context="B03 업로드", project_id=project_id) + except (OSError, ValueError) as exc: + return JSONResponse(status_code=400, content={"status": "error", "message": str(exc)}) + except Exception: + logger.exception( + "B03 업로드 상태 조회 실패: project_id=%s session_id=%s", + project_id, + session_id, + ) + return JSONResponse( + status_code=500, + content={"status": "error", "message": "업로드 상태 조회 중 오류가 발생했습니다."}, + ) diff --git a/B03_FileInput/B03_FileInput_Router_Helpers.py b/B03_FileInput/B03_FileInput_Router_Helpers.py new file mode 100644 index 00000000..7b895adb --- /dev/null +++ b/B03_FileInput/B03_FileInput_Router_Helpers.py @@ -0,0 +1,267 @@ +"""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 _missing_required_file_types(file_types: set[str], las_free: bool = False) -> list[str]: + missing = 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)) + # LAS 없는 설계(도엽등고선 기반, 2026-08-30)는 LAS 필수를 면제한다. + 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)) + 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, + )