낱개 출고 자동분석 업로드 예외 처리 보강

- 주문 날짜(custom_date) 형식 검증을 진입부로 이동 (400 응답)
- 엑셀 파싱/템플릿 컬럼 누락은 500이 아닌 400으로 분리
- 빈 행 자동 스킵으로 빈 Order 레코드 저장 방지
- safe_int_quantity 헬퍼 도입: 수량/세트 구성수량의 None·비숫자·
  '1.0' 같은 문자열도 견고하게 정수 변환
- itemcode_db 엔진 생성/조회 실패 시 단품 처리로 폴백하고
  치명적 오류 대신 warning 로그로 기록
- 세트 구성 단품이 single_items 에 없을 때 보강하여 결과 행 누락 방지
- 마스터 미등록 상품코드를 모아 warning 로그로 가시화
- 완료 시 단품/미등록 코드 수를 success 로그에 함께 기록

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

app/routers/orders.py

@@ -83,6 +83,29 @@ def normalize_template_column(value) -> str:
     return str(value).replace("\n", "").replace(" ", "").strip()
 
 
+def safe_int_quantity(value, default: int = 0) -> int:
+    """수량 값을 안전하게 정수로 변환. 비숫자/빈값/None 은 default 반환.
+    엑셀에서 1.0 같은 float, '1 '·'2개' 같은 공백 포함 문자열도 견고하게 처리."""
+    if value is None:
+        return default
+    if isinstance(value, bool):
+        return default
+    if isinstance(value, (int, float)):
+        try:
+            if pd.isna(value):
+                return default
+            return int(value)
+        except (ValueError, OverflowError, TypeError):
+            return default
+    text = str(value).strip()
+    if not text:
+        return default
+    try:
+        return int(float(text))
+    except (ValueError, TypeError):
+        return default
+
+
 def clean_excel_value(value) -> str | None:
     if value is None or pd.isna(value):
         return None
@@ -566,10 +589,18 @@ def upload_file(
     current_user = Depends(get_current_user),
     db: Session = Depends(get_db)
 ):
+    # 0. 주문 날짜(custom_date) 검증 — 잘못된 형식이면 400
+    try:
+        order_date_dt = datetime.strptime(custom_date, "%Y-%m-%d")
+    except (ValueError, TypeError):
+        raise HTTPException(status_code=400, detail="주문 날짜 형식이 올바르지 않습니다 (YYYY-MM-DD).")
+
     # 1. 파일 확장자 검증
+    if not files:
+        raise HTTPException(status_code=400, detail="업로드할 파일이 없습니다.")
     for file in files:
         filename = file.filename or ""
-        if not filename.endswith(".xls") and not filename.endswith(".xlsx"):
+        if not (filename.endswith(".xls") or filename.endswith(".xlsx")):
             raise HTTPException(status_code=400, detail="XLS 또는 XLSX 파일만 업로드할 수 있습니다.")
 
     filenames_str = ", ".join([f.filename for f in files])
@@ -580,17 +611,33 @@ def upload_file(
         {"filenames": filenames_str, "custom_date": custom_date}
     )
 
+    # 2. 파일 파싱 — 형식/템플릿 오류는 400 으로 분리
     db_records = []
     try:
-        # 2. 파일 파싱 및 데이터베이스 레코드 임시 적재
         for file in files:
             contents = file.file.read()
-            df = pd.read_excel(io.BytesIO(contents))
+            try:
+                df = pd.read_excel(io.BytesIO(contents))
+            except Exception as parse_err:
+                raise HTTPException(
+                    status_code=400,
+                    detail=f"'{file.filename}' 파일을 엑셀로 읽을 수 없습니다: {parse_err}"
+                )
             df = df.where(pd.notnull(df), None)
-            column_map = find_template_columns(list(df.columns))
+            try:
+                column_map = find_template_columns(list(df.columns))
+            except ValueError as ve:
+                raise HTTPException(
+                    status_code=400,
+                    detail=f"'{file.filename}' {ve}"
+                )
 
-            for index, row in df.iterrows():
+            for _index, row in df.iterrows():
                 rlist = list(row)
+                # 모든 셀이 비어있는 행은 스킵
+                if all(v is None or (isinstance(v, str) and not v.strip()) for v in rlist):
+                    continue
+
                 recipient_phone = row_value(rlist, column_map, "recipient_phone")
                 recipient_mobile = row_value(rlist, column_map, "recipient_mobile")
                 tracking_number = row_value(rlist, column_map, "tracking_number")
@@ -621,8 +668,14 @@ def upload_file(
                     upload_date=datetime.now().strftime("%Y-%m-%d %H:%M:%S")
                 )
                 db_records.append(order)
