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 відео з підтримкою прискорення
"""
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
print(f"???? TikTok генератор ініціалізовано:")
print(f" • Мінімальна тривалість частини: {self.min_part_duration} секунд")
print(f" • Цільова тривалість частини: {self.target_part_duration} секунд")
print(f" • Підтримка прискорення відео")
def trim_silence_from_audio(self, audio_segment, silence_threshold_db=-40, chunk_size_ms=100):
"""Видаляє мовчання з початку та кінця аудіо"""
if len(audio_segment) == 0:
return audio_segment
# Видаляємо мовчання з початку
start_trim = 0
for i in range(0, len(audio_segment), chunk_size_ms):
chunk = audio_segment[i:i + chunk_size_ms]
if chunk.dBFS > silence_threshold_db:
start_trim = max(0, i - 200) # Залишаємо 200ms перед першим звуком
break
# Видаляємо мовчання з кінця
end_trim = len(audio_segment)
for i in range(len(audio_segment) - chunk_size_ms, 0, -chunk_size_ms):
if i < 0:
break
chunk = audio_segment[i:i + chunk_size_ms]
if chunk.dBFS > silence_threshold_db:
end_trim = min(len(audio_segment), i + chunk_size_ms + 200) # Залишаємо 200ms після останнього звуку
break
if start_trim < end_trim:
trimmed = audio_segment[start_trim:end_trim]
removed_start = start_trim / 1000.0
removed_end = (len(audio_segment) - end_trim) / 1000.0
if removed_start > 0.2 or removed_end > 0.2:
print(f"✂️ Видалено мовчання: початок {removed_start:.1f}с, кінець {removed_end:.1f}с")
return trimmed
else:
return audio_segment
def apply_video_speedup(self, input_video_path, output_video_path, speed_factor):
"""Прискорює TikTok відео зі збереженням якості"""
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:
# Для відео з аудіо використовуємо atempo фільтр
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)
]
result = 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}")
print(f"FFmpeg stderr: {e.stderr}")
# У разі помилки копіюємо оригінал
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):
"""Розділяє історію на частини оптимальної тривалості"""
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:
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
preview_audio = self.tts_func(preview_text)
preview_duration = preview_audio.duration_seconds + 0.5
# Формуємо частини
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)
# Середня швидкість: 2.5 слова/секунда або 12 символів/секунда
sentence_duration = max(words_count / 2.5, chars_count / 12.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)} частин")
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.5, chars_count / 12.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"))
if not bg_videos:
raise ValueError("Не знайдено фонових відео!")
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)
# Отримуємо інформацію про відео
probe_cmd = [
"ffprobe", "-v", "quiet", "-print_format", "json",
"-show_format", "-show_streams", str(bg_video)
]
try:
probe_result = subprocess.run(probe_cmd, capture_output=True, text=True, check=True)
probe_data = json.loads(probe_result.stdout)
video_duration = float(probe_data['format']['duration'])
except:
# Якщо не вдалося отримати тривалість, використовуємо moviepy
clip = VideoFileClip(str(bg_video))
video_duration = clip.duration
clip.close()
remaining_duration = duration - duration_collected
use_duration = min(video_duration, remaining_duration)
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, capture_output=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"""
# Об'єднуємо всі 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, capture_output=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, capture_output=True)
return preview_video
def combine_tiktok_segments(self, story_id, part_number, preview_video, story_clips):
"""Об'єднує всі сегменти TikTok відео"""
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, capture_output=True)
return combined_path
def create_tiktok_part(self, story_part, story_id, horizontal_preview_path, selected_template, speed_settings=None):
"""
Створює одну частину TikTok відео з правильною обробкою прискорення
ВИПРАВЛЕНО: Не додає зайве мовчання при прискоренні
"""
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)
# ВАЖЛИВО: Очищуємо мовчання з обох частин
audio_preview = self.trim_silence_from_audio(audio_preview)
audio_story = self.trim_silence_from_audio(audio_story)
# Об'єднуємо з короткою паузою
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)
# ВАЖЛИВО: Очищуємо мовчання
audio_story = self.trim_silence_from_audio(audio_story)
final_audio = audio_story
preview_duration = 0
# Розраховуємо тривалості
story_duration = audio_story.duration_seconds
natural_duration = final_audio.duration_seconds
print(f"⏱️ Натуральна тривалість аудіо: {natural_duration:.1f}с")
# Обробляємо прискорення
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) # Максимальне прискорення
print(f"???? Цільова тривалість: {target_duration}с, потрібне прискорення: {speed_factor:.2f}x")
elif speed_settings.get('speed'):
# Режим фіксованого прискорення
speed_factor = speed_settings['speed']
print(f"⚡ Фіксоване прискорення: {speed_factor:.2f}x")
if speed_factor > 1.0:
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 = natural_duration
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 сегменти точно під тривалість аудіо story
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 відео"""
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)
# Застосовуємо субтитри
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
margin_v = 250
margin_lr = 100
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)
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 відео з правильною обробкою прискорення
ВИПРАВЛЕНО: Точно синхронізує відео з аудіо
"""
temp_output = self.temp_folder / f"temp_final_{random.randint(1000, 9999)}.mp4"
print(f"???? Фінальна збірка TikTok відео...")
# Отримуємо точну тривалість аудіо
audio_segment = AudioSegment.from_wav(audio_path)
audio_duration = audio_segment.duration_seconds
print(f"⏱️ Тривалість аудіо: {audio_duration:.1f}с")
try:
# Спочатку об'єднуємо відео з аудіо та точно обрізаємо під тривалість аудіо
subprocess.run([
"ffmpeg", "-y",
"-i", str(combined_video),
"-i", str(audio_path),
"-t", str(audio_duration), # ВАЖЛИВО: обрізаємо до точної тривалості аудіо
"-c:v", "h264_nvenc", "-preset", "p4", "-b:v", "8M", "-r", "60",
"-c:a", "aac", "-b:a", "128k",
"-shortest", # Використовуємо найкоротший stream
"-movflags", "+faststart",
str(temp_output)
], check=True, capture_output=True)
# Потім застосовуємо прискорення якщо потрібно
if apply_speedup and speed_factor > 1.0:
print(f"⚡ Застосування прискорення {speed_factor:.2f}x...")
self.apply_video_speedup(temp_output, output_path, speed_factor)
temp_output.unlink() # Видаляємо тимчасовий файл
else:
shutil.move(str(temp_output), str(output_path))
# Перевіряємо фінальну тривалість
try:
final_clip = VideoFileClip(str(output_path))
final_duration = final_clip.duration
final_clip.close()
expected_duration = audio_duration / speed_factor if apply_speedup else audio_duration
print(f"✅ Фінальна тривалість відео: {final_duration:.1f}с (очікувалось: {expected_duration:.1f}с)")
if abs(final_duration - expected_duration) > 1.0:
print(f"⚠️ УВАГА: Різниця в тривалості більше 1 секунди!")
except Exception as e:
print(f"⚠️ Не вдалося перевірити тривалість відео: {e}")
except subprocess.CalledProcessError as e:
print(f"❌ Помилка при фінальній збірці: {e}")
raise
finally:
if temp_output.exists():
temp_output.unlink()
# Функції поза класом
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 trim_silence_from_audio_standalone(audio_segment, silence_threshold_db=-40, chunk_size_ms=100):
"""Функція для очищення мовчання з початку та кінця"""
if len(audio_segment) == 0:
return audio_segment
# Видаляємо мовчання з початку
start_trim = 0
for i in range(0, len(audio_segment), chunk_size_ms):
chunk = audio_segment[i:i + chunk_size_ms]
if chunk.dBFS > silence_threshold_db:
start_trim = max(0, i - 200) # Залишаємо 200ms перед першим звуком
break
# Видаляємо мовчання з кінця
end_trim = len(audio_segment)
for i in range(len(audio_segment) - chunk_size_ms, 0, -chunk_size_ms):
if i < 0:
break
chunk = audio_segment[i:i + chunk_size_ms]
if chunk.dBFS > silence_threshold_db:
end_trim = min(len(audio_segment), i + chunk_size_ms + 200) # Залишаємо 200ms після останнього звуку
break
if start_trim < end_trim:
trimmed = audio_segment[start_trim:end_trim]
removed_start = start_trim / 1000.0
removed_end = (len(audio_segment) - end_trim) / 1000.0
if removed_start > 0.2 or removed_end > 0.2:
print(f"✂️ Видалено мовчання: початок {removed_start:.1f}с, кінець {removed_end:.1f}с")
return trimmed
else:
return audio_segment
def coqui_tts_simple(text: str) -> AudioSegment:
"""Простий TTS без цензури з очищенням мовчання"""
text_hash = hashlib.md5(text.encode()).hexdigest()
raw_path = temp_folder / f"raw_{text_hash}.wav"
clean_path = temp_folder / f"audio_{text_hash}.wav"
if clean_path.exists():
audio = AudioSegment.from_wav(clean_path)
return trim_silence_from_audio_standalone(audio)
try:
tts.tts_to_file(text=text, speaker=voice, file_path=raw_path)
# Очищуємо аудіо від шумів та нормалізуємо гучність
subprocess.run([
"ffmpeg", "-y", "-i", str(raw_path),
"-af", "afftdn,volume=3dB,silenceremove=1:0:-50dB:1:0.5:-50dB",
str(clean_path)
], check=True, capture_output=True)
audio = AudioSegment.from_wav(clean_path)
return trim_silence_from_audio_standalone(audio)
except Exception as e:
print(f"❌ Помилка TTS: {e}")
# Повертаємо порожнє аудіо у разі помилки
return AudioSegment.silent(duration=1000)
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():
audio = AudioSegment.from_wav(final_audio_path)
return trim_silence_from_audio_standalone(audio)
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 = trim_silence_from_audio_standalone(final_audio)
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 частини"""
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 відео"""
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 історію з підтримкою прискорення
ВИПРАВЛЕНО: Покращена логіка обробки прискорення
"""
# Читаємо файли
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