From 54c8c369f839f543d7f499e91d64440bcb990fec Mon Sep 17 00:00:00 2001 From: taozi5211 Date: Tue, 10 Mar 2026 01:36:12 +0800 Subject: [PATCH] feat(skill): add video-to-keyframes workflow --- README.md | 1 + README.zh-CN.md | 1 + skills/video-to-keyframes/SKILL.md | 57 ++ .../resources/requirements.txt | 2 + .../scripts/extract_frames_and_describe.py | 259 +++++++ .../scripts/generate_daily_folder.py | 20 + .../resources/scripts/run_video_workflow.py | 134 ++++ .../resources/scripts/select_keyframes.py | 699 ++++++++++++++++++ 8 files changed, 1173 insertions(+) create mode 100644 skills/video-to-keyframes/SKILL.md create mode 100644 skills/video-to-keyframes/resources/requirements.txt create mode 100644 skills/video-to-keyframes/resources/scripts/extract_frames_and_describe.py create mode 100644 skills/video-to-keyframes/resources/scripts/generate_daily_folder.py create mode 100644 skills/video-to-keyframes/resources/scripts/run_video_workflow.py create mode 100644 skills/video-to-keyframes/resources/scripts/select_keyframes.py diff --git a/README.md b/README.md index b7b50de..b80c43d 100644 --- a/README.md +++ b/README.md @@ -99,6 +99,7 @@ This section will list available skills as they are added. | [cn-punctuation-checker](skills/cn-punctuation-checker/SKILL.md) | Checks Chinese text for incorrect English punctuation marks and supports batch fixing. | Chinese Copy Editing, Punctuation Correction | Stable | | [wechat-mini-program-development](skills/wechat-mini-program-development/SKILL.md) | WeChat mini-program development skill with standard project structure, request wrapper, and API management. | WeChat Mini-Program Development, Project Scaffolding | Stable | | [kz-article-deep-analysis](skills/kz-article-deep-analysis/SKILL.md) | Deeply interpret non-academic articles (blogs, essays, commentary) and output a structured analysis report (core issue, thesis, argument map, cognitive gains). | Reading, Article Analysis | Stable | +| [video-to-keyframes](skills/video-to-keyframes/SKILL.md) | Extracts video frames, detects cuts/segments, selects candidate keyframes, and generates review HTML galleries. | Video Analysis, Keyframe Selection, Storyboard Screening | Stable | > Tip: To add your skill to this catalog, update this table in your PR. diff --git a/README.zh-CN.md b/README.zh-CN.md index 7761c25..bd363f0 100644 --- a/README.zh-CN.md +++ b/README.zh-CN.md @@ -100,6 +100,7 @@ description: 简要描述这个技能的功能和使用场景 | [cn-punctuation-checker](skills/cn-punctuation-checker/SKILL.md) | 检查中文文案中错误使用的英文标点符号，并支持批量修复。 | 中文文案润色, 标点纠错 | Stable | | [wechat-mini-program-development](skills/wechat-mini-program-development/SKILL.md) | 微信小程序开发专用技能，提供标准项目结构、请求封装和 API 管理。 | 微信小程序开发, 项目脚手架 | Stable | | [kz-article-deep-analysis](skills/kz-article-deep-analysis/SKILL.md) | 深度解读非学术类文章（博客、随笔、评论），输出结构化分析报告（核心议题、核心主张、论证拓扑、认知增量）。 | 深度阅读, 文章分析 | Stable | +| [video-to-keyframes](skills/video-to-keyframes/SKILL.md) | 抽取视频帧、检测转场与分段、筛选候选关键帧，并生成可复筛的 HTML 画廊。 | 视频分析, 关键帧筛选, 分镜初筛 | Stable | > 提示：要把你的技能加入此目录，请在 PR 中更新此表格。 diff --git a/skills/video-to-keyframes/SKILL.md b/skills/video-to-keyframes/SKILL.md new file mode 100644 index 0000000..3638d6e --- /dev/null +++ b/skills/video-to-keyframes/SKILL.md @@ -0,0 +1,57 @@ +--- +name: "video-to-keyframes" +description: "Extracts video frames, detects cuts/segments, selects candidate keyframes, and generates review HTML galleries. Invoke when users ask for keyframes/cuts/segmentation/storyboard screening." +--- + +# 视频转关键帧（video-to-keyframes） + +把用户提供的视频转成“候选帧池 → 转场/分段 → 候选关键帧集 → 复筛画廊页”，并把产物落盘到当天文件夹，方便后续分镜与生成。 + +## 何时调用 + +- 用户提供视频并说：抽帧/拆帧/关键帧/候选关键帧/镜头拆分/转场点/分段/分镜初筛 +- 用户希望按固定工作流落盘，需要可复现的目录与文件（frames.json、cuts.json、segments.json、gallery.html 等） + +## 依赖 + +- Python 3.10+ +- numpy +- opencv-python + +## 输入 + +- 视频文件路径（必填） +- 当天文件夹路径（可选，默认用视频所在目录；推荐 YYYY-MM-DD） +- 抽帧间隔（建议：30s≈1fps；变化快≈2fps） + +## 输出（固定规范） + +在 `<当天文件夹>` 下生成： + +- `<当天文件夹>\_frames_<视频名>_<间隔>\`：候选帧池目录 + - `f_*.jpg`：抽帧图片 + - `frames.csv / frames.json / top_keep.json / meta.json` + - `\_keyframe_candidates\`：候选关键帧集目录 + - `cuts.json`：转场点 + - `segments.json`：分段与每段代表帧 + - `segments_gallery.html`：分段可视化（每段1张代表帧） + - `gallery.html`：候选关键帧画廊（逐个复筛） + - `candidates.csv / candidates.json` + - `selected.txt`：人工/AI复筛后的最终候选ID（每行一个 cand_id） + - `prompt_pack.html`：复筛+提示词协作页（夜间模式，一键复制） +- `<当天文件夹>\<视频名>_拆分.txt`：汇总（转场点、分段、每段代表帧文件名） + +## 一键运行（推荐） + +```powershell +python .\skills\video-to-keyframes\resources\scripts\run_video_workflow.py "<视频路径>" --day-folder "<当天文件夹>" --every-seconds 0.5 --max-frames 600 +``` + +注意：一键运行只负责产出文件，不等于完成复筛；必须打开 `gallery.html` 做人工/AI语义复筛，并把最终选择写入 `selected.txt`。 + +## 复筛要点（简版） + +- 先看 `segments_gallery.html`：确认每段代表帧是否合理、分段是否过碎 +- 再看 `gallery.html`：挑 6-12 张最“代表内容且可复现”的帧（不要只挑清晰但信息弱的帧） +- 将 cand_id 写入 `selected.txt`（每行一个三位数字或逗号分隔均可） + diff --git a/skills/video-to-keyframes/resources/requirements.txt b/skills/video-to-keyframes/resources/requirements.txt new file mode 100644 index 0000000..cb3738c --- /dev/null +++ b/skills/video-to-keyframes/resources/requirements.txt @@ -0,0 +1,2 @@ +numpy +opencv-python diff --git a/skills/video-to-keyframes/resources/scripts/extract_frames_and_describe.py b/skills/video-to-keyframes/resources/scripts/extract_frames_and_describe.py new file mode 100644 index 0000000..5c48e20 --- /dev/null +++ b/skills/video-to-keyframes/resources/scripts/extract_frames_and_describe.py @@ -0,0 +1,259 @@ +import argparse +import csv +import json +import math +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import Any + +import cv2 +import numpy as np + + +@dataclass +class FrameInfo: + index: int + timestamp_s: float + file: str + width: int + height: int + sharpness: float + brightness: float + contrast: float + saturation: float + motion: float | None + suggested_keep: bool + description: str + + +def _ensure_dir(p: Path) -> None: + p.mkdir(parents=True, exist_ok=True) + + +def _save_image(out_dir: Path, stem: str, frame, jpeg_params: list[int]) -> Path: + out_file = out_dir / f"{stem}.jpg" + ok, buf = cv2.imencode(".jpg", frame, jpeg_params) + if not ok: + out_file = out_dir / f"{stem}.png" + ok, buf = cv2.imencode(".png", frame) + if not ok: + raise SystemExit("写入图片失败") + data = np.asarray(buf).tobytes() + out_file.write_bytes(data) + return out_file + + +def _fmt_ts(seconds: float) -> str: + if seconds < 0: + seconds = 0 + ms = int(round((seconds - int(seconds)) * 1000)) + s = int(seconds) % 60 + m = (int(seconds) // 60) % 60 + h = int(seconds) // 3600 + return f"{h:02d}-{m:02d}-{s:02d}-{ms:03d}" + + +def _lap_var(gray) -> float: + return float(cv2.Laplacian(gray, cv2.CV_64F).var()) + + +def _mean_std(gray) -> tuple[float, float]: + m, s = cv2.meanStdDev(gray) + return float(m[0][0]), float(s[0][0]) + + +def _mean_saturation(bgr) -> float: + hsv = cv2.cvtColor(bgr, cv2.COLOR_BGR2HSV) + s = hsv[:, :, 1] + return float(s.mean()) + + +def _motion_score(prev_gray, gray) -> float: + diff = cv2.absdiff(prev_gray, gray) + return float(diff.mean()) + + +def _desc(brightness: float, contrast: float, sharpness: float, saturation: float, motion: float | None) -> str: + parts: list[str] = [] + + if brightness < 70: + parts.append("偏暗") + elif brightness > 180: + parts.append("偏亮") + else: + parts.append("曝光正常") + + if contrast < 25: + parts.append("对比偏低") + elif contrast > 70: + parts.append("对比偏高") + else: + parts.append("对比适中") + + if sharpness < 60: + parts.append("偏糊") + elif sharpness > 200: + parts.append("很清晰") + else: + parts.append("清晰度正常") + + if saturation < 40: + parts.append("色彩寡淡") + elif saturation > 140: + parts.append("色彩浓烈") + else: + parts.append("色彩适中") + + if motion is not None: + if motion < 2.0: + parts.append("画面稳定") + elif motion > 10.0: + parts.append("运动幅度大") + else: + parts.append("有一定运动") + + return "，".join(parts) + + +def _bool_keep(sharpness: float, brightness: float, contrast: float, min_sharpness: float, bmin: float, bmax: float) -> bool: + if sharpness < min_sharpness: + return False + if brightness < bmin or brightness > bmax: + return False + if contrast < 10: + return False + return True + + +def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("video", help="视频路径") + parser.add_argument("--out", default=None, help="输出目录，默认在视频同目录下创建 _frames_<文件名>") + parser.add_argument("--every-seconds", type=float, default=0.5, help="按时间间隔抽帧") + parser.add_argument("--max-frames", type=int, default=600, help="最多抽取帧数，防止视频过长") + parser.add_argument("--start", type=float, default=0.0, help="起始秒") + parser.add_argument("--end", type=float, default=None, help="结束秒（不含）") + parser.add_argument("--jpeg-quality", type=int, default=92, help="JPEG质量 0-100") + parser.add_argument("--min-sharpness", type=float, default=80.0, help="最小清晰度阈值（拉普拉斯方差）") + parser.add_argument("--brightness-min", type=float, default=60.0, help="亮度下限") + parser.add_argument("--brightness-max", type=float, default=200.0, help="亮度上限") + args = parser.parse_args() + + video_path = Path(args.video).expanduser().resolve() + if not video_path.exists(): + raise SystemExit(f"视频不存在：{video_path}") + + out_dir = Path(args.out).expanduser().resolve() if args.out else (video_path.parent / f"_frames_{video_path.stem}") + _ensure_dir(out_dir) + + cap = cv2.VideoCapture(str(video_path)) + if not cap.isOpened(): + raise SystemExit("无法打开视频") + + fps = float(cap.get(cv2.CAP_PROP_FPS) or 0.0) + frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT) or 0) + width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH) or 0) + height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT) or 0) + duration_s = (frame_count / fps) if fps > 0 else 0.0 + + start_s = max(float(args.start), 0.0) + end_s = float(args.end) if args.end is not None else duration_s + end_s = min(end_s, duration_s) if duration_s > 0 else end_s + if end_s <= start_s: + raise SystemExit("end 必须大于 start") + + step_s = max(float(args.every_seconds), 0.05) + target_ts = start_s + extracted = 0 + prev_gray = None + rows: list[FrameInfo] = [] + + jpeg_params = [int(cv2.IMWRITE_JPEG_QUALITY), int(max(0, min(100, args.jpeg_quality)))] + + while extracted < int(args.max_frames) and target_ts < end_s: + if fps > 0: + cap.set(cv2.CAP_PROP_POS_MSEC, target_ts * 1000.0) + ret, frame = cap.read() + if not ret or frame is None: + break + + gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY) + sharpness = _lap_var(gray) + brightness, contrast = _mean_std(gray) + saturation = _mean_saturation(frame) + motion = _motion_score(prev_gray, gray) if prev_gray is not None else None + + keep = _bool_keep( + sharpness=sharpness, + brightness=brightness, + contrast=contrast, + min_sharpness=float(args.min_sharpness), + bmin=float(args.brightness_min), + bmax=float(args.brightness_max), + ) + description = _desc(brightness, contrast, sharpness, saturation, motion) + + stem = f"f_{extracted:05d}_t{_fmt_ts(target_ts)}" + out_file = _save_image(out_dir, stem, frame, jpeg_params) + + rows.append( + FrameInfo( + index=extracted, + timestamp_s=float(target_ts), + file=str(out_file), + width=width, + height=height, + sharpness=sharpness, + brightness=brightness, + contrast=contrast, + saturation=saturation, + motion=motion, + suggested_keep=bool(keep), + description=description, + ) + ) + + prev_gray = gray + extracted += 1 + target_ts += step_s + + cap.release() + + meta: dict[str, Any] = { + "video": str(video_path), + "fps": fps, + "frame_count": frame_count, + "width": width, + "height": height, + "duration_s": duration_s, + "start_s": start_s, + "end_s": end_s, + "every_seconds": step_s, + "max_frames": int(args.max_frames), + "extracted": extracted, + "out_dir": str(out_dir), + "columns": [f.name for f in FrameInfo.__dataclass_fields__.values()], + } + + (out_dir / "meta.json").write_text(json.dumps(meta, ensure_ascii=False, indent=2), encoding="utf-8") + (out_dir / "frames.json").write_text( + json.dumps([asdict(r) for r in rows], ensure_ascii=False, indent=2), encoding="utf-8" + ) + + with (out_dir / "frames.csv").open("w", encoding="utf-8-sig", newline="") as f: + w = csv.DictWriter(f, fieldnames=meta["columns"]) + w.writeheader() + for r in rows: + w.writerow(asdict(r)) + + kept = [r for r in rows if r.suggested_keep] + kept_sorted = sorted(kept, key=lambda r: (r.sharpness, -abs(128 - r.brightness), r.contrast), reverse=True) + top = kept_sorted[: min(30, len(kept_sorted))] + (out_dir / "top_keep.json").write_text(json.dumps([asdict(r) for r in top], ensure_ascii=False, indent=2), encoding="utf-8") + + print(str(out_dir)) + + +if __name__ == "__main__": + main() + diff --git a/skills/video-to-keyframes/resources/scripts/generate_daily_folder.py b/skills/video-to-keyframes/resources/scripts/generate_daily_folder.py new file mode 100644 index 0000000..e824ecc --- /dev/null +++ b/skills/video-to-keyframes/resources/scripts/generate_daily_folder.py @@ -0,0 +1,20 @@ +import argparse +from datetime import datetime +from pathlib import Path + + +def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("base_dir", nargs="?", default=".") + parser.add_argument("--format", default="%Y-%m-%d") + args = parser.parse_args() + + base_dir = Path(args.base_dir).expanduser().resolve() + folder_name = datetime.now().strftime(args.format) + target_dir = base_dir / folder_name + target_dir.mkdir(parents=True, exist_ok=True) + print(str(target_dir)) + + +if __name__ == "__main__": + main() diff --git a/skills/video-to-keyframes/resources/scripts/run_video_workflow.py b/skills/video-to-keyframes/resources/scripts/run_video_workflow.py new file mode 100644 index 0000000..00b5cd4 --- /dev/null +++ b/skills/video-to-keyframes/resources/scripts/run_video_workflow.py @@ -0,0 +1,134 @@ +import argparse +import json +import subprocess +import sys +from pathlib import Path + + +def _fmt_ts(seconds: float) -> str: + if seconds < 0: + seconds = 0 + ms = int(round((seconds - int(seconds)) * 1000)) + s = int(seconds) % 60 + m = (int(seconds) // 60) % 60 + h = int(seconds) // 3600 + return f"{h:02d}:{m:02d}:{s:02d}.{ms:03d}" + + +def _tag_seconds(v: float) -> str: + s = f"{v:.3f}".rstrip("0").rstrip(".") + return s.replace(".", "p") + "s" + + +def _run(args: list[str]) -> None: + p = subprocess.run(args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, encoding="utf-8", errors="replace") + if p.returncode != 0: + raise SystemExit(p.stdout.strip() or "运行失败") + + +def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("video", help="视频路径") + parser.add_argument("--day-folder", default=None, help="当天文件夹（YYYY-MM-DD），默认用视频所在目录") + parser.add_argument("--every-seconds", type=float, default=0.5, help="抽帧间隔秒") + parser.add_argument("--max-frames", type=int, default=600, help="最多抽帧数") + parser.add_argument("--jpeg-quality", type=int, default=92, help="JPEG质量") + parser.add_argument("--hamming", type=int, default=10, help="去重阈值（dHash汉明距离）") + parser.add_argument("--max-cands", type=int, default=30, help="候选关键帧数量上限") + parser.add_argument("--cut-thr", type=int, default=30, help="转场阈值：前后大量变化（dHash距离）") + parser.add_argument("--stable-thr", type=int, default=31, help="稳定阈值：前后相似（窗口内最大距离）") + parser.add_argument("--stable-window", type=int, default=1, help="稳定窗口：转场前后各看多少个相邻差值") + parser.add_argument("--min-gap", type=float, default=1.0, help="转场最小间隔（秒）") + parser.add_argument("--min-seg-len", type=float, default=2.0, help="最短分段长度（秒），过短会合并") + args = parser.parse_args() + + video_path = Path(args.video).expanduser().resolve() + if not video_path.exists(): + raise SystemExit("视频不存在") + + day_folder = Path(args.day_folder).expanduser().resolve() if args.day_folder else video_path.parent + day_folder.mkdir(parents=True, exist_ok=True) + + frames_dir = day_folder / f"_frames_{video_path.stem}_{_tag_seconds(float(args.every_seconds))}" + frames_dir.mkdir(parents=True, exist_ok=True) + + root = Path(__file__).resolve().parent + extract_py = root / "extract_frames_and_describe.py" + select_py = root / "select_keyframes.py" + if not extract_py.exists() or not select_py.exists(): + raise SystemExit("缺少脚本文件") + + _run( + [ + sys.executable, + str(extract_py), + str(video_path), + "--every-seconds", + str(float(args.every_seconds)), + "--max-frames", + str(int(args.max_frames)), + "--jpeg-quality", + str(int(args.jpeg_quality)), + "--out", + str(frames_dir), + ] + ) + + _run( + [ + sys.executable, + str(select_py), + str(frames_dir / "frames.json"), + "--hamming", + str(int(args.hamming)), + "--max-cands", + str(int(args.max_cands)), + "--cut-thr", + str(int(args.cut_thr)), + "--stable-thr", + str(int(args.stable_thr)), + "--stable-window", + str(int(args.stable_window)), + "--min-gap", + str(float(args.min_gap)), + "--min-seg-len", + str(float(args.min_seg_len)), + ] + ) + + meta = json.loads((frames_dir / "meta.json").read_text(encoding="utf-8")) + cuts = json.loads((frames_dir / "_keyframe_candidates" / "cuts.json").read_text(encoding="utf-8")) + segs = json.loads((frames_dir / "_keyframe_candidates" / "segments.json").read_text(encoding="utf-8")) + + cut_ts = [float(c["cut_t"]) for c in cuts] + cut_line = " / ".join([_fmt_ts(t) for t in cut_ts]) if cut_ts else "无" + + lines: list[str] = [] + lines.append(f"视频：{video_path.name}") + lines.append( + f"参数：{meta.get('fps')}fps，约{meta.get('frame_count')}帧，时长约{float(meta.get('duration_s') or 0.0):.2f}秒" + ) + lines.append("拆分依据：转场=前后相似且中间大量变化（基于dHash相似度，抽帧后检测）") + lines.append( + f"判定规则：cut_thr={int(args.cut_thr)}，stable_thr={int(args.stable_thr)}，stable_window={int(args.stable_window)}，min_seg_len={float(args.min_seg_len)}s" + ) + lines.append("") + lines.append(f"转场点：{cut_line}") + lines.append("") + lines.append("分段：") + for s in segs: + start_t = float(s["start_t"]) + end_t = float(s["end_t"]) + rep_t = float(s["rep_t"]) + rep_file = Path(str(s["rep_file"])).name + lines.append(f"- 分段{s['seg_id']}：{_fmt_ts(start_t)} - {_fmt_ts(end_t)}（代表帧 t={rep_t:.2f}s，{rep_file}）") + + out_txt = day_folder / f"{video_path.stem}_拆分.txt" + out_txt.write_text("\n".join(lines) + "\n", encoding="utf-8") + + print(str(out_txt)) + + +if __name__ == "__main__": + main() + diff --git a/skills/video-to-keyframes/resources/scripts/select_keyframes.py b/skills/video-to-keyframes/resources/scripts/select_keyframes.py new file mode 100644 index 0000000..7440fb8 --- /dev/null +++ b/skills/video-to-keyframes/resources/scripts/select_keyframes.py @@ -0,0 +1,699 @@ +import argparse +import csv +import json +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +import cv2 +import numpy as np + + +@dataclass +class Cand: + group_id: int + cand_id: int + timestamp_s: float + src_file: str + out_file: str + score: float + sharpness: float + brightness: float + contrast: float + saturation: float + motion: float | None + description: str + + +@dataclass +class Cut: + index_left: int + index_right: int + t_left: float + t_right: float + cut_t: float + dhash_dist: int + + +@dataclass +class Segment: + seg_id: int + start_t: float + end_t: float + rep_t: float + rep_file: str + rep_score: float + frame_count: int + + +def _load_frames(frames_json: Path) -> list[dict[str, Any]]: + data = json.loads(frames_json.read_text(encoding="utf-8")) + if not isinstance(data, list): + raise SystemExit("frames.json 格式不正确") + return data + + +def _read_gray(path: Path) -> np.ndarray: + b = cv2.imdecode(np.frombuffer(path.read_bytes(), np.uint8), cv2.IMREAD_GRAYSCALE) + if b is None: + raise SystemExit(f"无法读取图片：{path}") + return b + + +def _dhash64(gray: np.ndarray) -> int: + small = cv2.resize(gray, (9, 8), interpolation=cv2.INTER_AREA) + diff = small[:, 1:] > small[:, :-1] + bits = diff.flatten() + h = 0 + for i, v in enumerate(bits): + if v: + h |= 1 << i + return int(h) + + +def _hamming(a: int, b: int) -> int: + return int((a ^ b).bit_count()) + + +def _norm(v: float, lo: float, hi: float) -> float: + if hi <= lo: + return 0.5 + x = (v - lo) / (hi - lo) + if x < 0: + return 0.0 + if x > 1: + return 1.0 + return float(x) + + +def _score(row: dict[str, Any], stats: dict[str, float]) -> float: + sharp = float(row.get("sharpness") or 0.0) + bright = float(row.get("brightness") or 0.0) + contr = float(row.get("contrast") or 0.0) + sat = float(row.get("saturation") or 0.0) + motion = row.get("motion") + mot = float(motion) if motion is not None else stats["motion_med"] + + s1 = _norm(np.log1p(sharp), stats["logsharp_min"], stats["logsharp_max"]) + s2 = 1.0 - abs(bright - stats["bright_med"]) / max(stats["bright_mad"] * 3.0, 30.0) + if s2 < 0: + s2 = 0.0 + if s2 > 1: + s2 = 1.0 + s3 = _norm(contr, stats["contr_p10"], stats["contr_p90"]) + s4 = _norm(sat, stats["sat_p10"], stats["sat_p90"]) + s5 = 1.0 - _norm(mot, stats["motion_p10"], stats["motion_p90"]) + + return float(0.45 * s1 + 0.25 * s2 + 0.15 * s3 + 0.05 * s4 + 0.10 * s5) + + +def _percentile(vals: list[float], p: float) -> float: + if not vals: + return 0.0 + a = np.array(vals, dtype=np.float64) + return float(np.percentile(a, p)) + + +def _median(vals: list[float]) -> float: + if not vals: + return 0.0 + return float(np.median(np.array(vals, dtype=np.float64))) + + +def _mad(vals: list[float], med: float) -> float: + if not vals: + return 0.0 + a = np.array(vals, dtype=np.float64) + return float(np.median(np.abs(a - med))) + + +def _build_stats(rows: list[dict[str, Any]]) -> dict[str, float]: + sharp = [float(r.get("sharpness") or 0.0) for r in rows] + logsharp = [float(np.log1p(x)) for x in sharp] + bright = [float(r.get("brightness") or 0.0) for r in rows] + contr = [float(r.get("contrast") or 0.0) for r in rows] + sat = [float(r.get("saturation") or 0.0) for r in rows] + mot = [float(r.get("motion") or 0.0) for r in rows if r.get("motion") is not None] + + bright_med = _median(bright) + motion_med = _median(mot) if mot else 0.0 + return { + "logsharp_min": min(logsharp) if logsharp else 0.0, + "logsharp_max": max(logsharp) if logsharp else 1.0, + "bright_med": bright_med, + "bright_mad": _mad(bright, bright_med), + "contr_p10": _percentile(contr, 10.0), + "contr_p90": _percentile(contr, 90.0), + "sat_p10": _percentile(sat, 10.0), + "sat_p90": _percentile(sat, 90.0), + "motion_p10": _percentile(mot, 10.0) if mot else 0.0, + "motion_p90": _percentile(mot, 90.0) if mot else 1.0, + "motion_med": motion_med, + } + + +def _copy_bytes(src: Path, dst: Path) -> None: + dst.write_bytes(src.read_bytes()) + + +def _detect_cuts( + frames: list[dict[str, Any]], + cut_thr: int, + stable_thr: int, + stable_window: int, + min_gap_s: float, +) -> list[Cut]: + n = len(frames) + if n < (stable_window * 2 + 2): + return [] + + ts = [float(f.get("timestamp_s") or 0.0) for f in frames] + hs = [int(f.get("__dhash")) for f in frames] + + diffs = [0] * n + for i in range(1, n): + diffs[i] = _hamming(hs[i - 1], hs[i]) + + cuts: list[Cut] = [] + for i in range(max(1, stable_window), n - stable_window): + d = diffs[i] + if d < cut_thr: + continue + pre = diffs[i - stable_window : i] + post = diffs[i + 1 : i + 1 + stable_window] + if not pre or not post: + continue + if max(pre) > stable_thr: + continue + if max(post) > stable_thr: + continue + t_left = ts[i - 1] + t_right = ts[i] + cuts.append( + Cut( + index_left=i - 1, + index_right=i, + t_left=t_left, + t_right=t_right, + cut_t=(t_left + t_right) / 2.0, + dhash_dist=int(d), + ) + ) + + cuts.sort(key=lambda c: c.cut_t) + if not cuts: + return [] + + merged: list[Cut] = [cuts[0]] + for c in cuts[1:]: + prev = merged[-1] + if c.cut_t - prev.cut_t < min_gap_s: + if c.dhash_dist > prev.dhash_dist: + merged[-1] = c + else: + merged.append(c) + return merged + + +def _segments_from_cuts(frames: list[dict[str, Any]], cuts: list[Cut]) -> list[Segment]: + ts = [float(f.get("timestamp_s") or 0.0) for f in frames] + n = len(frames) + boundaries = [0] + [c.index_right for c in cuts] + [n] + segs: list[Segment] = [] + for si in range(len(boundaries) - 1): + a = boundaries[si] + b = boundaries[si + 1] + chunk = frames[a:b] + if not chunk: + continue + rep = max(chunk, key=lambda x: float(x.get("__score") or 0.0)) + segs.append( + Segment( + seg_id=si + 1, + start_t=ts[a], + end_t=ts[b - 1], + rep_t=float(rep.get("timestamp_s") or 0.0), + rep_file=str(rep.get("__abs_file") or ""), + rep_score=float(rep.get("__score") or 0.0), + frame_count=len(chunk), + ) + ) + return segs + + +def _merge_short_segments(frames: list[dict[str, Any]], cuts: list[Cut], min_len_s: float) -> list[Cut]: + if min_len_s <= 0: + return cuts + ts = [float(f.get("timestamp_s") or 0.0) for f in frames] + cuts = sorted(list(cuts), key=lambda c: c.cut_t) + while True: + boundaries = [0] + [c.index_right for c in cuts] + [len(frames)] + short_idx = None + for si in range(len(boundaries) - 1): + a = boundaries[si] + b = boundaries[si + 1] + if b - a <= 0: + continue + dur = ts[b - 1] - ts[a] + if dur < min_len_s: + short_idx = si + break + if short_idx is None: + return cuts + left_cut_i = short_idx - 1 + right_cut_i = short_idx + choices = [] + if 0 <= left_cut_i < len(cuts): + choices.append(left_cut_i) + if 0 <= right_cut_i < len(cuts): + choices.append(right_cut_i) + if not choices: + return cuts + drop_i = min(choices, key=lambda i: cuts[i].dhash_dist) + cuts.pop(drop_i) + + +def _write_gallery(out_dir: Path, cands: list[Cand]) -> None: + rels = [(Path(c.out_file).name, c) for c in cands] + rows = [] + for fn, c in rels: + rows.append( + f"" + f"{c.cand_id:03d}" + f"{c.group_id:03d}" + f"{c.timestamp_s:.2f}s" + f"{c.score:.3f}" + f"{c.description}" + f"" + f"{fn}" + f"" + ) + html = ( + "" + "关键帧候选" + "" + "" + "

