init: dbx-corm 프로젝트 Docker 기반 마이그레이션
주요 변경사항: - DB 접속정보를 환경변수(.env) 방식으로 분리 - orderlist_app → orderlist_db 변경 (main.py, routers/returns.py) - 네이버/카페24 API 키도 환경변수로 분리 - Dockerfile / docker-compose.yml 작성 (외부 PostgreSQL 네트워크 연결) - .gitignore: 민감파일(.env, cafe24_tokens.json, manual-ordering.json) 제외 - 불필요 파일 정리 (venv, __pycache__, 일회성 스크립트, xlsx 등) - README.md / .env.example / push_to_gitea.bat 추가 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
+155
@@ -0,0 +1,155 @@
|
||||
import os
|
||||
import requests
|
||||
import time
|
||||
import bcrypt
|
||||
import base64
|
||||
|
||||
CLIENT_ID = os.getenv("NAVER_CLIENT_ID", "")
|
||||
CLIENT_SECRET = os.getenv("NAVER_CLIENT_SECRET", "")
|
||||
|
||||
def get_access_token():
|
||||
timestamp = str(int(time.time() * 1000))
|
||||
pwd = f"{CLIENT_ID}_{timestamp}"
|
||||
|
||||
hashed = bcrypt.hashpw(pwd.encode('utf-8'), CLIENT_SECRET.encode('utf-8'))
|
||||
client_secret_sign = base64.urlsafe_b64encode(hashed).decode('utf-8')
|
||||
|
||||
url = "https://api.commerce.naver.com/external/v1/oauth2/token"
|
||||
payload = {
|
||||
"client_id": CLIENT_ID,
|
||||
"timestamp": timestamp,
|
||||
"client_secret_sign": client_secret_sign,
|
||||
"grant_type": "client_credentials",
|
||||
"type": "SELF"
|
||||
}
|
||||
|
||||
# URL encoded post
|
||||
headers = {
|
||||
"Content-Type": "application/x-www-form-urlencoded"
|
||||
}
|
||||
|
||||
response = requests.post(url, data=payload, headers=headers)
|
||||
if response.status_code == 200:
|
||||
return response.json().get('access_token')
|
||||
else:
|
||||
raise Exception(f"Failed to get token: {response.text}")
|
||||
|
||||
def get_payment_done_orders(token):
|
||||
url = "https://api.commerce.naver.com/external/v1/pay-order/seller/product-orders/last-changed-statuses"
|
||||
headers = {
|
||||
"Authorization": f"Bearer {token}"
|
||||
}
|
||||
|
||||
all_statuses = []
|
||||
|
||||
# 네이버 API는 lastChangedFrom과 lastChangedTo의 간격이 최대 24시간입니다.
|
||||
# 따라서 최근 7일의 데이터를 가져오기 위해 24시간 단위로 반복 조회합니다.
|
||||
now = time.time()
|
||||
for days_ago in range(7, 0, -1):
|
||||
start_time = now - (days_ago * 24 * 3600)
|
||||
end_time = now - ((days_ago - 1) * 24 * 3600)
|
||||
|
||||
if days_ago == 1:
|
||||
end_time = now
|
||||
|
||||
from_date = time.strftime('%Y-%m-%dT%H:%M:%S.000+09:00', time.localtime(start_time))
|
||||
to_date = time.strftime('%Y-%m-%dT%H:%M:%S.000+09:00', time.localtime(end_time))
|
||||
|
||||
params = {
|
||||
"lastChangedFrom": from_date,
|
||||
"lastChangedTo": to_date,
|
||||
"lastChangedType": "PAYED"
|
||||
}
|
||||
|
||||
response = requests.get(url, headers=headers, params=params)
|
||||
if response.status_code == 200:
|
||||
data = response.json()
|
||||
statuses = data.get('data', {}).get('lastChangeStatuses', [])
|
||||
all_statuses.extend(statuses)
|
||||
else:
|
||||
raise Exception(f"Failed to fetch orders ({from_date}): {response.text}")
|
||||
|
||||
return all_statuses
|
||||
|
||||
def get_product_order_details(token, product_order_ids):
|
||||
if not product_order_ids:
|
||||
return []
|
||||
|
||||
url = "https://api.commerce.naver.com/external/v1/pay-order/seller/product-orders/query"
|
||||
headers = {
|
||||
"Authorization": f"Bearer {token}",
|
||||
"Content-Type": "application/json"
|
||||
}
|
||||
payload = {
|
||||
"productOrderIds": product_order_ids
|
||||
}
|
||||
response = requests.post(url, headers=headers, json=payload)
|
||||
if response.status_code == 200:
|
||||
return response.json().get('data', [])
|
||||
else:
|
||||
raise Exception(f"Failed to fetch order details: {response.text}")
|
||||
|
||||
def dispatch_orders(token, dispatch_list):
|
||||
"""
|
||||
dispatch_list 형태:
|
||||
[
|
||||
{
|
||||
"productOrderId": "...",
|
||||
"deliveryMethod": "DELIVERY",
|
||||
"deliveryCompanyCode": "CJGLS",
|
||||
"trackingNumber": "..."
|
||||
}
|
||||
]
|
||||
"""
|
||||
url = "https://api.commerce.naver.com/external/v1/pay-order/seller/product-orders/dispatch"
|
||||
headers = {
|
||||
"Authorization": f"Bearer {token}",
|
||||
"Content-Type": "application/json"
|
||||
}
|
||||
|
||||
# 발송일은 현재 시간으로 셋팅
|
||||
dispatch_date = time.strftime('%Y-%m-%dT%H:%M:%S+09:00')
|
||||
for item in dispatch_list:
|
||||
item["dispatchDate"] = dispatch_date
|
||||
|
||||
payload = {
|
||||
"dispatchProductOrders": dispatch_list
|
||||
}
|
||||
|
||||
response = requests.post(url, headers=headers, json=payload)
|
||||
if response.status_code == 200:
|
||||
return response.json()
|
||||
else:
|
||||
raise Exception(f"Failed to dispatch orders: {response.text}")
|
||||
|
||||
def confirm_orders(token, product_order_ids):
|
||||
"""주문건을 발주확인(배송준비중) 상태로 변경합니다."""
|
||||
if not product_order_ids:
|
||||
return {}
|
||||
|
||||
url = "https://api.commerce.naver.com/external/v1/pay-order/seller/product-orders/confirm"
|
||||
headers = {
|
||||
"Authorization": f"Bearer {token}",
|
||||
"Content-Type": "application/json"
|
||||
}
|
||||
payload = {
|
||||
"productOrderIds": product_order_ids
|
||||
}
|
||||
|
||||
response = requests.post(url, headers=headers, json=payload)
|
||||
if response.status_code == 200:
|
||||
return response.json()
|
||||
else:
|
||||
raise Exception(f"Failed to confirm orders: {response.text}")
|
||||
|
||||
def get_product_details(token, product_id):
|
||||
"""지정된 상품의 전체 상세 정보(옵션 설정 포함)를 조회합니다."""
|
||||
url = f"https://api.commerce.naver.com/external/v2/products/channel-products/{product_id}"
|
||||
headers = {
|
||||
"Authorization": f"Bearer {token}"
|
||||
}
|
||||
response = requests.get(url, headers=headers)
|
||||
|
||||
if response.status_code == 200:
|
||||
return response.json()
|
||||
return {}
|
||||
Reference in New Issue
Block a user