skill-video-lapian/scripts/publish.py

#!/usr/bin/env python3
"""publish.py — 上传报告整套产物到 S3 个人空间, 生成公开URL.

用法:
  python3 publish.py --workdir /tmp/lapian
路径规范(硬性, 全英文小写):
  user/<userid>/report/lapian/<平台>/<YYYYMMDD>/
    report.html · video.mp4 · audio.wav · transcript.json
    frames/f0001.jpg... · assets/(analysis.json/comments.json/meta.json)
报告内资源引用全部相对路径 → 整个前缀自包含, 换域名也能用。
返回: report.html 的公开URL(并对每个产物做 HTTP 校验)。
"""
import argparse
import datetime
import json
import os
import sys
import urllib.request

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from common import get_userid, public_url, s3_put_file  # noqa: E402

MANIFEST_CT = {
    ".html": "text/html; charset=utf-8",
    ".json": "application/json; charset=utf-8",
    ".mp4": "video/mp4",
    ".wav": "audio/wav",
    ".jpg": "image/jpeg",
    ".jpeg": "image/jpeg",
    ".png": "image/png",
}


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--workdir", required=True)
    ap.add_argument("--date", default="", help="覆盖日期(默认今天, YYYYMMDD)")
    args = ap.parse_args()
    wd = args.workdir

    report = os.path.join(wd, "report.html")
    if not os.path.exists(report):
        raise SystemExit("[publish] 缺少 report.html, 先跑 build_report.py")
    meta = (json.load(open(os.path.join(wd, "meta.json"), encoding="utf-8"))
            if os.path.exists(os.path.join(wd, "meta.json")) else {})
    platform = meta.get("platform") or "douyin"
    userid = get_userid()
    day = args.date or datetime.date.today().strftime("%Y%m%d")
    base = f"user/{userid}/report/lapian/{platform}/{day}"
    print(f"[publish] 目标前缀: {base}/  (identity userid={userid})")

    uploads = []  # (local, key)
    uploads.append((report, "report.html"))
    for name in ("video.mp4", "audio.wav", "transcript.json"):
        p = os.path.join(wd, name)
        if os.path.exists(p):
            uploads.append((p, name))
    fdir = os.path.join(wd, "frames")
    if os.path.isdir(fdir):
        for f in sorted(os.listdir(fdir)):
            if f.endswith(".jpg"):
                uploads.append((os.path.join(fdir, f), f"frames/{f}"))
    for name in ("analysis.json", "comments.json", "meta.json"):
        p = os.path.join(wd, name)
        if os.path.exists(p):
            uploads.append((p, f"assets/{name}"))

    urls = []
    for local, rel in uploads:
        key = f"{base}/{rel}"
        ct = MANIFEST_CT.get(os.path.splitext(rel)[1].lower())
        url = s3_put_file(local, key, ct)
        urls.append((rel, url))
        print(f"[publish] ✓ {rel} ({os.path.getsize(local)/1024:.0f} KB)")

    # 逐个 HTTP 校验(外部状态回读, 不只信 PUT 返回)
    fails = []
    for rel, url in urls:
        try:
            req = urllib.request.Request(url, method="HEAD")
            with urllib.request.urlopen(req, timeout=30) as r:
                if r.status != 200:
                    fails.append((rel, r.status))
        except Exception as ex:
            fails.append((rel, str(ex)[:60]))
    if fails:
        print("[publish] WARN 以下产物校验失败:", fails)
    else:
        print(f"[publish] 全部 {len(urls)} 个产物公开可访问 ✓")

    report_url = public_url(f"{base}/report.html")
    print("\n=== 公开访问 ===")
    print(report_url)
    json.dump({"report_url": report_url, "prefix": base,
               "files": [{"name": rel, "url": u} for rel, u in urls]},
              open(os.path.join(wd, "publish.json"), "w", encoding="utf-8"),
              ensure_ascii=False, indent=2)


if __name__ == "__main__":
    main()