refactor(B03): 파일 입력 라우터 700줄 초과 분리 — 보조·청크 업로드
872줄 한 파일을 셋으로 나눔 (동작 불변, 순수 분리). - `B03_FileInput_Router.py` 309줄 — 파일 업로드·현황·워크플로 엔드포인트 - `B03_FileInput_Router_Chunks.py` 390줄 — 세션 생성·조각 전송·마무리·진행 조회 (자체 APIRouter 를 본체가 `include_router` 로 붙여 경로 문자열 불변) - `B03_FileInput_Router_Helpers.py` 268줄 — 필수 파일 판정·중복 지문 판별·단계 기록·알림 검증: 라우트 7개 경로·메서드 동일(`GET upload-overview|upload-status|workflow-state`, `POST files|upload-sessions|chunks|finalize`), 공용 브라우저에서 노선 5종 실제 업로드 성공 (shp metadata preview_path 포함), ruff check 통과, tmp/tests 378 passed Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5
@@ -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):
|
||||
|
||||
@@ -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": "업로드 상태 조회 중 오류가 발생했습니다."},
|
||||
)
|
||||
@@ -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,
|
||||
)
|
||||
Reference in New Issue
Block a user