feat: skill-video-lapian v1 平台级拉片技能(动态身份+S3发布)

- scripts/: fetch_video/transcribe/fetch_comments/frame_analysis/build_report/publish + common.py(stdlib SigV4)
- templates/report.html.j2 暗色手机自适应报告(分段跳转/词频情绪/VOC/相对路径)
- references/: methodology/voc-api/s3-upload
- 零硬编码凭据: FEME_* env > fmode-identity.json > FMODE_* 兼容

README.md

@@ -0,0 +1,79 @@
+# skill-video-lapian
+
+平台级**短视频拉片技能**: 下载 → 转写 → 抽帧 → 方法论逐段分析 → HTML 报告 → S3 个人空间发布。
+
+## 命名规则(全平台技能以后都遵守)
+
+```
+skill-<领域>-<功能>
+skill-video-lapian = 处理 video, 功能是 拉片(shot-by-shot breakdown)
+其他示例: skill-audio-zhuanxie / skill-image-haibao / skill-data-report
+```
+
+## 平台级设计(核心)
+
+任何数字生命克隆本仓库, 只配**自己的身份参数**就能跑全流程, 产物发布到**自己的** S3 个人空间:
+
+```bash
+# 飞马身份(每个生命不同)
+export FEME_USERID=<Parse objectId>       # 如 sr2WiPsDyQ / pODNJqbURI / dhv9yTqK6K
+export FEME_SESSION_TOKEN="r:..."         # 飞马会话
+export FEME_NEWAPI_TOKEN="sk-..."         # API计费(VOC网关/转写/调模型都用它)
+export FEME_STUDIO_URL=http://server.fmode.cn:<studio端口>
+
+# S3 个人空间
+export S3_BUCKET=storage-s3-nkkj
+export S3_REGION=cn-north-4
+export S3_ENDPOINT=https://obs.cn-north-4.myhuaweicloud.com
+# AK/SK: 优先 <STUDIO_URL>/api/storage/credentials 取临时凭证(1小时, 限prefix);
+# 过渡期未上线, 直接用容器 env 的 CLOUD_SDK_AK / CLOUD_SDK_SK
+```
+
+零硬编码: 全部凭据/路径从 env 或 `/opt/data/fmode-identity.json` 读取
+(优先级 `FEME_*` env > identity.json > FMODE_* 兼容旧名)。
+
+## 快速开始
+
+```bash
+export FEME_USERID=... FEME_SESSION_TOKEN=... FEME_NEWAPI_TOKEN=...
+WD=/tmp/lapian-test
+python3 scripts/fetch_video.py --aweme-id 7677548065384877346 --out-dir $WD
+python3 scripts/fetch_comments.py --aweme-id 7677548065384877346 --workdir $WD   # 可选
+python3 scripts/transcribe.py --workdir $WD
+python3 scripts/frame_analysis.py --workdir $WD --max-frames 12
+python3 scripts/build_report.py --workdir $WD --title "拉片分析:示例"
+python3 scripts/publish.py --workdir $WD   # → 公开URL
+```
+
+依赖: python3.8+(纯标准库) / ffmpeg / node+npx。无需 pip 安装任何包。
+
+## 目录
+
+```
+SKILL.md                  # 技能入口: 触发词+流程+参数说明
+references/
+  methodology.md          # 拉片方法论(薛辉/郑经说/三把刀/小希/南门)
+  voc-api.md              # VOC采集接口(搜索/详情/评论)
+  s3-upload.md            # S3发布规范(virtual-hosted/ACL/相对路径)
+scripts/
+  common.py               # 共享库: 身份参数/VOC/S3 SigV4(stdlib)
+  fetch_video.py          # 下载视频(VOC网关或直链/本地)
+  transcribe.py           # fmode-listen 转写(分段时间戳)
+  fetch_comments.py       # 评论区采集(可选)
+  frame_analysis.py       # 抽帧+豆包视觉逐帧点评
+  build_report.py         # 生成HTML报告(内置mini-Jinja, 无第三方依赖)
+  publish.py              # S3上传+公开URL+逐产物校验
+templates/
+  report.html.j2          # 报告模板(暗色/手机自适应/相对路径)
+```
+
+## 报告示例
+
+听show·丽江酒吧《云南人的物欲》实测报告:
+`user/sr2WiPsDyQ/report/lapian/douyin/20260829/report.html`
+(端到端自测记录见 git log / 会话历史)
+
+## License / 治理
+
+- 内部技能, 随数字生命体系分发; 二进制产物只进 S3, 不进 git
+- 修改方法论请同步 `/opt/data/knowledge-base/编导/` 源文件

SKILL.md

@@ -0,0 +1,106 @@
+---
+name: skill-video-lapian
+description: 平台级短视频拉片技能:下载→转写→抽帧→方法论逐段分析→HTML报告→S3个人空间发布。动态身份参数(每个数字生命用自己飞马身份/S3空间),零硬编码。触发词:拉片、拆解视频、逐帧分析、视频复盘。
+---
+
+# skill-video-lapian — 平台级拉片技能
+
+> 命名规则(全平台技能遵守): `skill-<领域>-<功能>`
+> skill-video-**lapian** = 处理 **video**, 功能是 **拉片**(逐段逐帧拆解)。
+> 本技能是**平台级**: 任何数字生命克隆本仓库后, 用**自己的身份参数**即可跑全流程,
+> 产物发布到**自己的** S3 个人空间。**零硬编码凭据**。
+
+## 何时用
+
+- 用户发来抖音/短视频链接说「拉片 / 拆解 / 逐帧分析 / 复盘这条为什么火」
+- 编导知识库沉淀: 把爆款视频拆成结构+技法+可复用模板
+- 批量研究某账号/某赛道的内容打法
+
+## 前置条件(生命周期初始化时配好)
+
+```bash
+# 飞马身份(每个生命不同, 见各自容器 /opt/data/fmode-identity.json)
+export FEME_USERID=sr2WiPsDyQ            # ← 换成你自己的 Parse objectId
+export FEME_SESSION_TOKEN="r:..."        # 飞马会话
+export FEME_NEWAPI_TOKEN="sk-..."        # API计费token(调VOC网关/转写/模型全走它)
+export FEME_STUDIO_URL="http://server.fmode.cn:<studio端口>"   # 自己的studio端点
+# S3 (过渡期可用容器env; studio /api/storage/credentials 上线后自动切换)
+export CLOUD_SDK_AK=...   export CLOUD_SDK_SK=...
+export S3_BUCKET=storage-s3-nkkj  S3_REGION=cn-north-4
+export S3_ENDPOINT=https://obs.cn-north-4.myhuaweicloud.com
+```
+
+依赖: `python3.8+`(纯标准库, 无需 pip 装包) · `ffmpeg/ffprobe` · `node/npx`(fmode-listen)
+身份参数优先级: `FEME_*` 环境变量 > `fmode-identity.json` > 兼容旧名(FMODE_*)。
+**S3 只能写 `user/<自己的FEME_USERID>/` 前缀, 写他人前缀=越权。**
+
+## 全流程(5 步, 每步可独立重跑, 幂等)
+
+```bash
+REPO=/opt/data/git-repos/skill-video-lapian   # 克隆后放哪都行
+WD=/tmp/lapian-$(date +%s)                     # 工作目录
+
+# 1. 拿视频(三选一): 抖音短链 / 视频ID / 已有本地文件
+python3 $REPO/scripts/fetch_video.py --url "https://v.douyin.com/xxxx/" --out-dir $WD
+python3 $REPO/scripts/fetch_video.py --aweme-id 7677548065384877346 --out-dir $WD
+python3 $REPO/scripts/fetch_video.py --local /path/to/video.mp4 --out-dir $WD --platform douyin
+#    可选: 采评论区VOC(报告自动带上)
+python3 $REPO/scripts/fetch_comments.py --aweme-id 7677548065384877346 --workdir $WD --max 100
+
+# 2. 转写(fmode-listen → 讯飞LFASR, 分段时间戳, 毫秒)
+python3 $REPO/scripts/transcribe.py --workdir $WD
+
+# 3. 抽帧+豆包视觉逐帧点评(计费可控: --fps/--max-frames, 或 --skip-vision 只抽帧)
+python3 $REPO/scripts/frame_analysis.py --workdir $WD --fps 0.15 --max-frames 12
+
+# 4. 生成报告(暗色主题/手机自适应/分段点击跳转播放/词频情绪/VOC/相对路径)
+python3 $REPO/scripts/build_report.py --workdir $WD --title "拉片分析:XXX"
+
+# 5. 发布到自己的S3个人空间 → 得公开URL
+python3 $REPO/scripts/publish.py --workdir $WD
+# → https://storage-s3-nkkj.obs.cn-north-4.myhuaweicloud.com/user/<FEME_USERID>/report/lapian/douyin/<YYYYMMDD>/report.html
+```
+
+## 产物结构(S3 路径规范, 全英文小写)
+
+```
+user/<userid>/report/lapian/<平台简称>/<YYYYMMDD>/
+├── report.html      ← 报告入口(资源全部相对路径, 目录自包含)
+├── video.mp4 / audio.wav / transcript.json
+├── frames/f0001.jpg...
+└── assets/          # analysis.json / comments.json / meta.json
+```
+平台简称映射: douyin / xiaohongshu / weixin / bilibili / tiktok / kuaishou / shortdrama。
+发布后对每个产物做 HEAD 校验, 全 200 才算成功。
+
+## 报告包含
+
+1. 原视频播放器置顶(粘性顶栏), 每段「⏵播放这一段」跳转对应时间
+2. 逐段可折叠卡片: 时间戳+台词+抽帧截图(画面描述)+技法标签(导师·技法)+可复用模板
+3. 整体结构表(起承转合)+核心发现
+4. 词频TOP/情绪分布/语速(纯CSS条形图, 零外网依赖)
+5. 评论区VOC: 正/中/负分布+高赞TOP5+每条洞察
+6. 3 条可执行建议
+
+## 分析依据
+
+- `references/methodology.md` — 薛辉/郑经说/三把刀/小希/南门 技法表(点评提示词的来源)
+- `references/voc-api.md` — VOC 网关接口(搜索/详情/评论, 解析层级与坑)
+- `references/s3-upload.md` — S3 发布规范(virtual-hosted/ACL/相对路径/凭证回退)
+
+## 实现要点(读脚本前先看)
+
+- `scripts/common.py` — 身份参数解析/VOC调用/S3 SigV4 纯stdlib实现(不依赖boto3, OBS必须virtual-hosted寻址)
+- `scripts/build_report.py` — 内置 mini-Jinja(stdlib), 无需安装 jinja2;
+  LLM 点评失败自动回退规则启发式, 永不中断流程
+- 转写短路: transcript.json 已存在则跳过(转写按分钟计费, 别重复花钱)
+- 视频下载完整校验: ffprobe 时长异常即报错, 禁止截断件进入分析
+
+## 坑(实测)
+
+- VOC 网关 401 先查 token 是否 `sk-` 开头且 Bearer 前缀完整
+- OBS path-style 上传报 `VirtualHostDomainRequired` → bucket 必须放进 Host
+- urllib 对该网关要先设 UA(默认 UA 偶发被 WAF 拦, curl 不拦)
+- fmode-listen 是异步转写, 长音频耐心等(脚本已设 30 分钟超时)
+- 豆包视觉模型: 用 `doubao-seed-2-0-pro-260215`(当前账号唯一可用豆包pro), DeepSeek 系无视觉
+- 同一视频当天重跑: publish.py 幂等覆盖同日期目录, 直接重传即可

