Elektroroller-Forum.de

Code: Alles auswählen

#!/usr/bin/env bash
set -u
set -o pipefail

# Black Tea Motorbikes – YouTube transcript archive
#
# Requirements:
#   - yt-dlp
#   - python3
#
# Downloads NO video/audio.
# Archives one preferred ORIGINAL caption track per video plus metadata/index.
#
# Re-running is safe:
#   - existing subtitle files are detected by YouTube video ID
#   - index rows are updated in place
#   - errors do not stop the complete run

CHANNEL_URL="https://www.youtube.com/@blackteamotorbikes/videos"

OUT_DIR="${1:-black-tea-youtube}"
RAW_DIR="$OUT_DIR/raw"
META_DIR="$OUT_DIR/meta"
WORK_DIR="$OUT_DIR/.work"
INDEX_CSV="$OUT_DIR/index.csv"
ERROR_LOG="$OUT_DIR/errors.log"
VIDEO_LIST="$WORK_DIR/video-ids.txt"

# Conservative pause to reduce HTTP 429 risk.
SLEEP_BETWEEN_VIDEOS=3
SLEEP_AFTER_ERROR=8

mkdir -p "$RAW_DIR" "$META_DIR" "$WORK_DIR"
touch "$ERROR_LOG"

command -v yt-dlp >/dev/null 2>&1 || {
    echo "ERROR: yt-dlp not found. Install it first (e.g. brew install yt-dlp)." >&2
    exit 1
}
command -v python3 >/dev/null 2>&1 || {
    echo "ERROR: python3 not found." >&2
    exit 1
}

if [[ ! -f "$INDEX_CSV" ]]; then
    printf '%s\n' \
      'published,title,youtube_id,url,duration_seconds,caption_language,caption_type,filename,status' \
      > "$INDEX_CSV"
fi

# v2 accidentally wrote literal "\t" into the flat-playlist rows, which could
# also create malformed CSV rows. Remove only clearly malformed legacy rows.
INDEX_CSV="$INDEX_CSV" python3 - <<'PY'
import csv
import os
from pathlib import Path

path = Path(os.environ["INDEX_CSV"])
if not path.exists():
    raise SystemExit

with path.open("r", encoding="utf-8", newline="") as f:
    rows = list(csv.DictReader(f))

fields = [
    "published", "title", "youtube_id", "url", "duration_seconds",
    "caption_language", "caption_type", "filename", "status",
]

clean = []
for r in rows:
    vid = r.get("youtube_id", "")
    if "\\t" in vid or "\t" in vid or " " in vid:
        continue
    clean.append(r)

with path.open("w", encoding="utf-8", newline="") as f:
    w = csv.DictWriter(f, fieldnames=fields)
    w.writeheader()
    w.writerows(clean)
PY

log_error() {
    local video_id="$1"
    local message="$2"
    printf '%s\t%s\t%s\n' \
      "$(date '+%Y-%m-%d %H:%M:%S')" "$video_id" "$message" >> "$ERROR_LOG"
}

csv_upsert() {
    ROW_PUBLISHED="$1" \
    ROW_TITLE="$2" \
    ROW_ID="$3" \
    ROW_URL="$4" \
    ROW_DURATION="$5" \
    ROW_LANG="$6" \
    ROW_TYPE="$7" \
    ROW_FILENAME="$8" \
    ROW_STATUS="$9" \
    INDEX_CSV="$INDEX_CSV" \
    python3 - <<'PY'
import csv
import os
from pathlib import Path

path = Path(os.environ["INDEX_CSV"])
video_id = os.environ["ROW_ID"]

rows = []
if path.exists():
    with path.open("r", encoding="utf-8", newline="") as f:
        rows = list(csv.DictReader(f))

rows = [r for r in rows if r.get("youtube_id") != video_id]
rows.append({
    "published": os.environ["ROW_PUBLISHED"],
    "title": os.environ["ROW_TITLE"],
    "youtube_id": video_id,
    "url": os.environ["ROW_URL"],
    "duration_seconds": os.environ["ROW_DURATION"],
    "caption_language": os.environ["ROW_LANG"],
    "caption_type": os.environ["ROW_TYPE"],
    "filename": os.environ["ROW_FILENAME"],
    "status": os.environ["ROW_STATUS"],
})

fields = [
    "published", "title", "youtube_id", "url", "duration_seconds",
    "caption_language", "caption_type", "filename", "status",
]

with path.open("w", encoding="utf-8", newline="") as f:
    w = csv.DictWriter(f, fieldnames=fields)
    w.writeheader()
    w.writerows(rows)
PY
}

