134 lines
6.7 KiB
Python
134 lines
6.7 KiB
Python
#!/usr/bin/env python3
|
||
import csv, io, json, os, re, sys
|
||
from datetime import datetime
|
||
from zoneinfo import ZoneInfo
|
||
|
||
if len(sys.argv) not in (3, 4):
|
||
raise SystemExit('usage: transform_tableau.py SPEND.csv YYYY-MM-DD [EVENTS.csv]')
|
||
src, target = sys.argv[1:3]
|
||
events_path = sys.argv[3] if len(sys.argv) == 4 else None
|
||
raw = open(src, 'rb').read()
|
||
if b'<html' in raw[:1000].lower():
|
||
raise SystemExit('Tableau authentication redirected to HTML')
|
||
text = raw.decode('utf-16') if raw.startswith((b'\xff\xfe', b'\xfe\xff')) else raw.decode('utf-8-sig')
|
||
delimiter = '\t' if text.count('\t') > text.count(',') else ','
|
||
matrix = list(csv.reader(io.StringIO(text), delimiter=delimiter))
|
||
|
||
aliases = {
|
||
'pkg': {'pkg', '维度1'}, 'source': {'source', 'platform'}, 'date': {'date', 'dt', '日期'},
|
||
'campaign': {'campaign name', 'campaign_name', '维度2'},
|
||
'impression': {'impression', 'impressions'}, 'click': {'click', 'clicks'},
|
||
'conversions': {'conversion', 'conversions'}, 'spend': {'spend', 'cost'},
|
||
}
|
||
header = [c.strip().lower() for c in matrix[0]] if matrix else []
|
||
has_header = any(c in aliases['pkg'] for c in header) and any(c in aliases['campaign'] for c in header)
|
||
|
||
def cell(row, field):
|
||
if has_header:
|
||
for i, name in enumerate(header):
|
||
if name in aliases[field]: return row[i] if i < len(row) else ''
|
||
return ''
|
||
positions = {'pkg':0, 'source':1, 'date':3, 'campaign':4, 'impression':5, 'click':6, 'conversions':7, 'spend':8}
|
||
i = positions[field]
|
||
return row[i] if i < len(row) else ''
|
||
|
||
def num(v):
|
||
try: return float(str(v or '0').replace(',', '').replace('%', ''))
|
||
except ValueError: return 0
|
||
|
||
def event_names(value):
|
||
value = str(value or '').strip()
|
||
if not value or value == '-': return []
|
||
return [x.strip().lower() for x in re.split(r'[,，;/；\n]+', value) if x.strip()]
|
||
|
||
events, covered_scopes = {}, set()
|
||
if events_path:
|
||
with open(events_path, encoding='utf-8-sig', newline='') as f:
|
||
for event_row in csv.DictReader(f):
|
||
dt = re.sub(r'[^0-9]', '', event_row.get('dt', ''))
|
||
if len(dt) == 8: dt = f'{dt[:4]}-{dt[4:6]}-{dt[6:]}'
|
||
channel = event_row.get('second_channel','').lower()
|
||
event_source = next((s for s in ('mintegral','bigo','kwai') if s in channel), '')
|
||
if not event_source: continue
|
||
covered_scopes.add((dt, event_source))
|
||
key = (dt, event_row.get('pkg_name','').strip(), event_source, event_row.get('offer_name','').strip(),
|
||
event_row.get('event_name','').strip().lower())
|
||
events[key] = events.get(key, 0) + num(event_row.get('event_num'))
|
||
|
||
config = [r for r in json.load(open('campaign_config.json', encoding='utf-8')) if r['status'].lower() == 'active']
|
||
|
||
def active_on(meta, date):
|
||
"""Respect the Campaign info remark such as 0912新上 for historical backfills."""
|
||
match = re.search(r'(?<!\d)(\d{2})(\d{2})\s*新上', str(meta.get('remark', '')))
|
||
if not match:
|
||
return True
|
||
return date >= f'{date[:4]}-{match.group(1)}-{match.group(2)}'
|
||
experiment = {}
|
||
for r in config:
|
||
if r['group'] == '实验组':
|
||
experiment.setdefault((r['pkg'], r['channel'].lower()), []).append((str(r['adping_id']), r))
|
||
# Kwai can occasionally return GEO in the pkg column for an experimental campaign.
|
||
# An adping ID is unique to the experiment, so use it as a guarded fallback and
|
||
# normalize the emitted pkg to the Campaign info configuration.
|
||
experiment_by_source = {}
|
||
for (_, source), items in experiment.items():
|
||
experiment_by_source.setdefault(source, []).extend(items)
|
||
scopes = {}
|
||
for r in config:
|
||
scopes.setdefault((r['pkg'], r['channel'].lower(), r['geo'].upper()), r)
|
||
|
||
def resolve_geo(pkg, source, campaign):
|
||
candidates = sorted({g for p, s, g in scopes if p == pkg and s == source}, key=len, reverse=True)
|
||
upper = campaign.upper()
|
||
# Some platform campaign names append a sequence directly to the GEO (for example MY2/TH1).
|
||
matches = [g for g in candidates if re.search(r'(^|[^A-Z0-9])' + re.escape(g) + r'(?=[^A-Z]|$)', upper)]
|
||
if len(matches) == 1: return matches[0]
|
||
return candidates[0] if len(candidates) == 1 else None
|
||
|
||
rows, unresolved = [], []
|
||
for raw_row in matrix[1:] if has_header else matrix:
|
||
pkg, source = cell(raw_row, 'pkg').strip(), cell(raw_row, 'source').strip().lower()
|
||
campaign = cell(raw_row, 'campaign').strip()
|
||
if not pkg or not source or not campaign: continue
|
||
date = cell(raw_row, 'date').strip() or target
|
||
if date != target: continue
|
||
exp = next((meta for adping, meta in experiment.get((pkg, source), [])
|
||
if re.search(r'(?<!\d)' + re.escape(adping) + r'(?!\d)', campaign)), None)
|
||
if not exp:
|
||
exp = next((meta for adping, meta in experiment_by_source.get(source, [])
|
||
if re.search(r'(?<!\d)' + re.escape(adping) + r'(?!\d)', campaign)), None)
|
||
if exp:
|
||
meta, group, shown, adping, pkg = exp, '实验组', campaign, str(exp['adping_id']), exp['pkg']
|
||
else:
|
||
geo = resolve_geo(pkg, source, campaign)
|
||
meta = scopes.get((pkg, source, geo)) if geo else None
|
||
if not meta:
|
||
unresolved.append(campaign)
|
||
continue
|
||
group, shown, adping = '对照组', 'General campaign', ''
|
||
if not active_on(meta, date):
|
||
continue
|
||
required = event_names(meta.get('kpi',''))
|
||
matched_events = [(date, pkg, source, campaign, name) for name in required
|
||
if (date, pkg, source, campaign, name) in events]
|
||
if not required or (date, source) not in covered_scopes or not matched_events:
|
||
event = None
|
||
else:
|
||
event = sum(events[key] for key in matched_events)
|
||
rows.append({'date': date, 'pkg':pkg, 'geo':meta['geo'], 'source':source,
|
||
'group':group, 'adpingId':adping, 'campaign':shown, 'status':'Active', 'kpi':meta.get('kpi',''),
|
||
'event':event,
|
||
'impression':num(cell(raw_row,'impression')), 'click':num(cell(raw_row,'click')),
|
||
'conversions':num(cell(raw_row,'conversions')), 'spend':num(cell(raw_row,'spend'))})
|
||
|
||
if not rows: raise SystemExit('No valid Tableau rows; data.json preserved')
|
||
old = json.load(open('data.json',encoding='utf-8')) if os.path.exists('data.json') else {'rows':[]}
|
||
old['rows'] = [r for r in old.get('rows',[]) if r.get('date') != target] + rows
|
||
old['latestDate'] = max(r['date'] for r in old['rows'])
|
||
old['generatedAt'] = datetime.now(ZoneInfo('Asia/Shanghai')).isoformat()
|
||
old['source'] = 'DSP Daily Data / Campaign info'
|
||
old['quality'] = {'unresolvedCount':len(unresolved), 'unresolvedExamples':unresolved[:20]}
|
||
with open('data.json.tmp','w',encoding='utf-8') as f: json.dump(old,f,ensure_ascii=False,separators=(',',':'))
|
||
os.replace('data.json.tmp','data.json')
|
||
print(json.dumps({'date':target,'rows':len(rows),'unresolved':len(unresolved)},ensure_ascii=False))
|