references/methodology.md

@@ -0,0 +1,74 @@
+# 拉片方法论(编导知识库要点 · 分析分段时对照使用)
+
+> 提炼自 /opt/data/knowledge-base/编导/(薛辉/郑经说/三把刀/编导小希/南门录像厅)。
+> 逐段点评时: 每段至少对上 1 条技法, 用「导师·技法」格式标注。
+
+## 一、薛辉小清新(短视频编导/IP 体系头部, 285万粉)
+
+| 技法 | 要点 | 识别特征 |
+|---|---|---|
+| 观点前置 | 第一句抛反常识观点, 停住人 | 开头即下判断, 无铺垫 |
+| 排比否定 | "既不…又不…也不…" 制造节奏 | 连续否定式排比 |
+| 不是A, 是B | 核心转折句式, 重定义 | "不是没X, 只是X都在Y里" |
+| 反击句式 | "有人说这是懒, 我觉得这是清醒" | 有人说X→我觉得Y |
+| 画面感细节 | 泉水不要钱/风免费吹/5块钱野花 | 具体到价格、动作、物件 |
+| 置景决定粉丝画像 | 背景直接决定吸什么人群 | 评价画面置景与目标人群匹配度 |
+| 收束升华 | 结尾回扣开头, 更高维定义 | "可能不是X, 而是Y" |
+| 系统化素材库 | 努力(搭系统) > 吃苦(蛮干) | 案例密度高, 信手拈来 |
+
+## 二、郑经说(功利型内容)
+
+| 技法 | 要点 |
+|---|---|
+| 人群分层 | 傻子/聪明人双轨: 流量款博泛人群, 付费款讲痛点 |
+| 功利型内容 | 教用户以极低成本获取极高收益, 天然有流量 |
+| 内容四原则 | 我相信/我想讲/我能想清楚/我能讲精炼——不迷信SOP |
+| 赚钱圣体 | 执行力→直播, 思考力→短视频(平台匹配) |
+
+## 三、三把刀(口播/成交文案, 86万粉)
+
+| 技法 | 要点 |
+|---|---|
+| 流量文案 vs 变现文案 | 流量讲观点(反认知), 变现讲痛点(迫切+解法) |
+| 平台基因 | 抖音=贪嗔痴算法; 视频号=真善美明牌 |
+| 痛点三段式 | 讲痛点→证明能解决→引导转化 |
+
+## 四、编导小希(操盘手, 前字节巨量课堂)
+
+| 技法 | 要点 |
+|---|---|
+| 前三秒画面钩子 | 钩子不一定来自第一句, 可以是第一个画面(猎奇+讲故事+期待感) |
+| 运动=流量密码 | 人物+运动开场, 完播率提升 |
+| 赛道决定上限 | 剧情/颜值类创作强但收入有天花板 |
+
+## 五、南门录像厅(剪辑思维, 126万粉)
+
+| 技法 | 要点 |
+|---|---|
+| 升格镜头 | 片子太平→慢动作制造节奏起伏 |
+| 素材资源库 | 剪辑最痛苦的是找素材, 建库解决 |
+
+## 六、逐段分析模板(报告 LLM 提示词的依据)
+
+对每个时间段检查 5 件事:
+1. **钩子力度** — 第一句/第一个画面是否停住人(小希: 画面钩子优先)
+2. **节奏控制** — 排比/对比/递进/升格(薛辉+南门)
+3. **画面匹配度** — 台词与置景/表情/动作是否互证(薛辉置景论)
+4. **情绪价值** — 共鸣/争议/认同(郑经说人群分层)
+5. **技法标签** — 用上表「导师·技法」命名
+
+## 七、评论区 VOC 分析维度
+
+- 分类: 认同 / 争议 / 补充 / 提问(三把刀: 认同是流量, 痛点是选题)
+- 高赞评论 = 免费的第二文案(可沉淀为下一条口播稿素材)
+- 最有共鸣的具体意象(如"社保交满就回云南") = 下一条视频的画面锚点
+
+## 八、结构评判基准(起承转合)
+
+| 段 | 功能 | 合格线 |
+|---|---|---|
+| 起(0-3s) | 观点/画面钩子 | 反常识 or 强画面 |
+| 承 | 解释+具体细节 | 至少2个画面感细节 |
+| 转 | 价值观对比/冲突 | 有明确对立面 |
+| 合 | 场景铺陈 | 情绪推到最高 |
+| 收 | 回扣+升华 | 闭环结构 |

references/s3-upload.md

@@ -0,0 +1,64 @@
+# S3 个人空间发布规范(存储铁律见 feme-identity-storage 技能)
+
+## 桶与路径
+
+```
+桶:   storage-s3-nkkj (华为云 OBS, cn-north-4)
+域名: 公网直接URL https://storage-s3-nkkj.obs.cn-north-4.myhuaweicloud.com/<key>
+      (s3.fmode.cn CNAME 生效后可换 https://s3.fmode.cn/<key>, 相对路径报告不受影响)
+铁律: 只能写 user/<我的userid>/ 前缀; 他人前缀 = 越权
+```
+
+## 拉片产物路径(硬性, 全英文小写)
+
+```
+user/<userid>/report/lapian/<平台简称>/<YYYYMMDD>/
+├── report.html           # 报告本体(入口)
+├── video.mp4             # 原视频
+├── audio.wav             # 音频
+├── transcript.json       # 带时间戳转写
+├── frames/f0001.jpg...   # 抽帧切片
+└── assets/               # 报告内嵌资源(analysis.json/comments.json/meta.json)
+```
+
+- 平台简称: douyin / xiaohongshu / weixin / bilibili / tiktok / kuaishou / shortdrama
+- 同一条视频重跑: 直接覆盖同日期目录(幂等); 换版本建议换日期目录
+
+## 相对路径(核心)
+
+报告 HTML 里所有资源引用一律**相对路径**:
+```html
+<source src="video.mp4">          <!-- 不是 https://... 绝对地址 -->
+<img src="frames/f0001.jpg">
+```
+→ 整个前缀自包含, s3.fmode.cn 生效/换桶/镜像到本地, 报告零改动可用。
+
+## 凭证策略(动态化)
+
+```
+优先: POST <FEME_STUDIO_URL>/api/storage/credentials
+      Header: Authorization: Bearer <FEME_SESSION_TOKEN>
+      Body:  {"prefix": "user/<userid>/report/lapian/"}
+      → {access, secret, securityToken, expires}  (限 prefix, 1小时)
+过渡: studio credentials API 未上线时(2026-08 实测 404),
+      回退容器级 CLOUD_SDK_AK / CLOUD_SDK_SK —— 仍只写自己 user/<userid>/ 前缀
+```
+`scripts/common.py get_storage_credentials()` 已封装该回退逻辑, 上线后零改动切换。
+
+## 上传技术要点(实测踩坑)
+
+1. **OBS 必须 virtual-hosted addressing**: bucket 放进 Host
+   (`storage-s3-nkkj.obs.cn-north-4.myhuaweicloud.com`),
+   path-style(把 bucket 放 URL 路径)会报 `VirtualHostDomainRequired`
+2. SigV4 签名: service=`s3`, region=`cn-north-4`,
+   头部至少签 `host;x-amz-content-sha256;x-amz-date`(临时凭证再加 x-amz-security-token)
+3. 公开读: 每个对象 PUT 时带 `x-amz-acl: public-read`(obsutil 老版没有 setacl,
+   boto3 需 `Config(s3={"addressing_style": "virtual"})`)
+4. Content-Type 要显式: html=`text/html; charset=utf-8`, mp4=`video/mp4`,
+   wav=`audio/wav`, jpg=`image/jpeg` —— 不然浏览器直接下载而非渲染
+5. 上传完对**每个**产物做 HEAD 校验(200 才算成功), publish.py 已内置
+6. 二进制绝不进 git(S3 只存产物, git 仓库只存代码/模板/文档)
+
+## 成本参考(2026-08 实测)
+
+1 条 1 分钟视频整套产物 ≈ 25MB(视频23MB+音频2MB+帧+html), OBS 流量费忽略不计。

references/voc-api.md

@@ -0,0 +1,64 @@
+# VOC 采集接口(经 server.fmode.cn 网关, 雨飏001容器 2026-08-28 实测可用)
+
+## 网关与鉴权
+
+```
+网关: https://server.fmode.cn/api/voc-social/<proxyPath>
+鉴权: Authorization: Bearer <FEME_NEWAPI_TOKEN>   # sk- 开头的 API 计费 token
+备选: x-api-key / body.token / query ?token=      # 服务端三通道都认, Bearer 最稳
+计费: 走 newapi 账套(每个生命自己的 token, 消耗自己余额)
+```
+
+注意:
+- 响应层级: **`data.aweme_list` / `data.aweme_detail`**, 不是 `data.data.aweme_list`
+- 错误分类: 401=token 无效(检查是否漏了 Bearer), 402/403=余额/权限, 429=风控限速(退避重试)
+- 所有请求带浏览器 UA; 平台风控时会 5xx, 客户端做 3 次指数退避
+
+## 抖音接口(已验证)
+
+### 1. 视频详情 — `douyin/web/fetch_one_video` (比 app/v3 可靠, 首选)
+```
+GET douyin/web/fetch_one_video?aweme_id=<视频ID>
+→ data.aweme_detail.{desc, author.nickname, author.signature, statistics,
+                      video.play_addr.url_list[], video.download_addr.url_list[],
+                      duration(ms), create_time, text_extra[].hashtag_name}
+```
+- 播放地址优先 `play_addr.url_list[0]`, 备用 `download_addr.url_list[0]`
+- CDN 带时效签名, 拿到就下, 别存 URL 过夜
+
+### 2. 搜索 — `douyin/app/v3/fetch_general_search`
+```
+GET douyin/app/v3/fetch_general_search?keyword=<关键词>&cursor=0
+→ data.data[] 或 data.aweme_list[], 每项含 aweme_id / desc / author / statistics
+```
+
+### 3. 评论 — `douyin/app/v3/fetch_video_comments`
+```
+GET douyin/app/v3/fetch_video_comments?aweme_id=<ID>&cursor=0&count=20
+→ data.comments[]{cid, text, digg_count, reply_comment_total,
+                   user.nickname, ip_label, create_time}
+翻页: cursor 累加返回条数; 返回空数组或 no more 即停
+```
+
+## 下载铁律
+- **完整下载**(流式读到底), 禁止 Range 截断 —— 拉片要全片
+- 下载后 `ffprobe` 验证时长与 meta.duration 一致
+- 命名: `<缓存目录>/<作者>/<aweme_id>.mp4`; 拉片工作目录用 `video.mp4`
+
+## 平台英文简称映射(S3 路径用)
+| 平台 | 简称 |
+|---|---|
+| 抖音 | douyin |
+| 小红书 | xiaohongshu |
+| 微信/视频号 | weixin |
+| B站 | bilibili |
+| TikTok | tiktok |
+| 快手 | kuaishou |
+| 短剧 | shortdrama |
+
+## Python 调用样例
+```python
+from scripts.common import voc_call, aweme_summary, extract_aweme_detail
+resp = voc_call("douyin/web/fetch_one_video", {"aweme_id": "7677548065384877346"})
+meta = aweme_summary(extract_aweme_detail(resp))   # desc/author/stats/play_urls
+```

scripts/build_report.py