关键帧候选画廊

" + "

建议做法：先在本页挑选要保留的 cand_id（左侧三位数），把编号写到 selected.txt（每行一个编号），再用后续脚本复制为最终关键帧集。

" + "" + + "\n".join(rows) + + "
cand_idgrouptscoredescimgfile
" + ) + (out_dir / "gallery.html").write_text(html, encoding="utf-8") + + +def _write_segments_gallery(out_dir: Path, cuts: list[Cut], segs: list[Segment]) -> None: + cut_line = " / ".join([f"{c.cut_t:.2f}s" for c in cuts]) if cuts else "无" + rows = [] + for s in segs: + rep_src = Path(s.rep_file) + rep_name = f"seg_{s.seg_id:02d}_t{float(s.rep_t):06.2f}_{rep_src.name}" + rep_dst = out_dir / rep_name + _copy_bytes(rep_src, rep_dst) + rows.append( + f"" + f"{s.seg_id:02d}" + f"{s.start_t:.2f}s - {s.end_t:.2f}s" + f"{s.frame_count}" + f"{s.rep_t:.2f}s" + f"{s.rep_score:.3f}" + f"" + f"{rep_name}" + f"" + ) + html = ( + "" + "分段结果" + "" + "" + "

分段结果（按转场切分）

" + f"

