Недавно OpenAi выпустила новую модель Sora 2, которая взорвала интернет благодаря бесплатному доступу и большим лимитом на генерации. Каждый день пользователю доступна генерация 30 видеороликов длительностью по 10 секунд или 15 видеороликов по 15 секунд.

Это привело меня к мысли, что на её основе можно начать раскрутку соцсетей органическим трафиком. Но как выделиться среди других?

Меняем Watermark

Первое, что меня озадачило - это наличие во всех роликах вотермарки Sora. Её наличие отпугивает потенциальных зрителей, демонстрируя низкое качество и потоковость контента.

На Github уже присутствовали решения, скрывающие вотермарку, но я решил написать своё. Для начала я просто написал сравнение по картинке с поиском зоны вотермарки с её последующим перекрытием.

def detect_watermark_zone(frame, watermark_template, threshold=0.3):
    result = cv2.matchTemplate(frame, watermark_template, cv2.TM_CCOEFF_NORMED)
    _, max_val, _, max_loc = cv2.minMaxLoc(result)
    if max_val >= threshold:
        h, w = watermark_template.shape[:2]
        return (*max_loc, w, h)
    return None

К обнаруженной зоне применялось две функции - blur + overlay. В результате вотермарка соры менялась на вотермарку моего канала.

В процессе тестирования на разных видеороликах выяснилось, что Sora использует 6 разных размеров вотермарки. Для эффективного обнаружения потребовалось добавить в программу все 6 примеров, а также функцию автоматического определения типа вотермарки.

def detect_best_watermark_type(input_path):
    cap = cv2.VideoCapture(input_path)
    total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
    sample_count = min(15, total_frames)  # анализируем первые 15 кадров
    scores = {t: [] for t in W_TYPES}

    for i in range(sample_count):
        ret, frame = cap.read()
        if not ret:
            break
        for t, path in W_TYPES.items():
            tmpl = cv2.imread(path)
            result = cv2.matchTemplate(frame, tmpl, cv2.TM_CCOEFF_NORMED)
            _, max_val, _, _ = cv2.minMaxLoc(result)
            scores[t].append(max_val)

    cap.release()
    avg_scores = {t: np.mean(v) if len(v) > 0 else 0 for t, v in scores.items()}
    best_type = max(avg_scores, key=avg_scores.get)

    if all(val < 0.5 for val in avg_scores.values()):
        best_type = 1

    print(f"[AutoDetect] Selected: {best_type} ({W_TYPES[best_type]}), score={avg_scores[best_type]:.3f}")
    return best_type

Итоговая точность получилась довольно высокая, но для большей эффективности были применены дополнительные оптимизации. В ходе исследования было определено, что позиция вотермарки меняется каждые 67 кадров, поэтому в алгоритм был внедрен автоматический пропуск кадров с простановкой предыдущей зоны, при совпадении зоны на пяти кадрах подрят.