@@ -0,0 +1,480 @@
+#!/usr/bin/env python3
+"""build_report.py — 生成拉片 HTML 报告(暗色主题/手机自适应/相对路径).
+
+用法:
+  python3 build_report.py --workdir /tmp/lapian --title "拉片分析:听show丽江酒吧"
+前置(可缺失, 缺了自动降级):
+  meta.json / transcript.json / frames.json / comments.json
+产物: workdir/report.html + workdir/analysis.json
+特性:
+  - 分段: 优先 LLM(编导方法论点评), 失败回退规则启发式, 永不中断
+  - 词频/情绪: 纯本地 n-gram + 情绪词典(无第三方依赖)
+  - 模板渲染: 内置 mini-Jinja(stdlib), 无需安装 jinja2
+"""
+import argparse
+import datetime
+import json
+import os
+import re
+import sys
+import urllib.request
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+from common import e as esc, fmt_ts  # noqa: E402
+
+ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+TEMPLATE = os.path.join(ROOT, "templates", "report.html.j2")
+TEXT_MODEL = os.environ.get("FEME_TEXT_MODEL", "deepseek-v4-flash")
+API_BASE = os.environ.get("FEME_API_BASE", "https://api.fmode.cn/v1")
+
+TAG_CLS = {"薛辉": "tag-blue", "郑经说": "tag-purple", "三把刀": "tag-green",
+           "小希": "tag-yellow", "南门": "tag-yellow", "志楠": "tag-green"}
+POS_KW = ("同感 真实 说得对 认同 共鸣 喜欢 爱 赞 通透 治愈 舒服 自由 幸福 羡慕 "
+          "向往 懂生活 回家 回云南 想去 泪目 豁达 乐观").split()
+NEG_KW = ("不认同 杠 胡说 假的 喷 讨厌 反对 无语 尬 难看 翻车 骗 贩卖焦虑 内卷 累 苦"
+          ).split()
+SENT_STOP = "的了是在和就有不人我都这也你们他们啊吧吗呢把被将还又再便而或如果所以因为但是"
+LLM_PROMPT = """你是短视频编导顾问(方法体系: 薛辉·观点前置/排比/画面感/收束升华; 郑经说·人群分层/功利型内容; 三把刀·流量文案vs变现文案; 编导小希·前三秒画面钩子)。对下面这条口播视频逐段拉片。
+
+视频数据: 时长{dur}s, 点赞{digg}, 评论{cmt}, 收藏{coll}, 分享{share}
+文案全文: {fulltext}
+
+逐段台词(带时间戳):
+{segment_lines}
+
+已抽帧画面参考:
+{frame_lines}
+
+严格输出 JSON(不要markdown围栏):
+{{"segments":[{{"i":0,"func":"起/承/转/合/收 之一","techniques":["导师·技法名"],"comment":"60-90字点评: 句式/节奏/画面匹配/情绪价值, 点名用了什么技法","reuse":"可直接套用的句式模板, 含{{占位符}}"}}],
+"suggestions":[{{"title":"建议标题","body":"40字内可执行动作"}}]}}
+要求: segments 数量与输入逐段台词一一对应; techniques 每段1-3个, 用"导师·技法"格式; suggestions 3条。"""
+
+
+# ----------------------------------------------------------- mini-Jinja ----
+_TOKEN = re.compile(r"({{.*?}}|{%.*?%})", re.S)
+
+
+def _lookup(val, key):
+    if isinstance(val, dict):
+        return val.get(key, "")
+    if isinstance(val, (list, tuple)):
+        if key == "length":
+            return len(val)
+        try:
+            return val[int(key)]
+        except (ValueError, IndexError):
+            return ""
+    return getattr(val, key, "")
+
+
+def _resolve(expr, ctx):
+    expr = expr.strip()
+    if re.fullmatch(r"-?\d+(\.\d+)?", expr):
+        return float(expr)
+    if expr in ("true", "True"):
+        return True
+    if expr in ("false", "False", "none", "None"):
+        return False
+    parts = expr.split(".")
+    val = ctx.get(parts[0], "")
+    for p in parts[1:]:
+        val = _lookup(val, p)
+    return val
+
+
+def _parse(tokens, i):
+    nodes = []
+    while i < len(tokens):
+        t = tokens[i]
+        if t.startswith("{%"):
+            stmt = t[2:-2].strip()
+            kw, _, rest = stmt.partition(" ")
+            if kw in ("endif", "endfor", "else"):
+                return nodes, i
+            if kw == "if":
+                branches, cond = [], rest.strip()
+                while True:
+                    body, i = _parse(tokens, i + 1)
+                    branches.append((cond, body))
+                    tk = tokens[i]
+                    tstmt = tk[2:-2].strip()
+                    tkw, _, trest = tstmt.partition(" ")
+                    if tkw == "elif":
+                        cond = trest.strip()
+                        continue
+                    if tkw == "else":
+                        body, i = _parse(tokens, i + 1)
+                        branches.append((None, body))
+                    i += 1  # consume endif
+                    break
+                nodes.append(("if", branches))
+            elif kw == "for":
+                var, itexpr = rest.strip().split(" in ", 1)
+                body, i = _parse(tokens, i + 1)
+                i += 1  # consume endfor
+                nodes.append(("for", var.strip(), itexpr.strip(), body))
+            else:
+                i += 1
+        elif t.startswith("{{"):
+            nodes.append(("expr", t[2:-2].strip()))
+            i += 1
+        else:
+            nodes.append(("text", t))
+            i += 1
+    return nodes, i
+
+
+def _exec(nodes, ctx, out):
+    for node in nodes:
+        if node[0] == "text":
+            out.append(node[1])
+        elif node[0] == "expr":
+            v = _resolve(node[1], ctx)
+            out.append("" if v is None else str(v))
+        elif node[0] == "if":
+            for cond, body in node[1]:
+                if cond is None or _resolve(cond, ctx):
+                    _exec(body, ctx, out)
+                    break
+        elif node[0] == "for":
+            _, var, itexpr, body = node
+            it = _resolve(itexpr, ctx)
+            items = it if isinstance(it, (list, tuple)) else ([it] if it else [])
+            n = len(items)
+            for idx, item in enumerate(items, 1):
+                sub = dict(ctx)
+                sub[var] = item
+                sub["loop"] = {"index": idx, "first": idx == 1, "last": idx == n}
+                _exec(body, sub, out)
+
+
+def render_template(template: str, ctx: dict) -> str:
+    tokens = _TOKEN.split(template)
+    nodes, _ = _parse(tokens, 0)
+    out = []
+    _exec(nodes, ctx, out)
+    return "".join(out)
+
+
+# ------------------------------------------------------------- analysis ----
+def load_json(path):
+    if os.path.exists(path):
+        try:
+            return json.load(open(path, encoding="utf-8"))
+        except Exception:
+            return None
+    return None
+
+
+def call_llm(prompt: str):
+    payload = {"model": TEXT_MODEL, "temperature": 0.3, "max_tokens": 4000,
+               "messages": [{"role": "user", "content": prompt}]}
+    req = urllib.request.Request(
+        f"{API_BASE}/chat/completions", data=json.dumps(payload).encode(),
+        headers={"Content-Type": "application/json",
+                 "Authorization": f"Bearer {os.environ.get('FEME_NEWAPI_TOKEN') or os.environ.get('FMODE_API_KEY','')}"})
+    with urllib.request.urlopen(req, timeout=180) as r:
+        resp = json.loads(r.read())
+    return resp["choices"][0]["message"]["content"]
+
+
+def parse_llm_json(text: str) -> dict:
+    text = text.strip()
+    if text.startswith("```"):
+        text = text.strip("`").lstrip("json").strip()
+    try:
+        return json.loads(text)
+    except Exception:
+        pass
+    if "{" in text and "}" in text:
+        try:
+            return json.loads(text[text.index("{"): text.rindex("}") + 1])
+        except Exception:
+            pass
+    return {}
+
+
+def tag_cls(label: str) -> str:
+    for k, cls in TAG_CLS.items():
+        if k in label:
+            return cls
+    return "tag-blue"
+
+
+def heuristic_techniques(i: int, n: int, text: str) -> list:
+    tags = []
+    if i == 0:
+        tags.append("薛辉·观点前置")
+        tags.append("小希·前三秒钩子")
+    if re.search(r"(既不|又不|也不|不急着|不是[^。，]*,?不是)", text):
+        tags.append("薛辉·排比")
+    if re.search(r"(有人说|不是.*而是|别人.*我们|他们.*我们)", text):
+        tags.append("薛辉·反击句式/对比")
+    if re.search(r"(块钱|免费|不要钱|吃|喝|买|菜市场|院子)", text):
+        tags.append("薛辉·画面感细节")
+    if i == n - 1:
+        tags.append("薛辉·收束升华")
+    return tags or ["志楠·节奏控制"]
+
+
+def fallback_comment(i, n, text) -> str:
+    if i == 0:
+        return "开场即抛出反常识观点, 承担前三秒留人任务; 句式短促, 信息密度高。"
+    if i == n - 1:
+        return "结尾回扣开头形成闭环, 用更高维度的定义完成价值升华。"
+    return "中段延续主线, 用具体细节支撑观点, 保持节奏与信息密度。"
+
+
+def chi_words(text: str, top: int = 8) -> list:
+    """无分词词典 → 2/3-gram 近似, 去停用字边, 去包含重复。"""
+    text = re.sub(r"[^\u4e00-\u9fff]", " ", text)
+    grams = {}
+    for w in text.split():
+        for n in (2, 3):
+            for k in range(len(w) - n + 1):
+                g = w[k:k + n]
+                grams[g] = grams.get(g, 0) + 1
+    cands = [(g, c) for g, c in grams.items()
+             if c >= 2 and g[0] not in SENT_STOP and g[-1] not in SENT_STOP]
+    cands.sort(key=lambda x: -x[1])
+    kept = []
+    for g, c in cands:
+        if not any(g in k for k, _ in kept):
+            kept.append((g, c))
+        if len(kept) >= top:
+            break
+    mx = kept[0][1] if kept else 1
+    return [{"word": g, "count": c, "pct": max(6, round(c / mx * 100))}
+            for g, c in kept]
+
+
+def sentiment_split(text: str) -> list:
+    sents = [s for s in re.split(r"[。！？；\n]", text) if s.strip()]
+    pos = sum(1 for s in sents if any(k in s for k in POS_KW))
+    neg = sum(1 for s in sents if any(k in s for k in NEG_KW))
+    neu = max(len(sents) - pos - neg, 0)
+    tot = max(pos + neg + neu, 1)
+    rows = [("正面", pos, "var(--green)"), ("中性", neu, "var(--yellow)"),
+            ("负面", neg, "var(--red)")]
+    out, acc = [], 0
+    for i, (label, v, color) in enumerate(rows):
+        pct = round(v / tot * 100) if i < 2 else max(0, 100 - acc)
+        acc += pct
+        out.append({"label": label, "pct": pct, "color": color})
+    return out
+
+
+def comment_insight(text: str) -> str:
+    if any(k in text for k in ("回去", "回云南", "回家", "社保")):
+        return "在外游子的身份共鸣——最能带动转发"
+    if any(k in text for k in ("同感", "真实", "说得对", "就是这样")):
+        return "强认同型评论, 印证文案戳中人群"
+    if any(k in text for k in ("不是", "而是", "——")):
+        return "评论本身有文案意识, 可沉淀为选题素材"
+    if any(k in text for k in ("但", "不过", "其实")):
+        return "补充视角, 可作为下一条视频的回应点"
+    return "氛围型互动"
+
+
+def classify_comment(text: str) -> str:
+    if any(k in text for k in NEG_KW):
+        return "neg"
+    if any(k in text for k in POS_KW):
+        return "pos"
+    return "neu"
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--workdir", required=True)
+    ap.add_argument("--title", default="")
+    args = ap.parse_args()
+    wd = args.workdir
+
+    meta = load_json(os.path.join(wd, "meta.json")) or {}
+    tr = load_json(os.path.join(wd, "transcript.json")) or {}
+    frames = load_json(os.path.join(wd, "frames.json")) or []
+    comments = load_json(os.path.join(wd, "comments.json")) or []
+
+    data = tr.get("data") or {}
+    segs = data.get("segments") or []
+    fulltext = data.get("text") or "".join(s.get("text", "") for s in segs)
+    if not segs and fulltext:  # 无时间戳转写 → 整体一段
+        segs = [{"bg": 0, "ed": int(meta.get("duration_s", 60) * 1000),
+                 "speaker": "0", "text": fulltext}]
+    if not segs:
+        raise SystemExit("[report] 没有转写内容, 先跑 transcribe.py")
+    duration_s = meta.get("duration_s") or (segs[-1]["ed"] / 1000)
+
+    # --- LLM 逐段点评(失败回退启发式) ---
+    llm_engine, llm_out = "规则启发式", {}
+    try:
+        seg_lines = "\n".join(
+            f"[{fmt_ts(s['bg'])}-{fmt_ts(s['ed'])}] {s.get('text','')}" for s in segs)
+        frame_lines = "\n".join(
+            f"{f['t_label']}: {f.get('caption','')}" for f in frames) or "(无)"
+        st = meta.get("stats") or {}
+        prompt = LLM_PROMPT.format(
+            dur=round(duration_s), digg=st.get("digg", "?"), cmt=st.get("comment", "?"),
+            coll=st.get("collect", "?"), share=st.get("share", "?"),
+            fulltext=fulltext[:1200], segment_lines=seg_lines, frame_lines=frame_lines)
+        raw = call_llm(prompt)
+        llm_out = parse_llm_json(raw)
+        if llm_out.get("segments"):
+            llm_engine = TEXT_MODEL
+            print(f"[report] LLM 点评 {len(llm_out['segments'])} 段 ({llm_engine})")
+        else:
+            print("[report] LLM 输出无法解析, 回退规则启发式")
+    except Exception as ex:
+        print(f"[report] LLM 调用失败({ex}), 回退规则启发式")
+
+    llm_segs = {int(s.get("i", i)): s for i, s in enumerate(llm_out.get("segments") or [])}
+
+    # --- 组装分析段 ---
+    n = len(segs)
+    analysis_segs = []
+    for i, s in enumerate(segs):
+        t0, t1 = int(s.get("bg", 0)), int(s.get("ed", 0))
+        text = s.get("text", "")
+        L = llm_segs.get(i) or {}
+        techniques = [t for t in (L.get("techniques") or
+                                  heuristic_techniques(i, n, text)) if t]
+        frame = min(frames, key=lambda f: abs(f["t_ms"] - t0),
+                    default=None) if frames else None
+        if frame and abs(frame["t_ms"] - t0) > 8000:
+            frame = None
+        analysis_segs.append({
+            "i": i, "bg": t0, "ed": t1,
+            "t_start": fmt_ts(t0), "t_end": fmt_ts(t1), "t_start_s": t0 // 1000,
+            "text": esc(text), "preview": esc(text[:24] + ("…" if len(text) > 24 else "")),
+            "techniques": [{"label": esc(t), "cls": tag_cls(t)} for t in techniques],
+            "func": esc(L.get("func") or ""),
+            "comment_html": esc(L.get("comment") or fallback_comment(i, n, text)),
+            "reuse": esc(L.get("reuse") or ""),
+            "frame": frame,
+        })
+
+    # --- 整体结构表 ---
+    stage_names = ["起", "承", "转", "合", "收"]
+    structure_rows = []
+    for idx, sg in enumerate(analysis_segs):
+        stage = sg["func"] or stage_names[min(int(idx / n * len(stage_names)),
+                                              len(stage_names) - 1)]
+        structure_rows.append({
+            "stage": esc(stage), "time": f"{sg['t_start']}—{sg['t_end']}",
+            "func": esc(re.sub(r"^.{0,3}[:：]", "", sg["func"]) if sg["func"]
+                        else ("开场留人" if idx == 0 else
+                              ("收束升华" if idx == n - 1 else "承接展开"))),
+            "techniques": " ".join(t["label"] for t in sg["techniques"]) or "—",
+        })
+    summary_html = esc(llm_out.get("structure_summary") or
+                       f"{n}段完成全片, 各段技法如上; 逐段细节见上方可折叠卡片。")
+
+    # --- 词频/情绪 ---
+    wf = chi_words(fulltext)
+    sentiment = sentiment_split(fulltext)
+    word_count = len(re.sub(r"\s", "", fulltext))
+    wpm = round(word_count / max(duration_s / 60, 0.1))
+
+    # --- 评论区 ---
+    comment_stats, top_comments, comment_keywords, comment_total = [], [], [], 0
+    if comments:
+        cls_counts = {"pos": 0, "neu": 0, "neg": 0}
+        for c in comments:
+            cls_counts[classify_comment(c.get("text", ""))] += 1
+        comment_total = len(comments)
+        rows = [("正面认同", cls_counts["pos"], "var(--green)"),
+                ("中性补充", cls_counts["neu"], "var(--yellow)"),
+                ("争议", cls_counts["neg"], "var(--red)")]
+        acc = 0
+        for i, (label, v, color) in enumerate(rows):
+            pct = round(v / comment_total * 100) if i < 2 else max(0, 100 - acc)
+            acc += pct
+            comment_stats.append({"label": label, "pct": pct, "color": color})
+        top_comments = [{"nickname": esc(c.get("nickname", "匿名")),
+                         "likes": c.get("likes", 0), "text": esc(c.get("text", "")),
+                         "insight": esc(comment_insight(c.get("text", "")))}
+                        for c in comments[:5]]
+        comment_keywords = chi_words(" ".join(c.get("text", "") for c in comments), top=6)
+
+    # --- 数据卡片 ---
+    st = meta.get("stats") or {}
+    stats_cards = [{"num": f"{duration_s:.0f}s", "label": "总时长"}]
+    if st.get("digg"):
+        stats_cards += [{"num": f"{st['digg']:,}", "label": "👍 点赞"},
+                        {"num": f"{st['comment']:,}", "label": "💬 评论"},
+                        {"num": f"{st['collect']:,}", "label": "🔖 收藏"},
+                        {"num": f"{st['share']:,}", "label": "↗ 分享"},
+                        {"num": f"{st['digg'] / max(st['comment'], 1):.0f}:1",
+                         "label": "赞评比"}]
+    else:
+        stats_cards += [{"num": len(comments), "label": "评论采集"},
+                        {"num": len(frames), "label": "抽帧"},
+                        {"num": n, "label": "分段"},
+                        {"num": word_count, "label": "总字数"},
+                        {"num": wpm, "label": "字/分"}]
+
+    # --- 建议 ---
+    suggestions = [{"title": esc(s.get("title", f"建议{ i + 1 }")),
+                    "body": esc(s.get("body", ""))}
+                   for i, s in enumerate(llm_out.get("suggestions") or [])]
+    if not suggestions:
+        hw = wf[0]["word"] if wf else "核心词"
+        pos_pct = next((r["pct"] for r in (comment_stats or sentiment)
+                        if r["label"] in ("正面认同", "正面")), 0)
+        suggestions = [
+            {"title": "强化记忆锚点", "body": f"高频词「{hw}」贯穿全片, 封面/标题/评论区置顶都应重复它, 形成账号记忆点。"},
+            {"title": "延续情绪线", "body": f"正面情绪约{pos_pct}%, 下一条保持同一情绪基调并升级一个具体场景细节。"},
+            {"title": "补强行动指令", "body": "结尾闭环后加一句轻互动引导(如「你家乡呢?」), 把共鸣转化为评论量。"},
+        ]
+
+    # --- 汇总上下文(所有动态值已转义) ---
+    author = meta.get("author") or "本地视频"
+    title = args.title or f"拉片分析:{author}"
+    today = datetime.date.today().strftime("%Y%m%d")
+    aweme_id = str(meta.get("aweme_id") or "local")
+    ctx = {
+        "meta": {
+            "title": esc(title),
+            "subtitle": esc(f"{(meta.get('desc') or '')[:42]} · {duration_s:.0f}秒 · "
+                            f"{n}段 · 方法论对照拆解"),
+            "author": esc(author), "author_signature": esc(meta.get("author_signature", "")),
+            "tags": [esc(t) for t in (meta.get("tags") or [])],
+            "publish_time": esc(meta.get("create_time", "")),
+            "aweme_id": esc(aweme_id), "platform": esc(meta.get("platform", "douyin")),
+            "generated_at": datetime.datetime.now().strftime("%Y-%m-%d %H:%M"),
+            "llm_engine": esc(llm_engine),
+            "report_id": esc(f"lapian-{meta.get('platform','douyin')}-{today}-{aweme_id[:10]}"),
+            "word_count": word_count, "wpm": wpm,
+            "sentence_count": len([s for s in re.split(r"[。！？]", fulltext) if s.strip()]),
+        },
+        "stats_cards": stats_cards,
+        "segments": analysis_segs,
+        "structure_rows": structure_rows,
+        "summary_html": summary_html,
+        "wordfreq": wf,
+        "sentiment": sentiment,
+        "comments": bool(comments),
+        "comment_total": comment_total,
+        "comment_stats": comment_stats,
+        "top_comments": top_comments,
+        "comment_keywords": comment_keywords,
+        "suggestions": suggestions,
+    }
+
+    html = render_template(open(TEMPLATE, encoding="utf-8").read(), ctx)
+    out_html = os.path.join(wd, "report.html")
+    open(out_html, "w", encoding="utf-8").write(html)
+    json.dump({"meta": {k: v for k, v in ctx["meta"].items()},
+               "segments": [{k: (v if k != "text" else v) for k, v in sg.items()
+                             if k not in ("frame",)} for sg in analysis_segs],
+               "suggestions": suggestions,
+               "llm_engine": llm_engine},
+              open(os.path.join(wd, "analysis.json"), "w", encoding="utf-8"),
+              ensure_ascii=False, indent=2, default=str)
+    print(f"[report] {out_html} ({os.path.getsize(out_html)/1024:.0f} KB)")
+
+
+if __name__ == "__main__":
+    main()

