182 lines
11 KiB
Python
182 lines
11 KiB
Python
#!/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'<w:rFonts w:ascii="Arial" w:hAnsi="Arial" w:eastAsia="Microsoft YaHei"/>'
|
||
f'<w:color w:val="{color}"/><w:sz w:val="{size}"/><w:szCs w:val="{size}"/>'
|
||
+ ("<w:b/><w:bCs/>" if bold else ""))
|
||
return f'<w:r><w:rPr>{props}</w:rPr><w:t xml:space="preserve">{escape(str(text))}</w:t></w:r>'
|
||
|
||
|
||
def para(text="", bold=False, color="1F2937", size=20, align=None, before=0,
|
||
after=90, keep_next=False, border=False):
|
||
ppr = [f'<w:spacing w:before="{before}" w:after="{after}" w:line="276" w:lineRule="auto"/>',
|
||
"<w:keepLines/>"]
|
||
if align:
|
||
ppr.append(f'<w:jc w:val="{align}"/>')
|
||
if keep_next:
|
||
ppr.append("<w:keepNext/>")
|
||
if border:
|
||
ppr.append('<w:pBdr><w:bottom w:val="single" w:sz="8" w:space="5" w:color="2E74B5"/></w:pBdr>')
|
||
return f'<w:p><w:pPr>{"".join(ppr)}</w:pPr>{run(text, bold, color, size)}</w:p>'
|
||
|
||
|
||
def rich(label, text):
|
||
props = '<w:spacing w:before="20" w:after="110" w:line="276" w:lineRule="auto"/>'
|
||
return f'<w:p><w:pPr>{props}</w:pPr>{run(label + "：", True, "0B2545", 20)}{run(text)}</w:p>'
|
||
|
||
|
||
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 = '<w:shd w:val="clear" w:fill="E8EEF5"/>' if header else ""
|
||
tcpr = (f'<w:tcW w:w="{width}" w:type="dxa"/>{fill}<w:vAlign w:val="center"/>'
|
||
'<w:tcMar><w:top w:w="85" w:type="dxa"/><w:left w:w="100" w:type="dxa"/>'
|
||
'<w:bottom w:w="85" w:type="dxa"/><w:right w:w="100" w:type="dxa"/></w:tcMar>')
|
||
return f'<w:tc><w:tcPr>{tcpr}</w:tcPr>{para(value, header, "1F3A5F" if header else "1F2937", 17, align, after=0)}</w:tc>'
|
||
|
||
|
||
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'<w:{edge} w:val="single" w:sz="4" w:color="CBD5E1"/>'
|
||
for edge in ("top", "left", "bottom", "right", "insideH", "insideV"))
|
||
props = f'<w:tblW w:w="9360" w:type="dxa"/><w:tblLayout w:type="fixed"/><w:tblBorders>{borders}</w:tblBorders>'
|
||
grid = "".join(f'<w:gridCol w:w="{width}"/>' for width in widths)
|
||
header = "".join(cell(value, width, True, aligns[i]) for i, (value, width) in enumerate(zip(headers, widths)))
|
||
xml_rows = [f'<w:tr><w:trPr><w:tblHeader/></w:trPr>{header}</w:tr>']
|
||
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'<w:tr><w:trPr><w:cantSplit/></w:trPr>{cells}</w:tr>')
|
||
return f'<w:tbl><w:tblPr>{props}</w:tblPr><w:tblGrid>{grid}</w:tblGrid>{"".join(xml_rows)}</w:tbl>'
|
||
|
||
|
||
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 = '<w:sectPr><w:pgSz w:w="12240" w:h="15840"/><w:pgMar w:top="1080" w:right="1440" w:bottom="1080" w:left="1440" w:header="720" w:footer="720"/></w:sectPr>'
|
||
document = f'<?xml version="1.0" encoding="UTF-8" standalone="yes"?><w:document xmlns:w="{W}"><w:body>{body}{sect}</w:body></w:document>'
|
||
styles = f'''<?xml version="1.0" encoding="UTF-8" standalone="yes"?><w:styles xmlns:w="{W}">
|
||
<w:docDefaults><w:rPrDefault><w:rPr><w:rFonts w:ascii="Arial" w:hAnsi="Arial" w:eastAsia="Microsoft YaHei"/><w:sz w:val="20"/><w:szCs w:val="20"/><w:color w:val="1F2937"/></w:rPr></w:rPrDefault></w:docDefaults>
|
||
<w:style w:type="paragraph" w:default="1" w:styleId="Normal"><w:name w:val="Normal"/><w:qFormat/></w:style></w:styles>'''
|
||
content_types = '''<?xml version="1.0" encoding="UTF-8" standalone="yes"?><Types xmlns="http://schemas.openxmlformats.org/package/2006/content-types"><Default Extension="rels" ContentType="application/vnd.openxmlformats-package.relationships+xml"/><Default Extension="xml" ContentType="application/xml"/><Override PartName="/word/document.xml" ContentType="application/vnd.openxmlformats-officedocument.wordprocessingml.document.main+xml"/><Override PartName="/word/styles.xml" ContentType="application/vnd.openxmlformats-officedocument.wordprocessingml.styles+xml"/><Override PartName="/docProps/core.xml" ContentType="application/vnd.openxmlformats-package.core-properties+xml"/></Types>'''
|
||
root_rels = '''<?xml version="1.0" encoding="UTF-8" standalone="yes"?><Relationships xmlns="http://schemas.openxmlformats.org/package/2006/relationships"><Relationship Id="rId1" Type="http://schemas.openxmlformats.org/officeDocument/2006/relationships/officeDocument" Target="word/document.xml"/><Relationship Id="rId2" Type="http://schemas.openxmlformats.org/package/2006/relationships/metadata/core-properties" Target="docProps/core.xml"/></Relationships>'''
|
||
doc_rels = '''<?xml version="1.0" encoding="UTF-8" standalone="yes"?><Relationships xmlns="http://schemas.openxmlformats.org/package/2006/relationships"><Relationship Id="rId1" Type="http://schemas.openxmlformats.org/officeDocument/2006/relationships/styles" Target="styles.xml"/></Relationships>'''
|
||
stamp = dt.datetime.now(dt.timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
|
||
core = f'''<?xml version="1.0" encoding="UTF-8" standalone="yes"?><cp:coreProperties xmlns:cp="http://schemas.openxmlformats.org/package/2006/metadata/core-properties" xmlns:dc="http://purl.org/dc/elements/1.1/" xmlns:dcterms="http://purl.org/dc/terms/" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"><dc:title>{escape(data.get("title", "Daily Campaign Triage"))}</dc:title><dc:creator>daily-campaign-triage</dc:creator><dcterms:created xsi:type="dcterms:W3CDTF">{stamp}</dcterms:created><dcterms:modified xsi:type="dcterms:W3CDTF">{stamp}</dcterms:modified></cp:coreProperties>'''
|
||
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)
|