# -*- 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__": sdspod_platform_sync.serve(name="platform-order-sdspod", cron="*/15 * * * *")