find_existing_subtitle() {
    VIDEO_ID="$1" RAW_DIR="$RAW_DIR" python3 - <<'PY'
import os
from pathlib import Path

vid = os.environ["VIDEO_ID"]
root = Path(os.environ["RAW_DIR"])

# We currently request VTT, but accept another subtitle sidecar if yt-dlp
# happens to return one in a future version.
allowed = {".vtt", ".srt", ".ttml", ".srv1", ".srv2", ".srv3", ".json3"}

for p in sorted(root.iterdir()) if root.exists() else []:
    if p.is_file() and f"[{vid}]" in p.name and p.suffix.lower() in allowed:
        print(p)
        break
PY
}

echo "1/3  Reading complete channel video list …"

# IMPORTANT: only print the ID. v2 used '\t' in the output template, but yt-dlp
# printed that literally rather than as a tab on this system.
if ! yt-dlp \
    --flat-playlist \
    --ignore-errors \
    --print '%(id)s' \
    "$CHANNEL_URL" > "$VIDEO_LIST" 2>>"$ERROR_LOG"
then
    echo "ERROR: Could not read channel video list. See $ERROR_LOG" >&2
    exit 1
fi

# Keep only plausible YouTube IDs and remove duplicates while preserving order.
python3 - "$VIDEO_LIST" <<'PY'
import re
import sys
from pathlib import Path

path = Path(sys.argv[1])
seen = set()
out = []

for line in path.read_text(encoding="utf-8").splitlines():
    vid = line.strip()
    if not re.fullmatch(r"[A-Za-z0-9_-]{11}", vid):
        continue
    if vid in seen:
        continue
    seen.add(vid)
    out.append(vid)

path.write_text("\n".join(out) + ("\n" if out else ""), encoding="utf-8")
PY

VIDEO_COUNT="$(grep -cve '^[[:space:]]*$' "$VIDEO_LIST" || true)"
if [[ "$VIDEO_COUNT" -eq 0 ]]; then
    echo "ERROR: Channel video list is empty. See $ERROR_LOG" >&2
    exit 1
fi

echo "     Found $VIDEO_COUNT videos."
echo "2/3  Downloading metadata + one caption track per video …"

CURRENT=0

# Channel listing is normally newest -> oldest.
# Process oldest -> newest on macOS without requiring GNU tac.
reverse_file() {
    if tail -r "$1" >/dev/null 2>&1; then
        tail -r "$1"
    elif command -v tac >/dev/null 2>&1; then
        tac "$1"
    else
        python3 - "$1" <<'PY'
import sys
from pathlib import Path
for line in reversed(Path(sys.argv[1]).read_text(encoding="utf-8").splitlines()):
    print(line)
PY
    fi
}

