#!/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

# ============ 执行时机判断（UTC+8，健壮版）============
# 平台每小时触发一次本脚本。策略：
#   当 UTC+8 已过 11 点(含11点) 且 data.json 里还没有"今天(UTC+8)"这一行时 -> 执行补跑；
#   否则跳过。这样不依赖平台精确在 11:05 触发，只要 11 点后某次触发到了就会补当天，且一天只跑一次。
#   设 FORCE_RUN=1 可无条件强制运行。
HOUR_CST=$(TZ='Asia/Shanghai' date +%H)
RUN_DATE=$(TZ='Asia/Shanghai' date '+%Y/%-m/%-d')          # 总表"日期"= 运行当天，格式 2026/9/3
FORCE="${FORCE_RUN:-0}"
if [ "$FORCE" != "1" ]; then
  # 1) 必须已过 11 点
  if [ "$((10#$HOUR_CST))" -lt 11 ]; then
    echo "[skip] 当前 UTC+8 $HOUR_CST 点，未到 11 点，跳过。"
    exit 0
  fi
  # 2) 今天是否已更新过（data.json 是否已有 RUN_DATE 这一行）
  if [ -f "$DATA_FILE" ]; then
    ALREADY=$(python3 -c "import json,sys
try:
    d=json.load(open('$DATA_FILE',encoding='utf-8'))
    print('1' if any(r.get('date')=='$RUN_DATE' for r in d.get('summary',[])) else '0')
except Exception:
    print('0')")
    if [ "$ALREADY" = "1" ]; then
      echo "[skip] 今天($RUN_DATE)已更新过，跳过。"
      exit 0
    fi
  fi
fi

# ============ 日期计算（UTC+8 前一天）============
# 前一天 UTC+8 的 00:00~24:00，换算成 ES 用的 UTC 时间窗口。
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 覆盖，否则新增；保留最近 3 天
SUMMARY_KEEP=3
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}
def dkey(s): return tuple(int(x) for x in s.split('/'))
old['summary']=[r for r in old['summary'] if r.get('date')!=run_date]+[row]
old['summary'].sort(key=lambda r:dkey(r['date']))
old['summary']=old['summary'][-SUMMARY_KEEP:]

# 明细：先删掉同 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 已更新"