scripts/common.py

@@ -0,0 +1,338 @@
+#!/usr/bin/env python3
+"""skill-video-lapian 共享库: 动态身份参数 + VOC网关 + S3 SigV4 上传.
+
+铁律: 所有凭据/端点从环境或 fmode-identity.json 读取, 绝不硬编码。
+每个数字生命用自己的 FEME_USERID / FEME_SESSION_TOKEN / FEME_NEWAPI_TOKEN。
+"""
+import datetime
+import hashlib
+import hmac
+import json
+import os
+import re
+import ssl
+import sys
+import urllib.error
+import urllib.parse
+import urllib.request
+
+UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
+      "(KHTML, like Gecko) Chrome/126.0 Safari/537.36")
+
+_IDENTITY_CACHE = None
+
+
+# ---------------------------------------------------------------- identity --
+def _load_identity_file():
+    """读取 /opt/data/fmode-identity.json (或 $FEME_IDENTITY_FILE)。"""
+    global _IDENTITY_CACHE
+    if _IDENTITY_CACHE is not None:
+        return _IDENTITY_CACHE
+    path = os.environ.get("FEME_IDENTITY_FILE", "/opt/data/fmode-identity.json")
+    data = {}
+    if os.path.exists(path):
+        try:
+            data = json.load(open(path, encoding="utf-8"))
+        except Exception as e:  # 文件坏了不致命, 继续用 env
+            print(f"[common] WARN 解析 {path} 失败: {e}", file=sys.stderr)
+    _IDENTITY_CACHE = data
+    return data
+
+
+def _identity_get(*names, secret=False):
+    """按优先级取身份参数: env(FEME_*) > env(兼容旧名) > fmode-identity.json。"""
+    env_map = {
+        "userid": ["FEME_USERID"],
+        "session_token": ["FEME_SESSION_TOKEN", "FMODE_SESSION_TOKEN"],
+        "newapi_token": ["FEME_NEWAPI_TOKEN", "FMODE_API_KEY", "FMODE_API_TOKEN"],
+        "studio_url": ["FEME_STUDIO_URL"],
+        "cloud_ak": ["CLOUD_SDK_AK"],
+        "cloud_sk": ["CLOUD_SDK_SK"],
+    }
+    for env_names in env_map.get(names[0], [names]):
+        for n in env_names:
+            v = os.environ.get(n, "").strip()
+            if v:
+                return v
+    ident = _load_identity_file()
+    v = (ident.get(names[0]) or ident.get(_camel(names[0])) or "").strip()
+    if v:
+        return v
+    return ""
+
+
+def _camel(snake):
+    return "".join(p.capitalize() for p in snake.split("_"))
+
+
+class IdentityError(RuntimeError):
+    pass
+
+
+def get_userid() -> str:
+    uid = _identity_get("userid")
+    if not uid:
+        raise IdentityError(
+            "缺少身份参数 FEME_USERID (或 fmode-identity.json.userid)。"
+            "每个数字生命必须用自己的 userid, 禁止借用他人空间。")
+    return uid
+
+
+def get_session_token() -> str:
+    tok = _identity_get("session_token")
+    if not tok:
+        raise IdentityError("缺少 FEME_SESSION_TOKEN (飞马会话 r:...)")
+    return tok
+
+
+def get_api_token() -> str:
+    """API 计费 token (sk-..., 调 VOC 网关 / fmode-listen / 视觉模型都用它)。"""
+    tok = _identity_get("newapi_token")
+    if not tok:
+        raise IdentityError("缺少 FEME_NEWAPI_TOKEN (sk-...)")
+    return tok
+
+
+def get_studio_url() -> str:
+    """自己的 studio 端点 (storage/credentials 上线后用)。"""
+    url = _identity_get("studio_url")
+    if url:
+        return url.rstrip("/")
+    uid = get_userid()
+    # 雨飏001 端口约定; 其他生命容器请显式 export FEME_STUDIO_URL
+    return "https://server.fmode.cn:19001" if uid == "sr2WiPsDyQ" else ""
+
+
+def get_storage_credentials(prefix: str = "") -> dict:
+    """优先走 studio /api/storage/credentials (用户级临时凭证);
+    未上线(过渡期)回退容器级 CLOUD_SDK_AK/SK。返回 {access_key, secret_key}。"""
+    studio = get_studio_url()
+    if studio:
+        try:
+            body = json.dumps({"prefix": prefix}).encode()
+            req = urllib.request.Request(
+                f"{studio}/api/storage/credentials", data=body, method="POST",
+                headers={"Content-Type": "application/json",
+                         "Authorization": f"Bearer {get_session_token()}"})
+            resp = json.loads(urllib.request.urlopen(req, timeout=15).read())
+            cred = resp.get("data") or resp
+            if cred.get("access") and cred.get("secret"):
+                return {"access_key": cred["access"], "secret_key": cred["secret"],
+                        "security_token": cred.get("securityToken") or "", "source": "studio"}
+        except urllib.error.HTTPError as e:
+            if e.code != 404:
+                print(f"[common] WARN credentials API {e.code}, 回退容器 AK/SK", file=sys.stderr)
+        except Exception as e:
+            print(f"[common] WARN credentials API 不可用({e}), 回退容器 AK/SK", file=sys.stderr)
+    ak, sk = _identity_get("cloud_ak"), _identity_get("cloud_sk")
+    if ak and sk:
+        return {"access_key": ak, "secret_key": sk, "security_token": "", "source": "container-env"}
+    raise IdentityError("拿不到 S3 凭证: studio credentials 不可用且无 CLOUD_SDK_AK/SK")
+
+
+# --------------------------------------------------------------- VOC gateway --
+def voc_call(proxy_path: str, params=None, method="GET", retries=3):
+    """调 https://server.fmode.cn/api/voc-social/<proxyPath>, Bearer sk-token。
+
+    解析层级注意: data.aweme_list (不是 data.data.aweme_list)。
+    返回 json 或 {"error": ...}。
+    """
+    base = os.environ.get("VOC_SOCIAL_GATEWAY", "https://server.fmode.cn/api/voc-social")
+    url = f"{base}/{proxy_path.lstrip('/')}"
+    if params and method.upper() == "GET":
+        qs = urllib.parse.urlencode({k: v for k, v in params.items() if v is not None})
+        if qs:
+            url += "?" + qs
+    headers = {"Authorization": f"Bearer {get_api_token()}",
+               "Accept": "application/json", "User-Agent": UA}
+    body = None
+    if method.upper() == "POST":
+        headers["Content-Type"] = "application/json"
+        body = json.dumps(params or {}).encode()
+    last = None
+    for attempt in range(retries):
+        req = urllib.request.Request(url, headers=headers, data=body,
+                                     method=method.upper())
+        try:
+            with urllib.request.urlopen(req, timeout=90) as r:
+                return json.loads(r.read().decode("utf-8", "replace"))
+        except urllib.error.HTTPError as e:
+            detail = e.read().decode("utf-8", "replace")[:300]
+            last = f"HTTP {e.code}: {detail}"
+            if e.code >= 500 and attempt < retries - 1:
+                continue
+            break
+        except Exception as e:
+            last = str(e)
+            if attempt < retries - 1:
+                continue
+            break
+    return {"error": last}
+
+
+def extract_aweme_detail(resp: dict) -> dict:
+    """从 fetch_one_video 响应提取 aweme 对象(容多种层级)。"""
+    data = resp.get("data") or {}
+    aweme = data.get("aweme_detail")
+    if not aweme:
+        lst = data.get("aweme_list") or []
+        aweme = lst[0] if lst else data
+    return aweme or {}
+
+
+def aweme_summary(aweme: dict) -> dict:
+    """提取报告需要的视频元数据。"""
+    st = aweme.get("statistics") or {}
+    author = aweme.get("author") or {}
+    video = aweme.get("video") or {}
+    play = (video.get("play_addr") or {}).get("url_list") or []
+    download = (video.get("download_addr") or {}).get("url_list") or []
+    tags = [t.get("hashtag_name") for t in (aweme.get("text_extra") or [])
+            if t.get("hashtag_name")]
+    ct = aweme.get("create_time")
+    return {
+        "aweme_id": aweme.get("aweme_id") or aweme.get("aweme_id_str") or "",
+        "desc": aweme.get("desc") or "",
+        "author": author.get("nickname") or "",
+        "author_signature": author.get("signature") or "",
+        "tags": tags,
+        "duration_ms": aweme.get("duration") or video.get("duration") or 0,
+        "create_time": datetime.datetime.fromtimestamp(ct).strftime("%Y-%m-%d %H:%M") if ct else "",
+        "stats": {
+            "digg": st.get("digg_count", 0), "comment": st.get("comment_count", 0),
+            "collect": st.get("collect_count", 0), "share": st.get("share_count", 0),
+            "play": st.get("play_count", 0),
+        },
+        "play_urls": play, "download_urls": download,
+    }
+
+
+# ------------------------------------------------------------------- ffmpeg --
+def ffprobe_duration(path: str) -> float:
+    import subprocess
+    out = subprocess.run(
+        ["ffprobe", "-v", "error", "-show_entries", "format=duration",
+         "-of", "default=noprint_wrappers=1:nokey=1", str(path)],
+        capture_output=True, text=True)
+    try:
+        return float(out.stdout.strip())
+    except ValueError:
+        return 0.0
+
+
+# ------------------------------------------------------------- S3 SigV4 PUT --
+# OBS(华为云) 必须 virtual-hosted addressing: bucket 放进 Host。
+S3_BUCKET = os.environ.get("S3_BUCKET", "storage-s3-nkkj")
+S3_REGION = os.environ.get("S3_REGION", "cn-north-4")
+S3_HOST = os.environ.get("S3_ENDPOINT", "https://obs.cn-north-4.myhuaweicloud.com") \
+    .replace("https://", "").rstrip("/")
+
+
+def _sigv4(method: str, key: str, payload: bytes, cred: dict,
+           content_type: str = "", query: str = "") -> dict:
+    """生成 OBS S3 兼容 SigV4 头。cred = get_storage_credentials()。"""
+    host = f"{S3_BUCKET}.{S3_HOST}"
+    t = datetime.datetime.now(datetime.timezone.utc)
+    amzdate = t.strftime("%Y%m%dT%H%M%SZ")
+    datestamp = t.strftime("%Y%m%d")
+    payload_hash = hashlib.sha256(payload).hexdigest()
+    headers = {"host": host, "x-amz-content-sha256": payload_hash, "x-amz-date": amzdate}
+    if content_type:
+        headers["content-type"] = content_type
+    if cred.get("security_token"):
+        headers["x-amz-security-token"] = cred["security_token"]
+    signed = ";".join(sorted(headers))
+    canonical_headers = "".join(f"{k}:{headers[k]}\n" for k in sorted(headers))
+    canonical = (f"{method}\n/{key}\n{query}\n{canonical_headers}\n{signed}\n{payload_hash}")
+    scope = f"{datestamp}/{S3_REGION}/s3/aws4_request"
+    sts = f"AWS4-HMAC-SHA256\n{amzdate}\n{scope}\n{hashlib.sha256(canonical.encode()).hexdigest()}"
+
+    def hm(k, m):
+        return hmac.new(k, m.encode(), hashlib.sha256).digest()
+
+    k = hm(hm(hm(hm(("AWS4" + cred["secret_key"]).encode(), datestamp), S3_REGION), "s3"),
+           "aws4_request")
+    sig = hmac.new(k, sts.encode(), hashlib.sha256).hexdigest()
+    auth = (f"AWS4-HMAC-SHA256 Credential={cred['access_key']}/{scope}, "
+            f"SignedHeaders={signed}, Signature={sig}")
+    out = {"Host": host, "Authorization": auth, "x-amz-date": amzdate,
+           "x-amz-content-sha256": payload_hash, "x-amz-acl": "public-read"}
+    if content_type:
+        out["Content-Type"] = content_type
+    if cred.get("security_token"):
+        out["x-amz-security-token"] = cred["security_token"]
+    return out
+
+
+def s3_put_bytes(key: str, payload: bytes, content_type: str = "application/octet-stream",
+                 public_read: bool = True) -> str:
+    """上传字节到 S3_BUCKET, 返回可公开访问的 URL。"""
+    cred = get_storage_credentials(f"user/{get_userid()}/")
+    hdrs = _sigv4("PUT", key, payload, cred, content_type)
+    if not public_read:
+        hdrs.pop("x-amz-acl", None)
+    url = f"https://{S3_BUCKET}.{S3_HOST}/{key}"
+    req = urllib.request.Request(url, data=payload, method="PUT", headers=hdrs)
+    with urllib.request.urlopen(req, timeout=300) as r:
+        if r.status not in (200, 201):
+            raise RuntimeError(f"S3 PUT {key} -> {r.status}")
+    return public_url(key)
+
+
+def s3_put_file(path: str, key: str, content_type: str = None) -> str:
+    ct = content_type or guess_content_type(path)
+    with open(path, "rb") as f:
+        return s3_put_bytes(key, f.read(), ct)
+
+
+def public_url(key: str) -> str:
+    """公开访问 URL。s3.fmode.cn CNAME 生效前用 OBS 原生域名。"""
+    if os.environ.get("S3_PUBLIC_BASE"):
+        return f"{os.environ['S3_PUBLIC_BASE'].rstrip('/')}/{key}"
+    return f"https://{S3_BUCKET}.{S3_HOST}/{key}"
+
+
+def guess_content_type(path: str) -> str:
+    import mimetypes
+    return mimetypes.guess_type(path)[0] or "application/octet-stream"
+
+
+# ----------------------------------------------------------------- helpers --
+def fmt_ts(ms: float) -> str:
+    s = int(ms / 1000)
+    return f"{s // 60}:{s % 60:02d}"
+
+
+def slugify_platform(name: str) -> str:
+    """平台英文简称映射(douyin/xiaohongshu/weixin/bilibili/tiktok/shortdrama...)。"""
+    m = {"抖音": "douyin", "douyin": "douyin", "tiktok": "tiktok",
+         "小红书": "xiaohongshu", "xiaohongshu": "xiaohongshu", "rednote": "xiaohongshu",
+         "微信": "weixin", "weixin": "weixin", "wechat": "weixin", "视频号": "weixin",
+         "b站": "bilibili", "哔哩哔哩": "bilibili", "bilibili": "bilibili",
+         "快手": "kuaishou", "kuaishou": "kuaishou", "短剧": "shortdrama",
+         "shortdrama": "shortdrama"}
+    return m.get((name or "").strip().lower(), "douyin")
+
+
+def download_file(url: str, dest: str, min_size: int = 100_000) -> int:
+    """流式下载(完整, 不 Range 截断), 返回字节数。"""
+    os.makedirs(os.path.dirname(dest) or ".", exist_ok=True)
+    req = urllib.request.Request(url, headers={"User-Agent": UA})
+    total = 0
+    with urllib.request.urlopen(req, timeout=300) as r, open(dest, "wb") as f:
+        while True:
+            chunk = r.read(1 << 20)
+            if not chunk:
+                break
+            f.write(chunk)
+            total += len(chunk)
+    if total < min_size:
+        os.remove(dest)
+        raise RuntimeError(f"下载不完整({total}B < {min_size}B), 已删除 {dest}")
+    return total
+
+
+def e(text) -> str:
+    """HTML 转义(模板里用)。"""
+    from html import escape
+    return escape(str(text or ""), quote=True)

