#!/usr/bin/env python3
"""Build a restrained four-section DOCX from the Skill's brief JSON contract."""
from __future__ import annotations
import argparse
import datetime as dt
import json
import sys
from pathlib import Path
from xml.sax.saxutils import escape
from zipfile import ZIP_DEFLATED, ZipFile
W = "http://schemas.openxmlformats.org/wordprocessingml/2006/main"
FORBIDDEN = ("核心判断", "规则一", "规则二", "Campaign层级", "层级判断")
def run(text, bold=False, color="1F2937", size=20):
props = (f''
f''
+ ("" if bold else ""))
return f'{props}{escape(str(text))}'
def para(text="", bold=False, color="1F2937", size=20, align=None, before=0,
after=90, keep_next=False, border=False):
ppr = [f'',
""]
if align:
ppr.append(f'')
if keep_next:
ppr.append("")
if border:
ppr.append('')
return f'{"".join(ppr)}{run(text, bold, color, size)}'
def rich(label, text):
props = ''
return f'{props}{run(label + "：", True, "0B2545", 20)}{run(text)}'
def heading(text, major=True):
return para(text, True, "0B2545" if major else "2E74B5", 28 if major else 23,
before=240 if major else 150, after=110, keep_next=True, border=major)
def cell(value, width, header=False, align="left"):
fill = '' if header else ""
tcpr = (f'{fill}'
''
'')
return f'{tcpr}{para(value, header, "1F3A5F" if header else "1F2937", 17, align, after=0)}'
def table(headers, rows, widths, aligns=None):
if sum(widths) != 9360 or len(headers) != len(widths):
raise ValueError("Every table width list must match headers and sum to 9360")
aligns = aligns or ["left"] * len(widths)
borders = "".join(f''
for edge in ("top", "left", "bottom", "right", "insideH", "insideV"))
props = f'{borders}'
grid = "".join(f'' for width in widths)
header = "".join(cell(value, width, True, aligns[i]) for i, (value, width) in enumerate(zip(headers, widths)))
xml_rows = [f'{header}']
for row in rows:
normalized = list(row) + [""] * (len(headers) - len(row))
cells = "".join(cell(normalized[i], width, False, aligns[i]) for i, width in enumerate(widths))
xml_rows.append(f'{cells}')
return f'{props}{grid}{"".join(xml_rows)}'
def rows(items, fields):
return [[str(item.get(field, "")) for field in fields] for item in items]
def validate(data):
required = ("analysis_date", "portfolio_kpis", "learning_pool", "stable_pool", "actions", "good_ads")
missing = [key for key in required if key not in data]
if missing:
raise ValueError("Missing brief keys: " + ", ".join(missing))
payload = json.dumps(data, ensure_ascii=False)
found = [term for term in FORBIDDEN if term in payload]
if found:
raise ValueError("Forbidden daily-brief content: " + ", ".join(found))
for item in data["stable_pool"]:
tier = str(item.get("tier", ""))
if tier and not any(tier.startswith(prefix) for prefix in ("S-优质", "A-稳定", "B-波动待优化", "C-劣质")):
raise ValueError(f"Invalid stable tier: {tier}")
def document_body(data):
title = data.get("title") or f'{data["analysis_date"]} Campaign Triage'
label = data.get("account_label", "广告投放日报")
mature = data.get("retention_mature_date")
date_line = f'投放数据日：{data["analysis_date"]}' + (f' ｜ D1次留成熟日：{mature}' if mature else "")
body = [para(label, True, "64748B", 18, after=55), para(title, True, "0B2545", 34, after=60),
para(date_line, color="475569", size=19, after=160, border=True)]
body += [heading("1. 是否达标"), table(
["指标", "实际值", "目标", "结论", "口径"],
rows(data["portfolio_kpis"], ["metric", "actual", "target", "verdict", "basis"]),
[1450, 1300, 1300, 1200, 4110], ["left", "center", "center", "center", "left"])]
if data.get("kpi_note"):
body.append(rich("口径说明", data["kpi_note"]))
body += [heading("2. 问题分析"), heading("学习期池排名", False)]
learning = data["learning_pool"]
if learning:
body.append(table(["排名", "Ad", "阶段", "成熟表现", "决定"],
rows(learning, ["rank", "ad", "stage", "evidence", "decision"]),
[650, 2300, 950, 3400, 2060], ["center", "left", "center", "left", "left"]))
else:
body.append(para("无", color="475569"))
body.append(heading("稳定期池排名", False))
stable = data["stable_pool"]
if stable:
body.append(table(["排名", "分级", "Ad", "判断", "决定"],
rows(stable, ["rank", "tier", "ad", "evidence", "decision"]),
[650, 1550, 2050, 3000, 2110], ["center", "center", "left", "left", "left"]))
else:
body.append(para("无", color="475569"))
body.append(heading("3. 建议动作"))
actions = data["actions"]
if actions:
body.append(table(["动作", "对象", "证据", "预算影响"],
rows(actions, ["action", "ad", "evidence", "budget_impact"]),
[1100, 2500, 4260, 1500], ["center", "left", "left", "center"]))
else:
body.append(para("无", color="475569"))
for note in data.get("action_notes", []):
body.append(rich(note.get("label", "说明"), note.get("text", "")))
body.append(heading("4. 近7天表现好的 Ads（3–5个）"))
good = data["good_ads"]
if good:
body.append(table(["排名", "Ad", "近7日数据", "为什么好", "下一步"],
rows(good, ["rank", "ad", "data", "why_good", "next_step"]),
[650, 1900, 2900, 2400, 1510], ["center", "left", "left", "left", "left"]))
else:
body.append(para("无符合标准的成熟 Ad；不降低标准凑数。", color="475569"))
if data.get("data_note"):
body += [heading("数据与决策说明", False), para(data["data_note"], color="475569", size=18)]
body.append(para("以上均为人工决策建议，未执行任何平台变更。", True, "0B2545", 19, before=60, after=0, border=True))
return "".join(body)
def write_docx(data, output: Path):
validate(data)
body = document_body(data)
sect = ''
document = f'{body}{sect}'
styles = f'''
'''
content_types = ''''''
root_rels = ''''''
doc_rels = ''''''
stamp = dt.datetime.now(dt.timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
core = f'''{escape(data.get("title", "Daily Campaign Triage"))}daily-campaign-triage{stamp}{stamp}'''
output.parent.mkdir(parents=True, exist_ok=True)
with ZipFile(output, "w", ZIP_DEFLATED) as archive:
for name, payload in (("[Content_Types].xml", content_types), ("_rels/.rels", root_rels),
("word/document.xml", document), ("word/styles.xml", styles),
("word/_rels/document.xml.rels", doc_rels), ("docProps/core.xml", core)):
archive.writestr(name, payload)
def main(argv=None):
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("input", type=Path, help="Brief JSON")
parser.add_argument("--output", type=Path, required=True)
args = parser.parse_args(argv)
data = json.loads(args.input.read_text(encoding="utf-8"))
write_docx(data, args.output)
print(f"Wrote {args.output}")
if __name__ == "__main__":
try:
main()
except (ValueError, json.JSONDecodeError) as exc:
print(f"error: {exc}", file=sys.stderr)
raise SystemExit(2)