skill-video-lapian/scripts/publish_via_obs.py

#!/usr/bin/env python3
"""publish_via_obs.py — publish.py 的跳板机备用通道。

适用场景: /api/storage/credentials 与 /api/storage/upload 均 404(未部署)、
容器无 CLOUD_SDK_AK/SK、无 obsutil 时 —— 借道 server.fmode.cn(云服务器)上的
obsutil(自带 nkkj 主账套 AK/SK)上传到 storage-s3-nkkj 桶。

用法: python3 publish_via_obs.py --workdir /tmp/lapian [--date 20260829]
产物清单与 publish.py 完全一致, 上传后逐个 HEAD 校验, 写 workdir/publish.json。
"""
import argparse, datetime, json, os, sys, tarfile, time, urllib.request
import paramiko

CLOUD = {"host": "139.159.253.131", "user": "root", "password": "bofang666."}
OBSUTIL = "/opt/obsutil/obsutil_linux_amd64_5.4.11/obsutil"
BUCKET = "storage-s3-nkkj"
ENDPOINT = "obs.cn-north-4.myhuaweicloud.com"
# 对外链接 host 规范(2026-08-30 硬性): 必须用 s3.fmode.cn, 禁用 OBS 原生域名
# (浏览器只有经 s3.fmode.cn 反代才能正常加载报告的相对路径资源)
PUB = "https://s3.fmode.cn"

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


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--workdir", required=True)
    ap.add_argument("--date", default=datetime.date.today().strftime("%Y%m%d"))
    ap.add_argument("--sub", default="", help="视频子目录(默认取meta.json的aweme_id, 防同日多视频覆盖)")
    args = ap.parse_args()
    wd = args.workdir
    sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
    from common import get_userid
    userid = get_userid()
    sub = args.sub
    if not sub:
        try:
            sub = (json.load(open(f"{wd}/meta.json")) or {}).get("aweme_id") or "video"
        except Exception:
            sub = "video"
    base = f"user/{userid}/report/lapian/douyin/{args.date}/{sub}"

    # 打包清单(与 publish.py 一致)
    files = ["report.html", "video.mp4", "audio.wav", "transcript.json"]
    files = [f for f in files if os.path.exists(f"{wd}/{f}")]
    files += [f"frames/{f}" for f in sorted(os.listdir(f"{wd}/frames"))] if os.path.isdir(f"{wd}/frames") else []
    pkg = "/tmp/lapian-pkg.tar.gz"
    asset_map = []
    for n in ("analysis.json", "comments.json", "meta.json", "llm_out.json"):
        if os.path.exists(f"{wd}/{n}"):
            asset_map.append((f"{wd}/{n}", f"assets/{n}"))
    with tarfile.open(pkg, "w:gz") as tf:
        for rel in files:
            tf.add(f"{wd}/{rel}", arcname=rel)
        for local, arc in asset_map:
            tf.add(local, arcname=arc)
    files += [arc for _, arc in asset_map]
    print(f"[pkg] {len(files)} files, {os.path.getsize(pkg)/1048576:.1f} MB")

    # SFTP(带重试, banner 抖动常见)
    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    for attempt in range(4):
        try:
            ssh.connect(CLOUD["host"], 22, CLOUD["user"], CLOUD["password"],
                        timeout=30, banner_timeout=30)
            break
        except Exception as ex:
            print(f"[ssh] attempt {attempt+1} failed: {type(ex).__name__}, retry 15s")
            time.sleep(15)
    else:
        sys.exit("[ssh] all retries failed")
    sftp = ssh.open_sftp()
    sftp.put(pkg, "/tmp/lapian-pkg.tar.gz")
    sftp.close()

    remote = f"""
set -e
rm -rf /tmp/lapian-pkg && mkdir -p /tmp/lapian-pkg
tar -xzf /tmp/lapian-pkg.tar.gz -C /tmp/lapian-pkg
cd /tmp/lapian-pkg
for f in {' '.join(files)}; do
  if {OBSUTIL} cp "$f" "obs://{BUCKET}/{base}/$f" -acl=public-read -e={ENDPOINT} -j=5 > /tmp/obs_last.log 2>&1; then
    echo "UP OK $f"
  else
    echo "UP FAIL $f: $(tail -2 /tmp/obs_last.log)"
  fi
done"""
    _, stdout, stderr = ssh.exec_command(remote, timeout=600)
    out = stdout.read().decode(errors="replace")
    print(out)
    ssh.close()
    fails = [l for l in out.splitlines() if "UP FAIL" in l]

    # 公网校验
    bad = []
    for rel in files:
        try:
            req = urllib.request.Request(f"{PUB}/{base}/{rel}", method="HEAD")
            with urllib.request.urlopen(req, timeout=25) as r:
                if r.status != 200:
                    bad.append((rel, r.status))
        except Exception as ex:
            bad.append((rel, str(ex)[:60]))
    report_url = f"{PUB}/{base}/report.html"
    print(f"\n[verify] {len(files)-len(bad)}/{len(files)} OK", f"fails={bad}" if bad else "ALL PASS")
    json.dump({"report_url": report_url, "prefix": base, "fails": bad + fails},
              open(f"{wd}/publish.json", "w"), ensure_ascii=False, indent=2)
    print("REPORT URL:", report_url)


if __name__ == "__main__":
    main()