scripts/fetch_comments.py

@@ -0,0 +1,63 @@
+#!/usr/bin/env python3
+"""fetch_comments.py — 采集评论区(可选, VOC 网关翻页).
+
+用法:
+  python3 fetch_comments.py --aweme-id 7677548065384877346 --workdir /tmp/lapian [--max 100]
+产物: workdir/comments.json = [{nickname,text,likes,...}...] (按点赞排序)
+"""
+import argparse
+import json
+import os
+import sys
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+from common import voc_call  # noqa: E402
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--aweme-id", required=True)
+    ap.add_argument("--workdir", required=True)
+    ap.add_argument("--max", type=int, default=100)
+    args = ap.parse_args()
+
+    comments, cursor, seen = [], 0, set()
+    while len(comments) < args.max:
+        resp = voc_call("douyin/app/v3/fetch_video_comments",
+                        {"aweme_id": args.aweme_id, "cursor": cursor, "count": 20})
+        if resp.get("error"):
+            print(f"[comments] 网关错误: {resp['error']}")
+            break
+        data = resp.get("data") or {}
+        batch = data.get("comments") or (data.get("data") or {}).get("comments") or []
+        if not batch:
+            break
+        new = 0
+        for c in batch:
+            cid = c.get("cid") or c.get("comment_id") or f"{c.get('user', {}).get('nickname')}|{c.get('create_time')}"
+            if cid in seen:
+                continue
+            seen.add(cid)
+            comments.append({
+                "nickname": (c.get("user") or {}).get("nickname") or "",
+                "text": c.get("text") or "",
+                "likes": c.get("digg_count") or 0,
+                "reply_count": c.get("reply_comment_total") or 0,
+                "time": c.get("create_time") or 0,
+                "ip_label": c.get("ip_label") or "",
+            })
+            new += 1
+        print(f"[comments] cursor={cursor} +{new} (累计 {len(comments)})")
+        cursor += len(batch)
+        if new == 0:
+            break
+
+    comments.sort(key=lambda x: -x["likes"])
+    out = os.path.join(args.workdir, "comments.json")
+    os.makedirs(args.workdir, exist_ok=True)
+    json.dump(comments, open(out, "w", encoding="utf-8"), ensure_ascii=False, indent=2)
+    print(f"[comments] {len(comments)} 条 → {out}")
+
+
+if __name__ == "__main__":
+    main()