转场点：{cut_line}

" + "" + + "\n".join(rows) + + "
segrangeframesrep_trep_scorerep_imgfile
" + ) + (out_dir / "segments_gallery.html").write_text(html, encoding="utf-8") + + +def _write_prompt_page(out_dir: Path, cuts: list[Cut], segs: list[Segment], cands: list[Cand]) -> None: + cut_line = " / ".join([f"{c.cut_t:.2f}s" for c in cuts]) if cuts else "无" + + seg_rows = [] + for s in segs: + rep_src = Path(s.rep_file) + rep_name = f"seg_{s.seg_id:02d}_t{float(s.rep_t):06.2f}_{rep_src.name}" + seg_rows.append( + f"" + f"S{s.seg_id:02d}" + f"{s.start_t:.2f}s - {s.end_t:.2f}s" + f"" + f"{rep_name}" + f"" + ) + + cand_rows = [] + for c in cands: + fn = Path(c.out_file).name + cand_rows.append( + f"" + f"{c.cand_id:03d}" + f"{c.group_id:03d}" + f"{c.timestamp_s:.2f}s" + f"{c.score:.3f}" + f"{c.description}" + f"" + f"{fn}" + f"" + ) + + q = [ + ("成片", "平台__；时长__s；分镜数（4-6默认）__；画幅__；风格参考__；禁用元素__"), + ("内容", "地点/场景__；主体__；动作__；情绪__；时代感/现代元素容忍度__"), + ("画面", "光线__；时段__；镜头语言（景别/机位/运镜）__；色调__；材质细节__"), + ("素材", "用户已有照片__张（分别是什么）；可补拍/可找素材的范围__"), + ] + q_rows = "".join([f"{a}{b}" for a, b in q]) + + html = f""" + + + + + 复筛与提示词 + + + +
+