def process_video(input_path, watermark_path, overlay_path, output_path):
    cap = cv2.VideoCapture(input_path)
    video_clip = VideoFileClip(input_path)
    frames = []
    audio = video_clip.audio  # может быть None
    fps = cap.get(cv2.CAP_PROP_FPS)
    watermark_template = cv2.imread(watermark_path)
    overlay_img = cv2.imread(overlay_path, cv2.IMREAD_UNCHANGED)  # RGBA
    frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
    positions = {}

    last_pos = (0, 0, 0, 0)
    ident_count = 0
    skip_frames = 0
    for i in range(frame_count):
        ret, frame = cap.read()
        if not ret:
            break
        if skip_frames > 0:
            positions[i + 1] = last_pos
            skip_frames -= 1
            continue
        zone = detect_watermark_zone(frame, watermark_template)
        if not zone:
            positions[i+1] = (0, 0, 0, 0)
            continue
        positions[i+1] = zone
        if last_pos != zone:
            last_pos = zone
            ident_count = 0
        else:
            ident_count += 1
        if ident_count >= 5:
            next_multiple = ((i + 67) // 67) * 67
            skip_frames = min(next_multiple - i, frame_count - i)
            ident_count = 0

В результате, скорость определения зоны вотермарки выросла почти в 10 раз, без снижения точности поиска.

Наложение субтитров поверх видео

Современная молодежь часто плохо фокусирует внимание и постоянно переключается между разными источниками контента. Для удержания внимания было решено внедрить поверх видео субтитры. В качестве основы использовалась библиотека с Github auto-subtitle . Она автоматически генерирует субтитры из голоса с помощью OpenAi Whisper.

Базовые настройки этой библиотеки не удовлетворяли требованиям, которые обычно используются в Tiktok и Shorts, в результате было применено несколько модификаций. Для субтитров был выбран определенный стиль и размер, распространенные у других создателей контента.

        style = (
            "Fontname=Bahnschrift SemiCondensed,"
            'Fontsize=14,'
            "ScaleX=0.8,"
            "Bold=1,"
            "MarginV=100,"
            "MarginL=50,"
            "MarginR=50,"
            "Outline=1.5,"
            'Shadow=0,'
            'BorderStyle=1'
        )

Также был добавлен ограничитель на вывод текста не более 5 слов за раз, чтобы не перегружать видео, и алгоритм раскраски некоторых слов, чтобы субтитры казались более привлекательными.

def color_srt(input_file: str, output_file: str):
    colors = ['lime', 'yellow']

    def colorize_text(text):
        words = text.split()
        n = len(words)
        if n <= 1:
            return text
        elif n == 2:
            words[1] = f'<font color="{random.choice(colors)}">{words[1]}</font>'
        elif n == 3:
            # красим центральное слово
            words[1] = f'<font color="{random.choice(colors)}">{words[1]}</font>'
        else:
            indices = list(range(1, n))
            to_color = random.sample(indices, random.randint(1, 2) if n >= 5 else 1)
            prev_i = None
            prev_color = None
            for i in to_color:
                if prev_i is not None and ((i == prev_i + 1) or (i == prev_i - 1)):
                    color = prev_color
                else:
                    color = random.choice([c for c in colors if c != prev_color]) if prev_color else random.choice(
                        colors)
                words[i] = f'<font color="{color}">{words[i]}</font>'
                prev_i = i
                prev_color = color
        return ' '.join(words)

В результате программа начала стабильно накладывать субтитры поверх каждого видеоролика.

Склейка видео

Алгоритмы youtube shorts и tiktok отдают предпочтения видеороликам продолжительностью 15-30 секунд. Было решено делать видео длительностью 30 секунд на основе двух Sora генераций. Для этого был написан простой модуль на основе Moviepy.

from moviepy import VideoFileClip, concatenate_videoclips


def concat_videos(video_paths, output_path="output.mp4"):
    clips = []
    for path in video_paths:
        clip = VideoFileClip(path)
        if clip.duration > 1:
            clip = clip.subclipped(0, clip.duration)
        clips.append(clip)

    final_clip = concatenate_videoclips(clips)
    final_clip.write_videofile(output_path, codec="libx264", audio_codec="aac")
    for c in clips:
        c.close()
    final_clip.close()

После этого была написана функция, объединяющая эти три операции в одну.

def mounting_video(video_name: str, subt_off=False):
    video_count = VIDEO_COUNT
    w_types = [0, 0, 0, 0]
    fix_only = (False, 3)
    for i in range(1, video_count+1):
        if fix_only[0]:
            i = fix_only[1]
        print(f'{i} видео начало обработку')
        if w_types[i-1] != -1:
            hide_watermark(PATHS.input + f'{i}.mp4', w_types[i-1])
        else:
            shutil.copy(PATHS.input + f'{i}.mp4', f'./input/m_{i}.mp4')
        if not subt_off:
            sys.argv = ['mounting_video.py', PATHS.tmp + f'm_{i}.mp4', '-o', PATHS.tmp]
            main()
        else:
            shutil.copy(PATHS.tmp + f'm_{i}.mp4', PATHS.tmp + f'sm_{i}.mp4')
        if fix_only[0]:
            break

    videos = [PATHS.tmp + f'sm_{i}.mp4' for i in range(1, video_count+1)]
    concat_videos(videos, PATHS.output + video_name)


if __name__ == '__main__':
    timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M")
    filename = f"file_{timestamp}.mp4"
    mounting_video(filename)

Изначально эта функция была рассчитана на склейку четырех видеороликов по 10 секунд, в процессе оптимизаций было решено заменить их количество на 2 по 15 секунд.

Автоматическая генерация сюжетов

Сначала я вручную писал промпты в чат гпт и вставлял ответы в сору. Через несколько дней мне это надоело, и я начал автоматизировать этот процесс с помощью библиотеки g4f.

def get_sora_prompts(pr_text):
    # new_text = prompt_modifications(pr_text)
    new_text = pr_text

    good_providers = [
        OIVSCodeSer0501
    ]
    print('Generate new prompt')
    response = client.chat.completions.create(
        model=g4f.models.default,
        provider=random.choice(good_providers),
        messages=[{"role": "user", "content": new_text}],
        web_search=False
    )
    print(f'Selected provider: {response.provider}')
    answer = response.choices[0].message.content
    start = answer.find('{')
    end = answer.rfind('}') + 1
    if start == -1 or end <= 0:
        return False
    js_text = answer[start:end]
    try:
        data = json.loads(js_text)
    except json.JSONDecodeError:
        return False
    required_keys = {"prompts", "title", "description", "tags"}
    if not required_keys.issubset(data.keys()):
        return False
    if not isinstance(data["prompts"], list) or len(data["prompts"]) != 2:
        return False
    for item in data["prompts"]:
        if not isinstance(item, str):
            return False
    print('Prompt successfully validated')
    save_correct_prompt(data)
    return True

Промпт составлен так, чтобы нейросеть возвращала ответ в следующем формате:

{
  "prompts": [
    "текст первого промпта",
    "текст второго промпта"
  ],
  "title": "кликбейтное название для видео + 3 тега через #",
  "description": "описание видео",
  "tags": "теги через запятую без #"
}

Это позволяет легко распарсить полученный json и провести несколько тестов на валидацию, что нейросеть не прислала в ответ отсебятину. Результат сохраняется в файл для дальнейшего использования.

Автоматическая генерация Sora

После автоматизации промптов напрашивалось автоматическое управление сайтом Sora, чтобы не тратить время на ручной ввод и загрузку результатов. Автоматизация проводилась средствами Python Playwright с аддоном stelth для обхода защиты cloudflare. Код получился довольно громоздким, посмотреть его вы можете на моем github. В программу заложена отказоустойчивость на случай перегрузки сервера или блокировки генерации за нарушение правил.

Написанный модуль использует два аккаунта параллельно через куки файлы, вводит в них промпты и качает полученные видео. Таким образом он делает 4 видеоролика по 15 секунд каждые 5 минут. Возможно дальнейшее масштабирование через увеличение количества аккаунтов.

Объединение модулей

После написания всех модулей я объединил их одной общей функцией, которая генерирует 10 видеороликов на одну кнопку.

import asyncio
import os
import time
from datetime import datetime
import shutil
from constants import PATHS, VIDEO_COUNT, FOLDER_VID_COUNT
from gpt_module import create_new_prompts
from sora_module import create_sora_videos
from mounting_video import mounting_video


def generate_video():
    subt_off = False

    create_new_prompts()
    is_all_videos = False
    while not is_all_videos:
        is_all_videos = asyncio.run(create_sora_videos())
        time.sleep(5)
    cur_date = datetime.now().strftime("%d%m")
    folder_name = cur_date
    folder_path = os.path.join(PATHS.sora_videos, folder_name)
    for i in range(1, FOLDER_VID_COUNT+1):
        result_name = f'{i}_{cur_date}.mp4'
        result_file = os.path.join(PATHS.output, result_name)
        if os.path.exists(result_file):
            print(f'Видео {i} уже создано')
            continue
        videos_path = os.path.join(folder_path, str(i))
        if len(os.listdir(videos_path)) < VIDEO_COUNT:
            print(f'Пропуск видео {i}, нехватает кусков')
            continue
        for filename in os.listdir(videos_path):
            source_file = os.path.join(videos_path, filename)
            destination_file = os.path.join(PATHS.input, filename)
            shutil.copy(source_file, destination_file)
        mounting_video(result_name, subt_off)


if __name__ == '__main__':
    generate_video()

В результате всего одно нажатие приводит к созданию 10 видеороликов длительность по 30 секунд каждый, с неповторяющимися уникальными сюжетами.

Создание всех видеороликов занимает примерно 30 минут. Основное время уходит на ожидание генераций от Sora2. При желании это можно масштабировать и параллелить, используя больше аккаунтов и проксей.

Автоматическая загрузка на YouTube

Последним шагом для полной автоматизации напрашивается загрузка видеороликов на Youtube и Tiktok. Это позволит запустить систему на удаленном сервере и забыть о ней на пол года, пока аккаунты набирают аудиторию без участия человека.

В автоматизации Youtube есть две методики. Первая это использование официального Api. Есть официальный репозиторий с подобным кодом. У этого метода есть недостаток - все видео попадают в черновик, а попытка публикации через Api приводит к автоматической блокировке видео. Это можно обойти договорившись с техподдержкой, но делать это я не пробовал.

Второй метод это автоматизация средствами Selenium или Playwright, в настоящее время я ещё не написал этот модуль, но он планируется к внедрению в ближайшие недели.

Итоги

В данный момент я веду каналы на Youtube и Tiktok, загружая в них по 10 видеороликов в день. Проект существует с 5 октября, поэтому качественной статистики пока нет. Алгоритм ютуб в среднем рекомендует 3 видео из 10, обеспечивая им примерно тысячу просмотров. Некоторые видео начинают рекомендоваться после задержки в несколько недель, повышая общие просмотры. В тикток рекомендуется почти каждое видео, но средние просмотры держатся в районе 300.

Проект пока недоступен на github, каждый день в него вносятся сильные корректировки. Также меняется системный промпт для тематики видеороликов, в поисках более качественных тем для повышения просмотров.