DSP-routing-daily: 每日总表+Kibana监测明细报告
This commit is contained in:
@@ -0,0 +1,206 @@
|
||||
#!/usr/bin/env bash
|
||||
# DSP-routing-daily 数据更新脚本
|
||||
# 每小时被服务器执行一次；仅在 UTC+8 11 点档真正跑，查询"前一天"(UTC+8)数据。
|
||||
# 产出 data.json：总表每天累加一行；Kibana 监测明细表按 adping_id 聚合，保留近14天。
|
||||
set -euo pipefail
|
||||
|
||||
cd "$(dirname "$0")"
|
||||
|
||||
# ============ 配置 ============
|
||||
API_TOKEN='eyJhbGciOiJIUzI1NiJ9.eyJqdGkiOiJkZTljM2IyMGZiYmE0ZTZiOTczODk3NzgwNDJmMGJmYSIsImlhdCI6MTc4ODc1MzI1OCwic3ViIjoidG9rZW4gYnkgdXNoYXJlaXQiLCJleHAiOjE3ODg4Mzk2NTh9.11wvyNVUw5XTZZJm0m5KNpeHSi528jQuCI1DfKqpaXQ'
|
||||
API_BASE='https://ms-api.sandssea.com/api/v1'
|
||||
KIBANA_ES='http://kibana-sg3.ushareit.org/elasticsearch'
|
||||
DATA_FILE='data.json'
|
||||
KEEP_DAYS=14
|
||||
|
||||
# ============ 11 点档判断（UTC+8）============
|
||||
# 服务器每小时触发；只有当 UTC+8 当前小时 == 11 时才真正执行。
|
||||
HOUR_CST=$(TZ='Asia/Shanghai' date +%H)
|
||||
FORCE="${FORCE_RUN:-0}"
|
||||
if [ "$HOUR_CST" != "11" ] && [ "$FORCE" != "1" ]; then
|
||||
echo "[skip] 当前 UTC+8 $HOUR_CST 点，非 11 点档，跳过。（设 FORCE_RUN=1 可强制运行）"
|
||||
exit 0
|
||||
fi
|
||||
|
||||
# ============ 日期计算（UTC+8 前一天）============
|
||||
# 前一天 UTC+8 的 00:00~24:00，换算成 ES 用的 UTC 时间窗口。
|
||||
RUN_DATE=$(TZ='Asia/Shanghai' date '+%Y/%-m/%-d') # 总表"日期"= 运行当天，格式 2026/9/3
|
||||
PREV_DATE_CST=$(TZ='Asia/Shanghai' date -v-1d '+%Y/%-m/%-d' 2>/dev/null || TZ='Asia/Shanghai' date -d 'yesterday' '+%Y/%-m/%-d')
|
||||
# 明细表 dt = 前一天(UTC+8)
|
||||
# ES 时间窗口(UTC): 前一天 00:00 CST = 前一天 -08:00 UTC；到 当天 00:00 CST
|
||||
FROM_UTC=$(TZ='Asia/Shanghai' date -v-1d '+%Y-%m-%dT00:00:00' 2>/dev/null || TZ='Asia/Shanghai' date -d 'yesterday 00:00' '+%Y-%m-%dT00:00:00')
|
||||
TO_UTC=$(TZ='Asia/Shanghai' date '+%Y-%m-%dT00:00:00')
|
||||
# 转成带 +08:00 偏移的 ISO，让 ES 自己按时区解析
|
||||
FROM_ISO="${FROM_UTC}+08:00"
|
||||
TO_ISO="${TO_UTC}+08:00"
|
||||
|
||||
echo "[run] RUN_DATE=$RUN_DATE 明细dt=$PREV_DATE_CST 窗口(CST)=$FROM_ISO ~ $TO_ISO"
|
||||
|
||||
# ============ 源1：原始订单 (camp_info/pageWithOffer, status=1 Active) ============
|
||||
# 原始订单 = Active 全部条数(不去重); pkg = 按 pkg_name+target_geo 去重后的条数
|
||||
echo "[src1] 拉取 camp_info/pageWithOffer 全量..."
|
||||
SRC1_JSON=$(python3 - "$API_BASE" "$API_TOKEN" <<'PYEOF'
|
||||
import sys,urllib.request,json,ssl,math
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
base,token=sys.argv[1],sys.argv[2]
|
||||
ctx=ssl.create_default_context(); ctx.check_hostname=False; ctx.verify_mode=ssl.CERT_NONE
|
||||
HDR={'authorization':token,'referer':'https://sands.hellay.net/','origin':'https://sands.hellay.net','accept':'application/json'}
|
||||
def get(url):
|
||||
req=urllib.request.Request(url,headers=HDR)
|
||||
with urllib.request.urlopen(req,timeout=120,context=ctx) as r:
|
||||
return json.load(r)
|
||||
SIZE=5000
|
||||
# status=1 即 Active（仅生效中的订单）；第一页拿 count
|
||||
first=get(f"{base}/camp_info/pageWithOffer?pageIndex=1&pageSize={SIZE}&status=1&exclude_midas=true")
|
||||
cnt=first.get('data',{}).get('count',0)
|
||||
pages=max(1, math.ceil(cnt/SIZE))
|
||||
def fetch(page):
|
||||
if page==1: return first.get('data',{}).get('list') or []
|
||||
d=get(f"{base}/camp_info/pageWithOffer?pageIndex={page}&pageSize={SIZE}&status=1&exclude_midas=true")
|
||||
return d.get('data',{}).get('list') or []
|
||||
total=0; seen=set()
|
||||
with ThreadPoolExecutor(max_workers=8) as ex:
|
||||
for lst in ex.map(fetch, range(1,pages+1)):
|
||||
for o in lst:
|
||||
total+=1
|
||||
pkg=(o.get('pkg_name') or '').strip()
|
||||
geo=(o.get('target_geo') or '').strip().upper()
|
||||
seen.add(pkg+'_'+geo)
|
||||
print(json.dumps({'raw_orders':total,'pkg_geo_distinct':len(seen)}))
|
||||
PYEOF
|
||||
)
|
||||
RAW_ORDERS=$(echo "$SRC1_JSON" | python3 -c "import sys,json;print(json.load(sys.stdin)['raw_orders'])")
|
||||
PKG_DISTINCT=$(echo "$SRC1_JSON" | python3 -c "import sys,json;print(json.load(sys.stdin)['pkg_geo_distinct'])")
|
||||
echo "[src1] 原始订单=$RAW_ORDERS pkg(去重)=$PKG_DISTINCT"
|
||||
|
||||
# ============ 源2：中转订单 (campaignOffer/list offerSource=5) ============
|
||||
echo "[src2] 拉取 campaignOffer offerSource=5 count..."
|
||||
RELAY_ORDERS=$(python3 - "$API_BASE" "$API_TOKEN" <<'PYEOF'
|
||||
import sys,urllib.request,json,ssl
|
||||
base,token=sys.argv[1],sys.argv[2]
|
||||
ctx=ssl.create_default_context(); ctx.check_hostname=False; ctx.verify_mode=ssl.CERT_NONE
|
||||
req=urllib.request.Request(f"{base}/campaignOffer/list?offerSource=5&pageSize=1&pageIndex=1",headers={'authorization':token,'referer':'https://sands.hellay.net/','origin':'https://sands.hellay.net'})
|
||||
with urllib.request.urlopen(req,timeout=90,context=ctx) as r:
|
||||
d=json.load(r)
|
||||
print(d.get('data',{}).get('count',0))
|
||||
PYEOF
|
||||
)
|
||||
echo "[src2] 中转订单=$RELAY_ORDERS"
|
||||
|
||||
# ============ 源3：Kibana 前一天 is_routing:1 ============
|
||||
echo "[src3] 查询 Kibana..."
|
||||
KIBANA_JSON=$(python3 - "$KIBANA_ES" "$FROM_ISO" "$TO_ISO" "$PREV_DATE_CST" <<'PYEOF'
|
||||
import sys,urllib.request,json
|
||||
es,frm,to,dt=sys.argv[1],sys.argv[2],sys.argv[3],sys.argv[4]
|
||||
CLICK=es+'/ads_san-adping_cold_click_*/_search'
|
||||
PB=es+'/ads_san-adping_cold_postback_*/_search'
|
||||
|
||||
ACCOUNT_MAP={102:'nativex_API',99:'adsplay_API',98:'clickpuff_API',97:'affle_API',96:'mobhuge_API',95:'altosum_API',94:'seanear_API',93:'stridemobi_API',92:'volo_API',91:'strongmetrics_API',87:'aba_API',86:'hope_api',85:'stackyads_API',84:'isummer_API',83:'vertex_API',82:'viking_API',80:'acmemobi_API',79:'flymobi_API',78:'motilal_API',77:'flatmedia_API',76:'hypertap_API',75:'Influx_API',74:'sync_API',73:'c2c_API',72:'tecdo_API',71:'yomiads_API',70:'goodjoy_API',69:'apkpure_API',68:'flatads_API',67:'adnov_API',66:'dolemobi_API',65:'witgala_API',64:'mejoy_API',63:'adwisdom_ddj_API',62:'vfineads_ddj_API',60:'Iogerads_API',56:'quarklio_API',55:'quarklio_ddj_API',54:'bidneuron_API',52:'mstoko_API',51:'hopemobi_API',23:'Macan_API',18:'Midas_API',11:'Dreammobi_API',9:'Flowbox_API'}
|
||||
|
||||
def es_post(url,body):
|
||||
req=urllib.request.Request(url,data=json.dumps(body).encode(),headers={'kbn-xsrf':'true','Content-Type':'application/json'})
|
||||
with urllib.request.urlopen(req,timeout=120) as r:
|
||||
return json.load(r)
|
||||
|
||||
flt=[{'term':{'is_routing':1}},{'range':{'@timestamp':{'gte':frm,'lt':to}}}]
|
||||
|
||||
# 总表: 有点击订单 = 不同 routing_campaign_id 个数 (click表)
|
||||
r=es_post(CLICK,{'size':0,'query':{'bool':{'filter':flt}},'aggs':{'rc':{'cardinality':{'field':'routing_campaign_id.keyword','precision_threshold':40000}}}})
|
||||
has_click_orders=r['aggregations']['rc']['value']
|
||||
|
||||
# 明细: 按 adping_id 聚合 (click表) -> clicks, revenue(sum payin), cost(sum deliver_price), 及维度
|
||||
# advertiser_payin/deliver_price 是字符串，用 script 转 double 求和
|
||||
agg={
|
||||
'by_adp':{'terms':{'field':'adping_id','size':100000},
|
||||
'aggs':{
|
||||
'rev':{'sum':{'script':{'source':"try{return Double.parseDouble(doc['advertiser_payin.keyword'].value)}catch(Exception e){return 0.0}"}}},
|
||||
'cost':{'sum':{'script':{'source':"try{return Double.parseDouble(doc['deliver_price.keyword'].value)}catch(Exception e){return 0.0}"}}},
|
||||
'rc':{'terms':{'field':'routing_campaign_id.keyword','size':1}},
|
||||
'cid':{'terms':{'field':'campaign_id.keyword','size':1}},
|
||||
'sc':{'terms':{'field':'second_channel.keyword','size':1}},
|
||||
'acc':{'terms':{'field':'advertiser_account_id','size':1}},
|
||||
'pkg':{'terms':{'field':'pkg_name.keyword','size':1}},
|
||||
}}
|
||||
}
|
||||
r=es_post(CLICK,{'size':0,'query':{'bool':{'filter':flt}},'aggs':agg})
|
||||
buckets=r['aggregations']['by_adp']['buckets']
|
||||
|
||||
# conversion: postback表 按 adping_id 计数
|
||||
r2=es_post(PB,{'size':0,'query':{'bool':{'filter':flt}},'aggs':{'by_adp':{'terms':{'field':'adping_id','size':100000}}}})
|
||||
conv_map={b['key']:b['doc_count'] for b in r2['aggregations']['by_adp']['buckets']}
|
||||
|
||||
def first(agg_obj):
|
||||
b=agg_obj.get('buckets',[])
|
||||
return b[0]['key'] if b else ''
|
||||
|
||||
detail=[]
|
||||
routing_ids_with_conv=set() # 有转化的 routing_campaign_id
|
||||
for b in buckets:
|
||||
adp=b['key']
|
||||
clicks=b['doc_count']
|
||||
rev=round(b['rev']['value'],6)
|
||||
cost=round(b['cost']['value'],6)
|
||||
conv=conv_map.get(adp,0)
|
||||
rc=str(first(b['rc']))
|
||||
cid=str(first(b['cid']))
|
||||
sc=str(first(b['sc']))
|
||||
accid=first(b['acc'])
|
||||
acc=ACCOUNT_MAP.get(accid,'')
|
||||
if not acc:
|
||||
acc = f'未知({accid})' if accid not in ('',None) else '未知'
|
||||
pkg=str(first(b['pkg']))
|
||||
cvr=round(conv/clicks,6) if clicks else 0
|
||||
gp=round(1-cost/rev,6) if rev else 0
|
||||
ecpc=round(rev/clicks,6) if clicks else 0
|
||||
detail.append({'dt':dt,'routing_campaign_id':rc,'campaign_id':cid,'adping_id':adp,
|
||||
'second_channel':sc,'account':acc,'pkg':pkg,'clicks':clicks,'conversion':conv,
|
||||
'cvr':cvr,'revenue':rev,'cost':cost,'gp':gp,'ecpc':ecpc})
|
||||
if conv>0 and rc:
|
||||
routing_ids_with_conv.add(rc)
|
||||
|
||||
# 无转化订单 = conversion=0 的 routing_campaign_id 个数
|
||||
# = 总(有点击的routing_campaign_id) - 有转化的routing_campaign_id
|
||||
no_conv_orders = int(has_click_orders) - len(routing_ids_with_conv)
|
||||
if no_conv_orders<0: no_conv_orders=0
|
||||
|
||||
print(json.dumps({'has_click_orders':int(has_click_orders),'no_conv_orders':no_conv_orders,'detail':detail}))
|
||||
PYEOF
|
||||
)
|
||||
HAS_CLICK=$(echo "$KIBANA_JSON" | python3 -c "import sys,json;print(json.load(sys.stdin)['has_click_orders'])")
|
||||
NO_CONV=$(echo "$KIBANA_JSON" | python3 -c "import sys,json;print(json.load(sys.stdin)['no_conv_orders'])")
|
||||
echo "[src3] 有点击订单=$HAS_CLICK 无转化订单=$NO_CONV 明细行=$(echo "$KIBANA_JSON" | python3 -c "import sys,json;print(len(json.load(sys.stdin)['detail']))")"
|
||||
|
||||
# ============ 合并写入 data.json（累加、明细保留14天）============
|
||||
python3 - "$DATA_FILE" "$RUN_DATE" "$PREV_DATE_CST" "$RAW_ORDERS" "$PKG_DISTINCT" "$RELAY_ORDERS" "$HAS_CLICK" "$NO_CONV" "$KEEP_DAYS" "$KIBANA_JSON" <<'PYEOF'
|
||||
import sys,json,os,datetime
|
||||
f,run_date,prev_dt,raw,pkg,relay,hasclick,noconv,keep,kib=sys.argv[1:11]
|
||||
raw,pkg,relay,hasclick,noconv,keep=int(raw),int(pkg),int(relay),int(hasclick),int(noconv),int(keep)
|
||||
kib=json.loads(kib)
|
||||
no_click = relay - hasclick # 无点击订单 = 中转订单 - 有点击订单
|
||||
|
||||
old={'summary':[],'detail':[],'updated_at':''}
|
||||
if os.path.exists(f):
|
||||
try:
|
||||
old=json.load(open(f,encoding='utf-8'))
|
||||
except Exception: pass
|
||||
old.setdefault('summary',[]); old.setdefault('detail',[])
|
||||
|
||||
# 总表：同一 run_date 覆盖，否则新增
|
||||
row={'date':run_date,'raw_orders':raw,'pkg':pkg,'relay_orders':relay,
|
||||
'has_click_orders':hasclick,'no_conv_orders':noconv,'no_click_orders':no_click}
|
||||
old['summary']=[r for r in old['summary'] if r.get('date')!=run_date]+[row]
|
||||
old['summary'].sort(key=lambda r:r['date'])
|
||||
|
||||
# 明细：先删掉同 dt 的旧数据，再加入本次；然后按 dt 保留最近 keep 天
|
||||
old['detail']=[d for d in old['detail'] if d.get('dt')!=prev_dt]+kib['detail']
|
||||
# 保留最近 keep 个不同 dt
|
||||
dts=sorted({d['dt'] for d in old['detail']}, key=lambda s: tuple(int(x) for x in s.split('/')))
|
||||
keep_dts=set(dts[-keep:])
|
||||
old['detail']=[d for d in old['detail'] if d['dt'] in keep_dts]
|
||||
|
||||
old['updated_at']=datetime.datetime.now(datetime.timezone(datetime.timedelta(hours=8))).strftime('%Y-%m-%d %H:%M:%S UTC+8')
|
||||
json.dump(old,open(f,'w',encoding='utf-8'),ensure_ascii=False,separators=(',',':'))
|
||||
print(f"[done] summary行数={len(old['summary'])} detail行数={len(old['detail'])} 保留dt={sorted(keep_dts)}")
|
||||
PYEOF
|
||||
|
||||
echo "[ok] data.json 已更新"
|
||||
Reference in New Issue
Block a user