Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 9 additions & 4 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -84,9 +84,12 @@ npm run dev

基础依赖:

- Node.js 20+
- Node.js >=20.19.0 或 >=22.12.0
- uv
- FFmpeg

媒体功能依赖:

- FFmpeg:用于稳定合并 YouTube 音视频、抽音频和导出剪辑。缺失时仍可安装并启动项目,下载会尝试单文件格式,但能力会受限。

可选依赖:

Expand Down Expand Up @@ -114,9 +117,9 @@ brew install node uv ffmpeg

用户需要自己先准备:

- Node.js / npm
- Node.js / npm:Node 需要 >=20.19.0 或 >=22.12.0
- uv
- FFmpeg
- FFmpeg:建议安装;缺失时 `npm run setup` 不再中断,但视频下载合并、抽音频和导出会受限

可选增强不会自动安装:

Expand All @@ -135,6 +138,8 @@ npm run doctor

它会检查基础依赖,并提示哪些可选工具缺失以及影响。

`npm run setup` 会在缺少 FFmpeg 时继续安装项目依赖;这是为了让干净机器先把应用跑起来。真正处理视频前仍建议安装 FFmpeg。

如果对方准备把 GitHub 链接丢给 Codex、Claude Code、Cursor 等本地 coding agent,让大模型帮忙安装,请看 [Setup With A Coding Agent](docs/AI_AGENT_SETUP.md)。里面有可直接复制的提示词,以及哪些依赖能自动装、哪些需要用户自己提供授权或 API key。

## 启动细节
Expand Down
151 changes: 123 additions & 28 deletions backend/app/downloads.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,15 @@
"audio": "bestaudio[ext=m4a]/bestaudio/best",
}

SINGLE_FILE_FORMATS = {
"720p": "b[height<=720][ext=mp4]/b[height<=720]/best[height<=720]/best",
"1080p": "b[height<=1080][ext=mp4]/b[height<=1080]/best[height<=1080]/best",
"best": "b[ext=mp4]/best",
"audio": "bestaudio[ext=m4a]/bestaudio/best",
}

MEDIA_SUFFIXES = {".mp4", ".m4a", ".mp3", ".mov", ".webm", ".mkv"}