复筛与提示词（本地一键复制页）

+

转场点：{cut_line}

+ +
+

第一步：让用户上传可用照片

+
+
+

把用户照片拖进这里，仅用于本页预览与对照（不上传到网络）。

+ +
+
+
+

要问用户的内容（复制给用户）：

+ + + {q_rows} + +
+
+
+
+ +
+

第二步：建议提供的帧（按转场切分的每段代表帧）

+

这些帧来自 segments.json；用于和用户照片对齐“缺什么补什么”。

+ + + + + + {"".join(seg_rows)} + +
段范围代表帧文件
+
+ +
+

第三步：已有关键帧/候选关键帧（复筛池）

+

从下面挑 6-12 张作为最终关键帧（4-6分镜×每分镜1-2张）。把 cand_id 用逗号写进“最终关键帧ID”。

+
+
+ + +
+ + +
+

提示：你也可以直接用已有的 selected.txt（每行一个编号），手动复制到这里。

+
+
+ + +
+
+ + + + + + {"".join(cand_rows)} + +
cand_idgrouptscoredescimgfile
+
+ +
+

第四步：分镜提示词（可直接复制）

+
+
+ + +
+
+ + +
+
+
+
+ + + + +""" + (out_dir / "prompt_pack.html").write_text(html, encoding="utf-8") + + +def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("frames_json", help="extract_frames_and_describe.py 产出的 frames.json 路径") + parser.add_argument("--out", default=None, help="输出目录，默认在 frames.json 同目录创建 _keyframes_candidates") + parser.add_argument("--hamming", type=int, default=10, help="dHash 汉明距离阈值（越小去重越强）") + parser.add_argument("--max-cands", type=int, default=30, help="最多保留多少候选关键帧") + parser.add_argument("--cut-thr", type=int, default=22, help="转场阈值：前后大量变化（dHash 汉明距离）") + parser.add_argument("--stable-thr", type=int, default=10, help="稳定阈值：前后相似（窗口内最大距离）") + parser.add_argument("--stable-window", type=int, default=3, help="稳定窗口：转场前后各看多少个相邻差值") + parser.add_argument("--min-gap", type=float, default=1.0, help="转场最小间隔（秒）") + parser.add_argument("--min-seg-len", type=float, default=1.5, help="最短分段长度（秒），过短会合并") + args = parser.parse_args() + + frames_json = Path(args.frames_json).expanduser().resolve() + rows = _load_frames(frames_json) + if not rows: + raise SystemExit("frames.json 为空") + + frames_dir = frames_json.parent + out_dir = Path(args.out).expanduser().resolve() if args.out else (frames_dir / "_keyframe_candidates") + out_dir.mkdir(parents=True, exist_ok=True) + + stats = _build_stats(rows) + + enriched = [] + for r in rows: + f = Path(str(r.get("file") or "")).expanduser() + if not f.is_absolute(): + f = (frames_dir / f).resolve() + r2 = dict(r) + r2["__abs_file"] = str(f) + r2["__score"] = _score(r2, stats) + enriched.append(r2) + + enriched.sort(key=lambda x: float(x.get("timestamp_s") or 0.0)) + + groups: list[list[dict[str, Any]]] = [] + cur: list[dict[str, Any]] = [] + last_hash: int | None = None + for r in enriched: + p = Path(r["__abs_file"]) + gray = _read_gray(p) + h = _dhash64(gray) + r["__dhash"] = h + if last_hash is None: + cur = [r] + last_hash = h + continue + if _hamming(last_hash, h) <= int(args.hamming): + cur.append(r) + else: + groups.append(cur) + cur = [r] + last_hash = h + if cur: + groups.append(cur) + + cuts = _detect_cuts( + frames=enriched, + cut_thr=int(args.cut_thr), + stable_thr=int(args.stable_thr), + stable_window=int(args.stable_window), + min_gap_s=float(args.min_gap), + ) + cuts = _merge_short_segments(enriched, cuts, float(args.min_seg_len)) + segs = _segments_from_cuts(enriched, cuts) + (out_dir / "cuts.json").write_text(json.dumps([c.__dict__ for c in cuts], ensure_ascii=False, indent=2), encoding="utf-8") + (out_dir / "segments.json").write_text( + json.dumps([s.__dict__ for s in segs], ensure_ascii=False, indent=2), encoding="utf-8" + ) + _write_segments_gallery(out_dir, cuts, segs) + + picks: list[dict[str, Any]] = [] + for g in groups: + best = max(g, key=lambda x: float(x.get("__score") or 0.0)) + picks.append(best) + + picks.sort(key=lambda x: float(x.get("__score") or 0.0), reverse=True) + picks = picks[: max(1, int(args.max_cands))] + picks.sort(key=lambda x: float(x.get("timestamp_s") or 0.0)) + + cands: list[Cand] = [] + for i, r in enumerate(picks, start=1): + src = Path(r["__abs_file"]) + group_id = 0 + for gi, g in enumerate(groups, start=1): + if r in g: + group_id = gi + break + out_name = f"c_{i:03d}_g{group_id:03d}_t{float(r.get('timestamp_s') or 0.0):06.2f}_{src.name}" + dst = out_dir / out_name + _copy_bytes(src, dst) + cands.append( + Cand( + group_id=group_id, + cand_id=i, + timestamp_s=float(r.get("timestamp_s") or 0.0), + src_file=str(src), + out_file=str(dst), + score=float(r.get("__score") or 0.0), + sharpness=float(r.get("sharpness") or 0.0), + brightness=float(r.get("brightness") or 0.0), + contrast=float(r.get("contrast") or 0.0), + saturation=float(r.get("saturation") or 0.0), + motion=float(r.get("motion")) if r.get("motion") is not None else None, + description=str(r.get("description") or ""), + ) + ) + + (out_dir / "candidates.json").write_text( + json.dumps([c.__dict__ for c in cands], ensure_ascii=False, indent=2), encoding="utf-8" + ) + with (out_dir / "candidates.csv").open("w", encoding="utf-8-sig", newline="") as f: + w = csv.DictWriter(f, fieldnames=list(Cand.__dataclass_fields__.keys())) + w.writeheader() + for c in cands: + w.writerow(c.__dict__) + + _write_gallery(out_dir, cands) + _write_prompt_page(out_dir, cuts, segs, cands) + (out_dir / "selected.txt").write_text("", encoding="utf-8") + print(str(out_dir)) + + +if __name__ == "__main__": + main() +