276 lines
11 KiB
Python
276 lines
11 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""平台订单提取(柔汇/子元素):写入 platform_order_data 表。
|
||
|
||
最小包装自 rouhui_extract_v3.py,对应 platform-order-rouhui。
|
||
"""
|
||
|
||
import time
|
||
import json
|
||
import hmac
|
||
import hashlib
|
||
import random
|
||
import requests
|
||
import urllib3
|
||
from urllib.parse import urlparse, parse_qs, quote
|
||
from collections import defaultdict
|
||
from datetime import date, timedelta, datetime
|
||
|
||
from prefect import flow
|
||
|
||
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
||
|
||
# ===================== 柔汇openapi配置 =====================
|
||
host_rouhui = "https://openapi.rouhuipod.com"
|
||
app_id_rouhui = "smt01zi783662727"
|
||
app_key_rouhui = "fd1bec5b190e5c52b6a15d9fe1086fe9"
|
||
redirect_uri_rouhui = "https://demo-callback.virtual-api.com/oauth/callback"
|
||
login_url_rouhui = f"{host_rouhui}/openapi/v2/oauth2/login"
|
||
token_url_rouhui = f"{host_rouhui}/openapi/v2/oauth2/token"
|
||
order_list_url_rouhui = "/openapi/v2/order"
|
||
|
||
PLATFORM_MAP = {
|
||
2: "Temu-半托管", 3: "亚马逊", 4: "Tiktok-自运营",
|
||
9: "SHEIN-半托管", 10: "Temu-全托管", 12: "Temu-本土店", 13: "Shein-自运营"
|
||
}
|
||
PLATFORM_ID_LIST = [2, 3, 4, 9, 10, 12, 13]
|
||
SHORT_PLATFORM_MAP = {
|
||
"Temu-半托管": "Temu", "亚马逊": "亚马逊", "Tiktok-自运营": "tiktok",
|
||
"SHEIN-半托管": "shein", "Temu-全托管": "Temu", "Temu-本土店": "Temu", "Shein-自运营": "shein"
|
||
}
|
||
|
||
rouhui_login_form = {
|
||
"_token": "merchant", "merchant": "smt01",
|
||
"username": "smt01", "password": "Sanmu111.",
|
||
"app_id": app_id_rouhui, "redirect_uri": redirect_uri_rouhui
|
||
}
|
||
|
||
LOOKBACK_DAYS = 30 # 回查天数
|
||
HISTORY_COMPLEMENT_MODE = False # True=历史补数模式；False=日常增量模式
|
||
|
||
def log_msg(msg, level="info"):
|
||
ts = time.strftime('%H:%M:%S')
|
||
tag = {"success":"[OK]","error":"[ERR]"}.get(level,"")
|
||
print(f"[{ts}] {tag} {msg}")
|
||
|
||
def format_date_only(date_str):
|
||
if not date_str: return ""
|
||
return str(date_str).split(" ")[0]
|
||
|
||
# ===================== 柔汇签名与登录 =====================
|
||
def get_rand(): return str(random.randint(100000, 999999))
|
||
def get_ts(): return str(int(time.time()))
|
||
|
||
def calc_sign(params):
|
||
clean = {}
|
||
for k, v in params.items():
|
||
if v not in (None, ""): clean[k] = v
|
||
buf = []
|
||
for k in sorted(clean.keys()):
|
||
v = clean[k]
|
||
if isinstance(v, (dict, list)):
|
||
v = json.dumps(v, ensure_ascii=False, separators=(',', ':'))
|
||
buf.append(f"{k}={v}")
|
||
return hmac.new(app_key_rouhui.encode("utf-8"), "&".join(buf).encode("utf-8"), hashlib.sha256).hexdigest()
|
||
|
||
def rouhui_get_token():
|
||
"""登录获取access_token"""
|
||
headers = {
|
||
"Content-Type": "application/x-www-form-urlencoded",
|
||
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
|
||
"Referer": f"{host_rouhui}/openapi/v2/oauth2?app_id={app_id_rouhui}&redirect_uri={quote(redirect_uri_rouhui)}",
|
||
"Origin": host_rouhui
|
||
}
|
||
try:
|
||
resp = requests.post(login_url_rouhui, data=rouhui_login_form, headers=headers, allow_redirects=False, verify=False, timeout=15)
|
||
redirect_url = resp.headers.get("Location")
|
||
if not redirect_url:
|
||
log_msg("柔汇登录无302跳转", "error"); return ""
|
||
code = parse_qs(urlparse(redirect_url).query)["code"][0]
|
||
log_msg("柔汇登录成功", "success")
|
||
except Exception as e:
|
||
log_msg(f"柔汇登录失败: {e}", "error"); return ""
|
||
|
||
params = {"code": code, "app_id": app_id_rouhui, "rand_str": get_rand(), "timestamp": get_ts()}
|
||
params["signature"] = calc_sign(params)
|
||
try:
|
||
resp = requests.post(token_url_rouhui, json=params, headers={"Content-Type": "application/json"}, timeout=10, verify=False)
|
||
data = resp.json()
|
||
if data.get("code") == 1:
|
||
log_msg("柔汇token获取成功", "success")
|
||
return data["data"]["access_token"]
|
||
log_msg(f"柔汇token失败: {data.get('message')}", "error")
|
||
except Exception as e:
|
||
log_msg(f"柔汇token网络异常: {e}", "error")
|
||
return ""
|
||
|
||
# ✅增加入参 begin_date,end_date，不再读取外部变量
|
||
def rouhui_fetch_platform_order(access_token, platform_id, begin_date, end_date):
|
||
"""拉取指定平台订单"""
|
||
bd = begin_date
|
||
ed = end_date
|
||
all_orders = []
|
||
page = 1
|
||
page_size = 200
|
||
headers = {"Content-Type": "application/json", "Authorization": f"Bearer {access_token}"}
|
||
full_name = PLATFORM_MAP[platform_id]
|
||
|
||
while True:
|
||
query = {"page": page, "page_size": page_size, "platform_id": platform_id,
|
||
"status": 3, "create_time": [bd, ed]}
|
||
common = {"app_id": app_id_rouhui, "rand_str": get_rand(), "timestamp": get_ts()}
|
||
all_params = {**query, **common}
|
||
all_params["signature"] = calc_sign(all_params)
|
||
|
||
try:
|
||
resp = requests.post(f"{host_rouhui}{order_list_url_rouhui}", json=all_params, headers=headers, timeout=15, verify=False)
|
||
data = resp.json()
|
||
if data.get("code") != 1:
|
||
log_msg(f"{full_name} 查询失败: {data.get('message')}", "error")
|
||
break
|
||
lst = data["data"].get("list", [])
|
||
if not lst: break
|
||
all_orders.extend(lst)
|
||
if len(lst) < page_size: break
|
||
page += 1
|
||
time.sleep(0.2)
|
||
except Exception as e:
|
||
log_msg(f"{full_name} page{page}异常: {e}", "error")
|
||
time.sleep(0.5)
|
||
|
||
return all_orders
|
||
|
||
# ===================== 主流程 =====================
|
||
@flow(name="platform-order-rouhui", log_prints=True)
|
||
def rouhui_platform_sync():
|
||
TODAY = date.today()
|
||
if HISTORY_COMPLEMENT_MODE:
|
||
# 历史补数模式： [T‑2*N , T‑N]
|
||
start_dt = TODAY - timedelta(days=LOOKBACK_DAYS * 2)
|
||
end_dt = TODAY - timedelta(days=LOOKBACK_DAYS)
|
||
else:
|
||
# 日常增量模式： [T‑N , T]
|
||
start_dt = TODAY - timedelta(days=LOOKBACK_DAYS)
|
||
end_dt = TODAY
|
||
|
||
START_DATE = start_dt.isoformat()
|
||
END_DATE = end_dt.isoformat()
|
||
|
||
log_msg(f"===== 子元素(柔汇)订单提取 v2.0 =====")
|
||
log_msg(f"查询时间: {START_DATE} ~ {END_DATE} (回查{LOOKBACK_DAYS}天)")
|
||
|
||
merged_detail = []
|
||
rouhui_stat = {}
|
||
rouhui_total_order = 0
|
||
rouhui_total_qty = 0
|
||
|
||
try:
|
||
rh_token = rouhui_get_token()
|
||
if not rh_token:
|
||
log_msg("柔汇登录失败，终止", "error")
|
||
return
|
||
|
||
for pid in PLATFORM_ID_LIST:
|
||
full_name = PLATFORM_MAP[pid]
|
||
short_name = SHORT_PLATFORM_MAP[full_name]
|
||
log_msg(f"\n拉取柔汇【{full_name}】(pid={pid})")
|
||
# ✅调用时把main内部日期传进去
|
||
order_list = rouhui_fetch_platform_order(rh_token, pid, START_DATE, END_DATE)
|
||
p_order_cnt = len(order_list)
|
||
p_qty = 0
|
||
rouhui_stat[short_name] = [p_order_cnt, 0]
|
||
|
||
if not order_list:
|
||
log_msg(f" {full_name}: 无订单")
|
||
continue
|
||
|
||
for order in order_list:
|
||
order_no = order.get("order_no", "")
|
||
store = order.get("store_name", "未知")
|
||
create_t = order.get("create_time", "")
|
||
create_date_only = format_date_only(create_t)
|
||
items = order.get("order_items", [])
|
||
total_qty = sum(i.get("third_qty", 1) for i in items)
|
||
p_qty += total_qty
|
||
merged_detail.append(["墨西哥", "墨西哥", short_name, store, order_no, total_qty, create_date_only, "子元素", "已发货"])
|
||
|
||
rouhui_stat[short_name][1] = p_qty
|
||
rouhui_total_order += p_order_cnt
|
||
rouhui_total_qty += p_qty
|
||
log_msg(f" {full_name}: {p_order_cnt}单, {p_qty}件")
|
||
|
||
except Exception as e:
|
||
log_msg(f"柔汇API异常: {e}", "error")
|
||
|
||
# 统一平台名称
|
||
for row in merged_detail:
|
||
if row[2].upper() == "TEMU": row[2] = "Temu"
|
||
|
||
# 拆分Tiktok/非Tiktok
|
||
tiktok_rows = [row for row in merged_detail if "tiktok" in row[2].lower()]
|
||
non_tiktok_rows = [row for row in merged_detail if "tiktok" not in row[2].lower()]
|
||
|
||
# 统计
|
||
nt_plat_stat = defaultdict(lambda: [0, 0])
|
||
nt_country_stat = defaultdict(lambda: [0, 0])
|
||
for row in non_tiktok_rows:
|
||
nt_plat_stat[row[2]][0] += 1; nt_plat_stat[row[2]][1] += row[5]
|
||
nt_country_stat[row[1]][0] += 1; nt_country_stat[row[1]][1] += row[5]
|
||
|
||
tk_plat_stat = defaultdict(lambda: [0, 0])
|
||
tk_country_stat = defaultdict(lambda: [0, 0])
|
||
for row in tiktok_rows:
|
||
tk_plat_stat[row[2]][0] += 1; tk_plat_stat[row[2]][1] += row[5]
|
||
tk_country_stat[row[1]][0] += 1; tk_country_stat[row[1]][1] += row[5]
|
||
|
||
total_qty = rouhui_total_qty
|
||
|
||
log_msg(f"\n========== 统计汇总 ==========")
|
||
log_msg(f"非Tiktok总件数：{sum(v[1] for v in nt_plat_stat.values())} 件")
|
||
log_msg(f"Tiktok总件数：{sum(v[1] for v in tk_plat_stat.values())} 件")
|
||
log_msg(f"全部商品总件数：{total_qty} 件")
|
||
|
||
# ===== 导入PostgreSQL =====
|
||
try:
|
||
import psycopg2
|
||
conn = psycopg2.connect(host="192.168.124.92", port=5432, user="postgres", password="Inkreach123456.", database="smt")
|
||
cur = conn.cursor()
|
||
cur.execute("""
|
||
CREATE TABLE IF NOT EXISTS platform_order_data (
|
||
id SERIAL PRIMARY KEY, site VARCHAR(50), country VARCHAR(50),
|
||
platform VARCHAR(50), store_name VARCHAR(200), order_no VARCHAR(100),
|
||
quantity INTEGER, order_date DATE, order_system VARCHAR(50),
|
||
order_status VARCHAR(100),
|
||
update_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||
UNIQUE(order_no, order_system)
|
||
)
|
||
""")
|
||
conn.commit()
|
||
|
||
now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
db_rows = [(r[0], r[1], r[2], r[3], r[4], r[5], r[6], r[7], r[8], now_str) for r in merged_detail]
|
||
|
||
sql = """
|
||
INSERT INTO platform_order_data (site, country, platform, store_name, order_no, quantity, order_date, order_system, order_status, update_time)
|
||
VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
|
||
ON CONFLICT (order_no, order_system) DO UPDATE SET
|
||
quantity = EXCLUDED.quantity,
|
||
order_date = EXCLUDED.order_date,
|
||
order_status = EXCLUDED.order_status,
|
||
update_time = EXCLUDED.update_time
|
||
"""
|
||
batch_size = 500
|
||
for i in range(0, len(db_rows), batch_size):
|
||
cur.executemany(sql, db_rows[i:i + batch_size])
|
||
conn.commit()
|
||
|
||
cur.execute("SELECT COUNT(*) FROM platform_order_data")
|
||
cnt = cur.fetchone()[0]
|
||
cur.close(); conn.close()
|
||
log_msg(f"已同步到PostgreSQL, 总{cnt}条", "success")
|
||
except Exception as e:
|
||
log_msg(f"PostgreSQL异常: {e}", "error")
|
||
|
||
log_msg("===== 提取完成 =====")
|
||
|
||
if __name__ == "__main__":
|
||
rouhui_platform_sync.serve(name="platform-order-rouhui", cron="*/15 * * * *") |