Один стример, который играет в PUBG, попросил делать из его записей короткие вертикальные ролики для Shorts, Reels и TikTok. Согласие на то, чтобы рассказать об этом, он дал, но имя я не называю: в статье он просто «стример». Сначала я нарезал клипы по речи и по расшифровке Whisper, и ему они не зашли: он написал, что я резал вообще не игровые моменты. Ему нужны моменты, где он сбил противника с ног или убил. Дальше пришлось учить скрипт находить такие моменты в записи на несколько часов.
Готового сигнала в записи Twitch нет: нет логов игры и нет доступа к данным матча, есть только картинка и звук. Зато сама игра пишет в нижней части экрана «Игрок (ВЫ) УБИВАЕТ игрока (ник), используя оружие» и показывает красный счётчик «1 УБИТО». Я читаю эту надпись как текст и по ней нахожу секунды событий. Дальше расскажу, как это устроено и на чём я спотыкался (кое-что работает неидеально до сих пор). Код взят из рабочих скриптов, лишнее я убрал.
Что на входе и что на выходе
Вход: ссылка на VOD (так называют запись прошедшего стрима на Twitch). Выход: вертикальные ролики 1080×1920, либо отдельными клипами, либо одним монтажом из нескольких моментов.
Конвейер состоит из четырёх шагов:
kill_events.py VOD → кадр в секунду → полоса с надписью → Tesseract → список событий pick_highlights.py события → серии подряд идущих → список клипов fetch_segment.py скачать из VOD только нужный отрезок render_clip.py отрезок → вертикальный ролик (ffmpeg)
Стек: yt-dlp для скачивания, ffmpeg для нарезки и вёрстки кадра, Tesseract с русским языком для распознавания, OpenCV и NumPy для подготовки картинки. Для разговорных клипов в проекте отдельно используется faster-whisper, но к этой статье он не относится.
Интерфейс игры у стримера русский, поэтому все шаблоны ниже ищут русские слова. Для английского интерфейса их пришлось бы переписать.
Почему нельзя просто скачать видео
Twitch отдаёт запись со скоростью около 1,3-1,5 МБ/с. Полный VOD в 1080p60 занимает около 8 ГБ и качается примерно полтора часа. Один раз потерпеть можно, но ждать столько на каждый стрим и на каждую правку скрипта я не собирался.
Поэтому скачивание устроено двумя способами. Для поиска событий я беру поток в 720p60 и читаю его на лету, ничего не сохраняя. Для самих клипов качаются только отрезки по 40-90 секунд, каждый за 30-60 секунд.
Отрезок скачивается так: yt-dlp -g печатает адрес плейлиста нужного формата, а ffmpeg читает из него только нужный интервал и копирует потоки без перекодирования:
m3u8 = subprocess.run(["yt-dlp", "-g", "-f", a.format, a.url], capture_output=True, text=True, check=True).stdout.strip() s = max(0, a.start - a.pad); e = a.end + a.pad subprocess.run(["ffmpeg", "-loglevel", "error", "-y", "-ss", str(s), "-to", str(e), "-i", m3u8, "-c", "copy", "-movflags", "+faststart", a.out], check=True) print(f"{s:.3f}")
Последняя строка нужна дальше по цепочке: скрипт печатает, с какой секунды VOD начинается сохранённый файл. Эта цифра (src-offset) требуется для точной резки: чтобы вырезать момент на 4210-й секунде, надо знать, где в файле находится эта секунда.
Две грабли, которые стоили мне потерянных списков и перекачанных файлов:
Если вызывать скачивание из цикла
while read, ffmpeg по умолчанию читает стандартный ввод и съедает оставшиеся строки списка. Лечится< /dev/nullна каждый вызов, а в питоне то же самое делаетstdin=subprocess.DEVNULL.Обрыв сети оставляет файл без
moov atom(это заголовок с описанием структуры видео, без него плеер файл не открывает). Мне такой отрезок однажды попался, и теперь я проверяю результат черезffprobeи битый файл качаю заново.
Где в кадре искать надпись
Сначала я пошёл по очевидному пути и искал ленту убийств в правом верхнем углу: туда игра выводит строки вида «кто кого». Детектор ничего не находил.
Оказалось, что крупную надпись о собственном убийстве игра у стримера выводит в нижней части кадра, и моя зона поиска её физически не пересекала. К тому же в ленте в углу нет его нокаутов (ноком называют ситуацию, когда противник сбит с ног, но ещё не добит), а ноки ему нужны в первую очередь. Надпись внизу содержит и ноки («СБИВАЕТ С НОГ», белым цветом), и киллы.
Зона задана долями кадра, чтобы не зависеть от разрешения:
BAND = dict(x=0.15, y=0.62, w=0.70, h=0.16) # доля кадра: полоса с надписью об убийстве
Это полоса от 62% до 78% высоты и от 15% до 85% ширины. Для другого стримера эти четыре числа придётся подбирать заново, и это единственная настройка под конкретный экран, которая нужна детектору.
Кадр в секунду из потока, без файла на диске
Кадры достаются так: yt-dlp качает фрагменты потока в 8 параллельных потоков и пишет в stdout, а ffmpeg на входе из пайпа берёт один кадр в секунду, приводит его к 1280×720 и вырезает нужную полосу. Результат идёт дальше как «сырое» видео, то есть последовательность картинок одинакового размера.
dl = subprocess.Popen( ["yt-dlp", "-f", fmt, "-N", "8", "--download-sections", sec, "--no-part", "-q", "-o", "-", url], stdout=subprocess.PIPE) ff = subprocess.Popen( ["ffmpeg", "-v", "error", "-i", "pipe:0", "-vf", f"fps={fps},scale={sw}:{sh},crop={w}:{h}:{x}:{y}", "-f", "rawvideo", "-pix_fmt", "bgr24", "-"], stdin=dl.stdout, stdout=subprocess.PIPE) dl.stdout.close() size = w * h * 3 while True: buf = ff.stdout.read(size) if len(buf) < size: break yield start + i / fps, np.frombuffer(buf, np.uint8).reshape(h, w, 3)
Размер одного кадра известен заранее (ширина × высота × 3 байта), поэтому читать поток можно ровными порциями. Если вместо ссылки передан файл на диске, тот же код читает его напрямую. Это нужно для проверки настроек на коротком куске или на одном скриншоте, чтобы не запускать скачивание на 45 минут.
scale я поставил перед crop после одной аварии: посреди одного из стримов у стримера поменялось разрешение картинки, и ffmpeg упал на crop, потому что вырезаемая область оказалась больше кадра. С предварительным scale на вход в crop всегда приходит один и тот же размер. Если поток всё же оборвался, основной цикл запоминает последнюю обработанную секунду и запускает чтение заново с этого места. Если за попытку не получено ни одного кадра, позиция сдвигается на 5 секунд, а после 40 таких попыток подряд цикл останавливается.
Почти всё время прохода съедает скачивание: запись длиной 2,5 часа в 720p проходит примерно за 45 минут, так что я запускаю её в фоне.
Про разрешение: в 480p текст рассыпается при бинаризации, поэтому 720p я считаю минимумом. Формат 480p качается втрое быстрее, но часть событий теряется. В скрипте формат по умолчанию 720p60, а параметр --format позволяет его поменять.
Подготовка картинки для Tesseract
Надпись полупрозрачная, лежит поверх живой картинки игры и состоит из белого текста и красного счётчика. Первая версия бинаризации брала серый кадр и отсекала всё ярче порога. Красный цвет при переводе в серый становится тёмным, и порог его съедал.
Теперь я строю для каждого кадра две маски, белую и красную:
def white_mask(band): g = cv2.cvtColor(band, cv2.COLOR_BGR2GRAY) return cv2.threshold(g, 175, 255, cv2.THRESH_BINARY)[1] def red_mask(band): b, g, r = band[:, :, 0].astype(int), band[:, :, 1].astype(int), band[:, :, 2].astype(int) return ((r > 110) & (r - g > 50) & (r - b > 50)).astype(np.uint8) * 255
Красный я определяю по разнице каналов: красного должно быть заметно больше, чем зелёного и синего, и тогда просто светлые участки игры в маску не попадают.
Дальше каждая маска увеличивается (белая в 2 раза, красная в 3), инвертируется в чёрный текст на белом и отдаётся в Tesseract:
def ocr(img, tmp, psm="6"): cv2.imwrite(tmp, img) r = subprocess.run(["tesseract", tmp, "-", "-l", "rus", "--psm", psm], capture_output=True) return r.stdout.decode("utf-8", "replace")
Режим --psm 6 говорит Tesseract, что перед ним один блок текста. Перед распознаванием стоит дешёвая проверка: если в белой маске меньше 300 светлых пикселей, текста в кадре нет, и кадр пропускается. Надпись о событии видна недолго, остальное время в полосе только игра, поэтому большая часть кадров до Tesseract не доходит.
Ещё одна ловушка: временный файл для Tesseract пришлось класть в папку проекта. Скрипты запускались в песочнице, где системная временная папка у питона и у Tesseract оказалась разной, и распознавание видело пустую картинку. Сначала я принял это за плохое качество кадров. Теперь в коде есть явная ошибка, если Tesseract упал, а раньше он молчал.
Главная ловушка: игра показывает чужие события
Игра пишет в ленту не только убийства стримера, но и чужие. Строка «Игрок (ВЫ) УБИВАЕТ игрока (ник)» значит, что убил он сам. Но точно так же выводятся строки про тиммейтов и про то, что убили самого стримера: «игрок (ник) УБИВАЕТ игрока (ВЫ)».
Из первых 10 найденных «моментов» его собственными были только 3. Остальные принадлежали тиммейтам или были случаями, когда убивали его самого. Правки к готовому монтажу это подтвердили: в одном ролике пришлось убрать последние 4 секунды, потому что там самого стримера нокают.
Я решил это правилом: «(ВЫ)» должно стоять в первых скобках строки и левее слова «игрока», тогда оно относится к тому, кто действует.
Проблема в том, что Tesseract плохо читает начало строки. Слово «Игрок» превращается в «РОК» или в «‘РОК», и привязываться к нему нельзя. Поэтому правило смотрит только на порядок скобок. Вот функция целиком:
BRACKET = re.compile(r"[\(\{\[]\s*([^\)\}\]]{1,24})[\)\}\]]") VICTIM = re.compile(r"[ИМН]ГР[ОД0][КХ][АЗ]") # «игрока» — дальше идёт жертва ANCHOR = ("ИСПОЛЬЗУЯ", "ИСПОЛЬЗУ", "ИСПОЛЬ") KILL = ("УБИ", "БИВ", "ЪИВ", "ЫБИ") KNOCK = ("ВЫРУБ", "НОКАУТ", "СБИ", "ОГЛУШ") FRIENDLY = ("СВОИМ", "СВОЕГО") # «огонь по своим» — не клип def classify(wtxt, rtxt=""): """Тип события по распознанной надписи, либо None, если это не её убийство.""" if not any(k in wtxt for k in ANCHOR) and not any(k in wtxt for k in KNOCK): return None m = VICTIM.search(wtxt) upto = m.start() if m else len(wtxt) first = BRACKET.search(wtxt) # (ВЫ) должно стоять в ПЕРВЫХ скобках и левее слова «игрока» if not first or "ВЫ" not in first.group(1) or first.end() > upto: return None if any(k in wtxt + rtxt for k in FRIENDLY): return None if any(k in wtxt for k in KNOCK): return "нок" if any(k in rtxt for k in KILL) or any(k in wtxt for k in KILL): return "килл" return "килл"
Что здесь важно:
Скобки Tesseract тоже иногда путает, поэтому регулярное выражение принимает все три вида: круглые, фигурные и квадратные.
Слово «игрока» в шаблоне
VICTIMзаписано с запасом по буквам ([ИМН]ГР[ОД0][КХ][АЗ]): распознавание подменяет часть букв похожими.Слова вида «УБИВАЕТ» и «СБИВАЕТ С НОГ» сверяются по обрывкам: «УБИ», «БИВ», «СБИ». Целое слово Tesseract читает с ошибками слишком часто.
Красная маска читается отдельным проходом и служит подсказкой для типа события, когда белый текст порвался.
Нок проверяется раньше килла, потому что обрывок «БИВ» из списка киллов входит и в слово «СБИВАЕТ».
Для повторных проверок я сделал дамп: параметр --dump-file сохраняет все распознанные строки с секундами, а --from-dump пересобирает события из дампа, не качая видео. Правило фильтрации теперь чинится за секунды вместо тех же 45 минут ожидания, и из всего, что я добавил в детектор, дамп окупился лучше всего.
Схлопывание повторов
Один и тот же килл может попасть в несколько соседних кадров подряд. Чтобы не получить несколько событий вместо одного, между событиями выдерживается пауза 6 секунд (--gap):
if t - last >= a.gap: events.append({"t": round(t, 1), "kind": kind, "text": wtxt[:90]}) json.dump(events, open(a.out, "w"), ensure_ascii=False, indent=1) last = t
Переменная last обновляется на каждом совпадении, даже если событие не записано. Пока надпись висит на экране, пауза продлевается, и новое событие считается только после того, как надпись исчезла хотя бы на 6 секунд. JSON перезаписывается после каждого события, поэтому при обрыве на третьем часу у меня остаётся всё найденное к этому моменту.
Из событий в клипы
Ноки и киллы чаще всего идут подряд, в одной перестрелке. Резать каждое по отдельности бессмысленно, поэтому pick_highlights.py собирает серии: события ближе 20 секунд друг к другу считаются одним моментом.
PRE, POST = 8.0, 4.0 # сколько секунд захватить до первого события и после последнего SERIES_GAP = 20.0 # события ближе этого — один момент MIN_LEN, MAX_LEN = 8.0, 45.0 for e in evs: if cur and e["t"] - cur[-1]["t"] > SERIES_GAP: series.append(cur); cur = [] cur.append(e) for s in series: start = max(0.0, s[0]["t"] - PRE) end = min(s[-1]["t"] + POST, start + MAX_LEN) if end - start < MIN_LEN: end = start + MIN_LEN
Рейтинг у серии пока простой: чем больше событий, тем выше, при равенстве раньше по времени. Громкость в рейтинг не входит, хотя в плане я её закладывал.
Вёрстка вертикального кадра
За кадр отвечает render_clip.py. Режимов несколько, потому что пожелания стримера менялись по ходу работы.
Режим game занимает всю вертикаль игрой: из горизонтального кадра вырезается центральная полоса 9:16 вокруг прицела, камеры и субтитров нет. Так выглядели его референсы.
Режим fit показывает весь горизонтальный кадр посередине, а сверху и снизу заполняет его же размытой увеличенной копией:
vf = (f"[0:v]{pre}split=2[bg][fg];" f"[bg]scale={W}:{H}:force_original_aspect_ratio=increase,crop={W}:{H}," f"boxblur={BLUR}:2[b];" f"[fg]scale={W}:-2[f];" f"[b][f]overlay=(W-w)/2:(H-h)/2[v]")
Режим split, под который стример в итоге попросил делать ролики (в клипах Twitch этот формат называется «Замостить»), ставит камеру сверху, а игру снизу:
vf = (f"[0:v]split=2[c][g];" f"[c]crop={CAM['w']}:{CAM['h']}:{CAM['x']}:{CAM['y']},scale={W}:{CAM_H}[cam];" f"[g]crop={GAME_CROP_W}:1080:{GAME_CROP_X}:0,scale={W}:{GAME_H}[game];" f"[cam][game]vstack,ass={ass}[v]")
Координаты камеры CAM заданы для кадра 1920×1080 и подобраны под оверлей конкретного стримера. Для другого их нужно измерять заново. Автоматического поиска камеры у меня нет, он в списке дел.
Субтитры в ass= нужны только для разговорных клипов. Чтобы они вжигались, нужен ffmpeg с библиотекой libass: обычная сборка у меня на этом падала, поэтому в скриптах стоит отдельная сборка с libass, на которую указывает константа FFMPEG.
Монтаж из нескольких убийств и неточность времени
Кроме отдельных клипов стример хотел один ролик из нескольких моментов подряд, по 1-2 секунды на кусок, как в его референсах. make_montage.py берёт каждый момент коротким куском: 2 секунды до события и 1,5 секунды после, а потом склеивает куски.
Дальше выяснилось, что время детектора неточное. Он смотрит один кадр в секунду, а на ранних кадрах, когда надпись только появилась, Tesseract иногда ничего не читает. В итоге найденное время лежало на 0,2-6 секунд позже реального появления надписи. На коротком куске в 3,5 секунды такой сдвиг может вывести убийство за пределы кадра.
Поэтому каждый момент я уточняю отдельно. Скрипт скачивает отрезок с запасом в 10 секунд назад и ищет в нём первый кадр, где надпись появилась, проверяя кадры с частотой 5 в секунду:
def refine(seg, off, t, kind, fps=5): """Первый кадр отрезка, где появилась надпись «Игрок (ВЫ) …». Проверка мягче, чем в детекторе: событие уже подтверждено, ищем только начало надписи. Для килла надписи о ноке пропускаем — часто перед добиванием тот же игрок сбит с ног.""" ... hit = "ВЫ" in wtxt and any(k in wtxt for k in ("ИГРОК", "ИСПОЛЬЗ")) if hit and not (kind == "килл" and any(k in wtxt for k in KNOCK)): return ts print(f" ! надпись у {t:.0f} c не нашлась — беру время детектора") return t
Условие здесь намеренно мягче, чем в classify: достаточно слова «ВЫ» и одного из слов «ИГРОК» или «ИСПОЛЬЗ». Строгая проверка на отдельных кадрах не срабатывала, потому что Tesseract терял скобку. Мягкой проверке можно доверять, потому что мы уже знаем, что в этом диапазоне событие есть, и ищем только его начало. Для килла дополнительно пропускаются кадры с надписью о ноке: часто перед добиванием тот же противник сначала сбит с ног, и более ранний кадр принадлежит нок-событию.
Если надпись так и не нашлась, скрипт пишет об этом и берёт время детектора. Такое случалось: один момент пришлось сдвигать руками, а в другом случае уточнение ухватило надпись на 9 секунд раньше нужной, и в кадр попал экран инвентаря.
Для версии с музыкой куски нарезаются по сетке битов. Период бита находит librosa.beat.beat_track, убийство ложится на 4-й бит куска, а в этот момент добавляются чёрно-белая вспышка на 0,35 секунды и быстрый наезд камеры:
zoom = f"if(gte(it,{te}),1+0.15*exp(-12*(it-{te})),1)" fx.append(f"[{i}:v]trim={s0}:{s0 + L + d},setpts=PTS-STARTPTS," f"zoompan=z='{zoom}':x='iw/2-iw/zoom/2':y='ih/2-ih/zoom/2':d=1:s=1080x1920:fps=60," f"hue=s=0:enable='between(t,{te},{te + 0.35})'," f"eq=brightness=0.3:enable='between(t,{te},{te + 0.06})'[v{i}]")
Куски соединяются фильтром xfade с переходом zoomin, звук игры идёт тише трека (множитель 0,45), а общий результат нормализуется loudnorm до -14 LUFS. Музыку я брал только с лицензией CC0 или CC BY: с условием «без производных» (-ND) трек нельзя класть под видео.
Когда я показал стримеру версию с музыкой и эффектами, ему понравился формат «Замостить», но музыку он попросил не накладывать: свою он подставляет сам. Пришлось добавить вариант без музыки, где куски идут подряд, звук игры остаётся, а в конце применяется только loudnorm. Он и стал основным.
Что получилось на реальных стримах
Я прогнал детектор на трёх записях стримера.
Запись длиной 2 часа 27 минут: 3 килла и 3 нока.
Ночной стрим длиной 4 часа 50 минут: 26 найденных событий. Три из них оказались случаями, когда сбили самого стримера: Tesseract порвал первые скобки, и правило про «(ВЫ)» не сработало. Я убрал их руками и получил 21 клип. По листу кадров (по три кадра с каждого клипа на одной картинке) проверил остальные: надпись о ноке или килле на месте у 20 из них, а в одном нашлась ошибка во времени, и момент я сдвинул руками.
Третья запись: 29 событий, 16 киллов и 13 ноков, во всех стоит «(ВЫ)». Самый плотный бой занял чуть больше шести с половиной минут и дал 10 событий. Из 16 киллов собрался монтаж без музыки длиной 57,8 секунды.
Монтаж из 9 киллов с ночного стрима в формате «Замостить» получился на 25,1 секунды.
Что осталось неидеальным
Ложные события. Правило про порядок скобок отсекает большинство чужих строк, но на втором длинном стриме три случая прошли, и я убрал их руками. Если запускать конвейер без присмотра, такие клипы будут уходить стримеру. Автоматического отсева «сбили меня» пока нет.
Неточное время. Детектор отстаёт от появления надписи до 6 секунд, а доуточнение иногда не находит надпись вообще или хватает слишком ранний кадр. Поэтому лист кадров перед отправкой я смотрю всегда.
Скорость. 45 минут на 2,5 часа записи, почти всё время уходит на скачивание.
Нет метрик качества. Точность и полноту детектора на размеченной выборке я не считал. Знаю только, что на трёх стримах он находил надписи, и знаю про найденные ошибки, потому что смотрел результат глазами.
Один язык и один оверлей. Шаблоны слов написаны под русский интерфейс игры, а зона
BANDподобрана под экран одного стримера.Окно вокруг события. В одном монтаже примерно через 36 секунд после килла открыт инвентарь, и клип смотрится хуже, чем мог бы. Проверять это автоматически я пока не умею.
Только офлайн. Стример спросил, можно ли делать клипы в прямом эфире. Теоретически тот же детектор можно направить на живой поток с задержкой в минуту-две, но я этого не делал и не проверял.
Итог в цифрах
Все цифры взяты из моих рабочих записей по трём стримам:
Скорость Twitch на скачивании: около 1,3-1,5 МБ/с. Полный VOD в 1080p60: около 8 ГБ и около полутора часов.
Проход детектора по записи в 2,5 часа в 720p60: около 45 минут.
Отрезок клипа в 40-90 секунд скачивается за 30-60 секунд.
Найдено событий: 6 на первом стриме (3 килла и 3 нока), 26 на втором длиной 4 часа 50 минут (из них 3 ложных), 29 на третьем (все по делу).
Расхождение времени детектора с реальным появлением надписи: от 0,2 до 6 секунд.
Монтаж из 9 киллов: 25,1 секунды, из 16 киллов: 57,8 секунды.
Если делать такое с нуля, я бы в первый же час написал режим «дамп строк и разбор из дампа» и проверял зону и правила на одном скриншоте из записи. Всё остальное, включая вёрстку и монтаж, оказалось проще, чем научить скрипт отличать собственный килл стримера от чужих.
