220 lines
12 KiB
Bash
Executable File
220 lines
12 KiB
Bash
Executable File
#!/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']
|
||
|
||
PAYIN_SRC="try{return Double.parseDouble(doc['advertiser_payin.keyword'].value)}catch(Exception e){return 0.0}"
|
||
PRICE_SRC="try{return Double.parseDouble(doc['deliver_price.keyword'].value)}catch(Exception e){return 0.0}"
|
||
|
||
def dims_aggs():
|
||
return {
|
||
'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}},
|
||
}
|
||
def first(agg_obj):
|
||
b=agg_obj.get('buckets',[])
|
||
return b[0]['key'] if b else ''
|
||
|
||
# click表: 按 adping_id -> clicks(计数) + 维度(优先来源)
|
||
rc1=es_post(CLICK,{'size':0,'query':{'bool':{'filter':flt}},
|
||
'aggs':{'by_adp':{'terms':{'field':'adping_id','size':100000},'aggs':dims_aggs()}}})
|
||
click_buckets=rc1['aggregations']['by_adp']['buckets']
|
||
|
||
# postback表: 按 adping_id -> conversion(计数) + revenue(sum payin) + cost(sum price) + 维度(备用)
|
||
rp=es_post(PB,{'size':0,'query':{'bool':{'filter':flt}},
|
||
'aggs':{'by_adp':{'terms':{'field':'adping_id','size':100000},
|
||
'aggs':dict(dims_aggs(), **{
|
||
'rev':{'sum':{'script':{'source':PAYIN_SRC}}},
|
||
'cost':{'sum':{'script':{'source':PRICE_SRC}}},
|
||
})}}})
|
||
pb_buckets=rp['aggregations']['by_adp']['buckets']
|
||
|
||
# 汇总各 adping_id
|
||
click_map={b['key']:b for b in click_buckets}
|
||
pb_map={b['key']:b for b in pb_buckets}
|
||
|
||
def acc_name(accid):
|
||
acc=ACCOUNT_MAP.get(accid,'')
|
||
if not acc:
|
||
acc = f'未知({accid})' if accid not in ('',None) else '未知'
|
||
return acc
|
||
|
||
def dims_from(b):
|
||
return {'rc':str(first(b['rc'])),'cid':str(first(b['cid'])),'sc':str(first(b['sc'])),
|
||
'acc':acc_name(first(b['acc'])),'pkg':str(first(b['pkg']))}
|
||
|
||
all_adp=set(click_map)|set(pb_map)
|
||
detail=[]
|
||
routing_ids_with_conv=set() # 有转化(postback出现)的 routing_campaign_id
|
||
for adp in all_adp:
|
||
cb=click_map.get(adp); pb=pb_map.get(adp)
|
||
clicks=cb['doc_count'] if cb else 0
|
||
conv=pb['doc_count'] if pb else 0
|
||
rev=round(pb['rev']['value'],6) if pb else 0.0 # revenue: postback逐条sum payin
|
||
cost=round(pb['cost']['value'],6) if pb else 0.0 # cost: postback逐条sum price
|
||
# 维度优先取 click表，缺则取 postback表
|
||
dm=dims_from(cb) if cb else dims_from(pb)
|
||
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':dm['rc'],'campaign_id':dm['cid'],'adping_id':adp,
|
||
'second_channel':dm['sc'],'account':dm['acc'],'pkg':dm['pkg'],
|
||
'clicks':clicks,'conversion':conv,'cvr':cvr,'revenue':rev,'cost':cost,'gp':gp,'ecpc':ecpc})
|
||
if conv>0 and dm['rc']:
|
||
routing_ids_with_conv.add(dm['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 已更新"
|