import io import json import math import os import threading from types import SimpleNamespace from typing import Optional from datetime import date, datetime import asyncio import httpx import pandas as pd from fastapi import APIRouter, Depends, HTTPException, File, Form, UploadFile from fastapi.responses import Response, StreamingResponse from pydantic import BaseModel from sqlalchemy.orm import Session from sqlalchemy import String, and_, cast, func, or_, text from urllib.parse import quote from app.database import SessionLocal, get_db, get_secondary_engine from app.models import Order from app.auth import get_current_user from app.event_logger import log_event from app.search_utils import ( can_use_trigram_fts, fts_phrase, normalize_phone, normalize_tracking, prefix_upper_bound, ) router = APIRouter() CS_RETURN_REQUEST_URL = os.getenv( "CS_RETURN_REQUEST_URL", "http://dbx-corm:8002/api/return/request", ) # 서비스 간 호출 인증용 토큰 (양쪽 .env 에 같은 값으로 박는다). # CORM 의 AuthGuard 가 이 헤더를 보고 세션 검사 우회. INTERNAL_SERVICE_TOKEN = os.getenv("INTERNAL_SERVICE_TOKEN", "") # Global dict to hold upload progress. In a real world scenario this should be in Redis or DB. upload_progress_state = {} TEMPLATE_COLUMN_ALIASES = { "order_date": ["주문날짜"], "sequence_num": ["번호"], "order_no": ["주문번호"], "recipient_name": ["수령인명"], "product_code": ["상품코드"], "product_name": ["상품명"], "order_quantity": ["수량"], "address": ["주소"], "postal_code": ["우편번호"], "recipient_phone": ["수령인전화번호", "수령인전화"], "recipient_mobile": ["수령인휴대폰"], "delivery_memo": ["배송시요구사항", "배송시요구"], "tracking_number": ["송장번호"], "vendor": ["쇼핑몰명"], "order_note": ["비고"], "order_no_mall": ["주문번호(쇼핑몰)"], "order_list_1": ["주문목록"], } TEMPLATE_COLUMN_FALLBACK_INDEXES = { "order_no_mall": 15, # Excel P column "order_list_1": 16, # Excel Q column } REQUIRED_TEMPLATE_FIELDS = [ "order_date", "order_no", "recipient_name", "product_code", "product_name", "order_quantity", "address", "recipient_mobile", "tracking_number", "vendor", ] 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 if isinstance(value, datetime): return value.strftime("%Y-%m-%d %H:%M:%S") if isinstance(value, date): return datetime.combine(value, datetime.min.time()).strftime("%Y-%m-%d %H:%M:%S") text = str(value).strip() return text or None def find_template_columns(raw_columns) -> dict[str, int]: normalized_columns = [normalize_template_column(column) for column in raw_columns] column_map = {} for field, aliases in TEMPLATE_COLUMN_ALIASES.items(): normalized_aliases = [normalize_template_column(alias) for alias in aliases] for alias in normalized_aliases: if alias in normalized_columns: column_map[field] = normalized_columns.index(alias) break for field, column_index in TEMPLATE_COLUMN_FALLBACK_INDEXES.items(): if field not in column_map and column_index < len(normalized_columns): column_map[field] = column_index missing_required = [ TEMPLATE_COLUMN_ALIASES[field][0] for field in REQUIRED_TEMPLATE_FIELDS if field not in column_map ] if missing_required: raise ValueError("템플릿 필수 컬럼이 없습니다: " + ", ".join(missing_required)) return column_map def row_value(row_values, column_map: dict[str, int], field: str) -> str | None: index = column_map.get(field) if index is None or index >= len(row_values): return None return clean_excel_value(row_values[index]) def normalize_address_key(value) -> str: if not value: return "" return str(value).strip(" \t\r\n\u00a0\u200b\u200c\u200d\ufeff") def is_manual_vendor(value) -> bool: return str(value or "").replace(" ", "") == "수동발주" def set_upload_state( user_id: str, *, status: str, message: str, upload_progress: int = 0, index_progress: int = 0, total_rows: int = 0, processed_rows: int = 0, indexed_rows: int = 0, ): upload_progress_state[user_id] = { "status": status, "message": message, "upload_progress": max(0, min(100, int(upload_progress))), "index_progress": max(0, min(100, int(index_progress))), "total_rows": int(total_rows or 0), "processed_rows": int(processed_rows or 0), "indexed_rows": int(indexed_rows or 0), } def set_upload_error(user_id: str, message: str): previous = upload_progress_state.get(user_id, {}) set_upload_state( user_id, status="error", message=message, upload_progress=previous.get("upload_progress", 0), index_progress=previous.get("index_progress", 0), total_rows=previous.get("total_rows", 0), processed_rows=previous.get("processed_rows", 0), indexed_rows=previous.get("indexed_rows", 0), ) def build_order_search_query( db: Session, *, name: Optional[str] = None, tracking: Optional[str] = None, phone: Optional[str] = None, address: Optional[str] = None, order_start_date: Optional[str] = None, order_end_date: Optional[str] = None, ): query = db.query(Order) params = {} is_sqlite = db.get_bind().dialect.name == "sqlite" if name: if is_sqlite and can_use_trigram_fts(name): params["name_fts"] = f"recipient_name:{fts_phrase(name)}" query = query.filter( text( "orders.id IN (" "SELECT rowid FROM orders_name_address_fts " "WHERE orders_name_address_fts MATCH :name_fts" ")" ) ) else: query = query.filter(Order.recipient_name.ilike(f"%{name}%")) if tracking: clean_tracking = normalize_tracking(tracking) if clean_tracking: tracking_upper = prefix_upper_bound(clean_tracking) if tracking_upper: query = query.filter( Order.tracking_number_normalized >= clean_tracking, Order.tracking_number_normalized < tracking_upper, ) else: query = query.filter(Order.tracking_number_normalized == clean_tracking) if phone: clean_phone = normalize_phone(phone) if clean_phone: if is_sqlite and can_use_trigram_fts(clean_phone): params["phone_fts"] = ( f"recipient_phone_normalized:{fts_phrase(clean_phone)} OR " f"recipient_mobile_normalized:{fts_phrase(clean_phone)}" ) query = query.filter( text( "orders.id IN (" "SELECT rowid FROM orders_phone_fts " "WHERE orders_phone_fts MATCH :phone_fts" ")" ) ) else: query = query.filter( or_( Order.recipient_phone_normalized.like(f"%{clean_phone}%"), Order.recipient_mobile_normalized.like(f"%{clean_phone}%"), ) ) if address: if is_sqlite and can_use_trigram_fts(address): params["address_fts"] = f"address:{fts_phrase(address)}" query = query.filter( text( "orders.id IN (" "SELECT rowid FROM orders_name_address_fts " "WHERE orders_name_address_fts MATCH :address_fts" ")" ) ) else: query = query.filter(Order.address.ilike(f"%{address}%")) if order_start_date: query = query.filter(Order.order_date >= f"{order_start_date} 00:00:00") if order_end_date: query = query.filter(Order.order_date <= f"{order_end_date} 23:59:59") if params: query = query.params(**params) has_filters = any([name, tracking, phone, address, order_start_date, order_end_date]) return query, has_filters def customer_key(name: str | None, phone: str | None): normalized_name = (name or "").strip() normalized_phone = normalize_phone(phone) if not normalized_name or not normalized_phone: return None return normalized_name, normalized_phone def order_note_key(order: Order) -> str: return (order.order_no_mall or order.order_no or "").strip() def apply_customer_notes(db: Session, orders: list[Order]): keys = { key for order in orders if (key := customer_key(order.recipient_name, order.recipient_mobile or order.recipient_phone)) } if not keys: for order in orders: order.customer_note = "" return names = sorted({name for name, _phone in keys}) phones = sorted({phone for _name, phone in keys}) note_rows = ( db.query( Order.recipient_name, Order.recipient_mobile_normalized, Order.recipient_phone_normalized, Order.customer_note, Order.note, ) .filter( or_( and_(Order.customer_note.isnot(None), Order.customer_note != ""), and_(Order.note.isnot(None), Order.note != ""), ) ) .filter(func.trim(Order.recipient_name).in_(names)) .filter( or_( Order.recipient_mobile_normalized.in_(phones), Order.recipient_phone_normalized.in_(phones), ) ) .order_by(Order.id.desc()) .all() ) notes_by_customer = {} for name, mobile_normalized, phone_normalized, customer_note, legacy_note in note_rows: note = customer_note or legacy_note or "" for phone_value in (mobile_normalized, phone_normalized): key = customer_key(name, phone_value) if key in keys and key not in notes_by_customer: notes_by_customer[key] = note for order in orders: key = customer_key(order.recipient_name, order.recipient_mobile or order.recipient_phone) order.customer_note = notes_by_customer.get(key, "") if key else "" def apply_order_notes(db: Session, orders: list[Order]): keys = sorted({key for order in orders if (key := order_note_key(order))}) if not keys: for order in orders: order.order_note = "" return note_rows = ( db.query( Order.order_no_mall, Order.order_no, Order.order_note, ) .filter(Order.order_note.isnot(None), Order.order_note != "") .filter( or_( Order.order_no_mall.in_(keys), Order.order_no.in_(keys), ) ) .order_by(Order.id.desc()) .all() ) notes_by_order = {} for order_no_mall, order_no, note in note_rows: for key in ((order_no_mall or "").strip(), (order_no or "").strip()): if key in keys and key not in notes_by_order: notes_by_order[key] = note or "" for order in orders: key = order_note_key(order) order.order_note = notes_by_order.get(key, "") if key else "" def order_to_search_item(order: Order): return { "id": order.id, "order_date": order.order_date, "sequence_num": order.sequence_num, "order_no": order.order_no, "order_no_mall": order.order_no_mall, "recipient_name": order.recipient_name, "product_code": order.product_code, "product_name": order.product_name, "order_quantity": order.order_quantity, "address": order.address, "postal_code": order.postal_code, "recipient_phone": order.recipient_phone, "recipient_mobile": order.recipient_mobile, "delivery_memo": order.delivery_memo, "tracking_number": order.tracking_number, "vendor": order.vendor, "order_list_1": order.order_list_1, "order_list_2": order.order_list_2, "note": order.note, "upload_date": order.upload_date, "address_order_count": getattr(order, "address_order_count", 0), "phone_order_count": getattr(order, "phone_order_count", 0), "customer_note": getattr(order, "customer_note", "") or "", "order_note": getattr(order, "order_note", "") or "", } def orders_to_excel_response(orders, filename: str, sheet_name: str = "Orders"): data = [] for o in orders: data.append({ "주문날짜": o.order_date, "수령인명": o.recipient_name, "상품코드": o.product_code, "상품명": o.product_name, "수량": o.order_quantity, "주소": o.address, "수령인 전화번호": o.recipient_phone, "휴대폰": o.recipient_mobile, "송장번호": o.tracking_number, "발주처": o.vendor, "주문번호": o.order_no, "주문번호(쇼핑몰)": o.order_no_mall, "주문목록": o.order_list_1, "추가날짜": o.upload_date, }) df = pd.DataFrame(data) output = io.BytesIO() with pd.ExcelWriter(output, engine="openpyxl") as writer: df.to_excel(writer, index=False, sheet_name=sheet_name) output.seek(0) encoded_filename = quote(filename) return Response( content=output.getvalue(), media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", headers={"Content-Disposition": f"attachment; filename*=UTF-8''{encoded_filename}"}, ) def run_upload_job(files_data: list[dict], user_id: str, user_data: dict): current_user = SimpleNamespace(**user_data) db = SessionLocal() filenames_str = ", ".join([f["filename"] for f in files_data]) try: db_records = [] total_rows = 0 # 1단계: 모든 파일 읽기 및 행 데이터 정규화 for f_idx, file_item in enumerate(files_data): contents = file_item["contents"] filename = file_item["filename"] set_upload_state( user_id, status="reading", message=f"[{f_idx + 1}/{len(files_data)}] 엑셀 파일을 읽는 중입니다 ({filename})...", ) df = pd.read_excel(io.BytesIO(contents)) df = df.where(pd.notnull(df), None) column_map = find_template_columns(list(df.columns)) file_rows = len(df) total_rows += file_rows set_upload_state( user_id, status="processing", message=f"[{f_idx + 1}/{len(files_data)}] {filename}의 엑셀 데이터를 정리하는 중...", total_rows=total_rows, ) for index, row in df.iterrows(): rlist = list(row) 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") address = row_value(rlist, column_map, "address") order = Order( order_date=row_value(rlist, column_map, "order_date"), sequence_num=row_value(rlist, column_map, "sequence_num"), order_no=row_value(rlist, column_map, "order_no"), order_no_mall=row_value(rlist, column_map, "order_no_mall"), recipient_name=row_value(rlist, column_map, "recipient_name"), product_code=row_value(rlist, column_map, "product_code"), product_name=row_value(rlist, column_map, "product_name"), order_quantity=row_value(rlist, column_map, "order_quantity"), address=address, address_normalized=normalize_address_key(address), postal_code=row_value(rlist, column_map, "postal_code"), recipient_phone=recipient_phone, recipient_mobile=recipient_mobile, recipient_phone_normalized=normalize_phone(recipient_phone), recipient_mobile_normalized=normalize_phone(recipient_mobile), delivery_memo=row_value(rlist, column_map, "delivery_memo"), tracking_number=tracking_number, tracking_number_normalized=normalize_tracking(tracking_number), vendor=row_value(rlist, column_map, "vendor"), order_list_1=row_value(rlist, column_map, "order_list_1"), order_note=row_value(rlist, column_map, "order_note"), upload_date=datetime.now().strftime("%Y-%m-%d %H:%M:%S") ) db_records.append(order) # 주기적으로 데이터 정리 진행률 갱신 (전체 누적 건수 기준) current_processed = len(db_records) if current_processed % 1000 == 0 and current_processed > 0: calc_progress = float(current_processed) / total_rows if total_rows else 1 set_upload_state( user_id, status="processing", message=f"전체 {current_processed}건 엑셀 데이터 정리 중...", upload_progress=calc_progress * 100, total_rows=total_rows, processed_rows=current_processed, ) # 2단계: FTS 인덱스 및 DB 저장 단계 set_upload_state( user_id, status="indexing", message="DB 저장과 검색 인덱스 반영을 시작합니다...", upload_progress=100, index_progress=0, total_rows=total_rows, processed_rows=total_rows, ) batch_size = 1000 for i in range(0, len(db_records), batch_size): chunk = db_records[i:i + batch_size] db.bulk_save_objects(chunk) indexed_rows = min(i + len(chunk), total_rows) index_progress = (indexed_rows / total_rows * 100) if total_rows else 100 set_upload_state( user_id, status="indexing", message=f"{indexed_rows}/{total_rows}건 DB 저장 및 검색 인덱스 반영 중...", upload_progress=100, index_progress=index_progress, total_rows=total_rows, processed_rows=total_rows, indexed_rows=indexed_rows, ) db.commit() set_upload_state( user_id, status="completed", message=f"완료: {len(db_records)}건 저장 및 검색 인덱스 반영", upload_progress=100, index_progress=100, total_rows=total_rows, processed_rows=total_rows, indexed_rows=total_rows, ) log_event( current_user, "파일 업로드 완료", "success", {"filenames": filenames_str, "uploaded_count": len(db_records)}, ) except ValueError as e: db.rollback() set_upload_error(user_id, str(e)) log_event(current_user, "파일 업로드 실패", "error", {"filenames": filenames_str, "error": str(e)}) except Exception as e: db.rollback() set_upload_error(user_id, f"에러 발생: {str(e)}") log_event(current_user, "파일 업로드 실패", "error", {"filenames": filenames_str, "error": str(e)}) finally: db.close() @router.post("/upload") def upload_file( files: list[UploadFile] = File(...), upload_token: str = Form(...), custom_date: str = Form(...), 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).") # 기존 데이터/검색 쿼리가 "YYYY-MM-DD HH:MM:SS" 문자열 비교를 사용하므로 # 새 업로드 데이터도 동일하게 시간 부분(00:00:00)을 붙여 저장한다. order_date_str = order_date_dt.strftime("%Y-%m-%d %H:%M:%S") # 1. 파일 확장자 검증 if not files: raise HTTPException(status_code=400, detail="업로드할 파일이 없습니다.") for file in files: filename = file.filename or "" 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]) log_event( current_user, "파일 업로드 시작 (낱개 출고 자동분석)", "started", {"filenames": filenames_str, "custom_date": custom_date} ) # 2. 파일 파싱 — 형식/템플릿 오류는 400 으로 분리 db_records = [] try: for file in files: contents = file.file.read() 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) 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(): 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") address = row_value(rlist, column_map, "address") order = Order( order_date=order_date_str, # 달력에서 고른 날짜 + 00:00:00 (검색 호환) sequence_num=row_value(rlist, column_map, "sequence_num"), order_no=row_value(rlist, column_map, "order_no"), order_no_mall=row_value(rlist, column_map, "order_no_mall"), recipient_name=row_value(rlist, column_map, "recipient_name"), product_code=row_value(rlist, column_map, "product_code"), product_name=row_value(rlist, column_map, "product_name"), order_quantity=row_value(rlist, column_map, "order_quantity"), address=address, address_normalized=normalize_address_key(address), postal_code=row_value(rlist, column_map, "postal_code"), recipient_phone=recipient_phone, recipient_mobile=recipient_mobile, recipient_phone_normalized=normalize_phone(recipient_phone), recipient_mobile_normalized=normalize_phone(recipient_mobile), delivery_memo=row_value(rlist, column_map, "delivery_memo"), tracking_number=tracking_number, tracking_number_normalized=normalize_tracking(tracking_number), vendor=row_value(rlist, column_map, "vendor"), order_list_1=row_value(rlist, column_map, "order_list_1"), order_note=row_value(rlist, column_map, "order_note"), 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. 데이터베이스 벌크 저장 — DB 오류는 500 try: batch_size = 1000 for i in range(0, len(db_records), batch_size): chunk = db_records[i:i + batch_size] db.bulk_save_objects(chunk) db.commit() log_event( current_user, "파일 업로드 DB 저장 완료", "success", {"filenames": filenames_str, "uploaded_count": len(db_records)} ) except Exception as e: db.rollback() 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: single_items: dict[str, dict] = {} set_components: dict[str, list[dict]] = {} # 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: try: with item_engine.connect() as conn: # 1) 단품 마스터 맵 구성 rs_single = conn.execute(text("SELECT item_code, sabangnet_code, name FROM single_items")) for row in rs_single: if not row[0]: continue single_items[row[0]] = { "sabangnet_code": row[1] or "", "name": row[2] or "" } # 2) 세트 구성품 맵 구성 rs_comp = conn.execute(text("SELECT set_code, single_code, quantity FROM set_components")) for row in rs_comp: s_code = row[0] single_code = row[1] if not s_code or not single_code: continue set_components.setdefault(s_code, []).append({ "single_code": single_code, # 마스터의 수량 컬럼도 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 # 수량은 비숫자/공백/None 모두 1로 폴백 (보존: 비어 있는 셀도 1건으로 본다) qty = safe_int_quantity(order.order_quantity, default=1) 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 = safe_int_quantity(comp["quantity"], default=1) single_qty_sums[s_code] = single_qty_sums.get(s_code, 0) + (qty * comp_qty) # 세트 구성 단품 정보가 single_items 에 없을 수도 있으므로 placeholder 만 보강. # 사방넷 코드는 비워서, 결과 엑셀 빌드 단계에서 제외 대상으로 인식되게 둔다. if s_code not in single_items: single_items[s_code] = {"sabangnet_code": "", "name": ""} else: # 마스터 미등록 코드: 사방넷 코드를 알 수 없으므로 placeholder 로만 추적. # 결과 엑셀에서는 제외된다. unknown_codes.add(prod_code) single_items[prod_code] = { "sabangnet_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 = [] skipped_no_sabang: list[str] = [] for s_code, total_qty in single_qty_sums.items(): s_info = single_items.get(s_code, {"sabangnet_code": "", "name": ""}) sabang_code = (s_info.get("sabangnet_code") or "").strip() if not sabang_code: skipped_no_sabang.append(s_code) continue excel_rows.append({ "상품코드[필수]": sabang_code, "가용수량": total_qty, "불용수량": s_info.get("name") or "" }) if skipped_no_sabang: log_event( current_user, "낱개 출고 분석 — 사방넷 코드 없는 항목 제외", "warning", {"skipped_count": len(skipped_no_sabang), "sample": skipped_no_sabang[:20]}, ) df_out = pd.DataFrame(excel_rows, columns=["상품코드[필수]", "가용수량", "불용수량"]) # 엑셀 파일 쓰기 (1행 헤더, 2행부터 데이터 — pandas to_excel 기본 동작) # 작성 직후 A~C 열 너비를 콘텐츠 길이에 맞춰 자동 조절한다. # (한글은 ASCII보다 폭이 약 2배이므로 가중치 적용) def _excel_display_len(value) -> int: s = "" if value is None else str(value) width = 0 for ch in s: width += 2 if ord(ch) > 127 else 1 return width excel_io = io.BytesIO() with pd.ExcelWriter(excel_io, engine='openpyxl') as writer: df_out.to_excel(writer, index=False, sheet_name="발주") worksheet = writer.sheets["발주"] from openpyxl.utils import get_column_letter for col_idx, col_name in enumerate(df_out.columns, start=1): header_w = _excel_display_len(col_name) data_w = max( (_excel_display_len(v) for v in df_out[col_name].values), default=0, ) # 최소 10, 최대 60 으로 클램프 + 여백 2 worksheet.column_dimensions[get_column_letter(col_idx)].width = ( min(60, max(10, max(header_w, data_w) + 2)) ) excel_io.seek(0) # 6. 다운로드 파일명: "MM월 DD일 (요일)_발주_(낱개 상품 출고).xls" weekdays = ["월요일", "화요일", "수요일", "목요일", "금요일", "토요일", "일요일"] weekday_str = weekdays[order_date_dt.weekday()] filename = f"{order_date_dt.strftime('%m')}월 {order_date_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), "excel_row_count": len(excel_rows), "unknown_code_count": len(unknown_codes), "skipped_no_sabang_count": len(skipped_no_sabang), }, ) 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)}") @router.get("/progress") async def upload_progress( upload_token: str, current_user = Depends(get_current_user) ): user_id = f"{current_user.id}:{upload_token}" async def event_generator(): while True: if user_id in upload_progress_state: state = upload_progress_state[user_id] yield f"data: {json.dumps(state, ensure_ascii=False)}\n\n" # Stop streaming connection if done or error if state["status"] in ["completed", "error"]: upload_progress_state.pop(user_id, None) break else: yield f"data: {json.dumps({'status': 'idle'}, ensure_ascii=False)}\n\n" await asyncio.sleep(0.5) return StreamingResponse(event_generator(), media_type="text/event-stream") @router.get("/search") async def search_orders( name: Optional[str] = None, tracking: Optional[str] = None, phone: Optional[str] = None, address: Optional[str] = None, order_start_date: Optional[str] = None, order_end_date: Optional[str] = None, page: int = 1, size: int = 27, db: Session = Depends(get_db), current_user = Depends(get_current_user) ): query, has_filters = build_order_search_query( db, name=name, tracking=tracking, phone=phone, address=address, order_start_date=order_start_date, order_end_date=order_end_date, ) # If no parameters are provided, return empty if not has_filters: return {"items": [], "total": 0, "page": page, "size": size} total = query.count() offset = (page - 1) * size orders = query.order_by(Order.order_date.desc(), Order.id.desc()).offset(offset).limit(size).all() page_address_keys = sorted({normalize_address_key(order.address) for order in orders if normalize_address_key(order.address)}) page_phones = sorted( { normalize_phone(order.recipient_mobile or order.recipient_phone) for order in orders if normalize_phone(order.recipient_mobile or order.recipient_phone) } ) address_date_counts = {} phone_date_counts = {} order_date_key = func.substr(Order.order_date, 1, 10) if page_address_keys: address_date_counts = dict( db.query( Order.address_normalized, func.count(func.distinct(order_date_key)), ) .filter(Order.address_normalized.in_(page_address_keys)) .group_by(Order.address_normalized) .all() ) if page_phones: phone_date_sets = {phone_value: set() for phone_value in page_phones} phone_rows = ( db.query( Order.recipient_mobile_normalized, Order.recipient_phone_normalized, order_date_key, ) .filter( or_( Order.recipient_mobile_normalized.in_(page_phones), Order.recipient_phone_normalized.in_(page_phones), ) ) .all() ) for mobile_normalized, phone_normalized, date_key in phone_rows: if mobile_normalized in phone_date_sets: if date_key: phone_date_sets[mobile_normalized].add(date_key) if phone_normalized in phone_date_sets: if date_key: phone_date_sets[phone_normalized].add(date_key) phone_date_counts = { phone_value: len(date_keys) for phone_value, date_keys in phone_date_sets.items() } for order in orders: address_key = normalize_address_key(order.address) order.address_order_count = address_date_counts.get(address_key, 0) if address_key else 0 display_phone_normalized = normalize_phone(order.recipient_mobile or order.recipient_phone) order.phone_order_count = phone_date_counts.get(display_phone_normalized, 0) if display_phone_normalized else 0 apply_customer_notes(db, orders) apply_order_notes(db, orders) return { "items": [order_to_search_item(order) for order in orders], "total": total, "page": page, "size": size, "total_pages": math.ceil(total / size) if size > 0 else 0 } @router.get("/export-search") async def export_search_orders( name: Optional[str] = None, tracking: Optional[str] = None, phone: Optional[str] = None, address: Optional[str] = None, order_start_date: Optional[str] = None, order_end_date: Optional[str] = None, db: Session = Depends(get_db), current_user = Depends(get_current_user) ): query, has_filters = build_order_search_query( db, name=name, tracking=tracking, phone=phone, address=address, order_start_date=order_start_date, order_end_date=order_end_date, ) if not has_filters: raise HTTPException(status_code=400, detail="검색 조건이 없습니다.") orders = query.order_by(Order.order_date.desc(), Order.id.desc()).all() if not orders: raise HTTPException(status_code=404, detail="다운로드할 검색 결과가 없습니다.") timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") log_event( current_user, "검색 결과 엑셀 다운로드", "success", { "download_count": len(orders), "name": name, "phone": phone, "address": address, "tracking": tracking, "start_date": order_start_date, "end_date": order_end_date, }, ) return orders_to_excel_response(orders, f"검색결과_{timestamp}.xlsx", "Search_Results") class DeleteOrdersRequest(BaseModel): order_ids: list[int] class UpdateOrderContactRequest(BaseModel): recipient_name: str address: str recipient_mobile: str customer_note: Optional[str] = None order_note: Optional[str] = None note: str = "" class DeleteSingleOrderRequest(BaseModel): password: str class ReturnRequestPayload(BaseModel): request_date: str request_type: str remarks: str = "" tracking_no: Optional[str] = None receiver_name: Optional[str] = None address: Optional[str] = None phone: Optional[str] = None @router.put("/{order_id}/contact") async def update_order_contact( order_id: int, req: UpdateOrderContactRequest, db: Session = Depends(get_db), current_user = Depends(get_current_user) ): order = db.query(Order).filter(Order.id == order_id).first() if not order: log_event(current_user, "주문 정보 수정", "error", {"order_id": order_id, "error": "not_found"}) raise HTTPException(status_code=404, detail="주문 데이터를 찾을 수 없습니다.") before = { "recipient_name": order.recipient_name, "address": order.address, "recipient_mobile": order.recipient_mobile, "customer_note": order.customer_note, "order_note": order.order_note, "note": order.note, } new_name = req.recipient_name.strip() or None new_phone = req.recipient_mobile.strip() or None new_phone_normalized = normalize_phone(new_phone) new_customer_note = ( req.customer_note if req.customer_note is not None else req.note ).strip() or None new_order_note = (req.order_note or "").strip() or None order.recipient_name = new_name order.address = req.address.strip() or None order.address_normalized = normalize_address_key(order.address) order.recipient_mobile = new_phone order.recipient_mobile_normalized = new_phone_normalized customer_note_updated_count = 1 customer_phone_normalized = new_phone_normalized or normalize_phone(order.recipient_phone) if new_name and customer_phone_normalized: customer_note_targets = ( db.query(Order) .filter(func.trim(Order.recipient_name) == new_name) .filter( or_( Order.recipient_mobile_normalized == customer_phone_normalized, Order.recipient_phone_normalized == customer_phone_normalized, ) ) .all() ) for target_order in customer_note_targets: target_order.customer_note = new_customer_note customer_note_updated_count = len(customer_note_targets) order_note_updated_count = 1 current_order_note_key = order_note_key(order) if current_order_note_key: order_note_targets = ( db.query(Order) .filter( or_( Order.order_no_mall == current_order_note_key, Order.order_no == current_order_note_key, ) ) .all() ) for target_order in order_note_targets: target_order.order_note = new_order_note order_note_updated_count = len(order_note_targets) order.customer_note = new_customer_note order.order_note = new_order_note db.commit() db.refresh(order) log_event( current_user, "주문 정보 수정", "success", { "order_id": order.id, "order_no": order.order_no, "order_no_mall": order.order_no_mall, "before": before, "after": { "recipient_name": order.recipient_name, "address": order.address, "recipient_mobile": order.recipient_mobile, "customer_note": order.customer_note, "order_note": order.order_note, "note": order.note, }, "customer_note_updated_count": customer_note_updated_count, "order_note_updated_count": order_note_updated_count, "submitted_customer_note_length": len( req.customer_note if req.customer_note is not None else req.note or "" ), "submitted_order_note_length": len(req.order_note or ""), }, ) return { "message": "주문 정보가 수정되었습니다.", "order_id": order.id, "customer_note": order.customer_note, "order_note": order.order_note, "note": order.note, "customer_note_updated_count": customer_note_updated_count, "order_note_updated_count": order_note_updated_count, } @router.post("/{order_id}/return-request") async def send_return_request( order_id: int, req: ReturnRequestPayload, db: Session = Depends(get_db), current_user = Depends(get_current_user), ): order = db.query(Order).filter(Order.id == order_id).first() if not order: log_event(current_user, "반품 신청 전송", "error", {"order_id": order_id, "error": "not_found"}) raise HTTPException(status_code=404, detail="주문 데이터를 찾을 수 없습니다.") payload = { "request_date": (req.request_date or "").strip(), "request_type": (req.request_type or "").strip(), "receiver_name": (req.receiver_name or order.recipient_name or "").strip(), "address": (req.address or order.address or "").strip(), "phone": (req.phone or order.recipient_mobile or order.recipient_phone or "").strip(), "tracking_no": (req.tracking_no or order.tracking_number or "").strip(), "mall": (order.vendor or "").strip(), "remarks": (req.remarks or "").strip(), } missing_fields = [ key for key in ("request_date", "request_type") if payload[key] == "" ] if missing_fields: log_event( current_user, "반품 신청 전송", "error", { "order_id": order_id, "missing_fields": missing_fields, "payload": payload, }, ) raise HTTPException( status_code=400, detail=f"반품 신청 필수값이 비어 있습니다: {', '.join(missing_fields)}", ) log_event( current_user, "반품 신청 전송 요청", "info", { "order_id": order_id, "endpoint": CS_RETURN_REQUEST_URL, "payload": payload, }, ) try: timeout = httpx.Timeout(connect=2.0, read=5.0, write=3.0, pool=2.0) headers = {} if INTERNAL_SERVICE_TOKEN: headers["X-Service-Token"] = INTERNAL_SERVICE_TOKEN async with httpx.AsyncClient(timeout=timeout, trust_env=False) as client: response = await client.post(CS_RETURN_REQUEST_URL, json=payload, headers=headers) response.raise_for_status() except httpx.HTTPStatusError as exc: response_text = exc.response.text[:1000] if exc.response is not None else "" log_event( current_user, "반품 신청 전송", "error", { "order_id": order_id, "status_code": exc.response.status_code if exc.response is not None else None, "response": response_text, "payload": payload, }, ) raise HTTPException( status_code=502, detail=f"CS 관리 프로그램 반품 신청 오류: HTTP {exc.response.status_code}", ) except httpx.RequestError as exc: log_event( current_user, "반품 신청 전송", "error", { "order_id": order_id, "endpoint": CS_RETURN_REQUEST_URL, "error": str(exc), "payload": payload, }, ) raise HTTPException(status_code=502, detail=f"CS 관리 프로그램 연결 오류: {str(exc)}") log_event( current_user, "반품 신청 전송", "success", { "order_id": order_id, "endpoint": CS_RETURN_REQUEST_URL, "payload": payload, "status_code": response.status_code, }, ) return { "message": "반품 신청 데이터가 CS 관리 프로그램으로 전송되었습니다.", "status_code": response.status_code, "payload": payload, } @router.post("/{order_id}/delete") @router.delete("/{order_id}") async def delete_single_order( order_id: int, req: DeleteSingleOrderRequest, db: Session = Depends(get_db), current_user = Depends(get_current_user) ): log_event(current_user, "주문 단건 삭제 요청", "info", {"order_id": order_id}) if req.password != "1225": log_event(current_user, "주문 단건 삭제", "error", {"order_id": order_id, "error": "invalid_password"}) raise HTTPException(status_code=403, detail="비밀번호가 올바르지 않습니다.") order = db.query(Order).filter(Order.id == order_id).first() if not order: log_event(current_user, "주문 단건 삭제", "error", {"order_id": order_id, "error": "not_found"}) raise HTTPException(status_code=404, detail="주문 데이터를 찾을 수 없습니다.") deleted_order = { "order_id": order.id, "order_date": order.order_date, "order_no": order.order_no, "order_no_mall": order.order_no_mall, "recipient_name": order.recipient_name, "recipient_mobile": order.recipient_mobile, "recipient_phone": order.recipient_phone, "product_code": order.product_code, "product_name": order.product_name, "order_quantity": order.order_quantity, "vendor": order.vendor, "note": order.note, } try: db.delete(order) db.commit() log_event(current_user, "주문 단건 삭제", "success", deleted_order) return {"message": "주문이 삭제되었습니다.", "deleted_count": 1, "order_id": order_id} except Exception as e: db.rollback() log_event(current_user, "주문 단건 삭제", "error", {**deleted_order, "error": str(e)}) raise HTTPException(status_code=500, detail=f"주문 삭제 오류: {str(e)}") @router.post("/delete") async def delete_orders( req: DeleteOrdersRequest, db: Session = Depends(get_db), current_user = Depends(get_current_user) ): if not current_user.is_admin: raise HTTPException(status_code=403, detail="Not authorized to delete data.") if not req.order_ids: return {"message": "No orders selected for deletion.", "deleted_count": 0} try: deleted_count = db.query(Order).filter(Order.id.in_(req.order_ids)).delete(synchronize_session=False) db.commit() return {"message": f"성공적으로 {deleted_count}건의 데이터를 삭제했습니다.", "deleted_count": deleted_count} except Exception as e: db.rollback() raise HTTPException(status_code=500, detail=str(e))