scripts/fetch_video.py

@@ -0,0 +1,101 @@
+#!/usr/bin/env python3
+"""fetch_video.py — 拉片第一步: 拿到完整视频.
+
+用法:
+  # 抖音分享链接 / 视频ID → VOC 网关取详情+完整下载
+  python3 fetch_video.py --url "https://v.douyin.com/xxxx/" --out-dir /tmp/lapian
+  python3 fetch_video.py --aweme-id 7677548065384877346 --platform douyin --out-dir /tmp/lapian
+  # 已有本地视频 → 直接登记
+  python3 fetch_video.py --local /path/video.mp4 --out-dir /tmp/lapian --platform douyin
+
+产物: <out-dir>/video.mp4 (完整) + meta.json (作者/统计/标签/时长)
+铁律: 视频必须完整下载, 用 ffprobe 验证时长, 不许 Range 截断。
+"""
+import argparse
+import json
+import os
+import re
+import subprocess
+import sys
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+from common import (aweme_summary, download_file, extract_aweme_detail,  # noqa: E402
+                    ffprobe_duration, get_api_token, slugify_platform, voc_call)
+
+
+def resolve_share_url(url: str) -> str:
+    """短链接 → 最终 URL(带重定向链)。"""
+    import urllib.request
+    req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
+    with urllib.request.urlopen(req, timeout=30) as r:
+        return r.geturl()
+
+
+def extract_aweme_id(final_url: str) -> str:
+    m = re.search(r"/(?:video|note)/(\d+)", final_url) or re.search(r"aweme_id=(\d+)", final_url)
+    if not m:
+        raise SystemExit(f"无法从 URL 提取视频ID: {final_url}")
+    return m.group(1)
+
+
+def fetch_meta(aweme_id: str, platform: str) -> dict:
+    """VOC 网关取视频详情(抖音)。其他平台留 TODO。"""
+    if platform != "douyin":
+        print(f"[fetch] WARN 平台 {platform} 详情接口未接入, 仅下载/登记")
+        return {}
+    resp = voc_call("douyin/web/fetch_one_video", {"aweme_id": aweme_id})
+    if resp.get("error"):
+        raise SystemExit(f"[fetch] VOC fetch_one_video 失败: {resp['error']}")
+    return aweme_summary(extract_aweme_detail(resp))
+
+
+def main():
+    ap = argparse.ArgumentParser(description="下载/登记待拉片视频")
+    ap.add_argument("--url", help="抖音分享短链或含视频ID的页面URL")
+    ap.add_argument("--aweme-id", help="视频ID(数字)")
+    ap.add_argument("--local", help="已有本地视频文件路径(跳过下载)")
+    ap.add_argument("--platform", default="douyin",
+                    help="平台英文简称: douyin/xiaohongshu/weixin/bilibili/tiktok/shortdrama")
+    ap.add_argument("--out-dir", required=True, help="工作目录(产物放这里)")
+    args = ap.parse_args()
+
+    os.makedirs(args.out_dir, exist_ok=True)
+    platform = slugify_platform(args.platform)
+    meta, video_path = {}, os.path.join(args.out_dir, "video.mp4")
+
+    if args.local:
+        video_path = os.path.abspath(args.local)
+        print(f"[fetch] 使用本地视频: {video_path}")
+    else:
+        if not (args.url or args.aweme_id):
+            ap.error("需要 --url / --aweme-id / --local 之一")
+        aweme_id = args.aweme_id
+        if args.url:
+            final = resolve_share_url(args.url)
+            aweme_id = extract_aweme_id(final)
+        print(f"[fetch] 视频ID: {aweme_id} (平台: {platform})")
+        meta = fetch_meta(aweme_id, platform)
+        urls = meta.get("play_urls") or meta.get("download_urls")
+        if not urls:
+            raise SystemExit("[fetch] 拿不到播放地址(可能视频被删/私密/风控)")
+        if os.path.exists(video_path) and ffprobe_duration(video_path) > 1:
+            print(f"[fetch] 复用已缓存 {video_path}")
+        else:
+            size = download_file(urls[0], video_path)
+            print(f"[fetch] 下载完成: {size/1e6:.1f} MB")
+        meta["aweme_id"] = meta.get("aweme_id") or aweme_id
+
+    dur = ffprobe_duration(video_path)
+    if dur <= 1:
+        raise SystemExit(f"[fetch] 视频时长异常({dur}s), 文件可能损坏")
+    print(f"[fetch] 时长验证通过: {dur:.1f}s")
+
+    meta.update({"video_path": os.path.abspath(video_path),
+                 "duration_s": round(dur, 2), "platform": platform})
+    meta_path = os.path.join(args.out_dir, "meta.json")
+    json.dump(meta, open(meta_path, "w", encoding="utf-8"), ensure_ascii=False, indent=2)
+    print(f"[fetch] meta.json 已写入: 作者={meta.get('author')} 统计={meta.get('stats')}")
+
+
+if __name__ == "__main__":
+    main()

scripts/frame_analysis.py