def run_authorized_download(video_id: int, authorization_note: str, quality: str = "1080p", include_subtitles: bool = True, include_thumbnail: bool = True) -> dict:
if not authorization_note.strip():
Expand All @@ -43,45 +52,49 @@ def run_authorized_download(video_id: int, authorization_note: str, quality: str
settings.download_dir.mkdir(parents=True, exist_ok=True)
slug = _slugify(video["title"])[:80] or f"video-{video_id}"
output_template = str(settings.download_dir / f"{video_id}-{slug}.%(ext)s")
command = [
*_ytdlp_command(),
"--newline",
"--no-playlist",
"--merge-output-format",
"mp4",
"-f",
QUALITY_FORMATS.get(quality, QUALITY_FORMATS["1080p"]),
"-o",
output_template,
"--print",
"after_move:filepath",
]
_add_js_runtime(command)
if include_thumbnail:
command.append("--write-thumbnail")
command.append(video["url"])
ffmpeg_path = shutil.which("ffmpeg")
command = _download_command(output_template, video["url"], quality, bool(ffmpeg_path))

try:
completed = subprocess.run(command, check=True, capture_output=True, text=True, timeout=60 * 60)
log = "\n".join(part for part in [completed.stdout, completed.stderr] if part)
log = _join_logs(_ffmpeg_fallback_log(quality, ffmpeg_path), _completed_log(completed))
output_path = _find_downloaded_media(completed.stdout, settings.download_dir, video_id)
if not output_path:
raise RuntimeError("下载结束但未找到媒体文件。")
warnings = []
except Exception as exc:
log = _join_logs(_ffmpeg_fallback_log(quality, ffmpeg_path), _exception_log(exc))
output_path = _find_downloaded_media(_exception_stdout(exc), settings.download_dir, video_id)
warnings = ["下载工具返回警告,已保留日志"] if output_path else []

subtitle_paths: dict[str, str] = {}
if output_path:
if include_thumbnail:
try:
thumbnail_log = _try_download_thumbnail(video["url"], output_template)
except Exception as exc:
thumbnail_log = f"封面下载失败但视频已保留:{_exception_log(exc)[-3000:]}"
if thumbnail_log:
log = _join_logs(log, thumbnail_log)
if thumbnail_log.startswith("封面下载失败"):
warnings.append("封面暂未拉取成功")
subtitle_paths: dict[str, str] = {}
if include_subtitles:
subtitle_paths, subtitle_log = _try_download_subtitles(video_id, video["url"], output_template)
try:
subtitle_paths, subtitle_log = _try_download_subtitles(video_id, video["url"], output_template)
except Exception as exc:
subtitle_paths = {}
subtitle_log = f"字幕下载失败但视频已保留:{_exception_log(exc)[-3000:]}"
if subtitle_log:
log = "\n".join(part for part in [log, subtitle_log] if part)
if not subtitle_paths:
warnings.append("字幕暂未拉取成功,将进入转写或等待手动字幕")
status = "completed"
message = "授权下载完成,已作为本地素材入库。"
if include_subtitles and not subtitle_paths:
message = "视频已下载;字幕暂未拉取成功,将进入转写或等待手动字幕。"
except Exception as exc:
log = getattr(exc, "stderr", "") or str(exc)
message = _success_message(warnings)
else:
output_path = ""
subtitle_paths = {}
status = "failed"
message = str(exc)
message = log[-1200:] or "下载结束但未找到媒体文件。"

with get_connection() as conn:
conn.execute(
Expand Down Expand Up @@ -136,6 +149,53 @@ def _ytdlp_command() -> list[str]:
return [sys.executable, "-m", "yt_dlp"]


def _download_command(output_template: str, url: str, quality: str, ffmpeg_available: bool) -> list[str]:
command = [
*_ytdlp_command(),
"--newline",
"--no-playlist",
]
if ffmpeg_available:
command.extend(["--merge-output-format", "mp4"])
command.extend(
[
"-f",
_quality_format(quality, ffmpeg_available),
"-o",
output_template,
"--print",
"after_move:filepath",
]
)
_add_js_runtime(command)
command.append(url)
return command


def _quality_format(quality: str, ffmpeg_available: bool) -> str:
formats = QUALITY_FORMATS if ffmpeg_available else SINGLE_FILE_FORMATS
return formats.get(quality, formats["1080p"])


def _try_download_thumbnail(url: str, output_template: str) -> str:
command = [
*_ytdlp_command(),
"--newline",
"--skip-download",
"--no-playlist",
"-o",
output_template,
"--write-thumbnail",
]
_add_js_runtime(command)
command.append(url)
completed = subprocess.run(command, capture_output=True, text=True, timeout=10 * 60)
log = _completed_log(completed)
if completed.returncode != 0:
return f"封面下载失败但视频已保留:{log[-3000:]}"
return log


def _try_download_subtitles(video_id: int, url: str, output_template: str) -> tuple[dict[str, str], str]:
command = [
*_ytdlp_command(),
Expand All @@ -162,17 +222,52 @@ def _find_downloaded_media(stdout: str, download_dir: Path, video_id: int) -> st
candidates = []
for line in stdout.splitlines():
path = Path(line.strip())
if path.exists() and path.suffix.lower() in {".mp4", ".m4a", ".mp3", ".mov", ".webm", ".mkv"}:
if _is_media_candidate(path):
candidates.append(path)
if candidates:
return str(candidates[-1])
files = sorted(
[path for path in download_dir.glob(f"{video_id}-*") if path.suffix.lower() in {".mp4", ".m4a", ".mp3", ".mov", ".webm", ".mkv"}],
[path for path in download_dir.glob(f"{video_id}-*") if _is_media_candidate(path)],
key=lambda path: path.stat().st_mtime,
)
return str(files[-1]) if files else ""


def _is_media_candidate(path: Path) -> bool:
name = path.name.lower()
return path.exists() and path.suffix.lower() in MEDIA_SUFFIXES and ".temp." not in name and not name.endswith(".part")


def _completed_log(completed: subprocess.CompletedProcess) -> str:
return _join_logs(completed.stdout or "", completed.stderr or "")


def _exception_log(exc: Exception) -> str:
stdout = getattr(exc, "stdout", "") or getattr(exc, "output", "")
stderr = getattr(exc, "stderr", "")
return _join_logs(stdout or "", stderr or "", str(exc))


def _exception_stdout(exc: Exception) -> str:
return str(getattr(exc, "stdout", "") or getattr(exc, "output", "") or "")


def _ffmpeg_fallback_log(quality: str, ffmpeg_path: str | None) -> str:
if ffmpeg_path or quality == "audio":
return ""
return "FFmpeg 未安装,已尝试下载单文件视频格式;如平台只提供分离音视频,仍可能失败。"


def _success_message(warnings: list[str]) -> str:
if warnings:
return f"视频已下载;{';'.join(warnings)}。"
return "授权下载完成,已作为本地素材入库。"


def _join_logs(*parts: str) -> str:
return "\n".join(part for part in parts if part)


def find_downloaded_subtitles(video_id: int) -> dict[str, str]:
subtitle_files = sorted(
[
Expand Down
4 changes: 2 additions & 2 deletions backend/app/system.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@ def system_status() -> dict:
"youtube_api": {
"ok": bool(settings.youtube_api_key),
"label": "YouTube Data API",
"message": "已配置,可用官方免费配额发现公开视频。" if settings.youtube_api_key else "未配置,将优先用本机 yt-dlp 搜索。",
"message": "已配置,可用官方免费配额发现公开视频。" if settings.youtube_api_key else "未配置,将依次尝试 opencli 和本机 yt-dlp 搜索。",
},
"yt_dlp": {
"ok": yt_dlp_ok,
Expand All @@ -35,7 +35,7 @@ def system_status() -> dict:
"ffmpeg": {
"ok": bool(ffmpeg_path),
"label": "FFmpeg",
"message": ffmpeg_path or "未安装,无法抽音频和转码。",
"message": ffmpeg_path or "未安装,下载会尝试单文件格式,但抽音频、合并和导出视频会受限。",
},
"argos": {
"ok": argos_ok,
Expand Down
118 changes: 118 additions & 0 deletions backend/tests/test_downloads.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,118 @@
from __future__ import annotations

import subprocess
from pathlib import Path

from app import downloads
from app.db import get_connection, init_db, now_iso
from app.downloads import run_authorized_download


class Completed:
def __init__(self, stdout: str = "", stderr: str = "", returncode: int = 0) -> None:
self.stdout = stdout
self.stderr = stderr
self.returncode = returncode


def _seed_video(monkeypatch, tmp_path: Path, title: str = "YouTube warning test") -> Path:
monkeypatch.setenv("TECH_PR_DB_PATH", str(tmp_path / "app.db"))
monkeypatch.setenv("DOWNLOAD_DIR", str(tmp_path / "downloads"))
monkeypatch.setenv("DOWNLOAD_ENGINE", "yt-dlp")
init_db()
timestamp = now_iso()
with get_connection() as conn:
conn.execute(
"""
INSERT INTO videos (
id, platform, external_id, url, title, published_at,
duration_seconds, view_count, like_count, interview_confidence,
priority_score, status, compliance_note, created_at, updated_at
)
VALUES (1, 'youtube', 'abc123', 'https://www.youtube.com/watch?v=abc123',
?, ?, 60, 0, 0, 0.5, 0, 'ready',
'metadata_only', ?, ?)
""",
(title, timestamp, timestamp, timestamp),
)
return tmp_path / "downloads"


def _output_path(command: list[str]) -> Path:
template = command[command.index("-o") + 1]
return Path(template.replace("%(ext)s", "mp4"))


def test_youtube_download_accepts_completed_file_after_tool_warning(monkeypatch, tmp_path: Path) -> None:
_seed_video(monkeypatch, tmp_path)

def fake_run(command: list[str], **_: object) -> Completed:
output = _output_path(command)
output.parent.mkdir(parents=True, exist_ok=True)
output.write_text("video", encoding="utf-8")
raise subprocess.CalledProcessError(
1,
command,
output=f"{output}\n",
stderr="WARNING: postprocess warning after media moved",
)

monkeypatch.setattr(downloads.subprocess, "run", fake_run)

result = run_authorized_download(1, "已授权本地测试", "1080p", include_subtitles=False, include_thumbnail=False)

task = result["task"]
assert task["status"] == "completed"
assert task["output_path"].endswith(".mp4")
assert "下载工具返回警告" in result["message"]


def test_thumbnail_failure_does_not_fail_media_download(monkeypatch, tmp_path: Path) -> None:
_seed_video(monkeypatch, tmp_path)
commands: list[list[str]] = []

def fake_run(command: list[str], **_: object) -> Completed:
commands.append(command)
if "--write-thumbnail" in command:
raise subprocess.TimeoutExpired(command, timeout=600)
output = _output_path(command)
output.parent.mkdir(parents=True, exist_ok=True)
output.write_text("video", encoding="utf-8")
return Completed(stdout=f"{output}\n")

monkeypatch.setattr(downloads.subprocess, "run", fake_run)

result = run_authorized_download(1, "已授权本地测试", "1080p", include_subtitles=False, include_thumbnail=True)

assert result["task"]["status"] == "completed"
assert "--write-thumbnail" not in commands[0]
assert any("--write-thumbnail" in command for command in commands[1:])
assert "封面" in result["message"]


def test_download_uses_single_file_format_when_ffmpeg_is_missing(monkeypatch, tmp_path: Path) -> None:
_seed_video(monkeypatch, tmp_path)
commands: list[list[str]] = []

def fake_which(name: str) -> str | None:
if name == "ffmpeg":
return None
if name == "node":
return "/usr/bin/node"
return None

def fake_run(command: list[str], **_: object) -> Completed:
commands.append(command)
output = _output_path(command)
output.parent.mkdir(parents=True, exist_ok=True)
output.write_text("video", encoding="utf-8")
return Completed(stdout=f"{output}\n")

monkeypatch.setattr(downloads.shutil, "which", fake_which)
monkeypatch.setattr(downloads.subprocess, "run", fake_run)

run_authorized_download(1, "已授权本地测试", "1080p", include_subtitles=False, include_thumbnail=False)

command = commands[0]
assert "--merge-output-format" not in command
assert "+" not in command[command.index("-f") + 1]
7 changes: 4 additions & 3 deletions docs/AI_AGENT_SETUP.md
Original file line number Diff line number Diff line change
Expand Up @@ -45,9 +45,10 @@ npm run dev

These must exist on the user's computer before the app can fully run:

- Node.js / npm
- Node.js / npm: Node must be >=20.19.0 or >=22.12.0
- uv
- FFmpeg

Media features also need FFmpeg for reliable YouTube audio/video merging, audio extraction, and clip export. `npm run setup` can continue without FFmpeg so the app can start, but real video processing will be degraded until FFmpeg is installed.

On macOS, a coding agent can usually install them with Homebrew:

Expand Down Expand Up @@ -132,7 +133,7 @@ https://github.com/arnoldhao/xiadown

1. Clone the repo.
2. Run `npm run doctor`.
3. Install missing required dependencies.
3. Install missing required dependencies, and install FFmpeg before real video processing.
4. Run `npm run setup`.
5. Run `npm run test`.
6. Run `npm run dev`.
Expand Down
Loading