scribe-json-to-ass

# ver.5 (Jul 2, 2025)

import msgspec
import pysubs2
import re
from itertools import groupby


class Word:
    def __init__(self, text, start, end, speaker):
        self.text = text
        self.start = start
        self.end = end
        self.speaker = speaker


def merge_consecutive_words(words, specified_text: list[str], new_text):
    """...with the same, specified text"""
    if not words:
        return []

    merged = []
    for to_merge, group in groupby(words, lambda w: w.text in specified_text):
        group_list = list(group)
        if to_merge:
            merged.append(Word(new_text, group_list[0].start, group_list[-1].end, group_list[0].speaker))
        else:
            merged.extend(group_list)
    return merged


def events_to_string(events):
    """see ssa events as if in an ass file"""
    ssa_file = pysubs2.SSAFile()
    ssa_file.events = events
    return (
        ssa_file.to_string(format_="ass")
        .split("\n[Events]\nFormat: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text\n")[1]
        .strip()
    )


SUBS_PATH = r""

PUNC_END = "。！？.!?"
PUNC_MID = "、，,… "
PUNC = PUNC_END + PUNC_MID
KEEP_PUNC = False

MAX_WORD_DURATION = 1  # in seconds
MIN_LEN_TO_SPLIT_MID = 15

scribe = msgspec.json.decode(open(rf"{SUBS_PATH}\scribe.json", "rb").read())  # from ElevenLabs's Python SDK
words = [Word(w["text"], w["start"], w["end"], w["speaker_id"]) for w in scribe["words"] if w["type"] != "audio_event"]


# clean up space/punctuation variants
for word in words:
    word.text = word.text.replace("　", " ")
for word in words:
    word.text = word.text.replace("...", "…")
words = merge_consecutive_words(words, ["・"], "…")
words = merge_consecutive_words(words, ["-", "-"], "—")


# break words if there's punctuation in the middle
processed_words = []
for word in words:
    text = word.text

    pattern = f"([{PUNC}])"
    segments = [s for s in re.split(pattern, text) if s]

    if len(segments) <= 1:
        processed_words.append(word)
    else:
        non_punct_len = sum(len(seg) for seg in segments if seg not in PUNC)
        start_time = word.start
        total_duration = word.end - word.start

        for segment in segments:
            new_word = Word(segment, start_time, start_time, word.speaker)
            if segment not in PUNC:
                segment_duration = total_duration * (len(segment) / non_punct_len) if non_punct_len > 0 else 0
                end_time = start_time + segment_duration
                new_word.end = end_time
                start_time = end_time
            processed_words.append(new_word)
words = processed_words


# form lines
lines: list[list[Word]] = []
current_line: list[Word] = []
for i, word in enumerate(words):
    current_line.append(word)

    is_last_word = i == len(words) - 1
    time_break = not is_last_word and word.end < words[i + 1].start
    speaker_change = not is_last_word and word.speaker != words[i + 1].speaker
    line_end = word.text and word.text[-1] in PUNC_END
    long_line_mid = word.text and word.text[-1] in PUNC_MID and sum(len(w.text) for w in current_line) >= MIN_LEN_TO_SPLIT_MID

    if is_last_word or time_break or speaker_change or line_end or long_line_mid:
        if any(w.text.strip() for w in current_line) and current_line[-1].end - current_line[0].start > 0:
            speaker = current_line[0].speaker

            if not KEEP_PUNC:
                # substitute punctuations with space
                punc_trans = str.maketrans(dict.fromkeys(PUNC, " "))
                for w in current_line:
                    w.text = w.text.translate(punc_trans)

                # merge consecutive spaces and trim leading/trailing spaces
                current_line = merge_consecutive_words(current_line, [" "], " ")
                start_idx, end_idx = 0, len(current_line) - 1
                while start_idx <= end_idx and not current_line[start_idx].text.strip():
                    start_idx += 1
                while start_idx <= end_idx and not current_line[end_idx].text.strip():
                    end_idx -= 1
                current_line = current_line[start_idx : end_idx + 1]

            if current_line:
                # apply MAX_WORD_DURATION limit to leading/trailing words
                if current_line[0].end - current_line[0].start > MAX_WORD_DURATION:
                    current_line[0].start = current_line[0].end - MAX_WORD_DURATION
                if current_line[-1].end - current_line[-1].start > MAX_WORD_DURATION:
                    current_line[-1].end = current_line[-1].start + MAX_WORD_DURATION

                lines.append(current_line)

        current_line: list[Word] = []


# create ass file and save
subs = pysubs2.SSAFile()
subs.styles["ja"] = pysubs2.SSAStyle(
    fontname="IPAexGothic",
    fontsize=40,
    primarycolor=pysubs2.Color(r=255, g=255, b=255, a=0),
    secondarycolor=pysubs2.Color(r=0, g=213, b=255, a=0),
    outlinecolor=pysubs2.Color(r=0, g=0, b=0, a=0),
    backcolor=pysubs2.Color(r=74, g=74, b=74, a=0),
    bold=True,
    alignment=pysubs2.Alignment.BOTTOM_CENTER,
    shadow=0,
)
subs.info["PlayResX"] = "1920"
subs.info["PlayResY"] = "1080"

flag_dict = {}  # for reporting issues, like words exceeding MAX_WORD_DURATION

for line in lines:
    karaoke_text = ""
    max_duration = 0
    for w in line:
        karaoke_text += f"{{\\kf{round(w.end * 100) - round(w.start * 100)}}}{w.text}"  # it's guaranteed that the next word starts at w.end (see time_break)
        max_duration = max(max_duration, w.end - w.start)
    if karaoke_text.strip():
        subs.append(
            pysubs2.SSAEvent(
                start=line[0].start * 1000,
                end=line[-1].end * 1000,
                text=karaoke_text.strip(),
                style="ja",
                name=line[0].speaker.lstrip("speaker_"),
            )
        )
        if max_duration > MAX_WORD_DURATION:
            flag_dict.setdefault(len(subs) - 1, []).append("duration")

for i, e in enumerate(subs):
    if i in flag_dict:
        print(f"[{i + 1}] {' '.join(flag_dict[i])}\n{events_to_string([e])}")

subs.save(rf"{SUBS_PATH}\scribe.ass")
subs.save(rf"{SUBS_PATH}\scribe.srt")