@@ -0,0 +1,146 @@
+#!/usr/bin/env python3
+"""frame_analysis.py — 抽帧 + 豆包视觉逐帧点评.
+
+用法:
+  python3 frame_analysis.py --workdir /tmp/lapian --fps 0.1 --max-frames 12
+  python3 frame_analysis.py --workdir /tmp/lapian --skip-vision   # 只抽帧, 不调视觉
+前置: workdir/video.mp4 (可选 workdir/transcript.json 用于对齐台词)
+产物: workdir/frames/f0001.jpg... + workdir/frames.json
+  frames.json = [{frame, file, t_ms, t_label, vision, caption}...]
+环境: FEME_NEWAPI_TOKEN — 调 api.fmode.cn 视觉模型(计费)
+"""
+import argparse
+import base64
+import json
+import os
+import subprocess
+import sys
+import urllib.request
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+from common import ffprobe_duration, fmt_ts, get_api_token  # noqa: E402
+
+VISION_MODEL = os.environ.get("FEME_VISION_MODEL", "doubao-seed-2-0-pro-260215")
+VISION_BASE = os.environ.get("FEME_API_BASE", "https://api.fmode.cn/v1")
+
+PROMPT = """你是短视频编导顾问, 对视频抽帧做画面拆解。用中文 JSON 输出(不要markdown代码块):
+{"shot":"景别(远景/全景/中景/近景/特写)","scene":"场景与置景一句话",
+ "subjects":"人物/主体在做什么(表情动作)","text_on_screen":"画面文字, 没有则空串",
+ "aesthetic":"画面技法一句话(光影/构图/节奏)"},
+只描述这一帧可见的信息。"""
+
+
+def extract_frames(video: str, outdir: str, fps: float, max_frames: int) -> list:
+    """均匀抽帧(先抽满 fps, 超出上限再等距丢弃), 返回 [{frame,file,t_ms}]。"""
+    os.makedirs(outdir, exist_ok=True)
+    subprocess.run(["ffmpeg", "-y", "-i", video, "-vf", f"fps={fps}",
+                    "-q:v", "3", os.path.join(outdir, "f%04d.jpg")],
+                   capture_output=True, text=True)
+    files = sorted(f for f in os.listdir(outdir) if f.endswith(".jpg"))
+    if len(files) > max_frames:
+        keep = [files[round(i * (len(files) - 1) / (max_frames - 1))]
+                for i in range(max_frames)]
+        for f in files:
+            if f not in keep:
+                os.remove(os.path.join(outdir, f))
+        files = keep
+    dur = ffprobe_duration(video) or 0.0
+    frames = []
+    for i, f in enumerate(files):
+        t_ms = int(dur * 1000 * i / max(len(files) - 1, 1))
+        frames.append({"frame": i + 1, "file": f, "t_ms": t_ms,
+                       "t_label": fmt_ts(t_ms)})
+    return frames
+
+
+def nearest_line(t_ms: int, segments: list) -> str:
+    for seg in segments:
+        if seg.get("bg", 0) <= t_ms < seg.get("ed", 0):
+            return seg.get("text", "")
+    return ""
+
+
+def call_vision(image_path: str) -> str:
+    """单帧 → 豆包视觉, 返回模型文本(JSON字符串或自然语言)。"""
+    b64 = base64.b64encode(open(image_path, "rb").read()).decode()
+    payload = {
+        "model": VISION_MODEL,
+        "messages": [{"role": "user", "content": [
+            {"type": "text", "text": PROMPT},
+            {"type": "image_url",
+             "image_url": {"url": f"data:image/jpeg;base64,{b64}"}}]}],
+        "temperature": 0.12, "max_tokens": 500,
+    }
+    req = urllib.request.Request(
+        f"{VISION_BASE}/chat/completions",
+        data=json.dumps(payload).encode(),
+        headers={"Content-Type": "application/json",
+                 "Authorization": f"Bearer {get_api_token()}"})
+    with urllib.request.urlopen(req, timeout=120) as r:
+        resp = json.loads(r.read())
+    return (resp["choices"][0]["message"]["content"] or "").strip()
+
+
+def parse_vision_json(text: str) -> dict:
+    """宽容解析模型输出(剥 markdown 围栏, 抠第一个 {...})。"""
+    text = text.strip()
+    if text.startswith("```"):
+        text = text.strip("`").lstrip("json").strip()
+    try:
+        return json.loads(text)
+    except Exception:
+        pass
+    if "{" in text and "}" in text:
+        try:
+            return json.loads(text[text.index("{"): text.rindex("}") + 1])
+        except Exception:
+            pass
+    return {}
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--workdir", required=True)
+    ap.add_argument("--fps", type=float, default=0.15, help="抽帧密度(默认0.15≈6.7s一帧)")
+    ap.add_argument("--max-frames", type=int, default=12, help="最多分析帧数(控制计费)")
+    ap.add_argument("--skip-vision", action="store_true", help="只抽帧, 不调视觉模型")
+    args = ap.parse_args()
+
+    video = os.path.join(args.workdir, "video.mp4")
+    if not os.path.exists(video):
+        raise SystemExit(f"[frames] 缺少 {video}")
+    outdir = os.path.join(args.workdir, "frames")
+    frames = extract_frames(video, outdir, args.fps, args.max_frames)
+    print(f"[frames] 抽帧 {len(frames)} 张 → {outdir}")
+
+    segs = []
+    tp = os.path.join(args.workdir, "transcript.json")
+    if os.path.exists(tp):
+        segs = (json.load(open(tp, encoding="utf-8")).get("data") or {}).get("segments") or []
+
+    for fr in frames:
+        fr["line"] = nearest_line(fr["t_ms"], segs)
+        if args.skip_vision:
+            fr["vision"] = {}
+            fr["caption"] = fr["line"][:60] or "(未调视觉)"
+            continue
+        img = os.path.join(outdir, fr["file"])
+        try:
+            raw = call_vision(img)
+            v = parse_vision_json(raw)
+            fr["vision"] = v
+            parts = [v.get(k, "") for k in ("scene", "subjects", "aesthetic")]
+            fr["caption"] = "；".join(p for p in parts if p) or raw[:120]
+            print(f"[frames] f{fr['frame']:04d} {fr['t_label']} ✓ {fr['caption'][:50]}")
+        except Exception as e:
+            fr["vision"], fr["caption"] = {}, f"(视觉识别失败: {e})"
+            print(f"[frames] f{fr['frame']:04d} {fr['t_label']} ✗ {e}")
+
+    out = os.path.join(args.workdir, "frames.json")
+    json.dump(frames, open(out, "w", encoding="utf-8"), ensure_ascii=False, indent=2)
+    ok = sum(1 for f in frames if f.get("vision"))
+    print(f"[frames] frames.json: {len(frames)} 帧(视觉成功 {ok})")
+
+
+if __name__ == "__main__":
+    main()

scripts/publish.py

@@ -0,0 +1,101 @@
+#!/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()

scripts/transcribe.py

@@ -0,0 +1,72 @@
+#!/usr/bin/env python3
+"""transcribe.py — 提取音频并用 fmode-listen 转写(讯飞 LFASR, 分段时间戳).
+
+用法:
+  python3 transcribe.py --workdir /tmp/lapian
+前置: workdir/video.mp4 (fetch_video.py 产物)
+产物: workdir/audio.wav + workdir/transcript.json
+  transcript.json = {code, data:{text, segments:[{bg,ed,speaker,text}]}}, 毫秒
+环境: FEME_NEWAPI_TOKEN(或 FMODE_API_KEY) — 传给 fmode-listen 的 FMODE_API_TOKEN
+"""
+import argparse
+import json
+import os
+import subprocess
+import sys
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+from common import get_api_token  # noqa: E402
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--workdir", required=True)
+    ap.add_argument("--language", default="autodialect",
+                    help="autodialect(中英+方言自动)/zh/en, 默认 autodialect")
+    args = ap.parse_args()
+
+    video = os.path.join(args.workdir, "video.mp4")
+    wav = os.path.join(args.workdir, "audio.wav")
+    out_json = os.path.join(args.workdir, "transcript.json")
+    if not os.path.exists(video):
+        raise SystemExit(f"[transcribe] 缺少 {video}, 先跑 fetch_video.py")
+
+    if not (os.path.exists(wav) and os.path.getsize(wav) > 1000):
+        print("[transcribe] 提取音频: 16kHz 单声道 wav")
+        r = subprocess.run(["ffmpeg", "-y", "-i", video, "-vn", "-ar", "16000",
+                            "-ac", "1", "-f", "wav", wav],
+                           capture_output=True, text=True)
+        if r.returncode != 0:
+            raise SystemExit(f"[transcribe] ffmpeg 失败: {r.stderr[-500:]}")
+    else:
+        print("[transcribe] 复用已提取的 audio.wav")
+
+    # 短路: 已有转写直接复用(转写计费, 别重复花钱)
+    if os.path.exists(out_json):
+        try:
+            old = json.load(open(out_json, encoding="utf-8"))
+            segs = (old.get("data") or {}).get("segments") or []
+            if segs:
+                print(f"[transcribe] 复用已有转写({len(segs)}段), 跳过计费")
+                return
+        except Exception:
+            pass
+
+    env = dict(os.environ, FMODE_API_TOKEN=get_api_token())
+    cmd = ["npx", "--yes", "fmode-listen@latest", "transcribe", "--", wav,
+           "--language", args.language, "--out", out_json]
+    print("[transcribe]", " ".join(cmd))
+    r = subprocess.run(cmd, env=env, capture_output=True, text=True, timeout=1800)
+    print(r.stdout[-800:])
+    if r.returncode != 0:
+        raise SystemExit(f"[transcribe] fmode-listen 失败: {r.stderr[-800:]}")
+    if not os.path.exists(out_json):
+        raise SystemExit("[transcribe] 转写产物缺失")
+
+    data = json.load(open(out_json, encoding="utf-8"))
+    segs = (data.get("data") or {}).get("segments") or []
+    print(f"[transcribe] 完成: {len(segs)} 段, 全文 {len((data.get('data') or {}).get('text',''))} 字")
+
+
+if __name__ == "__main__":
+    main()

templates/report.html.j2