+    except HTTPException:
+        raise
+    except Exception as e:
+        log_event(current_user, "파일 파싱 실패", "error", {"filenames": filenames_str, "error": str(e)})
+        raise HTTPException(status_code=400, detail=f"엑셀 파싱 실패: {str(e)}")
 
-        # 3. 데이터베이스 벌크 저장 및 커밋
+    # 3. 데이터베이스 벌크 저장 — DB 오류는 500
+    try:
         batch_size = 1000
         for i in range(0, len(db_records), batch_size):
             chunk = db_records[i:i + batch_size]
@@ -634,113 +687,152 @@ def upload_file(
             "success",
             {"filenames": filenames_str, "uploaded_count": len(db_records)}
         )
-
     except Exception as e:
         db.rollback()
-        log_event(current_user, "파일 업로드 실패", "error", {"filenames": filenames_str, "error": str(e)})
+        log_event(current_user, "파일 업로드 DB 저장 실패", "error", {"filenames": filenames_str, "error": str(e)})
         raise HTTPException(status_code=500, detail=f"데이터베이스 저장 실패: {str(e)}")
 
     # 4. 품목 마스터(itemcode_db) 연동 및 낱개 수량 환산 합계 계산
     try:
-        item_engine = get_secondary_engine("ITEM_DB_NAME")
+        single_items: dict[str, dict] = {}
-        
+        set_components: dict[str, list[dict]] = {}
-        single_items = {}
+
-        set_components = {}
+        # itemcode_db 연결 실패는 치명적이지 않게 처리:
-        
+        # 마스터를 못 읽으면 모든 상품을 단품으로 간주해서 그대로 합산한다.
+        try:
+            item_engine = get_secondary_engine("ITEM_DB_NAME")
+        except Exception as conn_err:
+            item_engine = None
+            log_event(
+                current_user,
+                "itemcode_db 엔진 생성 실패",
+                "warning",
+                {"error": str(conn_err)},
+            )
+
         if item_engine:
-            with item_engine.connect() as conn:
+            try:
-                # 1) 단품 마스터 맵 구성
+                with item_engine.connect() as conn:
-                rs_single = conn.execute(text("SELECT item_code, sabangnet_code, name FROM single_items"))
+                    # 1) 단품 마스터 맵 구성
-                for row in rs_single:
+                    rs_single = conn.execute(text("SELECT item_code, sabangnet_code, name FROM single_items"))
-                    single_items[row[0]] = {
+                    for row in rs_single:
-                        "sabangnet_code": row[1] or "",
+                        if not row[0]:
-                        "name": row[2] or ""
+                            continue
-                    }
+                        single_items[row[0]] = {
-                # 2) 세트 구성품 맵 구성
+                            "sabangnet_code": row[1] or "",
-                rs_comp = conn.execute(text("SELECT set_code, single_code, quantity FROM set_components"))
+                            "name": row[2] or ""
-                for row in rs_comp:
+                        }
-                    s_code = row[0]
+                    # 2) 세트 구성품 맵 구성
-                    if s_code not in set_components:
+                    rs_comp = conn.execute(text("SELECT set_code, single_code, quantity FROM set_components"))
-                        set_components[s_code] = []
+                    for row in rs_comp:
-                    set_components[s_code].append({
+                        s_code = row[0]
-                        "single_code": row[1],
+                        single_code = row[1]
-                        "quantity": row[2]
+                        if not s_code or not single_code:
-                    })
+                            continue
-        
+                        set_components.setdefault(s_code, []).append({
-        # 3) 낱개 환산 처리
+                            "single_code": single_code,
-        single_qty_sums = {}
+                            # 마스터의 수량 컬럼도 NULL/문자열 방어
+                            "quantity": safe_int_quantity(row[2], default=1),
+                        })
+            except Exception as db_err:
+                # 마스터 조회 실패해도 분석을 중단하지 않고 단품 처리로 폴백
+                log_event(
+                    current_user,
+                    "itemcode_db 조회 실패 — 단품 처리로 폴백",
+                    "warning",
+                    {"error": str(db_err)},
+                )
+
+        # 4-1) 낱개 환산 처리
+        single_qty_sums: dict[str, int] = {}
+        unknown_codes: set[str] = set()
         for order in db_records:
             prod_code = (order.product_code or "").strip()
             if not prod_code:
                 continue
-            
+
-            qty = 1
+            # 수량은 비숫자/공백/None 모두 1로 폴백 (보존: 비어 있는 셀도 1건으로 본다)
-            try:
+            qty = safe_int_quantity(order.order_quantity, default=1)
-                qty = int(order.order_quantity or 1)
+
-            except:
-                pass
-                
             if prod_code in single_items:
+                # 단품: 그대로 합산
                 single_qty_sums[prod_code] = single_qty_sums.get(prod_code, 0) + qty
             elif prod_code in set_components:
