475 lines
18 KiB
Python
475 lines
18 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""平台订单提取(SDSPOD/印美达):写入 platform_order_data 表。
|
||
|
||
最小包装自 sdspod_extract_v5.py,对应 platform-order-sdspod。
|
||
"""
|
||
|
||
import time
|
||
import json
|
||
import asyncio
|
||
import aiohttp
|
||
import math
|
||
import requests
|
||
import urllib3
|
||
from collections import defaultdict
|
||
from datetime import date, timedelta, datetime
|
||
|
||
from prefect import flow
|
||
|
||
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
||
|
||
# ===================== 共享配置 =====================
|
||
COMPANY_NO = "印美达"
|
||
EXTRA_INFO = "eyJpcCI6IjM4LjcxLjEyNy4xODkiLCJsb2NhdGlvbiI6eyJsYXQiOjM4Ljg4MzMsImxuZyI6LTc3fSwiYWRfaW5mbyI6eyJuYXRpb24iOiLnvo7lm70iLCJwcm92aW5jZSI6IiIsImNpdHkiOiIiLCJkaXN0cmljdCI6IiIsImFkY29kZSI6LTEsIm5hdGlvbl9jb2RlIjo4NDB9LCJmaW5nZXJwcmludElkIjoiN2M3YzllMDUxYWU2NDkzOTEzMGY5OWI0NjhiY2FhYTUiLCJ1c2VyQWdlbnQiOiJNb3ppbGxhLzUuMCAoV2luZG93cyBOVCAxMC4wOyBXaW4sgeDY0KSBBcHBsZVdlYktpdC81MzcuMzYgKEtIVE1MLCBsaWtlIEdl2tvKSBDaHJvbWUvMTM4LjAuMC4wIFNhZmFyaS81MzcuMzyiLCJocmVmIjoiaHR0cHM6Ly9tYW5hZ2Uuc2RzcG9kLmNvbS91c2VyL2xvZ2luIn0="
|
||
USERNAME_SD = "叶富华"
|
||
PASSWORD_SD = "Aa123456."
|
||
MAX_RETRY = 3
|
||
REQUEST_TIMEOUT = 30
|
||
|
||
# ===================== 国家映射（共用） =====================
|
||
COUNTRY_MAP = {
|
||
"United States": "美国", "America": "美国", "USA": "美国",
|
||
"Mexico": "墨西哥", "Canada": "加拿大",
|
||
"United Kingdom": "英国", "UK": "英国",
|
||
"Germany": "德国", "France": "法国", "Italy": "意大利",
|
||
"Spain": "西班牙", "Portugal": "葡萄牙",
|
||
"Netherlands": "荷兰", "Belgium": "比利时",
|
||
"Switzerland": "瑞士", "Sweden": "瑞典", "Norway": "挪威",
|
||
"Denmark": "丹麦", "Finland": "芬兰", "Poland": "波兰",
|
||
"Czech Republic": "捷克", "Czech": "捷克",
|
||
"Austria": "奥地利", "Hungary": "匈牙利",
|
||
"Romania": "罗马尼亚", "Greece": "希腊",
|
||
"Ireland": "爱尔兰", "Bulgaria": "保加利亚",
|
||
"Croatia": "克罗地亚", "Slovakia": "斯洛伐克",
|
||
"Slovenia": "斯洛文尼亚", "Lithuania": "立陶宛",
|
||
"Latvia": "拉脱维亚", "Estonia": "爱沙尼亚",
|
||
"Luxembourg": "卢森堡", "Malta": "马耳他",
|
||
"Cyprus": "塞浦路斯", "Iceland": "冰岛",
|
||
"Liechtenstein": "列支敦士登", "Monaco": "摩纳哥", "Andorra": "安道尔",
|
||
"Australia": "澳大利亚", "New Zealand": "新西兰",
|
||
"Japan": "日本", "South Korea": "韩国",
|
||
"Brazil": "巴西", "Argentina": "阿根廷",
|
||
"Chile": "智利", "Colombia": "哥伦比亚", "Peru": "秘鲁",
|
||
"India": "印度", "Russia": "俄罗斯", "Turkey": "土耳其",
|
||
"Israel": "以色列",
|
||
"UAE": "阿联酋", "United Arab Emirates": "阿联酋",
|
||
"Saudi Arabia": "沙特阿拉伯",
|
||
"Singapore": "新加坡", "Malaysia": "马来西亚",
|
||
"Indonesia": "印度尼西亚", "Thailand": "泰国",
|
||
"Philippines": "菲律宾", "Vietnam": "越南",
|
||
}
|
||
|
||
EUROPEAN_COUNTRIES = {
|
||
"德国", "法国", "意大利", "西班牙", "葡萄牙", "荷兰", "比利时",
|
||
"瑞士", "瑞典", "挪威", "丹麦", "芬兰", "波兰", "捷克", "奥地利",
|
||
"匈牙利", "罗马尼亚", "希腊", "爱尔兰", "保加利亚", "克罗地亚", "斯洛伐克",
|
||
"斯洛文尼亚", "立陶宛", "拉脱维亚", "爱沙尼亚", "卢森堡", "马耳他", "塞浦路斯",
|
||
"冰岛", "列支敦士登", "摩纳哥", "安道尔",
|
||
"Germany", "France", "Italy", "Spain", "Portugal",
|
||
"Netherlands", "Belgium", "Switzerland", "Sweden", "Norway", "Denmark",
|
||
"Finland", "Poland", "Czech Republic", "Austria", "Hungary", "Romania",
|
||
"Greece", "Ireland", "Bulgaria", "Croatia", "Slovakia", "Slovenia",
|
||
"Lithuania", "Latvia", "Estonia", "Luxembourg", "Malta", "Cyprus",
|
||
"Iceland", "Liechtenstein", "Monaco", "Andorra"
|
||
}
|
||
|
||
def to_chinese_country(name):
|
||
if not name: return ""
|
||
return COUNTRY_MAP.get(name.strip(), name.strip())
|
||
|
||
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 sd_login_sync():
|
||
ts = int(time.time() * 1000)
|
||
url = f"https://pod-api.sdspod.com/pod/auth/login?t={ts}"
|
||
headers = {"content-type": "application/json;charset=UTF-8"}
|
||
payload = {"no": COMPANY_NO, "password": PASSWORD_SD, "username": USERNAME_SD, "extraInfo": EXTRA_INFO}
|
||
for i in range(MAX_RETRY):
|
||
try:
|
||
resp = requests.post(url, headers=headers, json=payload, timeout=REQUEST_TIMEOUT)
|
||
data = resp.json()
|
||
if data.get("ret") == 0:
|
||
token = data.get("data", {}).get("accessToken")
|
||
if token:
|
||
log_msg("SDSPOD登录成功", "success")
|
||
return token
|
||
log_msg(f"SD登录失败: {data.get('msg')}", "error")
|
||
return ""
|
||
except Exception as e:
|
||
if i < MAX_RETRY - 1:
|
||
time.sleep(1)
|
||
return ""
|
||
|
||
# ===================== 数据库通用插入函数 =====================
|
||
def insert_orders_to_db(order_rows, order_system):
|
||
"""将统一格式的订单列表插入数据库，order_system 用于区分来源"""
|
||
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 = []
|
||
for r in order_rows:
|
||
# r = [站点, 国家, 平台, 店铺, 订单号, 数量, 日期, 订单系统, 状态]
|
||
order_date = r[6] if r[6] else None
|
||
db_rows.append((r[0], r[1], r[2], r[3], r[4], r[5], order_date, order_system, r[8], now_str))
|
||
|
||
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):
|
||
batch = db_rows[i:i + batch_size]
|
||
cur.executemany(sql, batch)
|
||
conn.commit()
|
||
|
||
cur.execute("SELECT COUNT(*) FROM platform_order_data")
|
||
cnt = cur.fetchone()[0]
|
||
cur.close()
|
||
conn.close()
|
||
log_msg(f"已同步到PostgreSQL, 总{cnt}条 (来源: {order_system})", "success")
|
||
return True
|
||
except Exception as e:
|
||
log_msg(f"PostgreSQL导入异常 ({order_system}): {str(e)}", "error")
|
||
return False
|
||
|
||
# ======================================================================
|
||
# 脚本1：按支付时间拉取最近30天订单（同步，productionType=1，仅一个商户）
|
||
# ======================================================================
|
||
def fetch_one_page_pay(token, page, pay_begin, pay_end):
|
||
url = "https://pod-api.sdspod.com/pod/orders/page"
|
||
headers = {"access-token": token}
|
||
params = {
|
||
"merchantIds": "35876",
|
||
"productionType": 1,
|
||
"page": page,
|
||
"size": 100,
|
||
"originType": "normal",
|
||
"payDateBegin": pay_begin,
|
||
"payDateEnd": pay_end,
|
||
"t": int(time.time() * 1000)
|
||
}
|
||
for attempt in range(MAX_RETRY):
|
||
try:
|
||
resp = requests.get(url, headers=headers, params=params, timeout=REQUEST_TIMEOUT)
|
||
return resp.json()
|
||
except Exception as e:
|
||
log_msg(f"支付单第{page}页请求异常:{str(e)}，重试 {attempt+1}/{MAX_RETRY}")
|
||
time.sleep(0.8)
|
||
return None
|
||
|
||
def fetch_all_orders_by_pay(token):
|
||
now = datetime.now()
|
||
pay_end = now.strftime("%Y-%m-%d 23:59:59")
|
||
pay_begin = (now - timedelta(days=30)).strftime("%Y-%m-%d 00:00:00")
|
||
|
||
|
||
# now = datetime.now()
|
||
# pay_end = (now - timedelta(days=120)).strftime("%Y-%m-%d 23:59:59")
|
||
# pay_begin = (now - timedelta(days=280)).strftime("%Y-%m-%d 00:00:00")
|
||
|
||
log_msg(f"[脚本1] 筛选支付时间：{pay_begin} ~ {pay_end}")
|
||
|
||
all_list = []
|
||
first_resp = fetch_one_page_pay(token, page=1, pay_begin=pay_begin, pay_end=pay_end)
|
||
if not first_resp:
|
||
log_msg("[脚本1] 第一页获取失败，终止拉取", "error")
|
||
return all_list
|
||
|
||
total_count = first_resp.get("totalCount", 0)
|
||
if total_count <= 0:
|
||
log_msg("[脚本1] 该时间段没有订单数据")
|
||
return all_list
|
||
|
||
total_pages = math.ceil(total_count / 100)
|
||
log_msg(f"[脚本1] 总订单:{total_count} 条，共{total_pages}页")
|
||
all_list.extend(first_resp.get("list", []))
|
||
|
||
for pg in range(2, total_pages + 1):
|
||
page_resp = fetch_one_page_pay(token, pg, pay_begin=pay_begin, pay_end=pay_end)
|
||
if page_resp and "list" in page_resp:
|
||
page_data = page_resp.get("list", [])
|
||
all_list.extend(page_data)
|
||
time.sleep(0.3)
|
||
log_msg(f"[脚本1] 拉取完成，合计 {len(all_list)} 条订单")
|
||
return all_list
|
||
|
||
def parse_order_item_pay(item):
|
||
"""
|
||
解析为统一格式：
|
||
[站点, 国家, 平台, 店铺名称, 订单号, 商品数量, 下单日期, 订单系统, 订单状态]
|
||
"""
|
||
order_no = item.get("no", "")
|
||
country_raw = item.get("address", {}).get("countryName", "")
|
||
country = to_chinese_country(country_raw)
|
||
store_name = item.get("merchantStore", {}).get("name", "")
|
||
platform = item.get("merchantStore", {}).get("platformName", "") # 可能不存在
|
||
gmt_created = item.get("gmtCreated", "")
|
||
order_date = gmt_created.split(" ")[0] if gmt_created else ""
|
||
|
||
total_qty = 0
|
||
for parcel in item.get("parcelList", []):
|
||
total_qty += parcel.get("totalQty", 0)
|
||
|
||
if country_raw in EUROPEAN_COUNTRIES or country in EUROPEAN_COUNTRIES:
|
||
country = "欧区"
|
||
site = country # 站点暂用国家
|
||
else:
|
||
site = country
|
||
|
||
status = str(item.get("status", ""))
|
||
|
||
return [site, country, platform, store_name, order_no, total_qty, order_date, "SDSPOD", status]
|
||
|
||
def main_script1():
|
||
log_msg("===== 脚本1：按支付时间拉取最近30天订单 =====")
|
||
token = sd_login_sync()
|
||
if not token:
|
||
log_msg("[脚本1] 登录失败，终止", "error")
|
||
return
|
||
|
||
orders = fetch_all_orders_by_pay(token)
|
||
if not orders:
|
||
log_msg("[脚本1] 没有订单数据")
|
||
return
|
||
|
||
parsed_rows = []
|
||
for item in orders:
|
||
if item.get("status") == 98:
|
||
continue
|
||
row = parse_order_item_pay(item)
|
||
parsed_rows.append(row)
|
||
|
||
log_msg(f"[脚本1] 有效订单: {len(parsed_rows)} 单")
|
||
if parsed_rows:
|
||
insert_orders_to_db(parsed_rows, "SDSPOD")
|
||
log_msg("===== 脚本1 完成 =====")
|
||
|
||
# ======================================================================
|
||
# 脚本2：按创建时间回查19天订单（异步，productionType=2，多商户）
|
||
# ======================================================================
|
||
MERCHANT_IDS = "35876,45637,52649,52699,35956,60756"
|
||
PRODUCTION_TYPE = 2
|
||
PAGE_SIZE_SD = 100
|
||
MAX_CONCURRENT = 5
|
||
LOOKBACK_DAYS = 7
|
||
|
||
def format_date_only(date_str):
|
||
if not date_str: return ""
|
||
return str(date_str).split(" ")[0]
|
||
|
||
async def sd_fetch_page(session, sem, params, pg, total):
|
||
async with sem:
|
||
for i in range(MAX_RETRY):
|
||
try:
|
||
timeout = aiohttp.ClientTimeout(total=REQUEST_TIMEOUT)
|
||
async with session.get("https://pod-api.sdspod.com/pod/orders/page", params=params, timeout=timeout) as resp:
|
||
data = await resp.json()
|
||
return data.get("list", [])
|
||
except:
|
||
if i < MAX_RETRY - 1:
|
||
await asyncio.sleep(0.5)
|
||
log_msg(f"SD第{pg}/{total}页拉取失败", "error")
|
||
return []
|
||
|
||
async def sd_fetch_batch(token, begin_str, end_str):
|
||
headers = {"access-token": token}
|
||
begin = f"{begin_str} 00:00:00"
|
||
end = f"{end_str} 23:59:59"
|
||
|
||
first_params = {
|
||
"page": 1, "size": PAGE_SIZE_SD, "orderTimeRangeType": "",
|
||
"createdDateBegin": begin, "createdDateEnd": end,
|
||
"merchantIds": MERCHANT_IDS, "productionType": PRODUCTION_TYPE,
|
||
"t": int(time.time() * 1000)
|
||
}
|
||
connector = aiohttp.TCPConnector(limit=MAX_CONCURRENT * 2, ttl_dns_cache=300)
|
||
async with aiohttp.ClientSession(headers=headers, connector=connector) as session:
|
||
timeout = aiohttp.ClientTimeout(total=REQUEST_TIMEOUT)
|
||
async with session.get("https://pod-api.sdspod.com/pod/orders/page", params=first_params, timeout=timeout) as resp:
|
||
first_data = await resp.json()
|
||
if first_data.get("ret") == 20001:
|
||
log_msg("Token过期", "error")
|
||
return []
|
||
total = first_data.get("totalCount", 0)
|
||
if total == 0:
|
||
return []
|
||
total_pages = math.ceil(total / PAGE_SIZE_SD)
|
||
log_msg(f" SD {begin_str}~{end_str}: {total}单/{total_pages}页")
|
||
|
||
first_list = first_data.get("list", [])
|
||
if total_pages == 1:
|
||
return first_list
|
||
|
||
sem = asyncio.Semaphore(MAX_CONCURRENT)
|
||
tasks = []
|
||
for pg in range(2, total_pages + 1):
|
||
pg_params = {
|
||
"page": pg, "size": PAGE_SIZE_SD, "orderTimeRangeType": "",
|
||
"createdDateBegin": begin, "createdDateEnd": end,
|
||
"merchantIds": MERCHANT_IDS, "productionType": PRODUCTION_TYPE,
|
||
"t": int(time.time() * 1000)
|
||
}
|
||
tasks.append(sd_fetch_page(session, sem, pg_params, pg, total_pages))
|
||
results = await asyncio.gather(*tasks)
|
||
all_data = first_list
|
||
for lst in results:
|
||
all_data.extend(lst)
|
||
return all_data
|
||
|
||
def sd_format_item(item):
|
||
address = item.get("address", {})
|
||
country = address.get("countryName", "")
|
||
store = item.get("merchantStore", {})
|
||
platform = store.get("platformName", "")
|
||
create_time = item.get("gmtCreated", "")
|
||
parcel_list = item.get("parcelList", [])
|
||
if parcel_list and isinstance(parcel_list, list):
|
||
qty = sum(p.get("totalQty", 0) for p in parcel_list)
|
||
else:
|
||
qty = 0
|
||
|
||
return [
|
||
to_chinese_country(country),
|
||
to_chinese_country(country),
|
||
platform,
|
||
store.get("name", "未知"),
|
||
item.get("no", ""),
|
||
qty,
|
||
format_date_only(create_time),
|
||
"SDSPOD",
|
||
str(item.get("status", ""))
|
||
]
|
||
|
||
async def main_script2():
|
||
log_msg("===== 脚本2：SDSPOD订单提取工具 v2.0 =====")
|
||
TODAY = date.today()
|
||
START_DATE = (TODAY - timedelta(days=LOOKBACK_DAYS)).isoformat()
|
||
END_DATE = TODAY.isoformat()
|
||
|
||
|
||
|
||
# TODAY = date.today()
|
||
# START_DATE = (TODAY - timedelta(days=180)).isoformat()
|
||
# END_DATE = (TODAY - timedelta(days=60)).isoformat()
|
||
log_msg(f"查询时间: {START_DATE} ~ {END_DATE} (回查{LOOKBACK_DAYS}天)")
|
||
|
||
sd_raw = []
|
||
start_d = datetime.strptime(START_DATE, "%Y-%m-%d").date()
|
||
end_d = datetime.strptime(END_DATE, "%Y-%m-%d").date()
|
||
d = start_d
|
||
batches = []
|
||
while d <= end_d:
|
||
batch_end = min(d + timedelta(days=2), end_d)
|
||
batches.append((d.isoformat(), batch_end.isoformat()))
|
||
d = batch_end + timedelta(days=1)
|
||
|
||
log_msg(f"共{len(batches)}批次拉取")
|
||
|
||
sd_token = sd_login_sync()
|
||
if not sd_token:
|
||
log_msg("登录失败，终止", "error")
|
||
return
|
||
|
||
for bi, (b_start, b_end) in enumerate(batches):
|
||
raw = await sd_fetch_batch(sd_token, b_start, b_end)
|
||
sd_raw.extend(raw)
|
||
if bi > 0 and bi % 5 == 0:
|
||
sd_token = sd_login_sync()
|
||
if not sd_token:
|
||
break
|
||
|
||
log_msg(f"API拉取总数: {len(sd_raw)} 条", "success")
|
||
|
||
sd_detail = []
|
||
for item in sd_raw:
|
||
if item.get("status") == 98:
|
||
continue
|
||
row = sd_format_item(item)
|
||
cn = row[1]
|
||
if cn in EUROPEAN_COUNTRIES:
|
||
row[0] = "欧区"
|
||
row[1] = "欧区"
|
||
sd_detail.append(row)
|
||
|
||
sd_total_order = len(sd_detail)
|
||
sd_total_qty = sum(r[5] for r in sd_detail)
|
||
log_msg(f"有效订单: {sd_total_order}单, {sd_total_qty}件", "success")
|
||
|
||
tiktok_rows = [r for r in sd_detail if "tiktok" in r[2].lower()]
|
||
non_tiktok_rows = [r for r in sd_detail if "tiktok" not in r[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 = sd_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} 件")
|
||
|
||
if sd_detail:
|
||
insert_orders_to_db(sd_detail, "SDSPOD")
|
||
|
||
log_msg("===== 脚本2 完成 =====")
|
||
|
||
# ======================================================================
|
||
# 主入口：先执行脚本1（支付订单），再执行脚本2（创建订单）
|
||
# ======================================================================
|
||
@flow(name="platform-order-sdspod", log_prints=True)
|
||
async def sdspod_platform_sync():
|
||
await asyncio.to_thread(main_script1)
|
||
await main_script2()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
# 直跑一次用于本地调试;勿改回 serve():同名 upsert 会覆盖 server 上挂 work pool 的正式 deployment
|
||
sdspod_platform_sync() |