# -*- coding: utf-8 -*- """平台订单提取(RIIN/银河):写入 platform_order_data 表。 最小包装自 riin_extract_v2.py,对应 platform-order-riin。 """ import time import json import hashlib import asyncio import aiohttp import urllib3 from collections import defaultdict from datetime import date, timedelta, datetime from prefect import flow urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) # ===================== 银河(RIIN)接口配置 ===================== RIIN_LOGIN_URL = "http://tshirt.riin.com/auth/api/auth/login" RIIN_ORDER_URL = "https://tshirt.riin.com/distribution/order/page" USERNAME_RIIN = "预上网" PASSWORD_RIIN = "Aa123456." SUB_ID_LIST = ["10368", "10283", "16905", "11502", "13615", "11531","12279"] MAX_RETRY = 3; REQUEST_TIMEOUT = 30; PAGE_SIZE_RIIN = 200 LOOKBACK_DAYS = 6 # 回查14天避免漏单 COUNTRY_MAP = { "United States":"美国","America":"美国","USA":"美国","Mexico":"墨西哥","Canada":"加拿大", "United Kingdom":"英国","UK":"英国","Germany":"德国","France":"法国","Italy":"意大利", "Spain":"西班牙","Portugal":"葡萄牙","Netherlands":"荷兰","Belgium":"比利时","Poland":"波兰", "Japan":"日本","South Korea":"韩国","Saudi Arabia":"沙特阿拉伯","Australia":"澳大利亚", "Brazil":"巴西","Sweden":"瑞典","Denmark":"丹麦","Austria":"奥地利","Czech Republic":"捷克", "Ireland":"爱尔兰","Switzerland":"瑞士","Norway":"挪威","Finland":"芬兰", "Hungary":"匈牙利","Romania":"罗马尼亚","Greece":"希腊","Bulgaria":"保加利亚", "Croatia":"克罗地亚","Slovakia":"斯洛伐克","Slovenia":"斯洛文尼亚", "Lithuania":"立陶宛","Latvia":"拉脱维亚","Estonia":"爱沙尼亚", "Luxembourg":"卢森堡","Malta":"马耳他","Cyprus":"塞浦路斯","Iceland":"冰岛", "Liechtenstein":"列支敦士登","Monaco":"摩纳哥","Andorra":"安道尔", "New Zealand":"新西兰","Argentina":"阿根廷","Chile":"智利","Colombia":"哥伦比亚", "Peru":"秘鲁","India":"印度","Russia":"俄罗斯","Turkey":"土耳其","Israel":"以色列", "UAE":"阿联酋","United Arab Emirates":"阿联酋","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 _platform_name(record): if record.get("temuHalf"): return "Temu" if record.get("temuFull"): return "TEMU全托管" if record.get("sheinOrder"): return "SHEIN" if record.get("tiktokOrder"): return "TikTok" if record.get("shopeeOrder"): return "Shopee" code = record.get("platformShopCode","") return code.split("_")[0] if code else "" # ===================== 银河(RIIN)登录与拉取 ===================== async def riin_login(session, username, password): payload = {"clientId":"WEB","username":username,"password":hashlib.md5(password.encode("utf-8")).hexdigest()} headers = {"content-type":"application/json;charset=UTF-8"} for retry in range(MAX_RETRY): try: async with session.post(RIIN_LOGIN_URL, json=payload, headers=headers, timeout=30) as resp: data = await resp.json(content_type=None) if not data.get("successful"): log_msg(f"银河登录失败: {data.get('message')}", "error"); return "" log_msg("银河登录成功", "success") return data["data"]["token"] except Exception as e: log_msg(f"银河登录重试{retry+1}: {e}", "error"); await asyncio.sleep(1) return "" def _riin_headers(token): return {"Accept":"application/json","Authorization":token,"Content-Type":"application/json", "Cookie":f"authorized-token-TRADE={token}","Origin":"http://tshirt.riin.com", "Referer":"http://tshirt.riin.com/","X-Time-Zone":"UTC+8"} async def riin_fetch_page(session, token, payload): for retry in range(MAX_RETRY): try: async with session.post(RIIN_ORDER_URL, json=payload, headers=_riin_headers(token), timeout=REQUEST_TIMEOUT) as resp: result = await resp.json(content_type=None) if not result.get("successful"): log_msg(f"银河API错误: {result.get('message')}", "error"); await asyncio.sleep(1); continue return result["data"] except Exception as e: log_msg(f"银河请求异常: {e}", "error"); await asyncio.sleep(1) return {} async def riin_fetch_all(token, start_date, end_date): base = {"timeType":3,"startTime":f"{start_date} 00:00:00","endTime":f"{end_date} 23:59:59", "dayRangeGroup":None,"hasRefund":False,"orderStatus":None,"shopCodeList":[],"subIdList":SUB_ID_LIST} all_records = []; page = 1 async with aiohttp.ClientSession() as session: while True: payload = {**base, "pageIndex":page, "pageSize":PAGE_SIZE_RIIN} data = await riin_fetch_page(session, token, payload) records = data.get("records",[]) or [] all_records.extend(records) total = data.get("total",0) if len(records) < PAGE_SIZE_RIIN or len(all_records) >= total: break page += 1 return all_records def riin_format(raw_list): rows = [] for r in raw_list: if str(r.get("orderStatus","")) == "13": # 过滤status=13 continue cn = to_chinese_country(r.get("receiverCountry","")) rows.append([cn, cn, _platform_name(r), r.get("platformShopName",""), r.get("platformOid",""), r.get("goodsTotalQty",0), # str(r.get("orderTime",""))[:10], "RIIN", str(r.get("orderTime","")), "RIIN", r.get("orderStatus","")]) return rows # ===================== 主流程 ===================== @flow(name="platform-order-riin", log_prints=True) async def riin_platform_sync(): TODAY = date.today() START_DATE = (TODAY - timedelta(days=LOOKBACK_DAYS)).isoformat() END_DATE = TODAY.isoformat() # END_DATE = (TODAY - timedelta(days=LOOKBACK_DAYS-60)).isoformat() log_msg(f"===== 银河(RIIN) v2.0 =====") log_msg(f"查询: {START_DATE} ~ {END_DATE} ({LOOKBACK_DAYS}天回查)") merged_detail = []; riin_total_order = 0; riin_total_qty = 0 try: async with aiohttp.ClientSession() as session: riin_token = await riin_login(session, USERNAME_RIIN, PASSWORD_RIIN) if not riin_token: return riin_raw = await riin_fetch_all(riin_token, START_DATE, END_DATE) merged_detail = riin_format(riin_raw) riin_total_order = len(merged_detail) riin_total_qty = sum(r[5] for r in merged_detail) log_msg(f"银河API: {riin_total_order}单, {riin_total_qty}件", "success") except Exception as e: log_msg(f"银河异常: {e}", "error") # 欧区 for row in merged_detail: if row[1] in EUROPEAN_COUNTRIES: row[0]="欧区"; row[1]="欧区" tiktok_rows = [r for r in merged_detail if "tiktok" in r[2].lower()] non_tiktok_rows = [r for r in merged_detail if "tiktok" not in r[2].lower()] # 统计 nt_plat_stat = defaultdict(lambda:[0,0]); nt_country_stat = defaultdict(lambda:[0,0]) for r in non_tiktok_rows: nt_plat_stat[r[2]][0]+=1; nt_plat_stat[r[2]][1]+=r[5] nt_country_stat[r[1]][0]+=1; nt_country_stat[r[1]][1]+=r[5] tk_plat_stat = defaultdict(lambda:[0,0]); tk_country_stat = defaultdict(lambda:[0,0]) for r in tiktok_rows: tk_plat_stat[r[2]][0]+=1; tk_plat_stat[r[2]][1]+=r[5] tk_country_stat[r[1]][0]+=1; tk_country_stat[r[1]][1]+=r[5] total_qty = riin_total_qty log_msg(f"\n非Tiktok: {sum(v[1] for v in nt_plat_stat.values())}件") for p in sorted(nt_plat_stat): log_msg(f" {p}: {nt_plat_stat[p][1]}件") # 导入PostgreSQL (ON CONFLICT DO UPDATE) 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") 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""" for i in range(0,len(rows),500): cur.executemany(sql,rows[i:i+500]); conn.commit() cur.execute("SELECT COUNT(*) FROM platform_order_data"); cnt=cur.fetchone()[0] cur.close(); conn.close() log_msg(f"已同步到DB, 总{cnt}条","success") except Exception as e: log_msg(f"DB异常: {e}","error") log_msg("===== 完成 =====") if __name__ == "__main__": # 直跑一次用于本地调试;勿改回 serve():同名 upsert 会覆盖 server 上挂 work pool 的正式 deployment riin_platform_sync()