+                # 세트: 구성 단품별로 풀어서 합산
                 for comp in set_components[prod_code]:
                     s_code = comp["single_code"]
-                    comp_qty = comp["quantity"]
+                    comp_qty = safe_int_quantity(comp["quantity"], default=1)
                     single_qty_sums[s_code] = single_qty_sums.get(s_code, 0) + (qty * comp_qty)
+                    # 세트 구성 단품 정보가 single_items 에 없을 수도 있으므로 보강
+                    if s_code not in single_items:
+                        single_items[s_code] = {"sabangnet_code": s_code, "name": ""}
             else:
-                # DB에 등록되어 있지 않은 코드는 단품으로 간주하고, 사방넷코드와 이름을 보존
+                # 마스터 미등록 코드: 단품으로 간주, 이름은 주문서의 상품명으로 채움
+                unknown_codes.add(prod_code)
                 single_items[prod_code] = {
                     "sabangnet_code": prod_code,
                     "name": order.product_name or ""
                 }
                 single_qty_sums[prod_code] = single_qty_sums.get(prod_code, 0) + qty
-                
+
+        if unknown_codes:
+            log_event(
+                current_user,
+                "낱개 출고 분석 — 마스터 미등록 코드 감지",
+                "warning",
+                {"unknown_count": len(unknown_codes), "sample": list(unknown_codes)[:20]},
+            )
+
         # 5. 결과 엑셀 데이터 빌드
+        # A열=사방넷코드, B열=합계수량, C열=이름. 헤더는 요청 사양(상품코드[필수]/가용수량/불용수량).
         excel_rows = []
         for s_code, total_qty in single_qty_sums.items():
             s_info = single_items.get(s_code, {"sabangnet_code": s_code, "name": ""})
             excel_rows.append({
                 "상품코드[필수]": s_info["sabangnet_code"] or s_code,
                 "가용수량": total_qty,
-                "불용수량": s_info["name"] or "" # C열 헤더는 "불용수량", 실데이터는 이름!
+                "불용수량": s_info["name"] or ""
             })
-            
+
-        df_out = pd.DataFrame(excel_rows)
+        df_out = pd.DataFrame(excel_rows, columns=["상품코드[필수]", "가용수량", "불용수량"])
-        if df_out.empty:
+
-            df_out = pd.DataFrame(columns=["상품코드[필수]", "가용수량", "불용수량"])
+        # 엑셀 파일 쓰기 (1행 헤더, 2행부터 데이터 — pandas to_excel 기본 동작)
-        else:
-            df_out = df_out[["상품코드[필수]", "가용수량", "불용수량"]]
-            
-        # 엑셀 파일 쓰기
         excel_io = io.BytesIO()
         with pd.ExcelWriter(excel_io, engine='openpyxl') as writer:
-            df_out.to_excel(writer, index=False)
+            df_out.to_excel(writer, index=False, sheet_name="발주")
         excel_io.seek(0)
-        
+
-        # 6. 다운로드 파일명 날짜 포맷 조합
+        # 6. 다운로드 파일명: "MM월 DD일 (요일)_발주_(낱개 상품 출고).xls"
-        try:
-            dt = datetime.strptime(custom_date, "%Y-%m-%d")
-        except:
-            dt = datetime.now()
-            
         weekdays = ["월요일", "화요일", "수요일", "목요일", "금요일", "토요일", "일요일"]
-        weekday_str = weekdays[dt.weekday()]
+        weekday_str = weekdays[order_date_dt.weekday()]
-        
+        filename = f"{order_date_dt.strftime('%m')}월 {order_date_dt.strftime('%d')}일 ({weekday_str})_발주_(낱개 상품 출고).xls"
-        # 형식 예: 05월 10일 (일요일)_발주_(낱개 상품 출고).xls
-        filename = f"{dt.strftime('%m')}월 {dt.strftime('%d')}일 ({weekday_str})_발주_(낱개 상품 출고).xls"
         encoded_filename = quote(filename)
-        
+
         headers = {
             "Content-Disposition": f"attachment; filename*=UTF-8''{encoded_filename}",
             "Access-Control-Expose-Headers": "Content-Disposition"
         }
-        
+
+        log_event(
+            current_user,
+            "낱개 출고 자동분석 완료",
+            "success",
+            {
+                "filenames": filenames_str,
+                "uploaded_count": len(db_records),
+                "single_code_count": len(single_qty_sums),
+                "unknown_code_count": len(unknown_codes),
+            },
+        )
+
         return StreamingResponse(
             excel_io,
             media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
             headers=headers
         )
-        
+
+    except HTTPException:
+        raise
     except Exception as e:
         log_event(current_user, "낱개 출고 분석 오류", "error", {"error": str(e)})
         raise HTTPException(status_code=500, detail=f"낱개 출고 분석 및 엑셀 생성 실패: {str(e)}")