king
7afe90646b
수동 일반 발주의 이름·연락처·주소를 문자 이력에 함께 저장하고 최근 전송 메시지 클릭 시 다시 불러오도록 개선 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
1811 lines
70 KiB
Python
1811 lines
70 KiB
Python
import os
|
|
import sys
|
|
import psycopg2
|
|
import psycopg2.extras
|
|
import time
|
|
import json
|
|
import re
|
|
import io
|
|
import math
|
|
import datetime
|
|
import configparser
|
|
import threading
|
|
import urllib.request
|
|
import urllib.parse
|
|
import uuid
|
|
from urllib.parse import quote
|
|
from dotenv import load_dotenv
|
|
from fastapi import FastAPI, HTTPException, Request, Body, UploadFile, File, Form, Query, BackgroundTasks
|
|
from fastapi.responses import HTMLResponse, JSONResponse, StreamingResponse, FileResponse, RedirectResponse
|
|
from starlette.middleware.base import BaseHTTPMiddleware
|
|
from starlette.middleware.sessions import SessionMiddleware
|
|
import pandas as pd
|
|
from naver_api import get_access_token, get_payment_done_orders, get_product_order_details, dispatch_orders, confirm_orders, get_product_details
|
|
import cafe24_api
|
|
from fastapi.staticfiles import StaticFiles
|
|
from fastapi.templating import Jinja2Templates
|
|
from pydantic import BaseModel
|
|
from typing import List, Optional, Dict, Any
|
|
|
|
# .env 로드
|
|
load_dotenv()
|
|
|
|
# ==========================================
|
|
# SSO / 서브경로 호스팅 설정
|
|
# ==========================================
|
|
APP_ROOT_PATH = os.getenv("APP_ROOT_PATH", "").rstrip("/")
|
|
MAIN_LOGIN_URL = os.getenv("MAIN_LOGIN_URL", "https://dbx.no1king.freeddns.org/login")
|
|
MAIN_LOGOUT_URL = os.getenv("MAIN_LOGOUT_URL", "https://dbx.no1king.freeddns.org/logout")
|
|
SESSION_USER_KEY = os.getenv("SESSION_USER_KEY", "user_email")
|
|
SESSION_NAME_KEY = os.getenv("SESSION_NAME_KEY", "user_name")
|
|
|
|
# 인증 우회 경로 — 정적 자원, 헬스, OAuth 콜백, 로그인/로그아웃 위임
|
|
# Starlette scope.path 는 root_path 가 제거된 형태로 들어오므로 prefix 없이 매칭.
|
|
_AUTH_EXEMPT_PREFIXES = (
|
|
"/static/",
|
|
"/health",
|
|
"/api/cafe24/callback",
|
|
"/admin/oauth/callback",
|
|
"/login",
|
|
"/logout",
|
|
"/favicon.ico",
|
|
)
|
|
|
|
|
|
def _require_session_secret() -> str:
|
|
secret = os.getenv("SESSION_SECRET_KEY", "").strip()
|
|
if not secret:
|
|
raise RuntimeError(
|
|
"SESSION_SECRET_KEY 환경변수가 설정되지 않았습니다. "
|
|
"openssl rand -hex 32 로 새 값을 만들어 .env 에 넣고 컨테이너를 재기동하세요. "
|
|
"dbx-main 과 SSO 하려면 양쪽 .env 에 같은 값이어야 합니다."
|
|
)
|
|
return secret
|
|
|
|
|
|
class AuthGuardMiddleware(BaseHTTPMiddleware):
|
|
"""세션에 user_email 이 없으면 차단.
|
|
|
|
- HTML 요청 (Accept: text/html 또는 / 같은 페이지) → dbx-main /login?next=... 302
|
|
- 그 외 (API 등) → 401 JSON
|
|
- _AUTH_EXEMPT_PREFIXES 에 해당하는 경로는 통과
|
|
"""
|
|
|
|
async def dispatch(self, request: Request, call_next):
|
|
# 환경별로 scope.path 가 root_path 를 포함할 수도, 안 할 수도 있어
|
|
# 항상 root_path 제외된 형태로 정규화한다.
|
|
raw = request.scope.get("path", "") or "/"
|
|
if APP_ROOT_PATH and raw.startswith(APP_ROOT_PATH + "/"):
|
|
path = raw[len(APP_ROOT_PATH):]
|
|
elif APP_ROOT_PATH and raw == APP_ROOT_PATH:
|
|
path = "/"
|
|
else:
|
|
path = raw
|
|
|
|
for prefix in _AUTH_EXEMPT_PREFIXES:
|
|
if path == prefix or path.startswith(prefix):
|
|
return await call_next(request)
|
|
|
|
# 서비스 간 호출 — X-Service-Token 헤더가 일치하면 세션 없이도 통과.
|
|
# NPM 은 외부에서 들어온 이 헤더를 빈 값으로 strip 해야 한다 (보안).
|
|
service_token = os.getenv("INTERNAL_SERVICE_TOKEN", "")
|
|
if service_token:
|
|
incoming_token = request.headers.get("x-service-token", "")
|
|
if incoming_token and incoming_token == service_token:
|
|
return await call_next(request)
|
|
|
|
email = request.session.get(SESSION_USER_KEY)
|
|
if email:
|
|
return await call_next(request)
|
|
|
|
# 미인증
|
|
accept = request.headers.get("accept", "")
|
|
wants_html = "text/html" in accept or not path.startswith("/api/")
|
|
if wants_html:
|
|
next_path = f"{APP_ROOT_PATH}{path}" if APP_ROOT_PATH else path
|
|
if request.scope.get("query_string"):
|
|
next_path = f"{next_path}?{request.scope['query_string'].decode('latin-1')}"
|
|
sep = "&" if "?" in MAIN_LOGIN_URL else "?"
|
|
target = f"{MAIN_LOGIN_URL}{sep}next={quote(next_path, safe='/')}"
|
|
return RedirectResponse(url=target, status_code=302)
|
|
return JSONResponse({"detail": "Not authenticated"}, status_code=401)
|
|
|
|
# ==========================================
|
|
# 데이터베이스 접속 설정 (환경변수 기반)
|
|
# ==========================================
|
|
_PG_HOST = os.getenv('POSTGRES_HOST', 'postgres-db')
|
|
_PG_PORT = int(os.getenv('POSTGRES_PORT', '5432'))
|
|
_PG_USER = os.getenv('POSTGRES_USER', 'king')
|
|
_PG_PASSWORD = os.getenv('POSTGRES_PASSWORD', '')
|
|
|
|
# 1. 메인 프로그램용 DB (품목코드)
|
|
DB_CONFIG = {
|
|
'host': _PG_HOST,
|
|
'port': _PG_PORT,
|
|
'dbname': os.getenv('ITEMCODE_DB', 'itemcode_db'),
|
|
'user': _PG_USER,
|
|
'password': _PG_PASSWORD,
|
|
}
|
|
|
|
# 2. 반품 데이터 저장용 DB
|
|
RETURN_DB_CONFIG = {
|
|
'host': _PG_HOST,
|
|
'port': _PG_PORT,
|
|
'dbname': os.getenv('RETURN_DB', 'return_db'),
|
|
'user': _PG_USER,
|
|
'password': _PG_PASSWORD,
|
|
}
|
|
|
|
# 3. 주문정보 조회용 DB
|
|
ORDER_DB_CONFIG = {
|
|
'host': _PG_HOST,
|
|
'port': _PG_PORT,
|
|
'dbname': os.getenv('ORDERLIST_DB', 'orderlist_db'),
|
|
'user': _PG_USER,
|
|
'password': _PG_PASSWORD,
|
|
}
|
|
# ==========================================
|
|
|
|
# Google Sheets
|
|
try:
|
|
import gspread
|
|
from google.oauth2.service_account import Credentials
|
|
except ImportError:
|
|
gspread = None
|
|
Credentials = None
|
|
|
|
app = FastAPI(title="CS Web Program")
|
|
|
|
# 미들웨어 등록 순서 주의:
|
|
# Starlette 은 add_middleware 를 "반대 순서로" 적용한다 (마지막에 추가한 것이 가장 바깥).
|
|
# AuthGuardMiddleware 가 request.session 을 읽으려면 SessionMiddleware 가
|
|
# 먼저 실행돼서 scope 에 session 을 주입해야 한다.
|
|
# 따라서 AuthGuard 를 먼저 추가 (내부) → Session 을 나중에 추가 (외부, 먼저 실행).
|
|
app.add_middleware(AuthGuardMiddleware)
|
|
app.add_middleware(
|
|
SessionMiddleware,
|
|
secret_key=_require_session_secret(),
|
|
session_cookie=os.getenv("SESSION_COOKIE_NAME", "session"),
|
|
same_site=os.getenv("SESSION_SAME_SITE", "lax"),
|
|
https_only=os.getenv("SESSION_HTTPS_ONLY", "0") == "1",
|
|
max_age=int(os.getenv("SESSION_MAX_AGE", "28800")),
|
|
path="/",
|
|
)
|
|
|
|
|
|
@app.get("/health/db")
|
|
async def health_db():
|
|
"""DB ping — 인증 불필요. dbx-main 의 헬스 배지가 이 경로를 폴링한다."""
|
|
try:
|
|
conn = psycopg2.connect(**DB_CONFIG, connect_timeout=3)
|
|
conn.close()
|
|
return JSONResponse({"ok": True, "detail": "ok"})
|
|
except Exception as e:
|
|
return JSONResponse({"ok": False, "detail": str(e)}, status_code=503)
|
|
|
|
|
|
@app.get("/login")
|
|
async def corm_login(request: Request, next: Optional[str] = None):
|
|
"""OMS 와 동일 패턴 — 자체 OAuth 없음. dbx-main 으로 위임."""
|
|
root = APP_ROOT_PATH
|
|
fallback = f"{root}/" if root else "/"
|
|
safe = next if (next and next.startswith("/") and not next.startswith("//") and not next.startswith("/\\")) else fallback
|
|
sep = "&" if "?" in MAIN_LOGIN_URL else "?"
|
|
target = f"{MAIN_LOGIN_URL}{sep}next={quote(safe, safe='/')}"
|
|
return RedirectResponse(url=target, status_code=302)
|
|
|
|
|
|
@app.get("/logout")
|
|
async def corm_logout():
|
|
"""전적으로 dbx-main 에 위임. CORM 은 세션을 직접 만지지 않는다 (쿠키 path 분리 방지)."""
|
|
return RedirectResponse(url=MAIN_LOGOUT_URL, status_code=302)
|
|
|
|
|
|
from routers.returns import router as returns_router
|
|
app.include_router(returns_router)
|
|
|
|
# Ensure directories exist
|
|
os.makedirs("static/css", exist_ok=True)
|
|
os.makedirs("static/js", exist_ok=True)
|
|
os.makedirs("templates", exist_ok=True)
|
|
|
|
app.mount("/static", StaticFiles(directory="static"), name="static")
|
|
templates = Jinja2Templates(directory="templates")
|
|
|
|
# Globals
|
|
SMS_HISTORY_FILE = 'sms_history.dat'
|
|
SMS_HISTORY_HEADERS = ["일시", "주문유형", "입금액", "주문아이템", "메시지 내용", "이름", "연락처", "주소"]
|
|
GSPREAD_CRED_FILE = 'manual-ordering.json'
|
|
MANUAL_SPREADSHEET_ID = '1QxFtjDurPPZd8NmtXBy4XEUIzkf93aXD5Mlx5kB33xw'
|
|
SETTINGS_SPREADSHEET_ID = '1k7SNEdIaRtGXzQNdJSQAiNSDRwzOePDRTLHY_mNCO8M'
|
|
MANUAL_HISTORY_FILE = 'manual_history.dat'
|
|
|
|
# ----- Utility Functions -----
|
|
def first_nonempty(data: dict, *keys: str) -> str:
|
|
for key in keys:
|
|
value = data.get(key)
|
|
if value is not None and str(value).strip():
|
|
return str(value)
|
|
return ""
|
|
|
|
def read_config(force_refresh=False):
|
|
global _CONFIG_ROWS_CACHE, _CONFIG_ROWS_CACHE_TIME
|
|
|
|
rows = []
|
|
|
|
# 1. 캐시 유효성 판단 (force_refresh가 아니고, 캐시가 존재하며, TTL이 지나지 않은 경우)
|
|
if not force_refresh and _CONFIG_ROWS_CACHE is not None:
|
|
if time.time() - _CONFIG_ROWS_CACHE_TIME < CONFIG_CACHE_TTL:
|
|
rows = _CONFIG_ROWS_CACHE
|
|
|
|
# 2. 캐시가 없거나 만료된 경우 구글 시트에서 직접 새로 로드
|
|
if not rows:
|
|
with _CONFIG_LOCK:
|
|
# 락 대기 중에 다른 스레드에서 캐시를 갱신했을 수 있으므로 다시 검사 (Double-Checked Locking)
|
|
if not force_refresh and _CONFIG_ROWS_CACHE is not None and time.time() - _CONFIG_ROWS_CACHE_TIME < CONFIG_CACHE_TTL:
|
|
rows = _CONFIG_ROWS_CACHE
|
|
else:
|
|
for _ in range(3):
|
|
try:
|
|
ws = get_settings_worksheet()
|
|
rows = ws.get_all_values()
|
|
break
|
|
except Exception as e:
|
|
time.sleep(1.5)
|
|
print(f"Google Sheets 불러오기 재시도 중...: {e}")
|
|
|
|
if rows:
|
|
_CONFIG_ROWS_CACHE = rows
|
|
_CONFIG_ROWS_CACHE_TIME = time.time()
|
|
|
|
# 3. 로드된 로우 데이터를 ConfigParser 형식으로 파싱
|
|
config = configparser.ConfigParser(allow_no_value=True)
|
|
config.optionxform = str
|
|
|
|
if rows and len(rows) > 1:
|
|
for row in rows:
|
|
if not row or not row[0].strip():
|
|
continue
|
|
section = row[0].strip()
|
|
if section == "대분류 (Section)":
|
|
continue
|
|
|
|
key = row[1].strip() if len(row) > 1 else ""
|
|
|
|
c_cells = [x.strip() for x in row[2:]]
|
|
while c_cells and not c_cells[-1]:
|
|
c_cells.pop()
|
|
val = " | ".join(c_cells)
|
|
|
|
if not config.has_section(section):
|
|
config.add_section(section)
|
|
|
|
if key:
|
|
if val:
|
|
config.set(section, key, val)
|
|
else:
|
|
config.set(section, key)
|
|
return config
|
|
|
|
def write_config(config):
|
|
global _CONFIG_ROWS_CACHE, _CONFIG_ROWS_CACHE_TIME
|
|
rows = [["대분류 (Section)", "항목 (Key)", "설정값 1", "설정값 2", "설정값 3", "설정값 4", "설정값 5"]]
|
|
for section in config.sections():
|
|
items = config.items(section)
|
|
if not items:
|
|
rows.append([section, "", ""])
|
|
else:
|
|
for k, v in items:
|
|
if v is None:
|
|
rows.append([section, k])
|
|
else:
|
|
v_parts = [p.strip() for p in v.split("|")]
|
|
rows.append([section, k] + v_parts)
|
|
|
|
for attempt in range(3):
|
|
try:
|
|
ws = get_settings_worksheet()
|
|
ws.clear()
|
|
ws.update(values=rows)
|
|
|
|
# 구글 시트에 저장이 성공하면 로컬 캐시도 즉시 최신화
|
|
with _CONFIG_LOCK:
|
|
_CONFIG_ROWS_CACHE = rows
|
|
_CONFIG_ROWS_CACHE_TIME = time.time()
|
|
return
|
|
except Exception as e:
|
|
print(f"Google Sheets 저장 시도 {attempt+1}/3 실패: {e}")
|
|
time.sleep(1.5)
|
|
|
|
print("Google Sheets에 설정을 저장하는 데 최종 실패했습니다.")
|
|
raise Exception("구글 시트 연동 오류 (할당량 초과). 잠시 후 다시 시도해주세요.")
|
|
|
|
# ----- API Endpoints -----
|
|
|
|
@app.get("/", response_class=HTMLResponse)
|
|
async def index(request: Request):
|
|
try:
|
|
with open("version_history.json", "r", encoding="utf-8") as f:
|
|
version_history = json.load(f)
|
|
except Exception as e:
|
|
print("버전 히스토리 로드 실패:", e)
|
|
version_history = []
|
|
|
|
return templates.TemplateResponse(
|
|
request=request,
|
|
name="index.html",
|
|
context={
|
|
"version_history": version_history,
|
|
"root_path": APP_ROOT_PATH,
|
|
"user_email": request.session.get(SESSION_USER_KEY, ""),
|
|
"user_name": request.session.get(SESSION_NAME_KEY, ""),
|
|
}
|
|
)
|
|
|
|
@app.get("/api/config")
|
|
async def get_config():
|
|
config = read_config()
|
|
data = {}
|
|
for section in config.sections():
|
|
data[section] = dict(config.items(section))
|
|
data[section]["__order"] = list(config.options(section))
|
|
return JSONResponse(data)
|
|
|
|
|
|
@app.post("/api/config/refresh")
|
|
async def refresh_config():
|
|
"""구글 시트로부터 실시간으로 설정을 동기화하여 캐시를 강제 갱신합니다."""
|
|
try:
|
|
read_config(force_refresh=True)
|
|
return {"status": "success", "message": "설정 캐시가 성공적으로 갱신되었습니다."}
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=f"설정 캐시 갱신 실패: {str(e)}")
|
|
|
|
|
|
|
|
class ProductGroup(BaseModel):
|
|
old_code: str
|
|
new_code: str
|
|
name: str
|
|
|
|
class ProductItem(BaseModel):
|
|
is_separator: bool = False
|
|
code: str = ""
|
|
name: str = ""
|
|
label: str = ""
|
|
price: int = 0
|
|
dprice: int = 0
|
|
color: str = ""
|
|
is_free: bool = False
|
|
|
|
class ProductsPayload(BaseModel):
|
|
groups: List[ProductGroup]
|
|
products: Dict[str, List[ProductItem]]
|
|
|
|
@app.post("/api/products/settings")
|
|
async def save_products_settings(payload: ProductsPayload):
|
|
config = read_config()
|
|
|
|
if config.has_section("GROUP_NAMES"):
|
|
config.remove_section("GROUP_NAMES")
|
|
config.add_section("GROUP_NAMES")
|
|
|
|
for g in payload.groups:
|
|
config.set("GROUP_NAMES", g.new_code, g.name)
|
|
|
|
for g in payload.groups:
|
|
if config.has_section(g.old_code):
|
|
config.remove_section(g.old_code)
|
|
if config.has_section(g.new_code):
|
|
config.remove_section(g.new_code)
|
|
|
|
for g in payload.groups:
|
|
config.add_section(g.new_code)
|
|
items = payload.products.get(g.new_code, [])
|
|
sep_idx = 1
|
|
for item in items:
|
|
if item.is_separator:
|
|
config.set(g.new_code, f"---separator_{sep_idx}---", "")
|
|
sep_idx += 1
|
|
else:
|
|
key = f"*{item.label}{item.color}" if item.is_free else f"{item.label}{item.color}"
|
|
val_parts = [item.code, item.name]
|
|
if item.price > 0 or item.dprice > 0:
|
|
val_parts.append(str(item.price))
|
|
if item.dprice > 0:
|
|
val_parts.append(str(item.dprice))
|
|
val = " | ".join(val_parts)
|
|
config.set(g.new_code, key, val)
|
|
|
|
write_config(config)
|
|
return {"status": "success"}
|
|
|
|
|
|
class OrderItem(BaseModel):
|
|
code: str
|
|
name: str
|
|
quantity: int
|
|
label: str
|
|
item_type: str
|
|
|
|
class OrderPayload(BaseModel):
|
|
customer_name: str
|
|
address: str
|
|
phone: str
|
|
comment: str
|
|
order_type: str # e.g., 'noolak', 'pason', 'bullyang', 'ellen', 'mira', 'normal'
|
|
no_lid: bool
|
|
items: List[OrderItem]
|
|
is_ellen: bool = False
|
|
total_amount: int = 0
|
|
|
|
def get_current_time_info():
|
|
korea_tz = datetime.timezone(datetime.timedelta(hours=9))
|
|
now = datetime.datetime.now(korea_tz)
|
|
time_str = "오후" if now.hour >= 12 else "오전"
|
|
day_names_kr = ["월요일", "화요일", "수요일", "목요일", "금요일", "토요일", "일요일"]
|
|
day_name = day_names_kr[now.weekday()]
|
|
date_str = now.strftime(f"%m월 %d일 ({day_name})")
|
|
datetime_str = now.strftime(f"%Y-%m-%d %H:%M:%S ({day_name})")
|
|
return date_str, time_str, datetime_str
|
|
|
|
def format_phone_number(num_str):
|
|
digits = re.sub(r'\D', '', num_str)
|
|
if len(digits) == 10: return f"{digits[:3]}-{digits[3:6]}-{digits[6:]}"
|
|
elif len(digits) == 11: return f"{digits[:3]}-{digits[3:7]}-{digits[7:]}"
|
|
elif len(digits) >= 8: return f"{digits[:4]}-{digits[4:]}"
|
|
return num_str
|
|
|
|
|
|
|
|
|
|
|
|
# ----- Google Sheets Config Cache & Global Sheets Cache -----
|
|
_CONFIG_ROWS_CACHE = None
|
|
_CONFIG_ROWS_CACHE_TIME = 0.0
|
|
_CONFIG_LOCK = threading.Lock()
|
|
CONFIG_CACHE_TTL = 600.0 # 캐시 유효 시간: 10분
|
|
|
|
gs_client_global = None
|
|
worksheet_manual_global = None
|
|
worksheet_settings_global = None
|
|
worksheet_sms_history_global = None
|
|
worksheet_manual_history_global = None
|
|
|
|
def get_gspread_client():
|
|
global gs_client_global
|
|
if gspread is None:
|
|
raise Exception("gspread 라이브러리가 없습니다.")
|
|
if gs_client_global is None:
|
|
scopes = ['https://www.googleapis.com/auth/spreadsheets']
|
|
creds = Credentials.from_service_account_file(GSPREAD_CRED_FILE, scopes=scopes)
|
|
gs_client_global = gspread.authorize(creds)
|
|
return gs_client_global
|
|
|
|
def get_manual_worksheet():
|
|
global worksheet_manual_global
|
|
if worksheet_manual_global is None:
|
|
client = get_gspread_client()
|
|
ss = client.open_by_key(MANUAL_SPREADSHEET_ID)
|
|
ws = ss.get_worksheet(0)
|
|
if ws.title == "시트1":
|
|
try:
|
|
ws.update_title("수동 발주")
|
|
except Exception:
|
|
pass
|
|
worksheet_manual_global = ws
|
|
return worksheet_manual_global
|
|
|
|
def get_settings_worksheet():
|
|
global worksheet_settings_global
|
|
if worksheet_settings_global is None:
|
|
client = get_gspread_client()
|
|
ss = client.open_by_key(SETTINGS_SPREADSHEET_ID)
|
|
try:
|
|
worksheet_settings_global = ss.worksheet("설정")
|
|
except gspread.exceptions.WorksheetNotFound:
|
|
worksheet_settings_global = ss.add_worksheet(title="설정", rows=500, cols=3)
|
|
return worksheet_settings_global
|
|
|
|
def get_sms_history_worksheet():
|
|
global worksheet_sms_history_global
|
|
if worksheet_sms_history_global is None:
|
|
client = get_gspread_client()
|
|
ss = client.open_by_key(SETTINGS_SPREADSHEET_ID)
|
|
try:
|
|
worksheet_sms_history_global = ss.worksheet("문자내역")
|
|
except gspread.exceptions.WorksheetNotFound:
|
|
worksheet_sms_history_global = ss.add_worksheet(title="문자내역", rows=1000, cols=10)
|
|
worksheet_sms_history_global.append_row(SMS_HISTORY_HEADERS)
|
|
return worksheet_sms_history_global
|
|
|
|
def get_manual_history_worksheet():
|
|
global worksheet_manual_history_global
|
|
if worksheet_manual_history_global is None:
|
|
client = get_gspread_client()
|
|
ss = client.open_by_key(SETTINGS_SPREADSHEET_ID)
|
|
try:
|
|
worksheet_manual_history_global = ss.worksheet("수동발주내역")
|
|
except gspread.exceptions.WorksheetNotFound:
|
|
worksheet_manual_history_global = ss.add_worksheet(title="수동발주내역", rows=1000, cols=3)
|
|
worksheet_manual_history_global.append_row(["일시", "고객명", "주문금액"])
|
|
return worksheet_manual_history_global
|
|
|
|
@app.on_event("startup")
|
|
async def startup_event():
|
|
init_db()
|
|
try:
|
|
print("구글 시트 연동 초기화 중...")
|
|
get_manual_worksheet()
|
|
get_settings_worksheet()
|
|
get_sms_history_worksheet()
|
|
get_manual_history_worksheet()
|
|
print("구글 시트 연동 완료!")
|
|
|
|
# 캐시 초기 적재 (Warm up cache)
|
|
print("구글 시트 설정 캐시 초기 적재 시작...")
|
|
read_config(force_refresh=True)
|
|
print("구글 시트 설정 캐시 적재 완료!")
|
|
except Exception as e:
|
|
print(f"구글 시트 연동 초기화 실패 (사용 전 권한을 확인하세요): {str(e)}")
|
|
|
|
|
|
@app.post("/api/order/submit")
|
|
async def submit_order_submit(payload: OrderPayload):
|
|
date_str, time_str, current_datetime_str = get_current_time_info()
|
|
c_tel = format_phone_number(payload.phone)
|
|
|
|
if payload.is_ellen:
|
|
title = "수동발주(엘렌)"
|
|
else:
|
|
title_map = {'noolak': '누락', 'pason': '파손', 'bullyang': '불량', 'normal': '수동발주'}
|
|
title = title_map.get(payload.order_type, "수동발주")
|
|
|
|
if not payload.customer_name:
|
|
title = "재구매"
|
|
|
|
try:
|
|
worksheet = get_manual_worksheet()
|
|
|
|
rows = []
|
|
lines = []
|
|
for i, item in enumerate(payload.items):
|
|
lid_suffix = "(뚜껑 없이 통만)" if payload.no_lid and item.item_type == "ORDER_SINGLE" else ""
|
|
sub_text = f"{title}, {item.name}{lid_suffix} {item.quantity}개{' 발송' if lid_suffix else ''}"
|
|
lines.append(f"{payload.customer_name} 고객님 {item.name}{lid_suffix} {item.quantity}개 발송, {title}")
|
|
|
|
c_comment = payload.comment or "빠른배송 부탁드립니다."
|
|
c_amount = payload.total_amount if i == len(payload.items) - 1 else ""
|
|
row_data = [current_datetime_str, "", c_amount, payload.customer_name, item.code, item.name, item.quantity, payload.address, "", c_tel, c_tel, c_comment, "", title, sub_text]
|
|
rows.append(row_data)
|
|
|
|
if rows:
|
|
worksheet.append_rows(rows, value_input_option='USER_ENTERED')
|
|
|
|
clipboard_text = "\r\n".join(lines) + "\r\n"
|
|
return {"status": "success", "message": f"{len(rows)}개 항목 구글 시트 입력 및 클립보드 준비 완료.", "clipboard_text": clipboard_text, "timestamp": current_datetime_str}
|
|
except Exception as e:
|
|
err_str = str(e)
|
|
if not err_str or err_str.strip() == "{}":
|
|
err_str = repr(e)
|
|
raise HTTPException(status_code=500, detail=f"수동발주 시트 접근 에러 (스프레드시트에 이메일 편집 권한을 추가했는지 확인해주세요!): {err_str}")
|
|
|
|
@app.post("/api/sheet/download_and_clear")
|
|
async def download_and_clear_sheet():
|
|
try:
|
|
scopes = ['https://www.googleapis.com/auth/spreadsheets', 'https://www.googleapis.com/auth/drive']
|
|
creds = Credentials.from_service_account_file(GSPREAD_CRED_FILE, scopes=scopes)
|
|
|
|
import google.auth.transport.requests
|
|
request = google.auth.transport.requests.Request()
|
|
creds.refresh(request)
|
|
access_token = creds.token
|
|
|
|
worksheet = get_manual_worksheet()
|
|
sheet_gid = worksheet.id
|
|
|
|
url = f"https://docs.google.com/spreadsheets/export?id={MANUAL_SPREADSHEET_ID}&exportFormat=xlsx"
|
|
req = urllib.request.Request(url, headers={"Authorization": f"Bearer {access_token}"})
|
|
with urllib.request.urlopen(req) as response:
|
|
content = response.read()
|
|
|
|
korea_tz = datetime.timezone(datetime.timedelta(hours=9))
|
|
now = datetime.datetime.now(korea_tz)
|
|
weekdays = ["월요일", "화요일", "수요일", "목요일", "금요일", "토요일", "일요일"]
|
|
weekday_str = weekdays[now.weekday()]
|
|
date_str = f"{now.strftime('%m')}월 {now.strftime('%d')}일 ({weekday_str})"
|
|
filename = f"{date_str} 5 수동발주.xlsx"
|
|
|
|
# Clear sheet values instantly
|
|
worksheet.batch_clear(["A2:Z10000"])
|
|
|
|
# Clear formatting (colors, borders, etc) from row 2 downwards
|
|
body = {
|
|
"requests": [
|
|
{
|
|
"updateCells": {
|
|
"range": {
|
|
"sheetId": worksheet.id,
|
|
"startRowIndex": 1
|
|
},
|
|
"fields": "userEnteredFormat"
|
|
}
|
|
}
|
|
]
|
|
}
|
|
worksheet.spreadsheet.batch_update(body)
|
|
|
|
encoded_filename = urllib.parse.quote(filename)
|
|
headers = {
|
|
"Content-Disposition": f"attachment; filename*=UTF-8''{encoded_filename}"
|
|
}
|
|
return StreamingResponse(
|
|
io.BytesIO(content),
|
|
media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
|
headers=headers
|
|
)
|
|
except Exception as e:
|
|
err_str = str(e)
|
|
if not err_str or err_str.strip() == "{}":
|
|
err_str = repr(e)
|
|
raise HTTPException(status_code=500, detail=f"저장 및 삭제 실패: {err_str}")
|
|
|
|
|
|
|
|
@app.get("/api/sms/history")
|
|
async def get_history():
|
|
local_history = {}
|
|
def merge_history_item(sheet_item, local_item):
|
|
if not local_item:
|
|
return sheet_item
|
|
merged = dict(sheet_item)
|
|
merged.update(local_item)
|
|
for key in ("custName", "custPhone", "custAddress", "customer_name", "phone", "address"):
|
|
if not merged.get(key) and sheet_item.get(key):
|
|
merged[key] = sheet_item[key]
|
|
return merged
|
|
|
|
if os.path.exists(SMS_HISTORY_FILE):
|
|
with open(SMS_HISTORY_FILE, 'r', encoding='utf-8') as f:
|
|
for line in f:
|
|
if line.strip():
|
|
try:
|
|
item = json.loads(line)
|
|
if "timestamp" in item:
|
|
local_history[item["timestamp"]] = item
|
|
except:
|
|
pass
|
|
|
|
history = []
|
|
try:
|
|
ws = get_sms_history_worksheet()
|
|
rows = ws.get_all_values()
|
|
if rows and len(rows) > 1:
|
|
for row in rows[1:]:
|
|
if not row or not row[0].strip():
|
|
continue
|
|
timestamp = row[0]
|
|
|
|
order_type = row[1] if len(row) > 1 else ""
|
|
deposit_amt = row[2] if len(row) > 2 else ""
|
|
items_str = row[3] if len(row) > 3 else ""
|
|
preview_text = row[4] if len(row) > 4 else ""
|
|
cust_name = row[5] if len(row) > 5 else ""
|
|
cust_phone = row[6] if len(row) > 6 else ""
|
|
cust_address = row[7] if len(row) > 7 else ""
|
|
preview_text = preview_text.replace(" ", "\n")
|
|
|
|
sheet_item = {
|
|
"timestamp": timestamp,
|
|
"orderType": order_type,
|
|
"deposit_amount_str": deposit_amt,
|
|
"items_str": items_str,
|
|
"previewText": preview_text,
|
|
"custName": cust_name,
|
|
"custPhone": cust_phone,
|
|
"custAddress": cust_address,
|
|
"customer_name": cust_name,
|
|
"phone": cust_phone,
|
|
"address": cust_address
|
|
}
|
|
history.append(merge_history_item(sheet_item, local_history.get(timestamp)))
|
|
return history
|
|
except Exception as e:
|
|
print(f"구글 시트에서 기록을 읽어오는 중 에러: {e}")
|
|
|
|
if os.path.exists(SMS_HISTORY_FILE):
|
|
return [json.loads(line) for line in open(SMS_HISTORY_FILE, 'r', encoding='utf-8') if line.strip()]
|
|
return []
|
|
|
|
@app.post("/api/sms/history")
|
|
async def add_history(data: dict):
|
|
with open(SMS_HISTORY_FILE, 'a', encoding='utf-8') as f:
|
|
json.dump(data, f, ensure_ascii=False)
|
|
f.write('\n')
|
|
|
|
# 구글 시트에 기록
|
|
try:
|
|
ws = get_sms_history_worksheet()
|
|
|
|
# 주문 품목 텍스트화
|
|
items_str = ""
|
|
items = data.get("items", [])
|
|
if items:
|
|
items_str = ", ".join([f"{i.get('label', '')}[{i.get('code', '')}] {i.get('qty', 0)}개" for i in items])
|
|
|
|
# 행 높이가 길어지지 않도록 줄바꿈을 띄어쓰기로 변경
|
|
preview_text = data.get("previewText", "").replace("\n", " ").replace("\r", "")
|
|
customer_name = first_nonempty(data, "custName", "customer_name", "customerName", "name")
|
|
customer_phone = first_nonempty(data, "custPhone", "phone", "customer_phone", "customerPhone")
|
|
customer_address = first_nonempty(data, "custAddress", "address", "customer_address", "customerAddress")
|
|
|
|
row = [
|
|
data.get("timestamp", ""),
|
|
data.get("orderType", ""),
|
|
data.get("deposit_amount_str", ""),
|
|
items_str,
|
|
preview_text,
|
|
customer_name,
|
|
customer_phone,
|
|
customer_address
|
|
]
|
|
next_row = len(ws.col_values(1)) + 1
|
|
ws.update(values=[row], range_name=f"A{next_row}:H{next_row}", value_input_option='USER_ENTERED')
|
|
except Exception as e:
|
|
print(f"구글 시트 문자내역 기록 실패: {e}")
|
|
|
|
return {"status": "success"}
|
|
|
|
@app.delete("/api/sms/history")
|
|
async def clear_history():
|
|
if os.path.exists(SMS_HISTORY_FILE):
|
|
os.remove(SMS_HISTORY_FILE)
|
|
|
|
try:
|
|
ws = get_sms_history_worksheet()
|
|
ws.batch_clear(["A2:Z10000"])
|
|
except Exception as e:
|
|
print(f"구글 시트 문자내역 전체 삭제 실패: {e}")
|
|
|
|
return {"status": "success"}
|
|
|
|
@app.delete("/api/sms/history/{index}")
|
|
async def delete_history_item(index: int):
|
|
try:
|
|
ws = get_sms_history_worksheet()
|
|
rows = ws.get_all_values()
|
|
timestamp_to_delete = None
|
|
if len(rows) > index + 1:
|
|
timestamp_to_delete = rows[index + 1][0] if len(rows[index+1]) > 0 else None
|
|
|
|
ws.delete_rows(index + 2)
|
|
|
|
if timestamp_to_delete and os.path.exists(SMS_HISTORY_FILE):
|
|
local_history = []
|
|
with open(SMS_HISTORY_FILE, 'r', encoding='utf-8') as f:
|
|
for line in f:
|
|
if line.strip():
|
|
try:
|
|
local_history.append(json.loads(line))
|
|
except:
|
|
pass
|
|
|
|
local_history = [item for item in local_history if item.get("timestamp") != timestamp_to_delete]
|
|
|
|
with open(SMS_HISTORY_FILE, 'w', encoding='utf-8') as f:
|
|
for entry in local_history:
|
|
json.dump(entry, f, ensure_ascii=False)
|
|
f.write('\n')
|
|
except Exception as e:
|
|
print(f"구글 시트 문자내역 항목 삭제 실패: {e}")
|
|
|
|
return {"status": "success"}
|
|
|
|
|
|
import sys
|
|
|
|
@app.get("/api/manual/history")
|
|
async def get_manual_history():
|
|
korea_tz = datetime.timezone(datetime.timedelta(hours=9))
|
|
today_prefix = datetime.datetime.now(korea_tz).strftime("%Y-%m-%d")
|
|
|
|
local_history = {}
|
|
if os.path.exists(MANUAL_HISTORY_FILE):
|
|
with open(MANUAL_HISTORY_FILE, 'r', encoding='utf-8') as f:
|
|
for line in f:
|
|
if line.strip():
|
|
try:
|
|
item = json.loads(line)
|
|
if item.get("timestamp", "").startswith(today_prefix):
|
|
local_history[item["timestamp"]] = item
|
|
except:
|
|
pass
|
|
|
|
history = []
|
|
try:
|
|
ws = get_manual_history_worksheet()
|
|
rows = ws.get_all_values()
|
|
if rows and len(rows) > 1:
|
|
for row in rows[1:]:
|
|
if not row or not row[0].strip():
|
|
continue
|
|
timestamp = row[0]
|
|
if not timestamp.startswith(today_prefix):
|
|
continue
|
|
|
|
if timestamp in local_history:
|
|
history.append(local_history[timestamp])
|
|
else:
|
|
customer_name = row[1] if len(row) > 1 else ""
|
|
total_amount_str = str(row[2]).replace(",", "") if len(row) > 2 else "0"
|
|
total_amount = int(total_amount_str) if total_amount_str.lstrip('-').isdigit() else total_amount_str
|
|
history.append({
|
|
"timestamp": timestamp,
|
|
"customer_name": customer_name,
|
|
"total_amount": total_amount
|
|
})
|
|
return history
|
|
except Exception as e:
|
|
print(f"구글 시트에서 수동발주 기록을 읽어오는 중 에러: {e}")
|
|
|
|
return list(local_history.values())
|
|
|
|
@app.post("/api/manual/history")
|
|
async def add_manual_history(data: dict):
|
|
local_history = []
|
|
korea_tz = datetime.timezone(datetime.timedelta(hours=9))
|
|
today_prefix = datetime.datetime.now(korea_tz).strftime("%Y-%m-%d")
|
|
|
|
if os.path.exists(MANUAL_HISTORY_FILE):
|
|
with open(MANUAL_HISTORY_FILE, 'r', encoding='utf-8') as f:
|
|
for line in f:
|
|
if line.strip():
|
|
try:
|
|
item = json.loads(line)
|
|
if item.get("timestamp", "").startswith(today_prefix):
|
|
local_history.append(item)
|
|
except:
|
|
pass
|
|
|
|
local_history.append(data)
|
|
|
|
with open(MANUAL_HISTORY_FILE, 'w', encoding='utf-8') as f:
|
|
for entry in local_history:
|
|
json.dump(entry, f, ensure_ascii=False)
|
|
f.write('\n')
|
|
|
|
try:
|
|
ws = get_manual_history_worksheet()
|
|
row = [
|
|
data.get("timestamp", ""),
|
|
data.get("customer_name", ""),
|
|
data.get("total_amount", 0)
|
|
]
|
|
ws.append_row(row, value_input_option='USER_ENTERED')
|
|
except Exception as e:
|
|
print(f"구글 시트 수동발주내역 기록 실패: {e}")
|
|
|
|
return {"status": "success"}
|
|
|
|
@app.delete("/api/manual/history/{timestamp}")
|
|
async def delete_manual_history(timestamp: str):
|
|
# 구글 시트에서 삭제
|
|
try:
|
|
ws = get_manual_worksheet()
|
|
rows = ws.get_all_values()
|
|
start_idx = None
|
|
end_idx = None
|
|
for i, row in enumerate(rows):
|
|
if row and row[0] == timestamp:
|
|
if start_idx is None:
|
|
start_idx = i + 1
|
|
end_idx = i + 1
|
|
if start_idx:
|
|
ws.delete_rows(start_idx, end_idx)
|
|
except Exception as e:
|
|
print(f"구글 시트 수동발주 삭제 실패: {e}")
|
|
|
|
try:
|
|
ws2 = get_manual_history_worksheet()
|
|
rows2 = ws2.get_all_values()
|
|
for i, row in enumerate(rows2):
|
|
if row and row[0] == timestamp:
|
|
ws2.delete_rows(i + 1)
|
|
break
|
|
except Exception as e:
|
|
print(f"구글 시트 수동발주내역 시트 삭제 실패: {e}")
|
|
|
|
# 로컬에서 삭제
|
|
if os.path.exists(MANUAL_HISTORY_FILE):
|
|
local_history = []
|
|
with open(MANUAL_HISTORY_FILE, 'r', encoding='utf-8') as f:
|
|
for line in f:
|
|
if line.strip():
|
|
try:
|
|
local_history.append(json.loads(line))
|
|
except:
|
|
pass
|
|
local_history = [item for item in local_history if item.get("timestamp") != timestamp]
|
|
with open(MANUAL_HISTORY_FILE, 'w', encoding='utf-8') as f:
|
|
for entry in local_history:
|
|
json.dump(entry, f, ensure_ascii=False)
|
|
f.write('\n')
|
|
return {"status": "success"}
|
|
|
|
@app.get("/api/fonts")
|
|
async def get_fonts():
|
|
fonts = set()
|
|
if sys.platform == 'win32':
|
|
try:
|
|
import winreg
|
|
for hkey in [winreg.HKEY_LOCAL_MACHINE, winreg.HKEY_CURRENT_USER]:
|
|
try:
|
|
key = winreg.OpenKey(hkey, r"SOFTWARE\Microsoft\Windows NT\CurrentVersion\Fonts")
|
|
for i in range(0, winreg.QueryInfoKey(key)[1]):
|
|
name, _, _ = winreg.EnumValue(key, i)
|
|
font_name = name.split(' (')[0].strip()
|
|
if font_name:
|
|
fonts.add(font_name)
|
|
winreg.CloseKey(key)
|
|
except Exception:
|
|
pass
|
|
except Exception as e:
|
|
print(f"폰트 불러오기 오류: {e}")
|
|
|
|
if not fonts:
|
|
fonts_list = ["Inter", "Malgun Gothic", "Arial", "sans-serif"]
|
|
else:
|
|
fonts_list = sorted(list(fonts))
|
|
return fonts_list
|
|
|
|
@app.post("/api/gift/settings")
|
|
async def save_gift_settings(payload: dict):
|
|
# 하위 호환성 (기존 rules만 오는 경우)
|
|
if "rules" in payload:
|
|
rules = payload["rules"]
|
|
app_settings = payload.get("app_settings", {})
|
|
else:
|
|
rules = payload
|
|
app_settings = {}
|
|
|
|
config = read_config()
|
|
section = "SMS_GIFT_RULES"
|
|
if section not in config:
|
|
config.add_section(section)
|
|
else:
|
|
config.remove_section(section)
|
|
config.add_section(section)
|
|
|
|
for k, v in rules.items():
|
|
if v:
|
|
config.set(section, k, str(v))
|
|
|
|
if app_settings:
|
|
app_section = "APP_SETTINGS"
|
|
if app_section not in config:
|
|
config.add_section(app_section)
|
|
else:
|
|
config.remove_section(app_section)
|
|
config.add_section(app_section)
|
|
|
|
for k, v in app_settings.items():
|
|
if v:
|
|
config.set(app_section, k, str(v))
|
|
|
|
write_config(config)
|
|
return {"status": "success"}
|
|
|
|
# ----- Code DB (SQLite) Data Definition -----
|
|
DB_PATH = "codes.db"
|
|
|
|
def init_db():
|
|
conn = psycopg2.connect(**DB_CONFIG, cursor_factory=psycopg2.extras.DictCursor)
|
|
cursor = conn.cursor()
|
|
|
|
# 단품 테이블
|
|
cursor.execute('''
|
|
CREATE TABLE IF NOT EXISTS single_items (
|
|
item_code TEXT PRIMARY KEY,
|
|
sabangnet_code TEXT,
|
|
name TEXT
|
|
)
|
|
''')
|
|
# 세트 테이블
|
|
cursor.execute('''
|
|
CREATE TABLE IF NOT EXISTS set_items (
|
|
item_code TEXT PRIMARY KEY,
|
|
sabangnet_code TEXT,
|
|
name TEXT
|
|
)
|
|
''')
|
|
# 세트 구성품
|
|
cursor.execute('''
|
|
CREATE TABLE IF NOT EXISTS set_components (
|
|
set_code TEXT,
|
|
single_code TEXT,
|
|
quantity INTEGER,
|
|
PRIMARY KEY (set_code, single_code),
|
|
FOREIGN KEY(set_code) REFERENCES set_items(item_code) ON DELETE CASCADE,
|
|
FOREIGN KEY(single_code) REFERENCES single_items(item_code) ON DELETE CASCADE
|
|
)
|
|
''')
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
# 반품 데이터베이스 (return_db) 초기화 시도
|
|
try:
|
|
conn_return = psycopg2.connect(**RETURN_DB_CONFIG, cursor_factory=psycopg2.extras.DictCursor)
|
|
cursor_return = conn_return.cursor()
|
|
|
|
# 반품 신청 테이블
|
|
cursor_return.execute('''
|
|
CREATE TABLE IF NOT EXISTS return_requests (
|
|
id SERIAL PRIMARY KEY,
|
|
request_date VARCHAR(20),
|
|
request_type VARCHAR(50),
|
|
receiver_name VARCHAR(100),
|
|
address TEXT,
|
|
phone VARCHAR(50),
|
|
tracking_no VARCHAR(100),
|
|
mall VARCHAR(100),
|
|
remarks TEXT,
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
|
)
|
|
''')
|
|
|
|
# 반품 입고 테이블
|
|
cursor_return.execute('''
|
|
CREATE TABLE IF NOT EXISTS return_receivings (
|
|
id SERIAL PRIMARY KEY,
|
|
receive_date VARCHAR(20),
|
|
tracking_no VARCHAR(100),
|
|
mall VARCHAR(100),
|
|
receiver_name VARCHAR(100),
|
|
phone VARCHAR(50),
|
|
remarks TEXT,
|
|
receive_status VARCHAR(50) DEFAULT '환불 대기',
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
|
)
|
|
''')
|
|
# 혹시 기존에 생성된 테이블에 receive_status 컬럼이 없다면 추가
|
|
try:
|
|
cursor_return.execute('ALTER TABLE return_receivings ADD COLUMN IF NOT EXISTS receive_status VARCHAR(50) DEFAULT \'환불 대기\'')
|
|
cursor_return.execute('ALTER TABLE return_receivings ADD COLUMN IF NOT EXISTS address TEXT')
|
|
except:
|
|
pass
|
|
conn_return.commit()
|
|
conn_return.close()
|
|
except Exception as e:
|
|
print(f"[Warning] return_db 초기화 실패. 데이터베이스가 아직 생성되지 않았거나 권한 문제가 있을 수 있습니다: {e}")
|
|
|
|
def get_db_conn():
|
|
conn = psycopg2.connect(**DB_CONFIG, cursor_factory=psycopg2.extras.DictCursor)
|
|
return conn
|
|
|
|
def get_return_db_conn():
|
|
conn = psycopg2.connect(**RETURN_DB_CONFIG, cursor_factory=psycopg2.extras.DictCursor)
|
|
return conn
|
|
|
|
def get_order_db_conn():
|
|
conn = psycopg2.connect(**ORDER_DB_CONFIG, cursor_factory=psycopg2.extras.DictCursor)
|
|
return conn
|
|
|
|
# Code API Models
|
|
class SingleItemAdd(BaseModel):
|
|
item_code: str
|
|
sabangnet_code: str
|
|
name: str
|
|
|
|
class SetComponentData(BaseModel):
|
|
single_code: str
|
|
quantity: int
|
|
|
|
class SetItemAdd(BaseModel):
|
|
item_code: str
|
|
components: List[SetComponentData]
|
|
sabangnet_code: str = ""
|
|
name: str = ""
|
|
|
|
@app.get("/api/codes/single")
|
|
async def get_single_items():
|
|
conn = get_db_conn()
|
|
cursor = conn.cursor()
|
|
cursor.execute("SELECT * FROM single_items ORDER BY item_code ASC")
|
|
rows = cursor.fetchall()
|
|
conn.close()
|
|
return {"status": "success", "data": [dict(r) for r in rows]}
|
|
|
|
@app.post("/api/codes/single")
|
|
async def add_single_item(payload: SingleItemAdd):
|
|
conn = get_db_conn()
|
|
cursor = conn.cursor()
|
|
try:
|
|
cursor.execute(
|
|
"INSERT INTO single_items (item_code, sabangnet_code, name) VALUES (%s, %s, %s)",
|
|
(payload.item_code, payload.sabangnet_code, payload.name)
|
|
)
|
|
conn.commit()
|
|
return {"status": "success"}
|
|
except psycopg2.IntegrityError:
|
|
conn.close()
|
|
raise HTTPException(status_code=400, detail="이미 존재하는 단품 코드입니다.")
|
|
finally:
|
|
if conn:
|
|
conn.close()
|
|
|
|
@app.put("/api/codes/single/{item_code}")
|
|
async def update_single_item(item_code: str, payload: SingleItemAdd):
|
|
conn = get_db_conn()
|
|
cursor = conn.cursor()
|
|
cursor.execute(
|
|
"UPDATE single_items SET sabangnet_code=%s, name=%s WHERE item_code=%s",
|
|
(payload.sabangnet_code, payload.name, item_code)
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
return {"status": "success"}
|
|
|
|
@app.delete("/api/codes/single/{item_code}")
|
|
async def delete_single_item(item_code: str):
|
|
conn = get_db_conn()
|
|
cursor = conn.cursor()
|
|
cursor.execute("DELETE FROM single_items WHERE item_code=%s", (item_code,))
|
|
conn.commit()
|
|
conn.close()
|
|
return {"status": "success"}
|
|
|
|
|
|
@app.get("/api/codes/set")
|
|
async def get_set_items():
|
|
conn = get_db_conn()
|
|
cursor = conn.cursor()
|
|
cursor.execute("SELECT * FROM set_items ORDER BY item_code ASC")
|
|
sets = cursor.fetchall()
|
|
|
|
result = []
|
|
for s in sets:
|
|
s_dict = dict(s)
|
|
cursor.execute("SELECT single_code, quantity FROM set_components WHERE set_code=%s", (s_dict["item_code"],))
|
|
comps = cursor.fetchall()
|
|
s_dict["components"] = [dict(c) for c in comps]
|
|
result.append(s_dict)
|
|
|
|
conn.close()
|
|
return {"status": "success", "data": result}
|
|
|
|
@app.post("/api/codes/set")
|
|
async def add_set_item(payload: SetItemAdd):
|
|
conn = get_db_conn()
|
|
cursor = conn.cursor()
|
|
try:
|
|
cursor.execute(
|
|
"INSERT INTO set_items (item_code, sabangnet_code, name) VALUES (%s, %s, %s)",
|
|
(payload.item_code, payload.sabangnet_code, payload.name)
|
|
)
|
|
for comp in payload.components:
|
|
cursor.execute(
|
|
"INSERT INTO set_components (set_code, single_code, quantity) VALUES (%s, %s, %s)",
|
|
(payload.item_code, comp.single_code, comp.quantity)
|
|
)
|
|
conn.commit()
|
|
return {"status": "success"}
|
|
except psycopg2.IntegrityError:
|
|
conn.rollback()
|
|
raise HTTPException(status_code=400, detail="중복되는 세트 코드이거나 구성품에 존재하지 않는 단품 코드가 있습니다.")
|
|
finally:
|
|
if conn: conn.close()
|
|
|
|
@app.put("/api/codes/set/{item_code}")
|
|
async def update_set_item(item_code: str, payload: SetItemAdd):
|
|
conn = get_db_conn()
|
|
cursor = conn.cursor()
|
|
try:
|
|
cursor.execute("UPDATE set_items SET sabangnet_code=%s, name=%s WHERE item_code=%s", (payload.sabangnet_code, payload.name, item_code))
|
|
cursor.execute("DELETE FROM set_components WHERE set_code=%s", (item_code,))
|
|
for comp in payload.components:
|
|
cursor.execute(
|
|
"INSERT INTO set_components (set_code, single_code, quantity) VALUES (%s, %s, %s)",
|
|
(item_code, comp.single_code, comp.quantity)
|
|
)
|
|
conn.commit()
|
|
return {"status": "success"}
|
|
except psycopg2.IntegrityError:
|
|
conn.rollback()
|
|
raise HTTPException(status_code=400, detail="구성품 업데이트 중 오류 발생. 유효한 단품 코드인지 확인해주세요.")
|
|
finally:
|
|
if conn: conn.close()
|
|
|
|
@app.delete("/api/codes/set/{item_code}")
|
|
async def delete_set_item(item_code: str):
|
|
conn = get_db_conn()
|
|
cursor = conn.cursor()
|
|
cursor.execute("DELETE FROM set_items WHERE item_code=%s", (item_code,))
|
|
conn.commit()
|
|
conn.close()
|
|
return {"status": "success"}
|
|
|
|
# ----- 스마트스토어 & 카페24 자동발주 공통 -----
|
|
|
|
EXCEL_COLUMNS = [
|
|
'주문날짜', '번호', '주문번호', '수령인명', '상품코드', '상품명', '수량',
|
|
'주소', '우편번호', '수령인 전화번호', '수령인 휴대폰', '배송시 요구사항',
|
|
'송장번호', '쇼핑몰명', '비고', '주문번호(쇼핑몰)', '주문목록'
|
|
]
|
|
|
|
def generate_order_excel(data, numeric_date, filename, is_cafe24=False, return_bytes=False):
|
|
df = pd.DataFrame(data, columns=EXCEL_COLUMNS)
|
|
|
|
address_replacements = {
|
|
"경기 ": "경기도 ",
|
|
"강원도 ": "강원특별자치도 ",
|
|
"경남 ": "경상남도 ",
|
|
"경북 ": "경상북도 ",
|
|
"광주 ": "광주광역시 ",
|
|
"대구 ": "대구광역시 ",
|
|
"대전 ": "대전광역시 ",
|
|
"부산 ": "부산광역시 ",
|
|
"서울 ": "서울특별시 ",
|
|
"울산 ": "울산광역시 ",
|
|
"인천 ": "인천광역시 ",
|
|
"전남 ": "전라남도 ",
|
|
"전북 ": "전라북도 ",
|
|
"충남 ": "충청남도 ",
|
|
"충북 ": "충청북도 "
|
|
}
|
|
|
|
if '주소' in df.columns:
|
|
for old_val, new_val in address_replacements.items():
|
|
df['주소'] = df['주소'].str.replace(old_val, new_val, regex=False)
|
|
|
|
df = df.sort_values(by=['주소', '주문번호']).reset_index(drop=True)
|
|
df['번호'] = range(1, len(df) + 1)
|
|
|
|
output = io.BytesIO()
|
|
with pd.ExcelWriter(output, engine='openpyxl') as writer:
|
|
df.to_excel(writer, index=False, sheet_name='Sheet1')
|
|
|
|
worksheet = writer.sheets['Sheet1']
|
|
from openpyxl.utils import get_column_letter
|
|
from openpyxl.styles import Alignment, PatternFill, Font
|
|
|
|
if is_cafe24:
|
|
header_fill = PatternFill(start_color="002060", end_color="002060", fill_type="solid")
|
|
else:
|
|
header_fill = PatternFill(start_color="009900", end_color="009900", fill_type="solid")
|
|
|
|
header_font = Font(name="나눔고딕", bold=True, color="FFFFFF")
|
|
data_font = Font(name="나눔고딕")
|
|
center_alignment = Alignment(horizontal="center", vertical="center")
|
|
|
|
for row in worksheet.iter_rows(min_row=1, max_row=worksheet.max_row, min_col=1, max_col=17):
|
|
for cell in row:
|
|
if cell.row == 1:
|
|
cell.fill = header_fill
|
|
cell.font = header_font
|
|
cell.alignment = center_alignment
|
|
else:
|
|
cell.font = data_font
|
|
|
|
duplicate_groups = {}
|
|
for i, rowTuple in df.iterrows():
|
|
key = (rowTuple['수령인명'], rowTuple['주소'])
|
|
if key not in duplicate_groups:
|
|
duplicate_groups[key] = []
|
|
duplicate_groups[key].append(i + 2)
|
|
|
|
color_60_lighter = PatternFill(start_color="B4C6E7", end_color="B4C6E7", fill_type="solid")
|
|
color_40_lighter = PatternFill(start_color="FFD966", end_color="FFD966", fill_type="solid")
|
|
current_toggle = True
|
|
|
|
for key, row_indices in duplicate_groups.items():
|
|
if len(row_indices) > 1:
|
|
fill = color_60_lighter if current_toggle else color_40_lighter
|
|
for r_idx in row_indices:
|
|
for c_idx in range(1, 18):
|
|
worksheet.cell(row=r_idx, column=c_idx).fill = fill
|
|
current_toggle = not current_toggle
|
|
|
|
worksheet.freeze_panes = "H2"
|
|
worksheet.auto_filter.ref = worksheet.dimensions
|
|
|
|
for idx, col in enumerate(df.columns):
|
|
col_letter = get_column_letter(idx + 1)
|
|
max_length = 0
|
|
for cell in worksheet[col_letter]:
|
|
if cell.value is not None:
|
|
val_str = str(cell.value)
|
|
length = sum(1.8 if ord(c) > 127 else 1.1 for c in val_str)
|
|
if length > max_length:
|
|
max_length = length
|
|
|
|
adjusted_width = max_length + 2
|
|
if adjusted_width > 60:
|
|
adjusted_width = 60
|
|
worksheet.column_dimensions[col_letter].width = adjusted_width
|
|
|
|
output.seek(0)
|
|
|
|
encoded_filename = urllib.parse.quote(filename)
|
|
if return_bytes:
|
|
return output.getvalue(), encoded_filename
|
|
|
|
headers = {
|
|
"Content-Disposition": f"attachment; filename*=UTF-8''{encoded_filename}"
|
|
}
|
|
return StreamingResponse(
|
|
output,
|
|
media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
|
headers=headers
|
|
)
|
|
|
|
def get_all_code_to_name_from_db():
|
|
import psycopg2
|
|
import psycopg2.extras
|
|
import os
|
|
code_to_name = {}
|
|
try:
|
|
conn = psycopg2.connect(**DB_CONFIG, cursor_factory=psycopg2.extras.DictCursor)
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT item_code, name FROM single_items")
|
|
for row in cur.fetchall():
|
|
code_to_name[row[0]] = row[1]
|
|
cur.execute("SELECT item_code, name FROM set_items")
|
|
for row in cur.fetchall():
|
|
code_to_name[row[0]] = row[1]
|
|
conn.close()
|
|
except Exception as e:
|
|
print(f"DB 매핑 오류: {e}")
|
|
return code_to_name
|
|
|
|
@app.post("/api/smartstore/start_download")
|
|
async def start_smartstore_download(background_tasks: BackgroundTasks, confirm: bool = Query(False)):
|
|
task_id = str(uuid.uuid4())
|
|
task_store[task_id] = {
|
|
"status": "processing",
|
|
"current": 0,
|
|
"total": 0,
|
|
"file_bytes": None,
|
|
"filename": None,
|
|
"error": None
|
|
}
|
|
|
|
def run_task(tid):
|
|
try:
|
|
token = get_access_token()
|
|
statuses = get_payment_done_orders(token)
|
|
|
|
if not statuses:
|
|
task_store[tid]["status"] = "error"
|
|
task_store[tid]["error"] = "새로운 결제완료(배송대기) 주문이 없습니다."
|
|
task_store[tid]["status_code"] = 404
|
|
return
|
|
|
|
po_ids = [s.get('productOrderId') for s in statuses if s.get('productOrderId')]
|
|
task_store[tid]["total"] = len(po_ids)
|
|
|
|
all_order_details = []
|
|
chunk_size = 100
|
|
for i in range(0, len(po_ids), chunk_size):
|
|
chunk = po_ids[i:i+chunk_size]
|
|
details = get_product_order_details(token, chunk)
|
|
all_order_details.extend(details)
|
|
task_store[tid]["current"] += len(chunk)
|
|
|
|
if confirm:
|
|
for i in range(0, len(po_ids), chunk_size):
|
|
chunk = po_ids[i:i+chunk_size]
|
|
confirm_orders(token, chunk)
|
|
|
|
data = []
|
|
now = datetime.datetime.now()
|
|
numeric_date = int(now.strftime('%Y%m%d'))
|
|
|
|
product_cache = {}
|
|
code_to_name = get_all_code_to_name_from_db()
|
|
|
|
for idx, detail in enumerate(all_order_details):
|
|
po = detail.get('productOrder', {})
|
|
order = detail.get('order', {})
|
|
shippingAddress = po.get('shippingAddress', {})
|
|
|
|
product_id = po.get('productId')
|
|
option_code = str(po.get('optionCode', ''))
|
|
seller_code = ''
|
|
|
|
if product_id:
|
|
if product_id not in product_cache:
|
|
product_cache[product_id] = get_product_details(token, product_id)
|
|
|
|
p_data = product_cache[product_id]
|
|
seller_code = p_data.get('originProduct', {}).get('detailAttribute', {}).get('sellerCodeInfo', {}).get('sellerManagementCode', '')
|
|
|
|
if not seller_code and option_code:
|
|
option_info = p_data.get('originProduct', {}).get('detailAttribute', {}).get('optionInfo', {})
|
|
found_opt_code = ''
|
|
for opt in option_info.get('optionCombinations', []):
|
|
if str(opt.get('id', '')) == option_code:
|
|
found_opt_code = opt.get('sellerManagerCode', '')
|
|
break
|
|
if not found_opt_code:
|
|
for opt in option_info.get('optionStandards', []):
|
|
if str(opt.get('id', '')) == option_code:
|
|
found_opt_code = opt.get('sellerManagerCode', '')
|
|
break
|
|
if found_opt_code:
|
|
seller_code = found_opt_code
|
|
|
|
final_quantity = int(po.get('quantity', 1))
|
|
if seller_code and '_' in seller_code:
|
|
parts = seller_code.rsplit('_', 1)
|
|
if len(parts) == 2 and parts[1].isdigit():
|
|
seller_code = parts[0]
|
|
final_quantity = final_quantity * int(parts[1])
|
|
|
|
base_address = shippingAddress.get('baseAddress', '')
|
|
detailed_address = shippingAddress.get('detailedAddress', '')
|
|
full_address = f"{base_address} {detailed_address}".strip()
|
|
|
|
final_item_code = seller_code or po.get('optionManageCode', '') or po.get('sellerProductCode', '')
|
|
original_product_name = po.get('productName', '')
|
|
product_name = code_to_name.get(final_item_code, original_product_name)
|
|
|
|
option_info = po.get('productOption', '')
|
|
if option_info:
|
|
order_list = f"{product_name} [{option_info}]=/{po.get('quantity', 1)}"
|
|
else:
|
|
order_list = f"{product_name}=/{po.get('quantity', 1)}"
|
|
|
|
row = {
|
|
'주문날짜': numeric_date,
|
|
'번호': idx + 1,
|
|
'주문번호': po.get('productOrderId', ''),
|
|
'수령인명': shippingAddress.get('name', ''),
|
|
'상품코드': final_item_code,
|
|
'상품명': product_name,
|
|
'수량': final_quantity,
|
|
'주소': full_address,
|
|
'우편번호': '',
|
|
'수령인 전화번호': shippingAddress.get('tel1', ''),
|
|
'수령인 휴대폰': shippingAddress.get('tel1', ''),
|
|
'배송시 요구사항': po.get('shippingMemo', ''),
|
|
'송장번호': '',
|
|
'쇼핑몰명': '스마트스토어',
|
|
'비고': '',
|
|
'주문번호(쇼핑몰)': order.get('orderId', ''),
|
|
'주문목록': order_list
|
|
}
|
|
data.append(row)
|
|
|
|
weekdays = ['월요일', '화요일', '수요일', '목요일', '금요일', '토요일', '일요일']
|
|
weekday_str = weekdays[now.weekday()]
|
|
if now.hour >= 15:
|
|
filename = f"{now.strftime('%m월 %d일')} ({weekday_str}) 1 미라네 스마트 스토어-3시.xlsx"
|
|
else:
|
|
filename = f"{now.strftime('%m월 %d일')} ({weekday_str}) 1 미라네 스마트 스토어-오전.xlsx"
|
|
|
|
file_bytes, enc_name = generate_order_excel(data, numeric_date, filename, return_bytes=True)
|
|
|
|
task_store[tid]["current"] = task_store[tid]["total"]
|
|
task_store[tid]["file_bytes"] = file_bytes
|
|
task_store[tid]["filename"] = enc_name
|
|
task_store[tid]["status"] = "completed"
|
|
|
|
except Exception as e:
|
|
task_store[tid]["status"] = "error"
|
|
task_store[tid]["error"] = str(e)
|
|
task_store[tid]["status_code"] = 500
|
|
|
|
background_tasks.add_task(run_task, task_id)
|
|
return {"status": "success", "task_id": task_id}
|
|
|
|
|
|
@app.post("/api/upload_invoices")
|
|
async def upload_invoices(file: UploadFile = File(...), deliveryCompanyCode: str = Form("CJGLS")):
|
|
try:
|
|
content = await file.read()
|
|
|
|
if file.filename.endswith('.xls'):
|
|
try:
|
|
df = pd.read_excel(io.BytesIO(content), engine='xlrd')
|
|
except Exception:
|
|
try:
|
|
df = pd.read_html(io.BytesIO(content))[0]
|
|
except Exception as e2:
|
|
return JSONResponse({"status": "error", "message": f"xls 파일 파싱 실패: {str(e2)}"}, status_code=400)
|
|
else:
|
|
df = pd.read_excel(io.BytesIO(content), engine='openpyxl')
|
|
|
|
if '주문번호(쇼핑몰)' not in df.columns or '송장번호' not in df.columns:
|
|
return JSONResponse({"status": "error", "message": "엑셀에 '주문번호(쇼핑몰)' 또는 '송장번호' 컬럼이 없습니다."}, status_code=400)
|
|
|
|
df = df[df['송장번호'].notna() & (df['송장번호'] != '')]
|
|
|
|
if df.empty:
|
|
return JSONResponse({"status": "error", "message": "송장번호가 입력된 데이터가 없습니다."}, status_code=400)
|
|
|
|
dispatch_list = []
|
|
for index, row in df.iterrows():
|
|
po_id = str(row['주문번호(쇼핑몰)']).replace('.0', '').strip()
|
|
tracking_num = str(row['송장번호']).replace('.0', '').strip()
|
|
|
|
if ',' in tracking_num:
|
|
tracking_num = tracking_num.split(',')[0].strip()
|
|
|
|
if po_id and tracking_num and tracking_num != 'nan':
|
|
dispatch_list.append({
|
|
"productOrderId": po_id,
|
|
"deliveryMethod": "DELIVERY",
|
|
"deliveryCompanyCode": deliveryCompanyCode,
|
|
"trackingNumber": tracking_num
|
|
})
|
|
|
|
if not dispatch_list:
|
|
return JSONResponse({"status": "error", "message": "유효한 송장 매핑 데이터가 없습니다."}, status_code=400)
|
|
|
|
token = get_access_token()
|
|
success_count = 0
|
|
fail_list = []
|
|
|
|
chunk_size = 100
|
|
for i in range(0, len(dispatch_list), chunk_size):
|
|
chunk = dispatch_list[i:i+chunk_size]
|
|
result = dispatch_orders(token, chunk)
|
|
|
|
results = result.get('data', {}).get('successProductOrderIds', [])
|
|
success_count += len(results)
|
|
|
|
fails = result.get('data', {}).get('failProductOrderInfos', [])
|
|
for f in fails:
|
|
fail_list.append(f)
|
|
|
|
return {
|
|
"status": "success",
|
|
"success_count": success_count,
|
|
"fail_count": len(fail_list),
|
|
"fails": fail_list,
|
|
"total_processed": len(dispatch_list)
|
|
}
|
|
|
|
except Exception as e:
|
|
return JSONResponse({"status": "error", "message": str(e)}, status_code=500)
|
|
|
|
from fastapi.responses import RedirectResponse
|
|
|
|
@app.get("/api/cafe24/login")
|
|
async def cafe24_login():
|
|
url = cafe24_api.get_auth_url()
|
|
return RedirectResponse(url)
|
|
|
|
@app.get("/api/cafe24/callback")
|
|
@app.get("/admin/oauth/callback")
|
|
async def cafe24_callback(code: str):
|
|
home = f"{APP_ROOT_PATH}/" if APP_ROOT_PATH else "/"
|
|
try:
|
|
cafe24_api.request_new_token(code)
|
|
return HTMLResponse(f"<script>alert('카페24 연동 성공!'); window.location.href='{home}';</script>")
|
|
except Exception as e:
|
|
return HTMLResponse(f"<script>alert('인증 실패: {e}'); window.location.href='{home}';</script>")
|
|
|
|
task_store = {}
|
|
|
|
@app.post("/api/cafe24/start_download")
|
|
async def start_cafe24_download(background_tasks: BackgroundTasks):
|
|
task_id = str(uuid.uuid4())
|
|
task_store[task_id] = {
|
|
"status": "processing",
|
|
"current": 0,
|
|
"total": 0,
|
|
"file_bytes": None,
|
|
"filename": None,
|
|
"error": None
|
|
}
|
|
|
|
def run_task(tid):
|
|
try:
|
|
total_count = cafe24_api.get_cafe24_orders_count(status="N20")
|
|
task_store[tid]["total"] = total_count
|
|
|
|
def progress_callback(current):
|
|
task_store[tid]["current"] = current
|
|
|
|
orders = cafe24_api.get_cafe24_orders(status="N20", progress_callback=progress_callback)
|
|
if not orders:
|
|
task_store[tid]["status"] = "error"
|
|
task_store[tid]["error"] = "새로운 배송준비중 주문이 없습니다."
|
|
return
|
|
|
|
data = []
|
|
now = datetime.datetime.now()
|
|
numeric_date = int(now.strftime('%Y%m%d'))
|
|
code_to_name = get_all_code_to_name_from_db()
|
|
exchange_customers = []
|
|
|
|
for idx, order in enumerate(orders):
|
|
items = order.get('items', [])
|
|
receivers = order.get('receivers', [])
|
|
receiver = receivers[0] if receivers else {}
|
|
|
|
has_exchange = False
|
|
for item in items:
|
|
if item.get('exchange_request_date') or item.get('exchange_date') or str(item.get('status_code', '')).startswith('E'):
|
|
has_exchange = True
|
|
break
|
|
|
|
if has_exchange:
|
|
exchange_customers.append({
|
|
"name": receiver.get('name', ''),
|
|
"phone": receiver.get('cellphone') or receiver.get('phone', ''),
|
|
"order_no": order.get('order_id', '')
|
|
})
|
|
continue
|
|
|
|
address_full = receiver.get('address_full', '')
|
|
if not address_full:
|
|
address_full = f"{receiver.get('address1', '')} {receiver.get('address2', '')}".strip()
|
|
full_address = address_full
|
|
|
|
for item in items:
|
|
if not item.get('status_code', '').startswith('N'):
|
|
continue
|
|
|
|
seller_code = item.get('custom_variant_code') or item.get('item_code') or item.get('custom_product_code') or item.get('product_code', '')
|
|
|
|
final_quantity = int(item.get('quantity', 1))
|
|
if seller_code and '_' in seller_code:
|
|
parts = seller_code.rsplit('_', 1)
|
|
if len(parts) == 2 and parts[1].isdigit():
|
|
seller_code = parts[0]
|
|
final_quantity = final_quantity * int(parts[1])
|
|
|
|
original_product_name = item.get('product_name', '')
|
|
product_name = code_to_name.get(seller_code, original_product_name)
|
|
original_quantity = item.get('quantity', 1)
|
|
|
|
option_info = item.get('option_value', '')
|
|
if option_info:
|
|
order_list = f"{original_product_name}/{option_info}/{original_quantity}"
|
|
else:
|
|
order_list = f"{original_product_name}/{original_quantity}"
|
|
|
|
row = {
|
|
'주문날짜': numeric_date,
|
|
'번호': 0,
|
|
'주문번호': item.get('order_item_code') or order.get('order_id', ''),
|
|
'수령인명': receiver.get('name', ''),
|
|
'상품코드': seller_code,
|
|
'상품명': product_name,
|
|
'수량': final_quantity,
|
|
'주소': full_address,
|
|
'우편번호': receiver.get('zipcode', ''),
|
|
'수령인 전화번호': receiver.get('phone', ''),
|
|
'수령인 휴대폰': receiver.get('cellphone', ''),
|
|
'배송시 요구사항': receiver.get('shipping_message', ''),
|
|
'송장번호': '',
|
|
'쇼핑몰명': '자사몰',
|
|
'비고': order.get('admin_additional_amount_name', ''),
|
|
'주문번호(쇼핑몰)': order.get('order_id', ''),
|
|
'주문목록': order_list
|
|
}
|
|
data.append(row)
|
|
|
|
weekdays = ['월요일', '화요일', '수요일', '목요일', '금요일', '토요일', '일요일']
|
|
weekday_str = weekdays[now.weekday()]
|
|
|
|
if now.hour >= 15:
|
|
filename = f"{now.strftime('%m월 %d일')} ({weekday_str}) 0 미라네 자사몰 발주-3시.xlsx"
|
|
else:
|
|
filename = f"{now.strftime('%m월 %d일')} ({weekday_str}) 0 미라네 자사몰 발주-오전.xlsx"
|
|
|
|
file_bytes, enc_name = generate_order_excel(data, numeric_date, filename, is_cafe24=True, return_bytes=True)
|
|
|
|
task_store[tid]["current"] = task_store[tid]["total"]
|
|
task_store[tid]["file_bytes"] = file_bytes
|
|
task_store[tid]["filename"] = enc_name
|
|
task_store[tid]["exchange_customers"] = exchange_customers
|
|
task_store[tid]["status"] = "completed"
|
|
|
|
except Exception as e:
|
|
status_code = 500
|
|
if "Authentication" in str(e) or "Please authenticate" in str(e) or "expired" in str(e):
|
|
status_code = 401
|
|
task_store[tid]["status"] = "error"
|
|
task_store[tid]["error"] = str(e)
|
|
task_store[tid]["status_code"] = status_code
|
|
|
|
background_tasks.add_task(run_task, task_id)
|
|
return {"status": "success", "task_id": task_id}
|
|
|
|
@app.get("/api/tasks/progress/{task_id}")
|
|
async def get_task_progress(task_id: str):
|
|
if task_id not in task_store:
|
|
return JSONResponse({"status": "error", "message": "작업을 찾을 수 없습니다."}, status_code=404)
|
|
|
|
task_data = task_store[task_id]
|
|
if task_data["status"] == "error":
|
|
status_code = task_data.get("status_code", 500)
|
|
return JSONResponse({"status": "error", "message": task_data["error"]}, status_code=status_code)
|
|
|
|
return {
|
|
"status": task_data["status"],
|
|
"current": task_data["current"],
|
|
"total": task_data["total"]
|
|
}
|
|
|
|
@app.get("/api/tasks/download_file/{task_id}")
|
|
async def download_task_file(task_id: str):
|
|
if task_id not in task_store or task_store[task_id]["status"] != "completed":
|
|
return JSONResponse({"status": "error", "message": "파일이 아직 준비되지 않았거나 만료되었습니다."}, status_code=404)
|
|
|
|
task_data = task_store[task_id]
|
|
output = io.BytesIO(task_data["file_bytes"])
|
|
enc_name = task_data["filename"]
|
|
|
|
headers = {
|
|
"Content-Disposition": f"attachment; filename*=UTF-8''{enc_name}"
|
|
}
|
|
|
|
# 메모리 정리 (1회 다운로드 후 삭제)
|
|
del task_store[task_id]
|
|
|
|
return StreamingResponse(
|
|
output,
|
|
media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
|
headers=headers
|
|
)
|
|
|
|
|
|
@app.post("/api/cafe24/upload_invoices")
|
|
async def cafe24_upload_invoices(file: UploadFile = File(...), deliveryCompanyCode: str = Form("0004")):
|
|
try:
|
|
content = await file.read()
|
|
if file.filename.endswith('.xls'):
|
|
try:
|
|
df = pd.read_excel(io.BytesIO(content), engine='xlrd')
|
|
except:
|
|
df = pd.read_html(io.BytesIO(content))[0]
|
|
else:
|
|
df = pd.read_excel(io.BytesIO(content), engine='openpyxl')
|
|
|
|
if '주문번호(쇼핑몰)' not in df.columns or '송장번호' not in df.columns:
|
|
return JSONResponse({"status": "error", "message": "엑셀에 '주문번호(쇼핑몰)' 또는 '송장번호' 컬럼이 없습니다."}, status_code=400)
|
|
|
|
df = df[df['송장번호'].notna() & (df['송장번호'] != '')]
|
|
if df.empty:
|
|
return JSONResponse({"status": "error", "message": "송장번호가 입력된 데이터가 없습니다."}, status_code=400)
|
|
|
|
dispatch_list = []
|
|
for index, row in df.iterrows():
|
|
po_id = str(row['주문번호(쇼핑몰)']).replace('.0', '').strip()
|
|
tracking_num = str(row['송장번호']).replace('.0', '').strip()
|
|
if ',' in tracking_num:
|
|
tracking_num = tracking_num.split(',')[0].strip()
|
|
|
|
if po_id and tracking_num and tracking_num != 'nan':
|
|
dispatch_list.append({
|
|
"order_id": po_id,
|
|
"tracking_no": tracking_num,
|
|
"shipping_company_code": deliveryCompanyCode
|
|
})
|
|
|
|
results = cafe24_api.update_cafe24_tracking(dispatch_list)
|
|
return {
|
|
"status": "success",
|
|
"success_count": len(results["success"]),
|
|
"fail_count": len(results["fail"]),
|
|
"fails": results["fail"],
|
|
"total_processed": len(dispatch_list)
|
|
}
|
|
except Exception as e:
|
|
status_code = 500
|
|
if "Authentication" in str(e) or "Please authenticate" in str(e) or "expired" in str(e):
|
|
status_code = 401
|
|
return JSONResponse({"status": "error", "message": str(e)}, status_code=status_code)
|
|
|
|
# ==========================================
|
|
# 반품 관리 API
|
|
# ==========================================
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
uvicorn.run(app, host="0.0.0.0", port=int(os.getenv("APP_PORT", "8002")))
|