1223 lines
45 KiB
Python
1223 lines
45 KiB
Python
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 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)
|
|
):
|
|
# 1. 파일 확장자 검증
|
|
for file in files:
|
|
filename = file.filename or ""
|
|
if not filename.endswith(".xls") and not 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}
|
|
)
|
|
|
|
db_records = []
|
|
try:
|
|
# 2. 파일 파싱 및 데이터베이스 레코드 임시 적재
|
|
for file in files:
|
|
contents = file.file.read()
|
|
df = pd.read_excel(io.BytesIO(contents))
|
|
df = df.where(pd.notnull(df), None)
|
|
column_map = find_template_columns(list(df.columns))
|
|
|
|
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=custom_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)
|
|
|
|
# 3. 데이터베이스 벌크 저장 및 커밋
|
|
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, "파일 업로드 실패", "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 = {}
|
|
set_components = {}
|
|
|
|
if item_engine:
|
|
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:
|
|
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]
|
|
if s_code not in set_components:
|
|
set_components[s_code] = []
|
|
set_components[s_code].append({
|
|
"single_code": row[1],
|
|
"quantity": row[2]
|
|
})
|
|
|
|
# 3) 낱개 환산 처리
|
|
single_qty_sums = {}
|
|
for order in db_records:
|
|
prod_code = (order.product_code or "").strip()
|
|
if not prod_code:
|
|
continue
|
|
|
|
qty = 1
|
|
try:
|
|
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"]
|
|
single_qty_sums[s_code] = single_qty_sums.get(s_code, 0) + (qty * comp_qty)
|
|
else:
|
|
# DB에 등록되어 있지 않은 코드는 단품으로 간주하고, 사방넷코드와 이름을 보존
|
|
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
|
|
|
|
# 5. 결과 엑셀 데이터 빌드
|
|
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열 헤더는 "불용수량", 실데이터는 이름!
|
|
})
|
|
|
|
df_out = pd.DataFrame(excel_rows)
|
|
if df_out.empty:
|
|
df_out = pd.DataFrame(columns=["상품코드[필수]", "가용수량", "불용수량"])
|
|
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)
|
|
excel_io.seek(0)
|
|
|
|
# 6. 다운로드 파일명 날짜 포맷 조합
|
|
try:
|
|
dt = datetime.strptime(custom_date, "%Y-%m-%d")
|
|
except:
|
|
dt = datetime.now()
|
|
|
|
weekdays = ["월요일", "화요일", "수요일", "목요일", "금요일", "토요일", "일요일"]
|
|
weekday_str = weekdays[dt.weekday()]
|
|
|
|
# 형식 예: 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"
|
|
}
|
|
|
|
return StreamingResponse(
|
|
excel_io,
|
|
media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
|
headers=headers
|
|
)
|
|
|
|
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))
|