while IFS= read -r VIDEO_ID; do
    [[ -z "$VIDEO_ID" ]] && continue

    CURRENT=$((CURRENT + 1))
    VIDEO_URL="https://www.youtube.com/watch?v=$VIDEO_ID"
    JSON_FILE="$META_DIR/$VIDEO_ID.info.json"
    CANDIDATES_FILE="$WORK_DIR/$VIDEO_ID.candidates.tsv"

    echo
    echo "[$CURRENT/$VIDEO_COUNT] $VIDEO_ID"

    # Fetch full metadata. This gives us title/date/duration and all subtitle
    # dictionaries without downloading media.
    if ! yt-dlp \
        --skip-download \
        --no-warnings \
        --dump-single-json \
        "$VIDEO_URL" > "$JSON_FILE.tmp" 2>>"$ERROR_LOG"
    then
        rm -f "$JSON_FILE.tmp"
        echo "  ERROR: metadata failed; continuing."
        log_error "$VIDEO_ID" "metadata download failed"
        csv_upsert "" "" "$VIDEO_ID" "$VIDEO_URL" "" "" "" "" "metadata_error"
        sleep "$SLEEP_AFTER_ERROR"
        continue
    fi
    mv "$JSON_FILE.tmp" "$JSON_FILE"

    # Extract basic metadata and build an ordered list of caption candidates.
    #
    # Language preference:
    #   de-orig -> en-orig -> any *-orig -> de -> en -> everything else
    #
    # For the SAME language, manually supplied subtitles are tried before
    # automatic captions. If one candidate fails, the next one is tried.
    JSON_FILE="$JSON_FILE" CANDIDATES_FILE="$CANDIDATES_FILE" python3 - <<'PY'
import json
import os
from pathlib import Path

src = Path(os.environ["JSON_FILE"])
dst = Path(os.environ["CANDIDATES_FILE"])

with src.open(encoding="utf-8") as f:
    d = json.load(f)

manual = d.get("subtitles") or {}
auto = d.get("automatic_captions") or {}

all_langs = {
    x for x in set(manual) | set(auto)
    if x and x != "live_chat"
}

def lang_rank(lang):
    if lang == "de-orig":
        return (0, lang)
    if lang == "en-orig":
        return (1, lang)
    if lang.endswith("-orig"):
        return (2, lang)
    if lang == "de":
        return (3, lang)
    if lang == "en":
        return (4, lang)
    return (5, lang)

rows = []
for lang in sorted(all_langs, key=lang_rank):
    if lang in manual:
        rows.append((lang, "manual"))
    if lang in auto:
        rows.append((lang, "auto"))

with dst.open("w", encoding="utf-8") as f:
    for lang, kind in rows:
        f.write(f"{lang}\t{kind}\n")
PY

    BASIC="$(python3 - "$JSON_FILE" <<'PY'
import json, sys
with open(sys.argv[1], encoding="utf-8") as f:
    d = json.load(f)
title = (d.get("title") or "").replace("\n", " ").replace("\t", " ")
published = d.get("upload_date") or ""
duration = "" if d.get("duration") is None else str(d.get("duration"))
print("\t".join([published, title, duration]))
PY
)"
    IFS=$'\t' read -r PUBLISHED TITLE DURATION <<< "$BASIC"

    echo "  $TITLE"

    # v2 may already have downloaded the correct VTT even though its lookup
    # failed. Detect by the real 11-character YouTube ID only.
    EXISTING_FILE="$(find_existing_subtitle "$VIDEO_ID")"
    if [[ -n "$EXISTING_FILE" && -f "$EXISTING_FILE" ]]; then
        echo "  Already present: $(basename "$EXISTING_FILE")"

        # Infer language from known candidates where possible.
        FOUND_LANG=""
        FOUND_TYPE=""
        while IFS=$'\t' read -r LANG KIND; do
            if [[ "$(basename "$EXISTING_FILE")" == *".$LANG."* ]]; then
                FOUND_LANG="$LANG"
                FOUND_TYPE="$KIND"
                break
            fi
        done < "$CANDIDATES_FILE"

        csv_upsert "$PUBLISHED" "$TITLE" "$VIDEO_ID" "$VIDEO_URL" "$DURATION" \
          "$FOUND_LANG" "$FOUND_TYPE" "$(basename "$EXISTING_FILE")" "ok"
        continue
    fi

    if [[ ! -s "$CANDIDATES_FILE" ]]; then
        echo "  No captions found; continuing."
        log_error "$VIDEO_ID" "no caption track found"
        csv_upsert "$PUBLISHED" "$TITLE" "$VIDEO_ID" "$VIDEO_URL" "$DURATION" \
          "" "" "" "no_captions"
        sleep "$SLEEP_BETWEEN_VIDEOS"
        continue
    fi

    SUCCESS=0
    ATTEMPTS=0

    while IFS=$'\t' read -r LANG CAPTION_TYPE; do
        [[ -z "$LANG" ]] && continue
        ATTEMPTS=$((ATTEMPTS + 1))

        echo "  Trying caption: $LANG ($CAPTION_TYPE)"

        SUB_ARGS=(
            --skip-download
            --sub-format "vtt/best"
            --sub-langs "$LANG"
        )

        if [[ "$CAPTION_TYPE" == "manual" ]]; then
            SUB_ARGS+=(--write-subs)
        else
            SUB_ARGS+=(--write-auto-subs)
        fi

        # Remove only stale subtitle sidecars belonging to THIS exact video ID.
        VIDEO_ID="$VIDEO_ID" RAW_DIR="$RAW_DIR" python3 - <<'PY'
