Made on
Tilda
# tiktok_processor.py
import os, re, shutil, subprocess, hashlib, random, time
from pathlib import Path
from datetime import datetime
from moviepy.editor import VideoFileClip
from TTS.api import TTS
from pydub import AudioSegment
from PIL import Image, ImageDraw, ImageFont
from faster_whisper import WhisperModel
import torch
import asyncio
import mouse
import json
# Імпорти модулів
from telegram_bot import send_video_preview
from preview_generator import PreviewGenerator
class TikTokVideoGenerator:
def __init__(self, source_folder, temp_folder, preview_generator, tts_func, asr_model,
tiktok_resolution=(1080, 1920), target_part_duration=105, min_part_duration=63):
"""
Ініціалізація генератора TikTok відео з підтримкою прискорення
Args:
source_folder: Папка з фоновими відео
temp_folder: Папка для тимчасових файлів
preview_generator: Об'єкт PreviewGenerator для створення превью
tts_func: Функція для TTS (передаємо coqui_tts)
asr_model: Модель для розпізнавання мовлення
tiktok_resolution: Розмір TikTok відео (ширина, висота)
target_part_duration: Цільова тривалість кожної частини у секундах
min_part_duration: Мінімальна тривалість частини
"""
self.source_folder = Path(source_folder)
self.temp_folder = Path(temp_folder)
self.preview_generator = preview_generator
self.tts_func = tts_func
self.asr_model = asr_model
self.tiktok_resolution = tiktok_resolution
self.target_part_duration = target_part_duration
self.min_part_duration = min_part_duration
self.fade_duration = 0.3 # Короткі fade для TikTok
print(f"???? TikTok генератор ініціалізовано:")
print(f" • Мінімальна тривалість частини: {self.min_part_duration} секунд")
print(f" • Цільова тривалість частини: {self.target_part_duration} секунд")
print(f" • Підтримка прискорення відео")
def calculate_extended_audio_duration(self, natural_duration, speed_factor):
"""
Розраховує потрібну тривалість аудіо для досягнення цільової після прискорення
Args:
natural_duration: Природна тривалість аудіо
speed_factor: Коефіцієнт прискорення
Returns:
float: Потрібна тривалість аудіо
"""
return natural_duration * speed_factor
def extend_audio_for_speedup(self, audio, target_duration):
"""
Розширює аудіо до потрібної тривалості додаванням мовчання
Args:
audio: AudioSegment об'єкт
target_duration: Цільова тривалість у секундах
Returns:
AudioSegment: Розширений аудіо
"""
current_duration = audio.duration_seconds
if target_duration <= current_duration:
return audio
silence_needed = target_duration - current_duration
silence = AudioSegment.silent(duration=int(silence_needed * 1000))
print(f"???? Розширення аудіо з {current_duration:.1f}с до {target_duration:.1f}с (+{silence_needed:.1f}с мовчання)")
return audio + silence
def apply_video_speedup(self, input_video_path, output_video_path, speed_factor):
"""
Прискорює TikTok відео зі збереженням якості та 60 FPS
Args:
input_video_path: Шлях до вхідного відео
output_video_path: Шлях для збереження прискореного відео
speed_factor: Коефіцієнт прискорення
Returns:
Path: Шлях до прискореного відео
"""
if speed_factor <= 1.0:
shutil.copy2(input_video_path, output_video_path)
return Path(output_video_path)
print(f"???? Прискорення TikTok відео в {speed_factor:.2f}x...")
temp_video = self.temp_folder / f"speedup_tiktok_{random.randint(1000, 9999)}.mp4"
try:
cmd = [
"ffmpeg", "-y", "-i", str(input_video_path),
"-filter_complex", f"[0:v]setpts={1/speed_factor}*PTS[v];[0:a]atempo={speed_factor}[a]",
"-map", "[v]", "-map", "[a]",
"-c:v", "h264_nvenc", "-preset", "p4", "-b:v", "8M", "-r", "60",
"-c:a", "aac", "-b:a", "128k",
"-movflags", "+faststart",
str(temp_video)
]
subprocess.run(cmd, check=True, capture_output=True, text=True)
shutil.move(str(temp_video), str(output_video_path))
print(f"✅ TikTok відео прискорено успішно")
return Path(output_video_path)
except subprocess.CalledProcessError as e:
print(f"❌ Помилка при прискоренні TikTok відео: {e}")
shutil.copy2(input_video_path, output_video_path)
return Path(output_video_path)
finally:
if temp_video.exists():
temp_video.unlink()
def split_story_into_parts(self, story_text, preview_text):
"""
Розділяє історію на частини оптимальної тривалості з урахуванням прискорення
Args:
story_text: Повний текст історії
preview_text: Текст preview (тільки для part1)
Returns:
list: Список частин з текстом і номером
"""
print(f"???? Розділення історії на TikTok частини...")
# Розділяємо історію на речення (покращена регулярка)
sentence_pattern = r'(?<=[.!?])\s+'
sentences = re.split(sentence_pattern, story_text.strip())
# Очищаємо та фільтруємо речення
clean_sentences = []
for sentence in sentences:
sentence = sentence.strip()
if sentence and len(sentence) > 5: # Мінімум 5 символів
# Додаємо крапку в кінці якщо немає розділового знаку
if not sentence[-1] in '.!?':
sentence += '.'
clean_sentences.append(sentence)
if not clean_sentences:
# Якщо не знайшли речення, розділяємо по параграфах
paragraphs = story_text.split('\n\n')
clean_sentences = [p.strip() for p in paragraphs if p.strip()]
if not clean_sentences:
# Останній варіант - розділити на блоки по словах
words = story_text.split()
clean_sentences = [' '.join(words[i:i+15]) + '.' for i in range(0, len(words), 15)]
print(f"???? Знайдено {len(clean_sentences)} речень для розділення")
# Розраховуємо тривалість preview (тільки для part1)
preview_audio = self.tts_func(preview_text)
preview_duration = preview_audio.duration_seconds + 0.5 # preview + пауза
# Формуємо частини з перевіркою мінімальної тривалості
parts = []
current_sentences = []
current_duration = 0
part_number = 1
for i, sentence in enumerate(clean_sentences):
# Оцінюємо тривалість речення
words_count = len(sentence.split())
chars_count = len(sentence)
sentence_duration = max(words_count / 2.0, chars_count / 10.0)
# Додаємо речення до поточної частини
current_sentences.append(sentence)
current_duration += sentence_duration
# Для part1 враховуємо preview
base_duration = preview_duration if part_number == 1 else 0
total_duration = current_duration + base_duration
# Перевіряємо умови для завершення частини
is_last_sentence = (i == len(clean_sentences) - 1)
exceeds_target = total_duration >= self.target_part_duration
meets_minimum = total_duration >= self.min_part_duration
if (exceeds_target and meets_minimum) or is_last_sentence:
part_text = ' '.join(current_sentences)
parts.append({
'number': part_number,
'text': part_text,
'preview_text': preview_text if part_number == 1 else None,
'estimated_duration': total_duration,
'has_preview': part_number == 1,
'sentence_count': len(current_sentences)
})
duration_status = "✅" if total_duration >= self.min_part_duration else "⚠️"
print(f" {duration_status} Частина {part_number}: {len(current_sentences)} речень, {total_duration:.1f}с")
# Починаємо нову частину
part_number += 1
current_sentences = []
current_duration = 0
print(f"✅ Історія розділена на {len(parts)} частин:")
for part in parts:
words_count = len(part['text'].split())
preview_marker = " (з preview)" if part['has_preview'] else ""
duration_status = "✅" if part['estimated_duration'] >= self.min_part_duration else "⚠️"
print(f" {duration_status} Частина {part['number']}: {words_count} слів, ~{part['estimated_duration']:.1f} сек{preview_marker}")
return parts
def merge_short_parts(self, parts):
"""Об'єднує тільки дуже короткі частини (менше 30 секунд)"""
print(f"???? Об'єднання критично коротких частин (менше 30с)...")
merged_parts = []
i = 0
critical_min_duration = 30
while i < len(parts):
current_part = parts[i].copy()
original_number = current_part['number']
if current_part['estimated_duration'] < critical_min_duration and i + 1 < len(parts):
next_part = parts[i + 1]
# Об'єднуємо тексти
current_part['text'] += ' ' + next_part['text']
# Перерахуємо тривалість
words_count = len(current_part['text'].split())
chars_count = len(current_part['text'])
new_duration = max(words_count / 2.0, chars_count / 10.0)
if current_part['has_preview']:
preview_audio = self.tts_func(current_part['preview_text'])
preview_duration = preview_audio.duration_seconds + 0.5
new_duration += preview_duration
current_part['estimated_duration'] = new_duration
current_part['sentence_count'] = current_part.get('sentence_count', 1) + next_part.get('sentence_count', 1)
if not current_part['has_preview'] and next_part['has_preview']:
current_part['preview_text'] = next_part['preview_text']
current_part['has_preview'] = True
i += 1
print(f" ???? Об'єднано критично короткі частини {original_number} і {next_part['number']} → {new_duration:.1f}с")
current_part['number'] = len(merged_parts) + 1
merged_parts.append(current_part)
i += 1
print(f"✅ Після об'єднання: {len(merged_parts)} частин")
return merged_parts
def create_background_segments_tiktok(self, story_id, part_number, duration):
"""Створює фонові відео сегменти для TikTok частини з 60 FPS"""
bg_videos = list(self.source_folder.glob("*.mp4"))
random.shuffle(bg_videos)
bg_iter = iter(bg_videos)
clips = []
duration_collected = 0
while duration_collected < duration:
try:
bg_video = next(bg_iter)
except StopIteration:
bg_videos_reset = list(self.source_folder.glob("*.mp4"))
random.shuffle(bg_videos_reset)
bg_iter = iter(bg_videos_reset)
bg_video = next(bg_iter)
clip = VideoFileClip(str(bg_video))
remaining_duration = duration - duration_collected
use_duration = min(clip.duration, remaining_duration)
clip.close()
if use_duration <= 0:
break
clip_output = self.temp_folder / f"{story_id}_part{part_number}_bg_{len(clips):02d}.mp4"
# Базові фільтри для вертикального відео з 60 FPS
base_filters = f"scale='max({self.tiktok_resolution[0]},iw*{self.tiktok_resolution[1]}/ih)':'max({self.tiktok_resolution[1]},ih*{self.tiktok_resolution[0]}/iw)',crop={self.tiktok_resolution[0]}:{self.tiktok_resolution[1]},fps=60"
# Додаємо fade ефекти
fade_filters = self.apply_safe_fade_filters(use_duration, base_filters)
subprocess.run([
"ffmpeg", "-y", "-i", str(bg_video), "-ss", "0", "-t", str(use_duration),
"-vf", fade_filters,
"-an", "-c:v", "h264_nvenc", "-preset", "p4", "-b:v", "8M", "-r", "60", str(clip_output)
], check=True)
clips.append(clip_output)
duration_collected += use_duration
return clips
def apply_safe_fade_filters(self, use_dur, base_filters):
"""Застосовує fade фільтри безпечно"""
fade_in_duration = min(self.fade_duration, use_dur * 0.3)
fade_out_duration = min(self.fade_duration, use_dur * 0.3)
if fade_in_duration + fade_out_duration > use_dur:
fade_in_duration = use_dur * 0.4
fade_out_duration = use_dur * 0.4
fade_out_start = max(0, use_dur - fade_out_duration)
fade_filters = []
if fade_in_duration > 0.05:
fade_filters.append(f"fade=t=in:st=0:d={fade_in_duration}")
if fade_out_duration > 0.05 and fade_out_start > fade_in_duration:
fade_filters.append(f"fade=t=out:st={fade_out_start}:d={fade_out_duration}")
all_filters = [base_filters] + fade_filters
return ",".join(all_filters)
def create_preview_with_overlay_tiktok(self, story_id, part_number, preview_clips, horizontal_preview_path, preview_duration):
"""Створює preview відео з накладеним вертикальним preview зображенням для TikTok з 60 FPS"""
# Об'єднуємо всі preview кліпи
if len(preview_clips) > 1:
preview_concat_list = self.temp_folder / f"{story_id}_part{part_number}_preview_concat.txt"
with open(preview_concat_list, "w", encoding="utf-8") as f:
for clip in preview_clips:
f.write(f"file '{clip.as_posix()}'\n")
preview_combined = self.temp_folder / f"{story_id}_part{part_number}_preview_combined.mp4"
subprocess.run([
"ffmpeg", "-y", "-f", "concat", "-safe", "0", "-i", str(preview_concat_list),
"-c:v", "h264_nvenc", "-b:v", "8M", "-preset", "p4", "-r", "60", str(preview_combined)
], check=True)
else:
preview_combined = preview_clips[0]
# Створюємо вертикальне preview зображення
preview_img_path = self.temp_folder / f"{story_id}_part{part_number}_preview_vertical.png"
self.preview_generator.draw_vertical_preview(horizontal_preview_path, preview_img_path)
# Накладаємо preview зображення на відео
preview_video = self.temp_folder / f"{story_id}_part{part_number}_preview_video.mp4"
subprocess.run([
"ffmpeg", "-y", "-i", str(preview_combined), "-i", str(preview_img_path),
"-filter_complex",
"[1:v]format=rgba[overlay];[0:v][overlay]overlay=(W-w)/2:(H-h)/2:format=auto",
"-t", str(preview_duration), "-c:v", "h264_nvenc", "-preset", "p4", "-b:v", "8M", "-r", "60", str(preview_video)
], check=True)
return preview_video
def combine_tiktok_segments(self, story_id, part_number, preview_video, story_clips):
"""Об'єднує всі сегменти TikTok відео з 60 FPS"""
if preview_video:
used_clips = [preview_video] + story_clips
else:
used_clips = story_clips
concat_list = self.temp_folder / f"{story_id}_part{part_number}_concat.txt"
with open(concat_list, "w", encoding="utf-8") as f:
for clip in used_clips:
f.write(f"file '{clip.as_posix()}'\n")
combined_path = self.temp_folder / f"{story_id}_part{part_number}_combined.mp4"
subprocess.run([
"ffmpeg", "-y", "-f", "concat", "-safe", "0", "-i", str(concat_list),
"-c:v", "h264_nvenc", "-b:v", "8M", "-preset", "p4", "-r", "60", str(combined_path)
], check=True)
return combined_path
def create_tiktok_part(self, story_part, story_id, horizontal_preview_path, selected_template, speed_settings=None):
"""
Створює одну частину TikTok відео з підтримкою прискорення
Args:
story_part: Словник з інформацією про частину
story_id: ID історії
horizontal_preview_path: Шлях до горизонтального preview
selected_template: Вибраний template
speed_settings: Налаштування прискорення {'enabled': bool, 'speed': float, 'target_duration': float}
Returns:
tuple: (video_path, audio_path, total_duration, is_accelerated)
"""
part_number = story_part['number']
story_text = story_part['text']
has_preview = story_part['has_preview']
print(f"???? Створення TikTok частини {part_number}...")
if speed_settings and speed_settings.get('enabled'):
print(f"???? З прискоренням: {speed_settings}")
print(f"???? Текст частини ({len(story_text)} символів): {story_text[:100]}...")
# Генеруємо базове аудіо для цієї частини
if has_preview:
# Тільки для part1 додаємо preview
preview_text = story_part['preview_text']
audio_preview = self.tts_func(preview_text)
audio_story = self.tts_func(story_text)
final_audio = audio_preview + AudioSegment.silent(duration=500) + audio_story
preview_duration = audio_preview.duration_seconds
else:
# Для інших частин тільки story
audio_story = self.tts_func(story_text)
final_audio = audio_story
preview_duration = 0
# Розраховуємо тривалості
story_duration = audio_story.duration_seconds
natural_duration = final_audio.duration_seconds
# Обробляємо прискорення
is_accelerated = False
speed_factor = 1.0
if speed_settings and speed_settings.get('enabled'):
if speed_settings.get('target_duration'):
# Режим цільової тривалості
target_duration = speed_settings['target_duration']
speed_factor = natural_duration / target_duration
speed_factor = min(speed_factor, 4.0) # Максимальне прискорення
elif speed_settings.get('speed'):
# Режим фіксованого прискорення
speed_factor = speed_settings['speed']
if speed_factor > 1.0:
# Розширюємо аудіо для прискорення
extended_duration = self.calculate_extended_audio_duration(natural_duration, speed_factor)
final_audio = self.extend_audio_for_speedup(final_audio, extended_duration)
is_accelerated = True
print(f"⚡ Підготовка до прискорення {speed_factor:.2f}x")
audio_path = self.temp_folder / f"{story_id}_part{part_number}_voice.wav"
final_audio.export(audio_path, format="wav")
total_duration = final_audio.duration_seconds
print(f"⏱️ Тривалість частини {part_number}: {total_duration:.1f}с")
# Створюємо відео сегменти
preview_video = None
if has_preview:
# Створюємо preview сегменти тільки для part1
preview_clips = self.create_background_segments_tiktok(story_id, part_number, preview_duration)
preview_video = self.create_preview_with_overlay_tiktok(
story_id, part_number, preview_clips, horizontal_preview_path, preview_duration
)
# Створюємо story сегменти
if is_accelerated:
# Якщо буде прискорення, створюємо відео для повної тривалості
story_video_duration = total_duration - preview_duration
else:
# Інакше використовуємо природну тривалість
story_video_duration = story_duration
story_clips = self.create_background_segments_tiktok(story_id, f"{part_number}_story", story_video_duration)
# Об'єднуємо сегменти
combined_video = self.combine_tiktok_segments(story_id, part_number, preview_video, story_clips)
return combined_video, audio_path, total_duration, is_accelerated, speed_factor
def apply_subtitles_ass_tiktok(self, video_path, chunks, story_id, part_number):
"""Додає субтитри для TikTok відео використовуючи ASS файл"""
if not chunks:
return video_path
print(f"???? Створення ASS субтитрів для частини {part_number}...")
# Створюємо ASS файл
ass_path = self.temp_folder / f"{story_id}_part{part_number}_subtitles.ass"
self.create_ass_subtitles_tiktok(chunks, ass_path)
# Застосовуємо субтитри з 60 FPS
output_video = self.temp_folder / f"{story_id}_part{part_number}_with_subtitles.mp4"
# Escape шлях для Windows
ass_path_escaped = str(ass_path).replace("\\", "/").replace(":", "\\:")
cmd = [
"ffmpeg", "-y",
"-i", str(video_path),
"-vf", f"ass='{ass_path_escaped}'",
"-c:v", "h264_nvenc", "-preset", "p4", "-b:v", "8M", "-r", "60",
"-c:a", "copy",
str(output_video)
]
try:
subprocess.run(cmd, check=True, capture_output=True, text=True)
print(f" ✅ Субтитри для частини {part_number} успішно додано")
return output_video
except subprocess.CalledProcessError as e:
print(f"❌ Помилка при додаванні субтитрів для частини {part_number}: {e}")
return video_path
def create_ass_subtitles_tiktok(self, chunks, ass_path):
"""Створює ASS файл субтитрів для TikTok з оптимізованими налаштуваннями"""
font_name = "Komika Axis"
font_size = 28 # Менший шрифт для TikTok
margin_v = 250 # Вертикальні відступи
margin_lr = 100 # Горизонтальні відступи
# ASS заголовок з налаштуваннями стилю
ass_content = f"""[Script Info]
Title: TikTok Subtitles
ScriptType: v4.00+
[V4+ Styles]
Format: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour, Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle, BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding
Style: Default,{font_name},{font_size},&H00FFFFFF,&H000000FF,&H00000000,&H80000000,1,0,0,0,100,100,0,0,1,3,0,5,{margin_lr},{margin_lr},{margin_v},1
[Events]
Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text
"""
# Додаємо субтитри
for chunk in chunks:
start_time = self.format_ass_time(chunk["start"])
end_time = self.format_ass_time(chunk["end"])
# Екрануємо спеціальні символи ASS
text = chunk["text"].replace("\\", "\\\\").replace("{", "\\{").replace("}", "\\}")
# Розбиваємо довгий текст на коротші рядки
words = text.split()
lines = []
current_line = ""
max_chars = 20 # Максимум символів на рядок
for word in words:
# Перевіряємо чи довге слово поміститься окремо
if len(word) > max_chars:
if current_line:
lines.append(current_line.strip())
current_line = ""
# Розбиваємо довге слово
while len(word) > max_chars:
lines.append(word[:max_chars-1] + "-")
word = word[max_chars-1:]
if word:
current_line = word
elif len(current_line + " " + word) <= max_chars:
current_line += (" " + word if current_line else word)
else:
if current_line:
lines.append(current_line.strip())
current_line = word
if current_line:
lines.append(current_line.strip())
text = "\\N".join(lines) # \\N - перенос рядка в ASS
ass_content += f"Dialogue: 0,{start_time},{end_time},Default,,0,0,0,,{text}\n"
# Записуємо файл
with open(ass_path, 'w', encoding='utf-8-sig') as f:
f.write(ass_content)
def format_ass_time(self, seconds):
"""Конвертує секунди в формат часу ASS (H:MM:SS.CC)"""
hours = int(seconds // 3600)
minutes = int((seconds % 3600) // 60)
secs = seconds % 60
return f"{hours}:{minutes:02d}:{secs:05.2f}"
def chunk_words(self, words, n=2):
"""Групує слова для субтитрів"""
chunks = []
i = 0
while i < len(words):
group = words[i:i+n]
text = " ".join([w["word"] for w in group])
start = group[0]["start"]
end = group[-1]["end"]
chunks.append({"text": text, "start": start, "end": end})
i += n
return chunks
def finalize_tiktok_video(self, combined_video, audio_path, output_path, apply_speedup=False, speed_factor=1.0):
"""
Фінальна збірка TikTok відео з аудіо, 60 FPS та прискоренням
Args:
combined_video: Шлях до відео без аудіо
audio_path: Шлях до аудіо файлу
output_path: Шлях для збереження фінального відео
apply_speedup: Чи застосовувати прискорення
speed_factor: Коефіцієнт прискорення
"""
temp_output = self.temp_folder / f"temp_final_{random.randint(1000, 9999)}.mp4"
print(f"???? Фінальна збірка TikTok відео з 60 FPS...")
# Спочатку об'єднуємо відео з аудіо
subprocess.run([
"ffmpeg", "-y", "-i", str(combined_video), "-i", str(audio_path),
"-c:v", "h264_nvenc", "-preset", "p4", "-b:v", "8M", "-r", "60",
"-c:a", "aac", "-shortest", "-movflags", "+faststart",
str(temp_output)
], check=True)
# Потім застосовуємо прискорення якщо потрібно
if apply_speedup and speed_factor > 1.0:
self.apply_video_speedup(temp_output, output_path, speed_factor)
temp_output.unlink() # Видаляємо тимчасовий файл
else:
shutil.move(str(temp_output), str(output_path))
def safe_filename(name):
"""Створює безпечну назву файлу"""
name = re.sub(r'[<>:"/\\|?*\x00-\x1F]', "_", name)
name = re.sub(r'\s+','_', name).strip("_")
name = name.rstrip(". ")
return name[:50]
def detect_censored_words(text):
"""Знаходить слова з зірочками для цензури"""
pattern = r'\b\w*\*+\w*\b'
return re.findall(pattern, text)
def split_text_with_censoring(text):
"""Розділяє текст на частини, виділяючи цензуровані слова"""
parts = []
pattern = r'(\b\w*\*+\w*\b)'
split_parts = re.split(pattern, text)
for part in split_parts:
if part.strip():
if re.match(r'\b\w*\*+\w*\b', part):
parts.append({"text": part, "is_censored": True})
else:
parts.append({"text": part, "is_censored": False})
return parts
def generate_beep_sound(beep_sound_path):
"""Генерує звук запікування якщо файл не існує"""
if not beep_sound_path.exists():
print("???? Генерація звуку запікування...")
beep = AudioSegment.sine(1000, duration=800)
beep = beep - 10
beep.export(beep_sound_path, format="wav")
print("✅ Звук запікування створено")
def create_tts_with_censoring(tts, temp_folder, beep_sound_path, voice):
"""Створює функцію TTS з підтримкою цензури"""
def coqui_tts_simple(text: str) -> AudioSegment:
"""Простий TTS без цензури"""
raw_path = temp_folder / f"raw_{hashlib.md5(text.encode()).hexdigest()}.wav"
clean_path = temp_folder / f"audio_{hashlib.md5(text.encode()).hexdigest()}.wav"
if clean_path.exists():
return AudioSegment.from_wav(clean_path)
tts.tts_to_file(text=text, speaker=voice, file_path=raw_path)
subprocess.run(["ffmpeg", "-y", "-i", str(raw_path), "-af", "afftdn,volume=3dB", str(clean_path)], check=True)
return AudioSegment.from_wav(clean_path)
def coqui_tts_with_censoring(text: str) -> AudioSegment:
"""TTS з підтримкою цензури"""
text_hash = hashlib.md5(text.encode()).hexdigest()
final_audio_path = temp_folder / f"final_audio_{text_hash}.wav"
if final_audio_path.exists():
return AudioSegment.from_wav(final_audio_path)
censored_words = detect_censored_words(text)
if not censored_words:
return coqui_tts_simple(text)
print(f"???? Знайдено цензуровані слова: {censored_words}")
generate_beep_sound(beep_sound_path)
beep_audio = AudioSegment.from_wav(beep_sound_path)
text_parts = split_text_with_censoring(text)
final_audio = AudioSegment.empty()
for part in text_parts:
if part["is_censored"]:
print(f" ???? Цензура: {part['text']} → *БІП*")
final_audio += beep_audio
else:
text_content = part["text"].strip()
if text_content:
part_audio = coqui_tts_simple(text_content)
final_audio += part_audio
if len(final_audio) == 0:
print("⚠️ Порожнє аудіо після цензури, використовуємо оригінальний TTS")
return coqui_tts_simple(text)
final_audio.export(final_audio_path, format="wav")
return final_audio
return coqui_tts_with_censoring
def get_interactive_speed_settings_tiktok(part_number, original_duration):
"""
Запитує у користувача налаштування прискорення для конкретної TikTok частини
Args:
part_number: Номер частини
original_duration: Оригінальна тривалість частини
Returns:
dict: Налаштування прискорення
"""
print(f"\n????️ Налаштування для TikTok частини {part_number}")
print(f"⏱️ Поточна тривалість: {original_duration:.1f} секунд")
choice = input("Прискорювати цю частину? (y/n): ").lower()
if choice != 'y':
return {'enabled': False, 'speed': 1.0, 'target_duration': None}
print("Виберіть спосіб:")
print("1. Фіксоване прискорення (наприклад, 1.5x)")
print("2. До цільової тривалості")
while True:
method = input("Ваш вибір (1-2): ").strip()
if method in ["1", "2"]:
break
print("❌ Введіть 1 або 2")
if method == "1":
while True:
try:
speed = float(input("Коефіцієнт прискорення (1.0-4.0): "))
if 1.0 <= speed <= 4.0:
break
print("❌ Коефіцієнт має бути від 1.0 до 4.0")
except ValueError:
print("❌ Введіть дійсне число!")
final_duration = original_duration / speed
print(f"✅ Швидкість {speed}x, фінальна тривалість: {final_duration:.1f}с")
return {'enabled': True, 'speed': speed, 'target_duration': None}
else:
while True:
try:
target = float(input("Цільова тривалість (15-300с): "))
if 15 <= target <= 300:
break
print("❌ Тривалість має бути від 15 до 300 секунд")
except ValueError:
print("❌ Введіть дійсне число!")
required_speed = original_duration / target
if required_speed > 4.0:
print(f"⚠️ Потрібне прискорення {required_speed:.1f}x перевищує максимум 4.0x")
required_speed = 4.0
final_duration = original_duration / required_speed
print(f"Буде використано максимальне прискорення 4.0x, тривалість: {final_duration:.1f}с")
else:
print(f"✅ Прискорення {required_speed:.1f}x до {target}с")
return {'enabled': True, 'speed': required_speed, 'target_duration': target}
def select_template_for_video(template_mode, default_template, preview_generator):
"""Вибирає template для поточного відео залежно від режиму"""
if template_mode == "fixed":
return default_template
elif template_mode == "interactive":
template_name, _ = preview_generator.select_template()
return template_name
else: # auto
return default_template
def get_tiktok_duration_settings():
"""
Дозволяє користувачу налаштувати мінімальну довжину TikTok відео
Returns:
tuple: (min_duration, target_duration)
"""
print("\n" + "="*60)
print("???? НАЛАШТУВАННЯ ТРИВАЛОСТІ TIKTOK ВІДЕО")
print("="*60)
print("Налаштуйте мінімальну тривалість частин TikTok відео:")
print("• Рекомендовано: 30-60 секунд")
print("• Мінімум: 15 секунд")
print("• Максимум: 300 секунд (5 хвилин)")
print("• ✨ ПІДТРИМКА ПРИСКОРЕННЯ: Можна створювати довші відео та прискорювати!")
print("="*60)
while True:
try:
min_duration_input = input("Введіть мінімальну тривалість частини (секунди, за замовчуванням 30): ").strip()
if not min_duration_input:
min_duration = 30
break
min_duration = int(min_duration_input)
if min_duration < 15:
print("❌ Мінімальна тривалість не може бути менше 15 секунд!")
continue
elif min_duration > 300:
print("❌ Мінімальна тривалість не може бути більше 300 секунд!")
continue
else:
break
except ValueError:
print("❌ Введіть число!")
# Автоматично розраховуємо цільову тривалість
target_duration = min(min_duration + 40, min_duration * 1.5)
target_duration = max(target_duration, min_duration + 10) # Мінімум +10 секунд
print(f"✅ Налаштування TikTok відео:")
print(f" • Мінімальна тривалість частини: {min_duration} секунд")
print(f" • Цільова тривалість частини: {target_duration:.0f} секунд")
print(f" • ???? Прискорення доступне для всіх частин!")
print(f" • Частини коротші за 30с будуть об'єднані")
return min_duration, int(target_duration)
def process_tiktok_story(story_path, tiktok_generator, horizontal_generator, selected_template, tts_func, asr_model, speed_config):
"""
Обробляє одну TikTok історію з підтримкою прискорення
Args:
story_path: Шлях до файлу з історією
tiktok_generator: Генератор TikTok відео
horizontal_generator: Генератор горизонтальних відео (для preview)
selected_template: Вибраний template
tts_func: Функція TTS
asr_model: Модель ASR
speed_config: Конфігурація прискорення
"""
# Читаємо файли
story_txt = story_path.read_text(encoding="utf-8").strip()
preview_path = story_path.with_name(f"{story_path.stem}_preview.txt")
description_path = story_path.with_name(f"{story_path.stem}_description.txt")
if not preview_path.exists() or not description_path.exists():
print(f"❌ Не знайдено preview або description для {story_path.name}")
return
preview_txt = preview_path.read_text(encoding="utf-8").strip()
description_txt = description_path.read_text(encoding="utf-8").strip()
if not story_txt or not preview_txt or not description_txt:
print(f"❌ Порожні файли для {story_path.name}")
return
story_id = hashlib.md5(story_txt.encode()).hexdigest()
video_title = safe_filename(preview_txt)
# Створюємо папку для результатів
output_dir = Path("K:/test_ffmpeg_drama/final_tiktok_videos") / video_title
output_dir.mkdir(parents=True, exist_ok=True)
print(f"???? Обробка TikTok історії: {story_path.name}")
print(f"???? Папка результатів: {output_dir}")
# Показуємо налаштування прискорення
if speed_config.get('mode') == '1':
print(f"⚡ Режим прискорення: Без прискорення")
elif speed_config.get('mode') == '2':
tiktok_settings = speed_config.get('tiktok', {})
if tiktok_settings.get('enabled'):
print(f"⚡ Режим прискорення: Фіксоване {tiktok_settings.get('speed')}x")
elif speed_config.get('mode') == '3':
tiktok_settings = speed_config.get('tiktok', {})
if tiktok_settings.get('enabled'):
print(f"⚡ Режим прискорення: До {tiktok_settings.get('target_duration')}с")
elif speed_config.get('mode') == '4':
print(f"⚡ Режим прискорення: Інтерактивний")
# Створюємо горизонтальне preview (потрібне для вертикального)
print(f"????️ Створення превью...")
audio_preview = tts_func(preview_txt)
# Створюємо preview зображення
horizontal_preview_path = tiktok_generator.temp_folder / f"{story_id}_tiktok_preview.png"
tiktok_generator.preview_generator.draw_horizontal_preview(
preview_txt, horizontal_preview_path, selected_template
)
# Розділяємо історію на частини
story_parts = tiktok_generator.split_story_into_parts(story_txt, preview_txt)
if not story_parts:
print(f"❌ Не вдалося розділити історію на частини")
return
# Об'єднуємо критично короткі частини
print(f"???? Перевірка критично коротких частин (менше 30 секунд)...")
very_short_parts = [p for p in story_parts if p['estimated_duration'] < 30]
if very_short_parts:
print(f"⚠️ Знайдено {len(very_short_parts)} критично коротких частин, об'єднуємо...")
story_parts = tiktok_generator.merge_short_parts(story_parts)
else:
print(f"✅ Всі частини довші за 30 секунд")
# Створюємо кожну частину
created_parts = []
for part in story_parts:
try:
print(f"\n???? Створення частини {part['number']}/{len(story_parts)}...")
# Отримуємо налаштування прискорення для цієї частини
part_speed_settings = None
if speed_config.get('mode') == '4': # Інтерактивний режим
part_speed_settings = get_interactive_speed_settings_tiktok(
part['number'], part['estimated_duration']
)
else:
# Використовуємо глобальні налаштування
part_speed_settings = speed_config.get('tiktok', {'enabled': False})
# Створюємо TikTok відео для цієї частини
combined_video, audio_path, total_duration, is_accelerated, speed_factor = tiktok_generator.create_tiktok_part(
part, story_id, horizontal_preview_path, selected_template, part_speed_settings
)
# Розпізнавання мовлення для субтитрів
print(f"???? Розпізнавання мовлення для частини {part['number']}...")
segments, _ = asr_model.transcribe(str(audio_path), word_timestamps=True)
# Створюємо субтитри
if part['has_preview']:
# Для part1 субтитри тільки для story частини (після preview та паузи)
story_start_time = audio_preview.duration_seconds + 0.5
story_words = []
for seg in segments:
for w in seg.words:
if w.start >= story_start_time:
story_words.append({"word": w.word.strip(), "start": w.start, "end": w.end})
else:
# Для інших частин субтитри для всього тексту
story_words = []
for seg in segments:
for w in seg.words:
story_words.append({"word": w.word.strip(), "start": w.start, "end": w.end})
chunks = tiktok_generator.chunk_words(story_words)
if chunks:
final_video = tiktok_generator.apply_subtitles_ass_tiktok(
combined_video, chunks, story_id, part['number']
)
else:
print(f"⚠️ Субтитри не знайдено для частини {part['number']}")
final_video = combined_video
# Фінальна збірка частини з прискоренням
part_filename = f"part{part['number']}.mp4"
part_output_path = output_dir / part_filename
tiktok_generator.finalize_tiktok_video(
final_video, audio_path, part_output_path,
apply_speedup=is_accelerated, speed_factor=speed_factor
)
# Розраховуємо фінальну тривалість після прискорення
if is_accelerated:
final_duration = total_duration / speed_factor
else:
final_duration = total_duration
created_parts.append({
'number': part['number'],
'path': part_output_path,
'original_duration': total_duration,
'final_duration': final_duration,
'speed_factor': speed_factor,
'is_accelerated': is_accelerated
})
print(f"✅ Частина {part['number']} створена: {part_output_path}")
if is_accelerated:
print(f"⏱️ Тривалість: {total_duration:.1f}с → {final_duration:.1f}с (прискорено в {speed_factor:.2f}x)")
else:
print(f"⏱️ Тривалість: {final_duration:.1f}с (без прискорення)")
except Exception as e:
print(f"❌ Помилка при створенні частини {part['number']}: {e}")
import traceback
traceback.print_exc()
# Зберігаємо допоміжні файли
if created_parts:
# Зберігаємо оригінальне preview
Image.open(horizontal_preview_path).convert("RGB").save(output_dir / "preview.jpg", "JPEG")
# Зберігаємо опис
(output_dir / "description.txt").write_text(description_txt, encoding="utf-8")
# Створюємо інформаційний файл про частини з інформацією про прискорення
parts_info = {
'total_parts': len(created_parts),
'story_title': video_title,
'min_duration_setting': tiktok_generator.min_part_duration,
'target_duration_setting': tiktok_generator.target_part_duration,
'speedup_enabled': any(part['is_accelerated'] for part in created_parts),
'parts': [
{
'number': part['number'],
'filename': f"part{part['number']}.mp4",
'original_duration': part['original_duration'],
'final_duration': part['final_duration'],
'speed_factor': part['speed_factor'],
'is_accelerated': part['is_accelerated']
}
for part in created_parts
]
}
(output_dir / "parts_info.json").write_text(
json.dumps(parts_info, indent=2, ensure_ascii=False),
encoding="utf-8"
)
# Очищуємо оригінальні файли
for f in [story_path, preview_path, description_path]:
f.write_text("", encoding="utf-8")
print(f"✅ TikTok історія завершена: {len(created_parts)} частин створено")
print(f"???? Результати збережено в: {output_dir}")
# Статистика прискорення
accelerated_count = sum(1 for part in created_parts if part['is_accelerated'])
if accelerated_count > 0:
print(f"???? Статистика прискорення:")
print(f" • Частин прискорено: {accelerated_count}/{len(created_parts)}")
total_original = sum(part['original_duration'] for part in created_parts)
total_final = sum(part['final_duration'] for part in created_parts)
overall_ratio = total_original / total_final if total_final > 0 else 1.0
print(f" • Загальне прискорення: {overall_ratio:.2f}x")
print(f" • Загальна тривалість: {total_original:.1f}с → {total_final:.1f}с")
# Надсилаємо превью в Telegram (тільки для першої частини)
try:
print(f"???? Надсилання превью в Telegram...")
preview_jpg_path = output_dir / "preview.jpg"
asyncio.run(send_video_preview(preview_txt, str(preview_jpg_path)))
print(f"✅ Превью успішно надіслано в Telegram")
except Exception as e:
print(f"❌ Помилка при надсиланні превью: {e}")
return created_parts
else:
print(f"❌ Жодної частини не створено для {story_path.name}")
return []
def process_tiktok_videos(tiktok_stories_dir, tiktok_generator, horizontal_generator, default_template, template_mode, preview_generator, tiktok_tts, asr_model, speed_config):
"""Обробляє TikTok відео з підтримкою прискорення"""
story_files = list(tiktok_stories_dir.glob("*.txt"))
story_files = [f for f in story_files if not f.name.endswith("_preview.txt") and not f.name.endswith("_description.txt")]
if not story_files:
print(f"❌ TikTok історій не знайдено в {tiktok_stories_dir}")
print("???? Переконайтесь що файли мають формат:")
print(" - story1.txt")
print(" - story1_preview.txt")
print(" - story1_description.txt")
return 0
print(f"???? Знайдено {len(story_files)} TikTok історій для обробки")
print(f"???? З підтримкою прискорення відео!")
print(f"⚙️ Налаштування: мінімум {tiktok_generator.min_part_duration}с, ціль {tiktok_generator.target_part_duration}с")
processed_count = 0
total_accelerated_parts = 0
for i, story in enumerate(story_files, 1):
print(f"\n{'='*60}")
print(f"???? [{i}/{len(story_files)}] Обробка TikTok історії → {story.name}")
print(f"???? З підтримкою прискорення")
print(f"{'='*60}")
try:
# Вибираємо template для цього відео
selected_template = select_template_for_video(template_mode, default_template, preview_generator)
if selected_template is None:
print("❌ Template не вибрано, пропускаємо відео")
continue
parts_created = process_tiktok_story(
story, tiktok_generator, horizontal_generator,
selected_template, tiktok_tts, asr_model, speed_config
)
if parts_created:
print(f"???? Історія {story.name} успішно оброблена!")
print(f"???? Створено частин: {len(parts_created)}")
# Статистика тривалості та прискорення
accelerated_in_story = [part for part in parts_created if part['is_accelerated']]
if accelerated_in_story:
print(f"???? Частин прискорено: {len(accelerated_in_story)}")
for part in accelerated_in_story:
print(f" • Частина {part['number']}: {part['original_duration']:.1f}с → {part['final_duration']:.1f}с ({part['speed_factor']:.2f}x)")
total_accelerated_parts += len(accelerated_in_story)
processed_count += 1
# Рух миші для запобігання сну комп'ютера
time.sleep(2)
mouse.move(random.randint(0, 1920), random.randint(0, 1080), duration=1)
time.sleep(2)
else:
print(f"❌ Не вдалося створити жодної частини для {story.name}")
except Exception as e:
print(f"❌ Критична помилка з {story.name}: {e}")
import traceback
traceback.print_exc()
# Загальна статистика
print(f"\n???? TikTok обробка завершена!")
print(f"???? Оброблено історій: {processed_count}")
if total_accelerated_parts > 0:
print(f"???? Всього частин прискорено: {total_accelerated_parts}")
return processed_count