@@ -0,0 +1,241 @@
+<!DOCTYPE html>
+<html lang="zh-CN" data-theme="dark">
+<head>
+<meta charset="UTF-8">
+<meta name="viewport" content="width=device-width, initial-scale=1.0">
+<title>{{ meta.title }} · 拉片报告</title>
+<style>
+  :root {
+    --bg:#0a0e17; --surface:#111827; --card:#1a2332; --border:#2a3a52;
+    --text:#e2e8f0; --muted:#94a3b8; --brand:#3b82f6; --brand-light:#60a5fa;
+    --green:#10b981; --yellow:#f59e0b; --red:#ef4444; --purple:#8b5cf6;
+  }
+  * { margin:0; padding:0; box-sizing:border-box; }
+  body { font-family:-apple-system,'PingFang SC','Microsoft YaHei',sans-serif;
+         background:var(--bg); color:var(--text); line-height:1.7; padding-bottom:80px; }
+
+  /* === 顶部视频播放器(点击分段跳转对应时间) === */
+  .video-bar { position:sticky; top:0; z-index:100; background:#000;
+               border-bottom:1px solid var(--border); }
+  .video-bar video { width:100%; max-height:56vw; display:block;
+                     object-fit:contain; background:#000; }
+  @media (min-width:600px){ .video-bar video { max-height:340px; } }
+  .time-indicator { position:absolute; bottom:8px; right:12px;
+                    background:rgba(0,0,0,.7); color:#fff; padding:2px 8px;
+                    border-radius:4px; font-size:.8rem; }
+  .main { max-width:960px; margin:0 auto; padding:0 16px; }
+
+  .head { text-align:center; margin:24px 0 8px; }
+  .head h1 { font-size:1.5rem; color:var(--brand-light); }
+  .head p { color:var(--muted); font-size:.9rem; margin-top:4px; }
+
+  .section { margin:32px 0; }
+  .section > h2 { font-size:1.25rem; font-weight:700; color:var(--brand-light);
+                  padding-bottom:10px; border-bottom:2px solid var(--brand);
+                  margin-bottom:16px; }
+
+  .grid3 { display:grid; grid-template-columns:repeat(3,1fr); gap:10px; margin:12px 0; }
+  .grid6 { display:grid; grid-template-columns:repeat(3,1fr); gap:10px; margin:12px 0; }
+  @media (min-width:600px){ .grid6 { grid-template-columns:repeat(6,1fr); } }
+  .stat { background:var(--card); border:1px solid var(--border); border-radius:10px;
+          padding:12px 8px; text-align:center; }
+  .stat .num { display:block; font-size:1.25rem; font-weight:700; color:var(--brand-light); }
+  .stat .label { font-size:.75rem; color:var(--muted); }
+
+  table { width:100%; border-collapse:collapse; margin:12px 0; font-size:.9rem; }
+  th, td { padding:10px 12px; text-align:left; border-bottom:1px solid var(--border); }
+  th { background:var(--surface); color:var(--brand-light); font-weight:600; }
+
+  /* === 可折叠分析分段 === */
+  .seg { background:var(--card); border:1px solid var(--border); border-radius:10px;
+         margin:12px 0; overflow:hidden; }
+  .seg-h { display:flex; align-items:center; gap:8px; padding:12px 14px; cursor:pointer;
+           user-select:none; }
+  .seg-h:active { background:var(--surface); }
+  .seg-h .t { color:var(--brand-light); font-weight:600; white-space:nowrap; font-size:.85rem; }
+  .seg-h .p { flex:1; color:var(--muted); font-size:.85rem; overflow:hidden;
+              text-overflow:ellipsis; white-space:nowrap; }
+  .seg-h .a { color:var(--muted); transition:transform .2s; }
+  .seg.open .seg-h .a { transform:rotate(180deg); }
+  .seg-b { display:none; padding:4px 14px 14px; }
+  .seg.open .seg-b { display:block; }
+  .quote { background:var(--surface); padding:12px; border-radius:8px; margin:8px 0;
+           border-left:3px solid var(--brand); font-style:italic; font-size:.98rem; }
+  .frame { width:100%; border-radius:8px; margin:8px 0; border:1px solid var(--border); }
+  .cap { color:var(--muted); font-size:.8rem; margin:-4px 0 8px; }
+  .play-btn { background:var(--brand); color:#fff; border:none; padding:6px 14px;
+              border-radius:6px; cursor:pointer; font-size:.85rem; margin:4px 0 10px; }
+  .analysis p { margin:8px 0; font-size:.93rem; color:var(--text); }
+  .analysis .muted { color:var(--muted); }
+  .tag { display:inline-block; padding:2px 8px; border-radius:4px; font-size:.75rem; margin:2px; }
+  .tag-blue { background:rgba(59,130,246,.15); color:var(--brand-light); }
+  .tag-green { background:rgba(16,185,129,.15); color:var(--green); }
+  .tag-yellow { background:rgba(245,158,11,.15); color:var(--yellow); }
+  .tag-purple { background:rgba(139,92,246,.15); color:var(--purple); }
+  .tag-red { background:rgba(239,68,68,.15); color:var(--red); }
+  code.reuse { display:block; color:var(--brand-light); background:var(--surface);
+               padding:8px 10px; border-radius:6px; font-size:.85rem; margin-top:6px;
+               overflow-x:auto; }
+
+  .callout { border-left:4px solid var(--brand); background:var(--surface);
+             padding:12px 14px; border-radius:0 8px 8px 0; margin:12px 0; font-size:.92rem; }
+  .callout.green { border-color:var(--green); }
+
+  /* === 纯CSS条形图 === */
+  .bar-row { display:flex; align-items:center; gap:8px; margin:6px 0; font-size:.85rem; }
+  .bar-row .w { width:6.5em; color:var(--muted); text-align:right; flex-shrink:0;
+                overflow:hidden; text-overflow:ellipsis; white-space:nowrap; }
+  .bar-row .track { flex:1; background:var(--surface); border-radius:4px; height:18px; }
+  .bar-row .fill { height:100%; border-radius:4px; background:var(--brand);
+                   min-width:2px; }
+  .bar-row .v { width:3.2em; color:var(--muted); font-size:.78rem; flex-shrink:0; }
+
+  .comment { background:var(--surface); border-radius:8px; padding:10px 12px; margin:8px 0; }
+  .comment .u { font-weight:600; font-size:.85rem; color:var(--brand-light); }
+  .comment .u .lk { color:var(--muted); font-weight:400; margin-left:8px; }
+  .comment .tx { font-size:.9rem; margin:4px 0; }
+  .comment .in { font-size:.8rem; color:var(--green); }
+
+  .foot { text-align:center; color:var(--muted); font-size:.75rem; margin-top:40px; }
+  .foot a { color:var(--brand-light); }
+</style>
+</head>
+<body>
+
+<div class="video-bar" id="playerBar">
+  <video id="mainVideo" controls playsinline preload="metadata">
+    <source src="video.mp4" type="video/mp4">
+  </video>
+  <div class="time-indicator" id="timeDisplay">0:00</div>
+</div>
+
+<div class="main">
+
+<div class="head">
+  <h1>🎬 {{ meta.title }}</h1>
+  <p>{{ meta.subtitle }}</p>
+</div>
+
+<div class="section">
+  <div class="grid6">
+    {% for s in stats_cards %}<div class="stat"><span class="num">{{ s.num }}</span><span class="label">{{ s.label }}</span></div>{% endfor %}
+  </div>
+  <table>
+    {% if meta.author %}<tr><td style="width:6em">作者</td><td><strong>{{ meta.author }}</strong>{% if meta.author_signature %} — {{ meta.author_signature }}{% endif %}</td></tr>{% endif %}
+    {% if meta.tags %}<tr><td>标签</td><td>{% for t in meta.tags %}<span class="tag tag-blue">#{{ t }}</span> {% endfor %}</td></tr>{% endif %}
+    <tr><td>发布时间</td><td>{{ meta.publish_time or "—" }}</td></tr>
+    <tr><td>报告信息</td><td>{{ meta.aweme_id or "local" }} · {{ meta.generated_at }} · 平台 {{ meta.platform }} · 分析引擎 {{ meta.llm_engine }}</td></tr>
+  </table>
+</div>
+
+<div class="section">
+  <h2>📝 逐段拉片({{ segments|length }}段 · 点击展开 · ⏵跳转播放)</h2>
+  {% for seg in segments %}
+  <div class="seg{% if loop.first %} open{% endif %}" id="seg{{ loop.index }}">
+    <div class="seg-h" onclick="toggleSeg(this)">
+      <span class="t">⏱ {{ seg.t_start }} — {{ seg.t_end }}</span>
+      <span class="p">{{ seg.preview }}</span>
+      <span class="a">▼</span>
+    </div>
+    <div class="seg-b">
+      <div class="quote">"{{ seg.text }}"</div>
+      <button class="play-btn" onclick="seekTo({{ seg.t_start_s }})">⏵ 播放这一段</button>
+      {% if seg.frame %}
+      <img class="frame" loading="lazy" src="frames/{{ seg.frame.file }}" alt="f{{ seg.frame.frame }}">
+      <p class="cap">🖼 抽帧 f{{ seg.frame.frame }} @ {{ seg.frame.t_label }}{% if seg.frame.caption %} — {{ seg.frame.caption }}{% endif %}</p>
+      {% endif %}
+      <div class="analysis">
+        {% if seg.techniques %}<p><strong>🎯 技法：</strong>{% for tc in seg.techniques %}<span class="tag {{ tc.cls }}">{{ tc.label }}</span> {% endfor %}</p>{% endif %}
+        <p>{{ seg.comment_html }}</p>
+        {% if seg.reuse %}<p class="muted"><strong>💡 可复用模板：</strong></p><code class="reuse">{{ seg.reuse }}</code>{% endif %}
+      </div>
+    </div>
+  </div>
+  {% endfor %}
+</div>
+
+<div class="section">
+  <h2>📐 整体结构(起承转合)</h2>
+  <table>
+    <tr><th>段</th><th>时间</th><th>功能</th><th>技法</th></tr>
+    {% for r in structure_rows %}<tr><td>{{ r.stage }}</td><td>{{ r.time }}</td><td>{{ r.func }}</td><td>{{ r.techniques }}</td></tr>{% endfor %}
+  </table>
+  {% if summary_html %}<div class="callout"><strong>🎯 核心发现：</strong>{{ summary_html }}</div>{% endif %}
+</div>
+
+{% if wordfreq %}
+<div class="section">
+  <h2>📊 词频TOP / 情绪分布</h2>
+  <div class="grid3">
+    <div class="stat"><span class="num">{{ meta.word_count }}</span><span class="label">总字数</span></div>
+    <div class="stat"><span class="num">{{ meta.wpm }}</span><span class="label">语速(字/分)</span></div>
+    <div class="stat"><span class="num">{{ meta.sentence_count }}</span><span class="label">句数</span></div>
+  </div>
+  <h3 style="margin:14px 0 8px;color:var(--text)">高频实词 TOP{{ wordfreq|length }}</h3>
+  {% for w in wordfreq %}
+  <div class="bar-row"><span class="w">{{ w.word }}</span><div class="track"><div class="fill" style="width:{{ w.pct }}%;background:var(--brand-light)"></div></div><span class="v">{{ w.count }}</span></div>
+  {% endfor %}
+  <h3 style="margin:14px 0 8px;color:var(--text)">情绪分布</h3>
+  {% for s in sentiment %}
+  <div class="bar-row"><span class="w">{{ s.label }}</span><div class="track"><div class="fill" style="width:{{ s.pct }}%;background:{{ s.color }}"></div></div><span class="v">{{ s.pct }}%</span></div>
+  {% endfor %}
+</div>
+{% endif %}
+
+{% if comments %}
+<div class="section">
+  <h2>💬 评论区VOC分析({{ comment_total }}条)</h2>
+  <div class="grid3">
+    {% for s in comment_stats %}<div class="stat"><span class="num" style="color:{{ s.color }}">{{ s.pct }}%</span><span class="label">{{ s.label }}</span></div>{% endfor %}
+  </div>
+  <h3 style="margin:14px 0 8px;color:var(--text)">高赞评论 TOP{{ top_comments|length }}</h3>
+  {% for c in top_comments %}
+  <div class="comment">
+    <div class="u">{{ c.nickname }}<span class="lk">👍{{ c.likes }}</span></div>
+    <div class="tx">{{ c.text }}</div>
+    {% if c.insight %}<div class="in">💡 {{ c.insight }}</div>{% endif %}
+  </div>
+  {% endfor %}
+  {% if comment_keywords %}
+  <h3 style="margin:14px 0 8px;color:var(--text)">评论关键词</h3>
+  {% for w in comment_keywords %}
+  <div class="bar-row"><span class="w">{{ w.word }}</span><div class="track"><div class="fill" style="width:{{ w.pct }}%;background:var(--green)"></div></div><span class="v">{{ w.count }}</span></div>
+  {% endfor %}
+  {% endif %}
+</div>
+{% endif %}
+
+{% if suggestions %}
+<div class="section">
+  <h2>🛠 可执行建议</h2>
+  {% for sg in suggestions %}<div class="callout green"><strong>{{ loop.index }}. {{ sg.title }}</strong> — {{ sg.body }}</div>{% endfor %}
+</div>
+{% endif %}
+
+<p class="foot">由 skill-video-lapian 生成 · 资源全部相对路径, 目录自包含 · 报告ID {{ meta.report_id }}</p>
+</div>
+
+<script>
+function toggleSeg(h){
+  var seg = h.parentElement;
+  seg.classList.toggle('open');
+  if (seg.classList.contains('open') && window.innerWidth < 600) {
+    setTimeout(function(){ seg.scrollIntoView({behavior:'smooth', block:'start'}); }, 60);
+  }
+}
+function seekTo(sec){
+  var v = document.getElementById('mainVideo');
+  if (!v) return;
+  try { v.currentTime = sec; } catch(e) {}
+  var p = v.play(); if (p && p.catch) p.catch(function(){});
+}
+(function(){
+  var v = document.getElementById('mainVideo'), d = document.getElementById('timeDisplay');
+  if (v && d) v.addEventListener('timeupdate', function(){
+    var m = Math.floor(v.currentTime/60), s = Math.floor(v.currentTime%60);
+    d.textContent = m + ':' + (s<10?'0':'') + s;
+  });
+})();
+</script>
+</body>
+</html>