king
065e81bf97
CS 통합 프로그램.ZIP(로컬 개발본 v9.0)의 기능 소스를 반영. DB 스키마는 기존과 동일해 마이그레이션 없이 그대로 사용한다. 가져온 것 - main.py / cafe24_api.py / templates / static: v9.0 기능 코드 - routers/coupang_milkrun.py, routers/mall_event.py (신규) - static/js/mall_event.js, static/js/milkrun_gsheet.js (신규) 운영 설정은 기존 것을 유지·재적용 - DB/카페24/네이버 접속정보를 하드코딩 대신 환경변수 기반으로 복원 - SSO(AuthGuardMiddleware, SessionMiddleware), /login, /logout, /health/db 복원 - APP_ROOT_PATH 서브경로 호스팅(root_path 템플릿 변수, app.js fetch 래퍼) 복원 - 구글시트 설정 인메모리 캐시(TTL 10분)와 /api/config/refresh 복원 Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
1124 lines
43 KiB
Python
1124 lines
43 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""자사몰 행사 - 카페24 주문/발주 파일 가공 모듈.
|
||
|
||
카페24에서 받은 파일을 업로드하면 행사 조건을 적용해
|
||
가공된 파일을 다시 내려받을 수 있게 한다.
|
||
|
||
현재 제공 기능
|
||
1) 선착순 사은품 추가 (발주주문서 xlsx 업로드 → 사은품 행이 추가된 xlsx)
|
||
2) 행사 주문 추출 (주문 CSV 업로드 → 조건에 맞는 행만 남긴 xlsx)
|
||
|
||
앞으로 다른 가공 기능(업로드 → 분석 → 변환 파일 다운로드)이 이 라우터에 계속 추가된다.
|
||
새 기능을 넣을 때는 `_first_come_*`, `_event_extract_*` 처럼 기능 접두사를 붙여 헬퍼를
|
||
분리하고, 엔드포인트 경로도 `/api/mall-event/<기능명>/...` 형태로 나눈다.
|
||
|
||
성능 메모
|
||
분석/미리보기는 값만 있으면 되므로 openpyxl read_only 모드로 한 번만 훑는다.
|
||
서식을 유지해야 하는 다운로드 단계에서만 쓰기 가능한 워크북을 연다.
|
||
같은 파일을 분석 → 미리보기 → 다운로드로 연달아 올리므로,
|
||
파일 내용 해시를 키로 스캔 결과를 짧게 캐싱해 재파싱을 피한다.
|
||
|
||
[기능 1] 발주주문서 열 위치 (카페24 양식 기준, 사용자 확정)
|
||
A: 주문날짜 C: 품목 순번(끝 숫자 증가 대상) E: 아이템코드
|
||
F: 상품이름 G: 수량 P: 주문번호 Q: 주문목록
|
||
|
||
[기능 2] 주문 CSV 열 위치 (사용자 확정)
|
||
B: 주문번호 K: 주문상품명(세트상품 포함) L, M: N열 수식의 피연산자
|
||
N: L×M 수식을 넣는 열
|
||
"""
|
||
|
||
import csv
|
||
import hashlib
|
||
import io
|
||
import json
|
||
import re
|
||
import time
|
||
import urllib.parse
|
||
from bisect import bisect_left
|
||
from collections import OrderedDict
|
||
from copy import copy
|
||
from datetime import date, datetime
|
||
|
||
from fastapi import APIRouter, File, Form, HTTPException, UploadFile
|
||
from fastapi.responses import StreamingResponse
|
||
from openpyxl import Workbook, load_workbook
|
||
from openpyxl.styles import Alignment, Font, PatternFill
|
||
from openpyxl.utils import get_column_letter
|
||
|
||
router = APIRouter(prefix="/api/mall-event", tags=["Mall Event"])
|
||
|
||
# --- 발주주문서 열 위치 (1-based) ---
|
||
COL_ORDER_DATE = 1 # A 주문날짜
|
||
COL_SEQ = 3 # C 품목 순번
|
||
COL_ITEM_CODE = 5 # E 아이템코드
|
||
COL_ITEM_NAME = 6 # F 상품이름
|
||
COL_QTY = 7 # G 수량
|
||
COL_ORDER_NO = 16 # P 주문번호
|
||
COL_ORDER_LIST = 17 # Q 주문목록
|
||
COL_LAST = COL_ORDER_LIST # 사은품 행에 색을 칠하는 범위: A~Q
|
||
|
||
GIFT_LABEL = "선착순 사은품"
|
||
YELLOW_FILL = PatternFill(start_color="FFFFFF00", end_color="FFFFFF00", fill_type="solid")
|
||
|
||
# --- 주문 CSV 열 위치 (1-based) ---
|
||
EE_COL_ORDER_NO = 2 # B 주문번호
|
||
EE_COL_PRODUCT = 11 # K 주문상품명(세트상품 포함)
|
||
EE_COL_L = 12 # L (N열 수식의 왼쪽 피연산자)
|
||
EE_COL_M = 13 # M (N열 수식의 오른쪽 피연산자)
|
||
EE_COL_N = 14 # N (=L×M 수식을 넣는 열)
|
||
EE_SPLIT_TOKEN = " -" # 주문상품명에서 이 문자열 앞까지를 상품명으로 본다
|
||
|
||
# --- 추출 결과 파일 서식 범위 (1-based) ---
|
||
EE_FORMAT_LAST_COL = 28 # AB. 제목행 서식/자동필터/열너비를 적용하는 마지막 열
|
||
EE_NUM_FIRST_COL = 12 # L. 숫자 변환 + 쉼표 서식 + 0은 빈 셀 처리 시작
|
||
EE_NUM_LAST_COL = 19 # S. 위 처리의 마지막 열
|
||
EE_DEDUPE_FIRST_COL = 15 # O. 같은 주문번호에서 중복 값을 지우는 범위 시작
|
||
EE_DEDUPE_LAST_COL = 19 # S. 위 처리의 마지막 열
|
||
EE_SORT_COL = 4 # D. 이 열 기준으로 텍스트 오름차순 정렬
|
||
|
||
EE_COMMA_FORMAT = "#,##0"
|
||
EE_COMMA_FORMAT_DECIMAL = "#,##0.##"
|
||
# 주문번호가 바뀔 때마다 두 색을 번갈아 칠해 주문 묶음을 눈으로 구분한다
|
||
EE_GROUP_FILLS = (
|
||
PatternFill(start_color="FFB4C6E7", end_color="FFB4C6E7", fill_type="solid"),
|
||
PatternFill(start_color="FFFFD966", end_color="FFFFD966", fill_type="solid"),
|
||
)
|
||
EE_HEADER_FILL = PatternFill(start_color="FF3182CE", end_color="FF3182CE", fill_type="solid")
|
||
EE_HEADER_FONT = Font(bold=True, color="FFFFFFFF")
|
||
EE_HEADER_ALIGN = Alignment(horizontal="center", vertical="center", wrap_text=False)
|
||
EE_MIN_COL_WIDTH = 8
|
||
EE_MAX_COL_WIDTH = 60
|
||
|
||
# 길고 장황한 카페24 제목을 짧게 바꾼다 (공백 차이는 무시하고 비교)
|
||
EE_HEADER_RENAMES = {
|
||
"결제일시(입금확인일)": "결제일시",
|
||
"총 배송비 (전체 품목에 표시)": "총 배송비",
|
||
"쿠폰 할인금액(최종)": "쿠폰 할인",
|
||
"사용한 적립금액(최종)": "사용 적립금",
|
||
"네이버 포인트": "N포인트",
|
||
"총 실결제금액(최초정보)": "총 실결제금액",
|
||
}
|
||
_EE_HEADER_RENAMES_LOOSE = {
|
||
re.sub(r"\s+", "", key): value for key, value in EE_HEADER_RENAMES.items()
|
||
}
|
||
|
||
# 픽셀 → 엑셀 열 너비 환산 기준 (사용자 확인: 145px = 너비 17.5)
|
||
EE_PIXELS_PER_WIDTH_UNIT = 8
|
||
EE_WIDTH_PADDING_PIXELS = 5
|
||
|
||
# 지정한 열은 픽셀 단위로 고정, 나머지는 내용에 맞춰 자동
|
||
EE_MANUAL_COL_PIXELS = {
|
||
1: 145, # A
|
||
2: 135, # B
|
||
3: 155, # C
|
||
4: 150, # D
|
||
5: 80, # E
|
||
6: 110, # F
|
||
7: 120, # G
|
||
8: 120, # H
|
||
12: 60, # L
|
||
}
|
||
EE_MANUAL_COL_PIXELS.update({col: 110 for col in range(13, 21)}) # M~T
|
||
|
||
# 헤더 행을 찾을 때 훑어볼 최대 행 수 (양식 위에 안내문이 몇 줄 있어도 잡히도록)
|
||
_HEADER_SCAN_ROWS = 15
|
||
|
||
# 업로드 → 미리보기 → 다운로드가 같은 파일로 이어지므로 스캔 결과를 잠시 들고 있는다.
|
||
_SCAN_CACHE_MAX = 4
|
||
_SCAN_CACHE_TTL_SECONDS = 600
|
||
_scan_cache = OrderedDict() # sha1 -> (저장시각, 스캔결과)
|
||
|
||
_DATE_FORMATS = (
|
||
"%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M", "%Y-%m-%d",
|
||
"%Y.%m.%d %H:%M:%S", "%Y.%m.%d %H:%M", "%Y.%m.%d",
|
||
"%Y/%m/%d %H:%M:%S", "%Y/%m/%d %H:%M", "%Y/%m/%d",
|
||
"%Y%m%d%H%M%S", "%Y%m%d",
|
||
)
|
||
|
||
|
||
# =====================================================================
|
||
# 공용 헬퍼
|
||
# =====================================================================
|
||
def cell_text(value):
|
||
"""셀 값을 비교/표시에 쓸 문자열로 정규화."""
|
||
if value is None:
|
||
return ""
|
||
if isinstance(value, bool):
|
||
return str(value)
|
||
if isinstance(value, float) and value.is_integer():
|
||
return str(int(value))
|
||
if isinstance(value, datetime):
|
||
return value.strftime("%Y-%m-%d %H:%M:%S")
|
||
if isinstance(value, date):
|
||
return value.strftime("%Y-%m-%d")
|
||
return str(value).strip()
|
||
|
||
|
||
def parse_order_date(value):
|
||
"""주문날짜 셀을 datetime으로. 해석 불가면 None."""
|
||
if isinstance(value, datetime):
|
||
return value
|
||
if isinstance(value, date):
|
||
return datetime(value.year, value.month, value.day)
|
||
text = cell_text(value)
|
||
if not text:
|
||
return None
|
||
for fmt in _DATE_FORMATS:
|
||
try:
|
||
return datetime.strptime(text, fmt)
|
||
except ValueError:
|
||
continue
|
||
return None
|
||
|
||
|
||
def first_segment(order_list_value):
|
||
"""주문목록 텍스트에서 첫 번째 '/' 앞부분만 잘라낸다."""
|
||
text = cell_text(order_list_value)
|
||
if not text:
|
||
return ""
|
||
return text.split("/")[0].strip()
|
||
|
||
|
||
def increment_trailing_number(value):
|
||
"""끝에 붙은 숫자를 한 단계 올린다. ('...-01' → '...-02', 3 → 4)
|
||
|
||
반환: (새 값, 성공 여부). 끝 숫자가 없으면 원본을 그대로 돌려주고 False.
|
||
"""
|
||
if value is None:
|
||
return None, False
|
||
if isinstance(value, bool):
|
||
return value, False
|
||
if isinstance(value, int):
|
||
return value + 1, True
|
||
if isinstance(value, float):
|
||
return (int(value) + 1) if value.is_integer() else value + 1, True
|
||
|
||
text = str(value)
|
||
matches = list(re.finditer(r"\d+", text))
|
||
if not matches:
|
||
return text, False
|
||
last = matches[-1]
|
||
digits = last.group(0)
|
||
bumped = str(int(digits) + 1).zfill(len(digits))
|
||
return text[:last.start()] + bumped + text[last.end():], True
|
||
|
||
|
||
def _validate_extension(filename):
|
||
name = (filename or "").lower()
|
||
if name.endswith(".xls"):
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail="구형 .xls 파일은 서식을 유지한 채 가공할 수 없습니다. "
|
||
"엑셀에서 '다른 이름으로 저장 → .xlsx'로 변환한 뒤 다시 업로드해주세요.",
|
||
)
|
||
if not name.endswith((".xlsx", ".xlsm")):
|
||
raise HTTPException(status_code=400, detail="엑셀 파일(.xlsx)만 업로드할 수 있습니다.")
|
||
|
||
|
||
def _value_at(values, column):
|
||
return values[column - 1] if len(values) >= column else None
|
||
|
||
|
||
def _is_header_values(values):
|
||
order_list_label = cell_text(_value_at(values, COL_ORDER_LIST))
|
||
order_date_label = cell_text(_value_at(values, COL_ORDER_DATE))
|
||
return "주문목록" in order_list_label or "주문날짜" in order_date_label
|
||
|
||
|
||
def excel_response(workbook, filename, extra_headers=None):
|
||
stream = io.BytesIO()
|
||
workbook.save(stream)
|
||
stream.seek(0)
|
||
encoded = urllib.parse.quote(filename)
|
||
headers = {"Content-Disposition": f"attachment; filename*=UTF-8''{encoded}"}
|
||
if extra_headers:
|
||
headers.update(extra_headers)
|
||
return StreamingResponse(
|
||
stream,
|
||
media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
||
headers=headers,
|
||
)
|
||
|
||
|
||
def _base_name(original_filename):
|
||
"""업로드된 파일 이름에서 경로와 확장자를 떼어낸다."""
|
||
base = (original_filename or "발주서").rsplit("/", 1)[-1].rsplit("\\", 1)[-1]
|
||
for ext in (".xlsx", ".xlsm", ".xls", ".csv"):
|
||
if base.lower().endswith(ext):
|
||
return base[: -len(ext)]
|
||
return base
|
||
|
||
|
||
def _output_name(original_filename, suffix):
|
||
return f"{_base_name(original_filename)}_{suffix}.xlsx"
|
||
|
||
|
||
def _same_name(original_filename):
|
||
"""업로드한 파일과 같은 이름으로 내려준다. (저장 형식이 xlsx이므로 확장자만 맞춤)"""
|
||
return f"{_base_name(original_filename)}.xlsx"
|
||
|
||
|
||
# =====================================================================
|
||
# 발주서 스캔 (read_only 1회 통과 + 해시 캐시)
|
||
# =====================================================================
|
||
def _cache_get(key):
|
||
entry = _scan_cache.get(key)
|
||
if entry is None:
|
||
return None
|
||
saved_at, scan = entry
|
||
if time.monotonic() - saved_at > _SCAN_CACHE_TTL_SECONDS:
|
||
_scan_cache.pop(key, None)
|
||
return None
|
||
_scan_cache.move_to_end(key)
|
||
return scan
|
||
|
||
|
||
def _cache_put(key, scan):
|
||
_scan_cache[key] = (time.monotonic(), scan)
|
||
_scan_cache.move_to_end(key)
|
||
while len(_scan_cache) > _SCAN_CACHE_MAX:
|
||
_scan_cache.popitem(last=False)
|
||
|
||
|
||
def _collect_row(row_index, values, orders, sequence):
|
||
"""한 행을 주문 묶음에 반영. 내용이 있는 행이면 True."""
|
||
if not any(v is not None and str(v).strip() != "" for v in values):
|
||
return False
|
||
|
||
order_no = cell_text(_value_at(values, COL_ORDER_NO))
|
||
segment = first_segment(_value_at(values, COL_ORDER_LIST))
|
||
if not order_no and not segment:
|
||
return True # 합계행 등: 데이터 범위에는 포함하되 주문으로 세지 않는다
|
||
|
||
key = order_no or f"__행{row_index}"
|
||
info = orders.get(key)
|
||
if info is None:
|
||
info = {
|
||
"key": key,
|
||
"order_no": order_no,
|
||
"first_row": row_index,
|
||
"last_row": row_index,
|
||
"order_date": parse_order_date(_value_at(values, COL_ORDER_DATE)),
|
||
"order_date_text": cell_text(_value_at(values, COL_ORDER_DATE)),
|
||
"seq_value": _value_at(values, COL_SEQ), # 마지막 행 값으로 계속 덮어씀
|
||
"segments": [],
|
||
"rows": [],
|
||
}
|
||
orders[key] = info
|
||
sequence.append(key)
|
||
else:
|
||
info["last_row"] = row_index
|
||
info["seq_value"] = _value_at(values, COL_SEQ)
|
||
if info["order_date"] is None:
|
||
parsed = parse_order_date(_value_at(values, COL_ORDER_DATE))
|
||
if parsed is not None:
|
||
info["order_date"] = parsed
|
||
info["order_date_text"] = cell_text(_value_at(values, COL_ORDER_DATE))
|
||
|
||
info["rows"].append(row_index)
|
||
if segment and segment not in info["segments"]:
|
||
info["segments"].append(segment)
|
||
return True
|
||
|
||
|
||
def scan_order_sheet(file_bytes, filename):
|
||
"""발주서를 값만 읽어 주문 단위로 묶는다. 같은 파일은 캐시에서 즉시 반환."""
|
||
cache_key = "first-come:" + hashlib.sha1(file_bytes).hexdigest()
|
||
cached = _cache_get(cache_key)
|
||
if cached is not None:
|
||
return cached
|
||
|
||
_validate_extension(filename)
|
||
try:
|
||
workbook = load_workbook(io.BytesIO(file_bytes), read_only=True, data_only=True)
|
||
except Exception as exc:
|
||
raise HTTPException(status_code=400, detail=f"엑셀 파일을 열 수 없습니다: {exc}")
|
||
|
||
try:
|
||
sheet = workbook.active
|
||
sheet_name = sheet.title
|
||
rows_iter = sheet.iter_rows(min_col=1, max_col=COL_LAST, values_only=True)
|
||
|
||
# 헤더 후보 구간만 먼저 버퍼링해서 헤더 행을 정한다
|
||
buffered = []
|
||
header_row = 1
|
||
for index, values in enumerate(rows_iter, start=1):
|
||
buffered.append((index, values))
|
||
if _is_header_values(values):
|
||
header_row = index
|
||
break
|
||
if index >= _HEADER_SCAN_ROWS:
|
||
break
|
||
|
||
orders = {}
|
||
sequence = []
|
||
last_row = header_row
|
||
|
||
for index, values in buffered:
|
||
if index > header_row and _collect_row(index, values, orders, sequence):
|
||
last_row = index
|
||
for index, values in enumerate(rows_iter, start=len(buffered) + 1):
|
||
if _collect_row(index, values, orders, sequence):
|
||
last_row = index
|
||
finally:
|
||
workbook.close()
|
||
|
||
if last_row <= header_row:
|
||
raise HTTPException(status_code=400, detail="발주 데이터가 없는 파일입니다. 파일을 확인해주세요.")
|
||
|
||
counts = {}
|
||
for key in sequence:
|
||
for segment in orders[key]["segments"]:
|
||
counts[segment] = counts.get(segment, 0) + 1
|
||
|
||
scan = {
|
||
"sheet_name": sheet_name,
|
||
"header_row": header_row,
|
||
"last_row": last_row,
|
||
"orders": orders,
|
||
"sequence": sequence,
|
||
"items": [{"name": name, "order_count": count} for name, count in sorted(counts.items())],
|
||
}
|
||
_cache_put(cache_key, scan)
|
||
return scan
|
||
|
||
|
||
# =====================================================================
|
||
# 기능 1) 선착순 사은품 추가
|
||
# =====================================================================
|
||
def _first_come_sort_key(info):
|
||
"""주문날짜 오름차순. 날짜를 못 읽은 주문은 뒤로 보내고 파일 등장 순서를 따른다."""
|
||
order_date = info["order_date"]
|
||
if order_date is None:
|
||
return (1, 0.0, info["first_row"])
|
||
return (0, order_date.timestamp(), info["first_row"])
|
||
|
||
|
||
def _first_come_plan(scan, selected_names, limit):
|
||
"""선착순 대상 주문과 사은품 행이 들어갈 위치를 계산한다. (파일을 열지 않는다)
|
||
|
||
사은품 행은 그 주문의 마지막 행 바로 아래에 삽입되므로,
|
||
위쪽에 먼저 삽입된 행 수만큼 아래 행들의 최종 번호가 밀린다.
|
||
"""
|
||
selected = set(selected_names)
|
||
matched = []
|
||
for key in scan["sequence"]:
|
||
info = scan["orders"][key]
|
||
hits = [s for s in info["segments"] if s in selected]
|
||
if hits:
|
||
matched.append((info, hits))
|
||
|
||
matched.sort(key=lambda pair: _first_come_sort_key(pair[0]))
|
||
picked = matched[:limit]
|
||
|
||
# 삽입 위치(원본 좌표) 오름차순 = 위에서 아래 순서
|
||
by_position = sorted(picked, key=lambda pair: pair[0]["last_row"])
|
||
position_of = {pair[0]["key"]: i for i, pair in enumerate(by_position)}
|
||
insert_rows_sorted = [pair[0]["last_row"] for pair in by_position]
|
||
|
||
def shifted(original_row):
|
||
"""앞서 삽입된 행 수를 반영한 최종 행 번호."""
|
||
return original_row + bisect_left(insert_rows_sorted, original_row)
|
||
|
||
plan = []
|
||
for info, hits in picked: # 선착순(날짜) 순서 유지
|
||
rank = position_of[info["key"]]
|
||
new_seq, seq_ok = increment_trailing_number(info["seq_value"])
|
||
plan.append({
|
||
"key": info["key"],
|
||
"order_no": info["order_no"],
|
||
"order_date": info["order_date_text"],
|
||
"source_row": info["last_row"],
|
||
"new_row": info["last_row"] + 1 + rank,
|
||
"seq": cell_text(new_seq),
|
||
"seq_ok": seq_ok,
|
||
"rows": info["rows"],
|
||
"matched": hits,
|
||
})
|
||
|
||
gift_order_rows = set()
|
||
for entry in plan:
|
||
for row in entry["rows"]:
|
||
gift_order_rows.add(shifted(row))
|
||
gift_order_rows.add(entry["new_row"])
|
||
|
||
return {
|
||
"matched_order_count": len(matched),
|
||
"plan": plan,
|
||
"gift_order_rows": sorted(gift_order_rows),
|
||
"final_last_row": scan["last_row"] + len(plan),
|
||
}
|
||
|
||
|
||
def _first_come_warnings(result, limit):
|
||
warnings = []
|
||
matched_total = result["matched_order_count"]
|
||
if matched_total == 0:
|
||
warnings.append("선택한 주문목록에 해당하는 주문을 찾지 못했습니다.")
|
||
elif matched_total < limit:
|
||
warnings.append(
|
||
f"조건에 맞는 주문이 {matched_total}건뿐이라 선착순 {limit}명을 채우지 못했습니다."
|
||
)
|
||
|
||
no_seq = [entry["order_no"] for entry in result["plan"] if not entry["seq_ok"]]
|
||
if no_seq:
|
||
preview = ", ".join(no_seq[:3])
|
||
more = f" 외 {len(no_seq) - 3}건" if len(no_seq) > 3 else ""
|
||
warnings.append(f"C열에 끝 숫자가 없어 순번을 올리지 못한 주문이 있습니다: {preview}{more}")
|
||
return warnings
|
||
|
||
|
||
def _bulk_shift_rows(sheet, insert_points):
|
||
"""insert_points(원본 행 번호)들의 '바로 아래'에 빈 행 한 줄씩을 한 번에 만든다.
|
||
|
||
openpyxl의 insert_rows는 호출할 때마다 아래쪽 셀 전체를 리스트로 만들고 정렬하므로
|
||
(Worksheet._move_cells), 삽입 지점마다 부르면 O(삽입건수 × 전체셀)이 되어 매우 느리다.
|
||
삽입 지점을 모아 한 번만 훑으면서 각 셀의 최종 행 번호를 계산해 옮긴다.
|
||
행 높이(row_dimensions)는 insert_rows도 옮겨주지 않으므로 여기서 같이 처리한다.
|
||
"""
|
||
points = sorted(insert_points)
|
||
if not points:
|
||
return
|
||
|
||
def shift_of(row):
|
||
return bisect_left(points, row) # 이 행보다 위에서 삽입된 행 수
|
||
|
||
moved_cells = {}
|
||
for (row, column), cell in sheet._cells.items():
|
||
shift = shift_of(row)
|
||
if shift:
|
||
row += shift
|
||
cell.row = row
|
||
moved_cells[(row, column)] = cell
|
||
sheet._cells = moved_cells
|
||
sheet._current_row = max((r for r, _c in moved_cells), default=0)
|
||
|
||
# 행 높이도 같은 규칙으로 이동 (없으면 건너뜀)
|
||
dimensions = sheet.row_dimensions
|
||
saved = [
|
||
(row, dim.height, dim.hidden, dim.customHeight)
|
||
for row, dim in dimensions.items()
|
||
]
|
||
if saved:
|
||
dimensions.clear()
|
||
for row, height, hidden, custom_height in saved:
|
||
new_row = row + shift_of(row)
|
||
dim = dimensions[new_row]
|
||
dim.height = height
|
||
dim.hidden = hidden
|
||
dim.customHeight = custom_height
|
||
# 새로 생긴 행은 바로 위(복사 원본) 행의 높이를 따라간다
|
||
for rank, point in enumerate(points):
|
||
source_row = point + rank
|
||
if source_row in dimensions and dimensions[source_row].height is not None:
|
||
target = dimensions[source_row + 1]
|
||
target.height = dimensions[source_row].height
|
||
target.customHeight = dimensions[source_row].customHeight
|
||
|
||
|
||
def _first_come_write(sheet, plan_result, gift):
|
||
"""계획대로 각 주문의 마지막 행 아래에 사은품 행을 끼워 넣는다."""
|
||
plan = plan_result["plan"]
|
||
if not plan:
|
||
return
|
||
|
||
# 먼저 필요한 빈 행을 한 번에 확보한다. 이후 각 사은품 행의 위치는
|
||
# 계획 단계에서 계산해 둔 new_row와 정확히 일치한다 (원본 행은 new_row - 1).
|
||
_bulk_shift_rows(sheet, [entry["source_row"] for entry in plan])
|
||
|
||
for entry in plan:
|
||
target_row = entry["new_row"]
|
||
source_row = target_row - 1
|
||
|
||
for col in range(1, COL_LAST + 1):
|
||
source_cell = sheet.cell(row=source_row, column=col)
|
||
target_cell = sheet.cell(row=target_row, column=col)
|
||
target_cell.value = source_cell.value
|
||
if source_cell.has_style:
|
||
target_cell._style = copy(source_cell._style)
|
||
|
||
seq_value, _seq_ok = increment_trailing_number(sheet.cell(row=source_row, column=COL_SEQ).value)
|
||
sheet.cell(row=target_row, column=COL_SEQ).value = seq_value
|
||
|
||
sheet.cell(row=target_row, column=COL_ITEM_CODE).value = gift["item_code"]
|
||
sheet.cell(row=target_row, column=COL_ITEM_NAME).value = gift["name"]
|
||
sheet.cell(row=target_row, column=COL_QTY).value = gift["qty"]
|
||
sheet.cell(row=target_row, column=COL_ORDER_LIST).value = " / ".join(entry["matched"] + [GIFT_LABEL])
|
||
|
||
for col in range(1, COL_LAST + 1):
|
||
sheet.cell(row=target_row, column=col).fill = YELLOW_FILL
|
||
|
||
|
||
def _prune_to_rows(sheet, header_row, keep_rows, last_row):
|
||
"""헤더와 keep_rows만 남기고 나머지 데이터 행을 지운다."""
|
||
keep = set(keep_rows)
|
||
to_delete = [row for row in range(header_row + 1, last_row + 1) if row not in keep]
|
||
if not to_delete:
|
||
return
|
||
|
||
groups = []
|
||
for row in to_delete:
|
||
if groups and groups[-1][0] + groups[-1][1] == row:
|
||
groups[-1][1] += 1
|
||
else:
|
||
groups.append([row, 1])
|
||
|
||
for start, count in reversed(groups):
|
||
sheet.delete_rows(start, count)
|
||
|
||
|
||
def _parse_selected(selected_json, empty_detail="상품을 최소 1개 선택해주세요."):
|
||
try:
|
||
parsed = json.loads(selected_json or "[]")
|
||
except json.JSONDecodeError:
|
||
raise HTTPException(status_code=400, detail="선택한 목록 값을 읽지 못했습니다.")
|
||
if not isinstance(parsed, list):
|
||
raise HTTPException(status_code=400, detail="선택한 목록 형식이 올바르지 않습니다.")
|
||
|
||
names = [str(v).strip() for v in parsed if str(v).strip()]
|
||
if not names:
|
||
raise HTTPException(status_code=400, detail=empty_detail)
|
||
return names
|
||
|
||
|
||
def _validate_gift(gift_item_code, gift_name, gift_qty, limit):
|
||
if limit < 1:
|
||
raise HTTPException(status_code=400, detail="선착순 인원은 1명 이상이어야 합니다.")
|
||
if gift_qty < 1:
|
||
raise HTTPException(status_code=400, detail="사은품 수량은 1개 이상이어야 합니다.")
|
||
code = (gift_item_code or "").strip()
|
||
if not code:
|
||
raise HTTPException(status_code=400, detail="사은품(세트 코드)을 선택해주세요.")
|
||
return {"item_code": code, "name": (gift_name or "").strip(), "qty": gift_qty}
|
||
|
||
|
||
@router.post("/first-come/analyze")
|
||
async def first_come_analyze(file: UploadFile = File(...)):
|
||
"""업로드된 발주서에서 주문목록(Q열) 후보를 뽑아 가나다순으로 돌려준다."""
|
||
file_bytes = await file.read()
|
||
scan = scan_order_sheet(file_bytes, file.filename)
|
||
return {
|
||
"status": "success",
|
||
"file_name": file.filename,
|
||
"sheet_name": scan["sheet_name"],
|
||
"header_row": scan["header_row"],
|
||
"total_rows": scan["last_row"] - scan["header_row"],
|
||
"total_orders": len(scan["sequence"]),
|
||
"order_list_items": scan["items"],
|
||
}
|
||
|
||
|
||
@router.post("/first-come/preview")
|
||
async def first_come_preview(
|
||
file: UploadFile = File(...),
|
||
selected_json: str = Form(...),
|
||
limit: int = Form(...),
|
||
gift_item_code: str = Form(...),
|
||
gift_name: str = Form(""),
|
||
gift_qty: int = Form(1),
|
||
):
|
||
"""다운로드 전에 어떤 주문에 사은품이 붙는지 미리 보여준다. (파일을 다시 쓰지 않는다)"""
|
||
selected_names = _parse_selected(selected_json, "사은품이 지급되는 주문을 최소 1개 선택해주세요.")
|
||
_validate_gift(gift_item_code, gift_name, gift_qty, limit)
|
||
|
||
file_bytes = await file.read()
|
||
scan = scan_order_sheet(file_bytes, file.filename)
|
||
result = _first_come_plan(scan, selected_names, limit)
|
||
|
||
added = [
|
||
{
|
||
"order_no": entry["order_no"],
|
||
"order_date": entry["order_date"],
|
||
"source_row": entry["source_row"],
|
||
"new_row": entry["new_row"],
|
||
"seq": entry["seq"],
|
||
"matched": entry["matched"],
|
||
}
|
||
for entry in result["plan"]
|
||
]
|
||
|
||
return {
|
||
"status": "success",
|
||
"matched_order_count": result["matched_order_count"],
|
||
"applied_count": len(result["plan"]),
|
||
"added": added,
|
||
"warnings": _first_come_warnings(result, limit),
|
||
}
|
||
|
||
|
||
@router.post("/first-come/download")
|
||
async def first_come_download(
|
||
file: UploadFile = File(...),
|
||
selected_json: str = Form(...),
|
||
limit: int = Form(...),
|
||
gift_item_code: str = Form(...),
|
||
gift_name: str = Form(""),
|
||
gift_qty: int = Form(1),
|
||
mode: str = Form("full"),
|
||
):
|
||
"""가공된 발주 파일을 내려준다.
|
||
|
||
mode=full : 원본 전체 + 각 주문 아래에 끼워 넣은 사은품 행
|
||
mode=gift : 사은품이 적용된 주문의 모든 행 + 사은품 행
|
||
"""
|
||
selected_names = _parse_selected(selected_json, "사은품이 지급되는 주문을 최소 1개 선택해주세요.")
|
||
gift = _validate_gift(gift_item_code, gift_name, gift_qty, limit)
|
||
|
||
file_bytes = await file.read()
|
||
scan = scan_order_sheet(file_bytes, file.filename)
|
||
result = _first_come_plan(scan, selected_names, limit)
|
||
|
||
if not result["plan"]:
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail="조건에 맞는 주문이 없어 사은품 행을 추가하지 못했습니다. 선택 항목을 확인해주세요.",
|
||
)
|
||
|
||
try:
|
||
workbook = load_workbook(io.BytesIO(file_bytes))
|
||
except Exception as exc:
|
||
raise HTTPException(status_code=400, detail=f"엑셀 파일을 열 수 없습니다: {exc}")
|
||
sheet = workbook.active
|
||
|
||
_first_come_write(sheet, result, gift)
|
||
|
||
if mode == "gift":
|
||
_prune_to_rows(sheet, scan["header_row"], result["gift_order_rows"], result["final_last_row"])
|
||
filename = _output_name(file.filename, "선착순사은품_대상주문")
|
||
else:
|
||
# 전체 발주 파일은 업로드한 파일 이름 그대로 내려준다
|
||
filename = _same_name(file.filename)
|
||
|
||
return excel_response(workbook, filename)
|
||
|
||
|
||
# =====================================================================
|
||
# 기능 2) 행사 주문 추출
|
||
# 주문 CSV를 올려 K열(주문상품명)로 상품을 고른 뒤,
|
||
# - product 모드: 그 상품이 담긴 행만
|
||
# - order 모드 : 그 상품을 산 주문번호(B열)의 모든 행
|
||
# 만 남긴 엑셀을 내려준다. N열에는 =L×M 수식을 넣고 노란색을 칠한다.
|
||
# =====================================================================
|
||
_CSV_ENCODINGS = ("utf-8-sig", "cp949", "utf-8")
|
||
|
||
|
||
def _decode_csv(file_bytes):
|
||
"""카페24 CSV는 UTF-8(BOM), CP949(EUC-KR), UTF-16 중 하나로 내려온다.
|
||
|
||
UTF-16 파일을 CP949로 잘못 읽으면 글자 사이에 NUL이 끼어 csv 파서가 죽으므로
|
||
BOM과 NUL 비율로 UTF-16을 먼저 걸러낸다.
|
||
"""
|
||
if file_bytes[:2] in (b"\xff\xfe", b"\xfe\xff"):
|
||
try:
|
||
return file_bytes.decode("utf-16"), "utf-16"
|
||
except UnicodeDecodeError:
|
||
pass
|
||
|
||
# BOM이 없는 UTF-16은 NUL 바이트가 많다는 특징으로 판별한다
|
||
head = file_bytes[:4096]
|
||
if head and head.count(0) > len(head) * 0.2:
|
||
for encoding in ("utf-16-le", "utf-16-be"):
|
||
try:
|
||
return file_bytes.decode(encoding), encoding
|
||
except UnicodeDecodeError:
|
||
continue
|
||
|
||
for encoding in _CSV_ENCODINGS:
|
||
try:
|
||
return file_bytes.decode(encoding), encoding
|
||
except UnicodeDecodeError:
|
||
continue
|
||
return file_bytes.decode("utf-8", errors="replace"), "utf-8(일부 깨짐)"
|
||
|
||
|
||
def _normalize_csv_text(text):
|
||
"""줄바꿈을 \\n으로 통일하고 NUL을 제거한다.
|
||
|
||
CR(\\r)만 쓰는 파일이나 UTF-16 잔재가 남으면 csv 파서가
|
||
'new-line character seen in unquoted field' 오류로 죽는다.
|
||
"""
|
||
if "\x00" in text:
|
||
text = text.replace("\x00", "")
|
||
if "\r" in text:
|
||
text = text.replace("\r\n", "\n").replace("\r", "\n")
|
||
return text.lstrip("")
|
||
|
||
|
||
def _sniff_delimiter(sample_line):
|
||
"""구분자 추정. 카페24는 쉼표지만 탭으로 받는 경우도 있어 대비한다."""
|
||
counts = {d: sample_line.count(d) for d in (",", "\t", ";", "|")}
|
||
best = max(counts, key=counts.get)
|
||
return best if counts[best] > 0 else ","
|
||
|
||
|
||
def _cell_at(row, column):
|
||
"""1-based 열 번호로 CSV 행에서 값을 꺼낸다. 없으면 빈 문자열."""
|
||
return row[column - 1].strip() if len(row) >= column else ""
|
||
|
||
|
||
def event_extract_product_name(product_value):
|
||
"""주문상품명에서 첫 번째 ' -' 앞부분만 잘라낸다."""
|
||
text = (product_value or "").strip()
|
||
if not text:
|
||
return ""
|
||
return text.split(EE_SPLIT_TOKEN)[0].strip()
|
||
|
||
|
||
def _to_number(value):
|
||
"""수식이 계산되도록 L·M 값을 숫자로 바꾼다. 숫자가 아니면 원본 그대로."""
|
||
text = (value or "").strip().replace(",", "")
|
||
if not text:
|
||
return value
|
||
try:
|
||
return int(text)
|
||
except ValueError:
|
||
pass
|
||
try:
|
||
return float(text)
|
||
except ValueError:
|
||
return value
|
||
|
||
|
||
def scan_event_csv(file_bytes, filename):
|
||
"""주문 CSV를 읽어 상품명 후보와 행/주문 정보를 뽑는다. 같은 파일은 캐시 사용.
|
||
|
||
예상 못한 오류도 화면에 이유가 보이도록 400으로 감싸서 돌려준다.
|
||
(그냥 두면 500이 되어 사용자에게는 원인 없는 실패로만 보인다)
|
||
"""
|
||
cache_key = "event-extract:" + hashlib.sha1(file_bytes).hexdigest()
|
||
cached = _cache_get(cache_key)
|
||
if cached is not None:
|
||
return cached
|
||
|
||
name = (filename or "").lower()
|
||
if not name.endswith(".csv"):
|
||
raise HTTPException(status_code=400, detail="CSV 파일(.csv)만 업로드할 수 있습니다.")
|
||
|
||
try:
|
||
return _scan_event_csv_uncached(file_bytes, cache_key)
|
||
except HTTPException:
|
||
raise
|
||
except Exception as exc:
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail=f"CSV를 분석하지 못했습니다 ({type(exc).__name__}: {exc}). "
|
||
"파일이 손상되었거나 예상과 다른 형식일 수 있습니다.",
|
||
)
|
||
|
||
|
||
def _scan_event_csv_uncached(file_bytes, cache_key):
|
||
text, encoding = _decode_csv(file_bytes)
|
||
text = _normalize_csv_text(text)
|
||
if not text.strip():
|
||
raise HTTPException(status_code=400, detail="내용이 비어 있는 파일입니다.")
|
||
|
||
lines = text.splitlines()
|
||
first_line = lines[0] if lines else ""
|
||
delimiter = _sniff_delimiter(first_line)
|
||
try:
|
||
all_rows = list(csv.reader(io.StringIO(text, newline=""), delimiter=delimiter))
|
||
except csv.Error as exc:
|
||
raise HTTPException(status_code=400, detail=f"CSV 형식을 읽지 못했습니다: {exc}")
|
||
if not all_rows:
|
||
raise HTTPException(status_code=400, detail="CSV에서 읽어낼 행이 없습니다.")
|
||
|
||
# 헤더 행: K열에 '주문상품명'이 들어간 줄. 못 찾으면 첫 줄을 헤더로 본다.
|
||
header_index = 0
|
||
for index, row in enumerate(all_rows[:_HEADER_SCAN_ROWS]):
|
||
if "주문상품명" in _cell_at(row, EE_COL_PRODUCT):
|
||
header_index = index
|
||
break
|
||
|
||
header = all_rows[header_index]
|
||
data_rows = [row for row in all_rows[header_index + 1:] if any(c.strip() for c in row)]
|
||
if not data_rows:
|
||
raise HTTPException(status_code=400, detail="헤더 아래에 주문 데이터가 없습니다.")
|
||
|
||
stats = {}
|
||
meta = []
|
||
orders_of_product = {}
|
||
for row in data_rows:
|
||
order_no = _cell_at(row, EE_COL_ORDER_NO)
|
||
product = event_extract_product_name(_cell_at(row, EE_COL_PRODUCT))
|
||
meta.append({"order_no": order_no, "product": product})
|
||
if not product:
|
||
continue
|
||
entry = stats.setdefault(product, {"name": product, "row_count": 0, "orders": set()})
|
||
entry["row_count"] += 1
|
||
if order_no:
|
||
entry["orders"].add(order_no)
|
||
orders_of_product.setdefault(product, set())
|
||
if order_no:
|
||
orders_of_product[product].add(order_no)
|
||
|
||
items = [
|
||
{"name": e["name"], "row_count": e["row_count"], "order_count": len(e["orders"])}
|
||
for e in sorted(stats.values(), key=lambda x: x["name"])
|
||
]
|
||
|
||
column_count = max([len(header)] + [len(r) for r in data_rows])
|
||
|
||
scan = {
|
||
"encoding": encoding,
|
||
"delimiter": delimiter,
|
||
"header_row": header_index + 1,
|
||
"header": header,
|
||
"rows": data_rows,
|
||
"meta": meta,
|
||
"items": items,
|
||
"orders_of_product": orders_of_product,
|
||
"column_count": column_count,
|
||
}
|
||
_cache_put(cache_key, scan)
|
||
return scan
|
||
|
||
|
||
def _event_extract_select(scan, selected_names, mode):
|
||
"""조건에 맞는 행 번호(데이터 행 기준 0-based)를 고른다."""
|
||
selected = set(selected_names)
|
||
matched_rows = [i for i, m in enumerate(scan["meta"]) if m["product"] in selected]
|
||
|
||
if mode == "order":
|
||
# 해당 상품을 산 주문번호(B열)와 같은 주문번호를 가진 모든 행을 남긴다
|
||
target_orders = {scan["meta"][i]["order_no"] for i in matched_rows if scan["meta"][i]["order_no"]}
|
||
keep = [i for i, m in enumerate(scan["meta"]) if m["order_no"] and m["order_no"] in target_orders]
|
||
else:
|
||
keep = matched_rows
|
||
|
||
order_numbers = {scan["meta"][i]["order_no"] for i in keep if scan["meta"][i]["order_no"]}
|
||
return {
|
||
"keep": keep,
|
||
"matched_row_count": len(matched_rows),
|
||
"row_count": len(keep),
|
||
"order_count": len(order_numbers),
|
||
}
|
||
|
||
|
||
def _is_number(value):
|
||
return isinstance(value, (int, float)) and not isinstance(value, bool)
|
||
|
||
|
||
def _pixels_to_column_width(pixels):
|
||
"""픽셀을 엑셀 열 너비 단위로 환산.
|
||
|
||
엑셀의 열 너비는 픽셀이 아니라 '기본 글꼴의 숫자 한 글자 폭' 단위라서
|
||
글꼴에 따라 배율이 달라진다. 이 프로그램이 쓰는 환경 기준으로
|
||
145px = 너비 17.5 이므로 1글자 = 8px, 좌우 여백 5px로 계산한다.
|
||
"""
|
||
return round(max(pixels - EE_WIDTH_PADDING_PIXELS, 0) / EE_PIXELS_PER_WIDTH_UNIT, 3)
|
||
|
||
|
||
def _rename_header(value):
|
||
"""장황한 카페24 제목을 짧은 제목으로 바꾼다. 목록에 없으면 원본 유지."""
|
||
text = (value or "").strip()
|
||
if not text:
|
||
return value
|
||
if text in EE_HEADER_RENAMES:
|
||
return EE_HEADER_RENAMES[text]
|
||
return _EE_HEADER_RENAMES_LOOSE.get(re.sub(r"\s+", "", text), text)
|
||
|
||
|
||
def _display_width(value):
|
||
"""열 너비 자동 맞춤용 글자 폭 추정. 한글은 영문보다 넓게 잡는다."""
|
||
if value is None or value == "":
|
||
return 0
|
||
if _is_number(value):
|
||
text = f"{value:,.0f}" if float(value).is_integer() else f"{value:,.2f}"
|
||
else:
|
||
text = str(value)
|
||
return sum(1.8 if ord(ch) > 127 else 1.1 for ch in text)
|
||
|
||
|
||
def _event_extract_build_workbook(scan, keep_indexes):
|
||
"""고른 행만 담아 서식까지 적용한 워크북을 만든다.
|
||
|
||
정렬 : D열 기준 텍스트 오름차순 (같은 값끼리는 원래 순서 유지)
|
||
A~AB : 1행은 가운데 정렬·파랑 배경·흰 글자, 자동 필터, A2 틀고정
|
||
열 너비는 A~H·L~T는 픽셀 고정, 나머지는 내용에 맞춰 자동
|
||
L~S : 문자로 들어온 숫자를 숫자로 바꾸고 쉼표 서식, 값이 0이면 빈 셀
|
||
O~S : 같은 주문번호(B열) 안에서 첫 행과 같은 값이면 지운다
|
||
배경 : 주문번호가 바뀔 때마다 #B4C6E7 / #FFD966 을 번갈아 칠하고,
|
||
마지막에 N열(=L×M 수식이 들어간 셀)만 노란색으로 덮어쓴다
|
||
N : =L×M 수식(계산값 아님)
|
||
"""
|
||
workbook = Workbook()
|
||
sheet = workbook.active
|
||
sheet.title = "행사 주문 추출"
|
||
|
||
column_count = max(scan["column_count"], EE_COL_N)
|
||
format_last = min(column_count, EE_FORMAT_LAST_COL)
|
||
widths = {}
|
||
|
||
def note_width(col, value):
|
||
width = _display_width(value)
|
||
if width > widths.get(col, 0):
|
||
widths[col] = width
|
||
|
||
# D열 텍스트 오름차순. 파이썬 정렬은 안정적이라 같은 값끼리는 원래 순서가 유지된다.
|
||
keep_indexes = sorted(
|
||
keep_indexes,
|
||
key=lambda index: _cell_at(scan["rows"][index], EE_SORT_COL),
|
||
)
|
||
|
||
# ---- 제목행 ----
|
||
for col in range(1, column_count + 1):
|
||
value = _rename_header(_cell_at(scan["header"], col)) or None
|
||
cell = sheet.cell(row=1, column=col, value=value)
|
||
if col <= format_last:
|
||
cell.fill = EE_HEADER_FILL
|
||
cell.font = EE_HEADER_FONT
|
||
cell.alignment = EE_HEADER_ALIGN
|
||
note_width(col, value)
|
||
|
||
# ---- 데이터 행 ----
|
||
first_seen = {} # (주문번호, 열) -> 그 주문에서 처음 만난 값
|
||
row_orders = [] # (엑셀 행번호, 주문번호) - 아래 배경색 칠하기에 쓴다
|
||
for offset, data_index in enumerate(keep_indexes):
|
||
excel_row = offset + 2
|
||
source = scan["rows"][data_index]
|
||
order_no = _cell_at(source, EE_COL_ORDER_NO)
|
||
row_orders.append((excel_row, order_no))
|
||
|
||
for col in range(1, column_count + 1):
|
||
if col == EE_COL_N:
|
||
continue # 아래에서 수식으로 채운다
|
||
|
||
raw = _cell_at(source, col)
|
||
|
||
# O~S: 같은 주문번호에서 이미 나온 값과 같으면 빈 셀로 둔다
|
||
if order_no and EE_DEDUPE_FIRST_COL <= col <= EE_DEDUPE_LAST_COL:
|
||
key = (order_no, col)
|
||
if key in first_seen:
|
||
if first_seen[key] == raw:
|
||
continue
|
||
elif raw != "":
|
||
first_seen[key] = raw
|
||
|
||
# L~S: 숫자로 변환 + 쉼표 서식, 0이면 빈 셀
|
||
if EE_NUM_FIRST_COL <= col <= EE_NUM_LAST_COL:
|
||
value = _to_number(raw)
|
||
if _is_number(value):
|
||
if value == 0:
|
||
continue
|
||
cell = sheet.cell(row=excel_row, column=col, value=value)
|
||
cell.number_format = (
|
||
EE_COMMA_FORMAT if float(value).is_integer() else EE_COMMA_FORMAT_DECIMAL
|
||
)
|
||
note_width(col, value)
|
||
continue
|
||
|
||
if raw != "":
|
||
sheet.cell(row=excel_row, column=col, value=raw)
|
||
note_width(col, raw)
|
||
|
||
# 계산된 값이 아니라 수식 자체를 그대로 넣는다 (배경색은 아래에서 한 번에)
|
||
cell = sheet.cell(row=excel_row, column=EE_COL_N)
|
||
cell.value = f"=L{excel_row}*M{excel_row}"
|
||
cell.number_format = EE_COMMA_FORMAT
|
||
# 수식 결과는 저장 시점에 알 수 없으므로 L×M을 미리 계산해 너비만 가늠한다
|
||
left = sheet.cell(row=excel_row, column=EE_COL_L).value
|
||
right = sheet.cell(row=excel_row, column=EE_COL_M).value
|
||
if _is_number(left) and _is_number(right):
|
||
note_width(EE_COL_N, left * right)
|
||
|
||
# ---- 주문번호 묶음별 배경색 (두 색을 번갈아) ----
|
||
previous_order = None
|
||
color_index = 0
|
||
for excel_row, order_no in row_orders:
|
||
if previous_order is not None and order_no != previous_order:
|
||
color_index = 1 - color_index # 주문번호가 바뀌면 다른 색으로
|
||
previous_order = order_no
|
||
fill = EE_GROUP_FILLS[color_index]
|
||
for col in range(1, format_last + 1):
|
||
sheet.cell(row=excel_row, column=col).fill = fill
|
||
|
||
# ---- 마지막으로 N열(수식이 들어간 셀)만 노란색으로 덮어쓴다 ----
|
||
for excel_row, _order_no in row_orders:
|
||
cell = sheet.cell(row=excel_row, column=EE_COL_N)
|
||
if cell.value not in (None, ""):
|
||
cell.fill = YELLOW_FILL
|
||
|
||
# ---- 열 너비: 지정한 열은 픽셀 고정, 나머지는 자동 ----
|
||
for col in range(1, format_last + 1):
|
||
pixels = EE_MANUAL_COL_PIXELS.get(col)
|
||
if pixels is not None:
|
||
width = _pixels_to_column_width(pixels)
|
||
else:
|
||
width = min(max(widths.get(col, 0) + 2, EE_MIN_COL_WIDTH), EE_MAX_COL_WIDTH)
|
||
sheet.column_dimensions[get_column_letter(col)].width = width
|
||
|
||
# ---- 자동 필터 + A2 틀고정 ----
|
||
last_letter = get_column_letter(format_last)
|
||
sheet.auto_filter.ref = f"A1:{last_letter}{sheet.max_row}"
|
||
if sheet.max_row > 1:
|
||
# 행은 이미 D열 오름차순으로 써 두었고, 필터에도 정렬 기준을 남겨 둔다
|
||
sort_letter = get_column_letter(EE_SORT_COL)
|
||
sheet.auto_filter.add_sort_condition(f"{sort_letter}2:{sort_letter}{sheet.max_row}")
|
||
sheet.freeze_panes = "A2"
|
||
|
||
return workbook
|
||
|
||
|
||
def _parse_extract_mode(mode):
|
||
if mode not in ("product", "order"):
|
||
raise HTTPException(status_code=400, detail="추출 조건이 올바르지 않습니다.")
|
||
return mode
|
||
|
||
|
||
@router.post("/event-extract/analyze")
|
||
async def event_extract_analyze(file: UploadFile = File(...)):
|
||
"""업로드된 주문 CSV에서 상품명 후보를 뽑아 가나다순으로 돌려준다."""
|
||
file_bytes = await file.read()
|
||
scan = scan_event_csv(file_bytes, file.filename)
|
||
|
||
warnings = []
|
||
if scan["column_count"] < EE_COL_N:
|
||
warnings.append(
|
||
f"CSV의 열이 {scan['column_count']}개뿐이라 L·M열 값이 비어 있을 수 있습니다. "
|
||
"N열 수식은 그대로 넣지만 결과가 0이 될 수 있습니다."
|
||
)
|
||
|
||
return {
|
||
"status": "success",
|
||
"file_name": file.filename,
|
||
"encoding": scan["encoding"],
|
||
"header_row": scan["header_row"],
|
||
"total_rows": len(scan["rows"]),
|
||
"total_orders": len({m["order_no"] for m in scan["meta"] if m["order_no"]}),
|
||
"column_count": scan["column_count"],
|
||
"product_items": scan["items"],
|
||
"warnings": warnings,
|
||
}
|
||
|
||
|
||
@router.post("/event-extract/download")
|
||
async def event_extract_download(
|
||
file: UploadFile = File(...),
|
||
selected_json: str = Form(...),
|
||
mode: str = Form("product"),
|
||
):
|
||
"""조건에 맞는 행만 남기고 N열에 =L×M 수식을 넣은 엑셀을 내려준다."""
|
||
selected_names = _parse_selected(selected_json)
|
||
_parse_extract_mode(mode)
|
||
|
||
file_bytes = await file.read()
|
||
scan = scan_event_csv(file_bytes, file.filename)
|
||
result = _event_extract_select(scan, selected_names, mode)
|
||
|
||
if not result["keep"]:
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail="조건에 맞는 주문이 없어 추출할 행이 없습니다. 선택 항목을 확인해주세요.",
|
||
)
|
||
|
||
workbook = _event_extract_build_workbook(scan, result["keep"])
|
||
suffix = "해당상품만" if mode == "product" else "해당주문"
|
||
# 화면에 추출 결과를 알려주기 위한 값. 파일과 함께 헤더로 실어 보낸다.
|
||
summary = {
|
||
"X-Extract-Matched-Rows": str(result["matched_row_count"]),
|
||
"X-Extract-Rows": str(result["row_count"]),
|
||
"X-Extract-Orders": str(result["order_count"]),
|
||
"X-Extract-Total-Rows": str(len(scan["rows"])),
|
||
}
|
||
return excel_response(workbook, _output_name(file.filename, f"행사추출_{suffix}"), summary)
|