import os
from pathlib import Path
vid = os.environ["VIDEO_ID"]
root = Path(os.environ["RAW_DIR"])
allowed = {".vtt", ".srt", ".ttml", ".srv1", ".srv2", ".srv3", ".json3"}
for p in root.iterdir() if root.exists() else []:
    if p.is_file() and f"[{vid}]" in p.name and p.suffix.lower() in allowed:
        p.unlink()
PY

        yt-dlp \
            "${SUB_ARGS[@]}" \
            --no-warnings \
            --output "$RAW_DIR/%(upload_date)s_%(title).120B [%(id)s].%(ext)s" \
            "$VIDEO_URL" >>"$WORK_DIR/download.log" 2>>"$ERROR_LOG"
        RC=$?

        DOWNLOADED_FILE="$(find_existing_subtitle "$VIDEO_ID")"

        if [[ $RC -eq 0 && -n "$DOWNLOADED_FILE" && -f "$DOWNLOADED_FILE" ]]; then
            echo "  Saved: $(basename "$DOWNLOADED_FILE")"
            csv_upsert "$PUBLISHED" "$TITLE" "$VIDEO_ID" "$VIDEO_URL" "$DURATION" \
              "$LANG" "$CAPTION_TYPE" "$(basename "$DOWNLOADED_FILE")" "ok"
            SUCCESS=1
            break
        fi

        if [[ $RC -eq 0 ]]; then
            echo "    No subtitle file produced; trying fallback."
            log_error "$VIDEO_ID" "no subtitle file produced for $LANG ($CAPTION_TYPE)"
        else
            echo "    Download failed; trying fallback."
            log_error "$VIDEO_ID" "caption download failed for $LANG ($CAPTION_TYPE), rc=$RC"
        fi

        # Small pause before trying another language/track.
        sleep 2
    done < "$CANDIDATES_FILE"

    if [[ $SUCCESS -eq 0 ]]; then
        echo "  ERROR: all $ATTEMPTS caption candidates failed; continuing."
        csv_upsert "$PUBLISHED" "$TITLE" "$VIDEO_ID" "$VIDEO_URL" "$DURATION" \
          "" "" "" "caption_error"
        sleep "$SLEEP_AFTER_ERROR"
    else
        sleep "$SLEEP_BETWEEN_VIDEOS"
    fi

done < <(reverse_file "$VIDEO_LIST")

echo
echo "3/3  Finished."
echo "     Raw captions: $RAW_DIR"
echo "     Metadata:     $META_DIR"
echo "     Index:        $INDEX_CSV"
echo "     Errors:       $ERROR_LOG"
echo
echo "Re-running the script is safe: existing subtitle files are skipped."