| |
|
|
| MIN_DISPLAY = 1.5 |
| MAX_DISPLAY = 7.0 |
| MERGE_MAX_CHARS = 72 |
| MAX_GAP = 1.5 |
| WRAP_CHARS = 42 |
| MAX_LINES = 2 |
| CPS_LIMIT = 17 |
|
|
| |
| |
| |
| WEAK_ENDINGS = { |
| 'in', 'my', 'your', 'his', 'her', 'our', 'their', 'its', 'the', 'a', 'an', |
| 'of', 'to', 'at', 'on', 'by', 'for', 'with', 'is', 'are', 'was', 'were', |
| 'and', 'or', 'but', 'so', 'if', 'that', 'this', 'we', 'you', 'he', 'she', 'they', |
| } |
|
|
|
|
| def format_timestamp(seconds: float) -> str: |
| ms = round((seconds % 1) * 1000) |
| s = int(seconds) % 60 |
| m = int(seconds // 60) % 60 |
| h = int(seconds // 3600) |
| return f"{h:02d}:{m:02d}:{s:02d},{ms:03d}" |
|
|
|
|
| SENTENCE_ENDERS = {'.', '!', '?'} |
| LEAD_WORD_BUDGET = 9 |
| LEAD_CHAR_BUDGET = 55 |
|
|
|
|
| def rebalance_boundaries(segments: list) -> list: |
| """ |
| مشکلی که این تابع حل می‌کنه: خودِ Whisper بعضی‌وقتا وسط یه عبارت به‌هم‌چسبیده |
| (مثل «in my | opinion» یا «of this | video») مرز segment می‌ذاره — نه سر یه |
| مکث طبیعی. چون این دو تیکه معمولاً segmentهای RAW جدا از هم می‌مونن (مثلاً چون |
| مجموع مدت‌زمانشون از MAX_DISPLAY بیشتره و merge_segments نمی‌تونه mergeشون کنه)، |
| این مرز اشتباه هیچ‌وقت به merge_segments یا split_long_segment نمی‌رسه که اصلاح بشه. |
| |
| این تابع، قبل از هر پردازش دیگه‌ای، مرز بین segmentهای پشت‌سرهم رو از دو جهت چک می‌کنه: |
| |
| جهتِ ۱ (رو به جلو): اگه segment اول با یه یا چند کلمه‌ی «ضعیف» و بدون علامت نگارشی |
| (in/my/this/your/...) تموم بشه، اون کلمه(ها) به ابتدای segment بعدی منتقل میشن. |
| مثال: «...building. In my» + «opinion, ...» → «...building.» + «In my opinion, ...» |
| |
| جهتِ ۲ (رو به عقب): اگه segment اول هنوز به یه پایانِ جمله‌ی واقعی (./!/؟) نرسیده باشه، |
| و ابتدای segment بعدی یه دنباله‌ی کوتاهه (حداکثر چند کلمه) که زود به یه پایانِ جمله‌ی |
| واقعی می‌رسه، اون دنباله به آخر segment اول منتقل میشه — چون یعنی داره همون جمله‌ی |
| ناتمومِ قبلی رو کامل می‌کنه، نه اینکه جمله‌ی جدیدی باشه. |
| مثال: «...TVs and other» + «electronics. I am curious...» → «...TVs and other electronics.» + «I am curious...» |
| |
| هر دو جهت فقط بین segmentهای هم‌گوینده (یا وقتی گوینده مشخص نیست) و با فاصله‌ی کوتاه انجام میشن. |
| """ |
| segments = [dict(s, words=list(s.get("words", []))) for s in segments] |
|
|
| for i in range(len(segments) - 1): |
| cur = segments[i] |
| nxt = segments[i + 1] |
|
|
| cur_words = cur["words"] |
| nxt_words = nxt["words"] |
| if not cur_words or not nxt_words: |
| continue |
|
|
| |
| if cur.get("speaker") is not None and nxt.get("speaker") is not None \ |
| and cur.get("speaker") != nxt.get("speaker"): |
| continue |
|
|
| |
| gap = nxt_words[0]["start"] - cur_words[-1]["end"] |
| if gap > MAX_GAP: |
| continue |
|
|
| moved_any = False |
|
|
| |
| while len(cur_words) > 1: |
| last = cur_words[-1] |
| stripped = last["word"].strip() |
| bare = stripped.strip(".,!?;:").lower() |
| has_punct = bool(stripped) and stripped[-1] in {'.', '!', '?', ',', ';', ':'} |
|
|
| if bare in WEAK_ENDINGS and not has_punct: |
| nxt_words.insert(0, cur_words.pop()) |
| moved_any = True |
| else: |
| break |
|
|
| |
| |
| |
| |
| |
| |
| PAUSE_CHARS = SENTENCE_ENDERS | {','} |
| cur_last = cur_words[-1]["word"].strip() |
| cur_at_pause = bool(cur_last) and cur_last[-1] in PAUSE_CHARS |
|
|
| if not cur_at_pause and len(nxt_words) > 1: |
| lead = [] |
| lead_chars = 0 |
| found_end = False |
| for w in nxt_words: |
| stripped = w["word"].strip() |
| if len(lead) >= LEAD_WORD_BUDGET or lead_chars + len(stripped) + 1 > LEAD_CHAR_BUDGET: |
| break |
| lead.append(w) |
| lead_chars += len(stripped) + 1 |
| if stripped and stripped[-1] in PAUSE_CHARS: |
| found_end = True |
| break |
|
|
| if found_end and len(lead) < len(nxt_words): |
| trial_words = cur_words + lead |
| trial_text = " ".join(w["word"].strip() for w in trial_words) |
| trial_seg = {"text": trial_text, "start": cur_words[0]["start"], |
| "end": lead[-1]["end"], "words": trial_words} |
| trial_pieces = split_long_segment(trial_seg) |
|
|
| if all(fits_in_two_lines(p["text"]) is not None for p in trial_pieces): |
| for _ in lead: |
| cur_words.append(nxt_words.pop(0)) |
| moved_any = True |
|
|
| |
| |
| if moved_any: |
| cur["text"] = " ".join(w["word"].strip() for w in cur_words) |
| cur["end"] = cur_words[-1]["end"] |
| nxt["text"] = " ".join(w["word"].strip() for w in nxt_words) |
| nxt["start"] = nxt_words[0]["start"] |
|
|
| return segments |
|
|
|
|
| def merge_segments(segments: list) -> list: |
| if not segments: |
| return [] |
|
|
| merged = [] |
| current = None |
|
|
| for seg in segments: |
| if current is None: |
| current = { |
| "text": seg["text"].strip(), |
| "start": seg["start"], |
| "end": seg["end"], |
| "words": list(seg.get("words", [])), |
| "speaker": seg.get("speaker"), |
| } |
| continue |
|
|
| gap = seg["start"] - current["end"] |
| combined_text = current["text"] + " " + seg["text"].strip() |
| combined_duration = seg["end"] - current["start"] |
| ends_sentence = current["text"].rstrip().endswith((".", "!", "?")) |
| current_duration = current["end"] - current["start"] |
| is_short = current_duration < 1.5 |
| same_speaker = current.get("speaker") is None or current.get("speaker") == seg.get("speaker") |
| too_fast = (len(current["text"]) / current_duration) > CPS_LIMIT if current_duration > 0 else True |
| fits_dialogue = len(current["text"]) <= WRAP_CHARS and len(seg["text"].strip()) <= WRAP_CHARS |
| already_dialogue = current.get("is_dialogue_block", False) |
|
|
| should_merge = ( |
| not ends_sentence |
| and combined_duration <= MAX_DISPLAY |
| and same_speaker |
| and gap <= MAX_GAP |
| and ( |
| is_short |
| or len(combined_text) <= MERGE_MAX_CHARS |
| ) |
| ) |
|
|
| should_pair_as_dialogue = ( |
| not already_dialogue |
| and not same_speaker |
| and too_fast |
| and gap <= MAX_GAP |
| and fits_dialogue |
| ) |
|
|
| if should_merge: |
| current["text"] = combined_text |
| current["end"] = seg["end"] |
| current["words"] += list(seg.get("words", [])) |
| elif should_pair_as_dialogue: |
| current["text"] = f"- {current['text']}\n- {seg['text'].strip()}" |
| current["end"] = seg["end"] |
| current["words"] += list(seg.get("words", [])) |
| current["is_dialogue_block"] = True |
| else: |
| merged.append(current) |
| current = { |
| "text": seg["text"].strip(), |
| "start": seg["start"], |
| "end": seg["end"], |
| "words": list(seg.get("words", [])), |
| "speaker": seg.get("speaker"), |
| } |
|
|
| if current: |
| merged.append(current) |
|
|
| return merged |
|
|
|
|
| def remove_duplicates(segments: list) -> list: |
| if not segments: |
| return [] |
| result = [segments[0]] |
| for seg in segments[1:]: |
| if seg["text"].strip() != result[-1]["text"].strip(): |
| result.append(seg) |
| return result |
|
|
|
|
| def fix_overlaps(segments: list) -> list: |
| fixed = [] |
| for seg in segments: |
| seg = dict(seg) |
| if fixed and seg["start"] < fixed[-1]["end"]: |
| seg["start"] = fixed[-1]["end"] + 0.001 |
| if seg["start"] >= seg["end"]: |
| seg["end"] = seg["start"] + MIN_DISPLAY |
| fixed.append(seg) |
| return fixed |
|
|
|
|
| def apply_min_display(segments: list) -> list: |
| result = [] |
| for i, seg in enumerate(segments): |
| seg = dict(seg) |
| duration = seg["end"] - seg["start"] |
| if duration < MIN_DISPLAY: |
| if i + 1 < len(segments): |
| next_start = segments[i + 1]["start"] |
| max_end = next_start - 0.080 |
| else: |
| max_end = seg["end"] + MAX_DISPLAY |
| new_end = seg["start"] + MIN_DISPLAY |
| seg["end"] = min(new_end, max(max_end, seg["end"])) |
| result.append(seg) |
| return result |
|
|
|
|
| def split_long_segment(seg: dict) -> list: |
| """ |
| فقط اگه punctuation نزدیک وسط جمله بود، به دو subtitle جداگانه بشکن. |
| اگه نقطه مناسبی نبود، نشکن — wrap_text در build_srt دو خطش می‌کنه. |
| بلوک‌های دیالوگِ دو-گوینده‌ای (is_dialogue_block) هیچ‌وقت شکسته نمی‌شن. |
| """ |
| if seg.get("is_dialogue_block"): |
| return [seg] |
|
|
| words = seg.get("words", []) |
| text = seg["text"].strip() |
|
|
| punct_chars = {'.', '!', '?', ',', ';', ':'} |
| sentence_enders = {'.', '!', '?'} |
| min_part_chars = 10 |
|
|
| |
| |
| |
| has_internal_sentence_end = any( |
| w["word"].strip() and w["word"].strip()[-1] in sentence_enders |
| for w in words[:-1] |
| ) |
|
|
| if not has_internal_sentence_end and (len(words) < 4 or fits_in_two_lines(text) is not None): |
| return [seg] |
|
|
| full_text = text |
| mid_char = len(full_text) / 2 |
|
|
| best_idx = -1 |
| best_score = float('-inf') |
| char_pos = 0 |
|
|
| for i, w in enumerate(words): |
| word_stripped = w["word"].strip() |
| this_word_end_char = char_pos + len(word_stripped) |
|
|
| if word_stripped and word_stripped[-1] in punct_chars and 0 < i < len(words) - 1: |
| |
| bare_word = word_stripped.strip(".,!?;:").lower() |
| part1_chars = this_word_end_char |
| part2_chars = len(full_text) - this_word_end_char |
|
|
| if bare_word not in WEAK_ENDINGS and part1_chars >= min_part_chars and part2_chars >= min_part_chars: |
| dist_from_mid = abs(this_word_end_char - mid_char) |
| proximity_score = 1.0 - (dist_from_mid / len(full_text)) |
| sentence_end_bonus = 0.5 if word_stripped[-1] in sentence_enders else 0.0 |
| score = proximity_score + sentence_end_bonus |
|
|
| if score > best_score: |
| best_score = score |
| best_idx = i + 1 |
|
|
| char_pos = this_word_end_char + 1 |
|
|
| |
| if best_idx == -1: |
| |
| if fits_in_two_lines(full_text) is not None: |
| return [seg] |
| |
| |
| |
| |
| |
| best_idx = None |
| best_diff = float('inf') |
| semantic_idx = None |
| semantic_diff = float('inf') |
| LINE_START_OK = WEAK_ENDINGS | {"and", "or", "but", "because", "so", "although", |
| "which", "who", "that", "when", "while", "if"} |
| for i in range(2, len(words) - 1): |
| part1 = " ".join(w["word"].strip() for w in words[:i]) |
| part2 = " ".join(w["word"].strip() for w in words[i:]) |
| if fits_in_two_lines(part1) is not None and fits_in_two_lines(part2) is not None: |
| diff = abs(len(part1) - len(part2)) |
| if diff < best_diff: |
| best_diff = diff |
| best_idx = i |
| |
| |
| |
| prev_bare = words[i - 1]["word"].strip() |
| next_bare = words[i]["word"].strip().strip(".,!?;:").lower() |
| is_semantic = (prev_bare and prev_bare[-1] in punct_chars) or (next_bare in LINE_START_OK) |
| if is_semantic and diff < semantic_diff: |
| semantic_diff = diff |
| semantic_idx = i |
| if semantic_idx is not None: |
| best_idx = semantic_idx |
| if best_idx is None: |
| return [seg] |
|
|
| words1 = words[:best_idx] |
| words2 = words[best_idx:] |
|
|
| if len(words1) < 2 or len(words2) < 2: |
| return [seg] |
|
|
| start2 = words2[0]["start"] |
| end1 = max(start2 - 0.080, words1[-1]["end"]) |
|
|
| if end1 - words1[0]["start"] < MIN_DISPLAY: |
| end1 = words1[0]["start"] + MIN_DISPLAY |
| if end1 >= start2: |
| end1 = start2 - 0.040 |
|
|
| seg1 = { |
| "text": " ".join(w["word"].strip() for w in words1), |
| "start": words1[0]["start"], |
| "end": end1, |
| "words": words1, |
| } |
| seg2 = { |
| "text": " ".join(w["word"].strip() for w in words2), |
| "start": start2, |
| "end": words2[-1]["end"], |
| "words": words2, |
| } |
|
|
| return split_long_segment(seg1) + split_long_segment(seg2) |
|
|
|
|
| def merge_short_fragments(segments: list) -> list: |
| """ |
| بعد از شکستنِ جمله‌های بلند، بعضی‌وقتا یه تیکه‌ی خیلی کوچیک باقی می‌مونه که مدت‌زمان |
| نمایشش زیر MIN_DISPLAY (۱.۵ ثانیه) میشه — نه چون واقعاً یه عبارت مستقله، بلکه چون |
| نقطه‌ی شکستن (punctuation) دقیقاً همون‌جا بوده. این تابع، به‌صورت یه پاسِ نهایی، |
| هر تیکه‌ی زیر ۱.۵ ثانیه رو در صورت امکان با همسایه‌ش (قبلی یا بعدی) یکی می‌کنه — |
| فقط وقتی که ادغام از سقف‌های نمایشی (MAX_DISPLAY و طول قابل‌قبول برای دو خط) رد نشه، |
| از رو یه پایانِ جمله‌ی واقعی (./!/؟) عبور نکنه، و از همه مهم‌تر: فاصله‌ی زمانیِ واقعی |
| بین دو تیکه از MAX_GAP بیشتر نباشه — یعنی اگه یه کلمه یا عبارتِ کوتاه چند ثانیه بعد از |
| تیکه‌ی قبلی پخش میشه (یه مکثِ واقعی وسطشونه)، حتی با اینکه کوتاهه، مستقل می‌مونه و به |
| تیکه‌ی قبلی نمی‌چسبه. |
| |
| نکته: این یه قانونِ کاملاً مکانیکیه (فقط بر اساس مدت‌زمان)، نه یه تشخیص معنایی از |
| اینکه عبارت «کامل» هست یا نه — پس ممکنه عبارت‌های کوتاهِ ولی کاملی (مثل «For example,») |
| رو هم با همسایه‌شون ادغام کنه. |
| |
| یه نکته‌ی مهمِ دیگه: این تابع همیشه هر دو جهت (چسبیدن به تیکه‌ی قبلی، یا چسبیدن به |
| تیکه‌ی بعدی) رو امتحان می‌کنه و هر کدوم که ترکیبِ کوتاه‌تر (جادارتر) بسازه رو انتخاب |
| می‌کنه — نه اینکه همیشه فقط به عقب بچسبونه. چون گاهی چسبوندنِ حریصانه به عقب باعث |
| میشه یه تیکه‌ی کوتاهِ *بعدی* بی‌جا بمونه (چون هر دو همسایه‌ش پر شدن)، در حالی که اگه |
| از اول به جلو می‌چسبید، هیچ‌کدوم بی‌جا نمی‌موندن. |
| """ |
| if not segments: |
| return [] |
|
|
| segments = [dict(s) for s in segments] |
|
|
| def can_merge(a: dict, b: dict) -> tuple: |
| """آیا a و b قابل ادغامن؟ اگه بله، (True, متنِ ترکیبی، طولِ متنِ ترکیبی) برمی‌گردونه.""" |
| if a["text"].rstrip().endswith((".", "!", "?")): |
| return False, None, None |
| gap = b["start"] - a["end"] |
| if gap > MAX_GAP: |
| return False, None, None |
| combined_duration = b["end"] - a["start"] |
| if combined_duration > MAX_DISPLAY: |
| return False, None, None |
| combined_text = a["text"].rstrip() + " " + b["text"].lstrip() |
| if fits_in_two_lines(combined_text) is None: |
| return False, None, None |
| return True, combined_text, len(combined_text) |
|
|
| def do_merge(a: dict, b: dict, combined_text: str) -> dict: |
| merged = dict(a) |
| merged["text"] = combined_text |
| merged["end"] = b["end"] |
| if "words" in a or "words" in b: |
| merged["words"] = list(a.get("words", [])) + list(b.get("words", [])) |
| return merged |
|
|
| result = [] |
| i = 0 |
| n = len(segments) |
| while i < n: |
| seg = segments[i] |
| seg_duration = seg["end"] - seg["start"] |
|
|
| if seg_duration >= MIN_DISPLAY: |
| result.append(seg) |
| i += 1 |
| continue |
|
|
| prev = result[-1] if result else None |
| nxt = segments[i + 1] if i + 1 < n else None |
|
|
| back_ok, back_text, back_len = (can_merge(prev, seg) if prev is not None else (False, None, None)) |
| fwd_ok, fwd_text, fwd_len = (can_merge(seg, nxt) if nxt is not None else (False, None, None)) |
|
|
| if back_ok and (not fwd_ok or back_len <= fwd_len): |
| merged = do_merge(prev, seg, back_text) |
| result[-1] = merged |
| i += 1 |
| elif fwd_ok: |
| merged = do_merge(seg, nxt, fwd_text) |
| result.append(merged) |
| i += 2 |
| else: |
| result.append(seg) |
| i += 1 |
|
|
| return result |
|
|
|
|
| def fits_in_two_lines(text: str) -> str | None: |
| """ |
| طبقِ استانداردِ واقعیِ زیرنویس (Netflix/Amazon/Disney+: حداکثر ۴۲ کاراکتر در هر خط، |
| حداکثر ۲ خط در هر cue)، سعی می‌کنه متن رو به ۱ یا ۲ خط بشکنه طوری که هیچ‌کدوم از |
| خط‌ها از WRAP_CHARS بیشتر نشه. |
| |
| بینِ نقاطِ معتبر (هر دو خط ≤۴۲)، صرفاً «متعادل‌ترین از نظرِ طول» رو انتخاب نمی‌کنه — |
| اول دنبالِ نقطه‌ای می‌گرده که از نظرِ نگارشی/معنایی هم معقول باشه (بعدِ یه ویرگول یا |
| علامتِ نگارشی، یا لااقل قبل از یه کلمه‌ی ربط‌دهنده مثل and/or/because)، تا یه عبارتِ |
| چسبیده (مثل «certain locations») وسط بریده نشه. فقط وقتی هیچ‌کدومِ نقاطِ معتبر چنین |
| ویژگی‌ای نداشتن، سراغِ متعادل‌ترینِ صرفاً از نظرِ طول می‌ره. |
| |
| اگه ممکن باشه، متنِ شکسته‌شده (یا خودِ متن، اگه از اول یه‌خطی بود) رو برمی‌گردونه. |
| اگه هیچ نقطه‌ی شکستنی وجود نداشته باشه که هر دو خط رو زیرِ ۴۲ نگه داره، None برمی‌گردونه — |
| یعنی این متن ذاتاً برای یه cueِ دوخطی خیلی بلنده و باید به چند subtitleِ جدا تقسیم بشه؛ |
| این تشخیص با یه عددِ اختیاریِ حدسی (مثل «۹۴ کاراکتر») جایگزین نمیشه. |
| """ |
| if len(text) <= WRAP_CHARS: |
| return text |
|
|
| words = text.split() |
|
|
| |
| |
| LINE_START_OK = WEAK_ENDINGS | {"and", "or", "but", "because", "so", "although", |
| "which", "who", "that", "when", "while", "if"} |
|
|
| valid = [] |
| for i in range(1, len(words)): |
| line1 = " ".join(words[:i]) |
| line2 = " ".join(words[i:]) |
| if len(line1) <= WRAP_CHARS and len(line2) <= WRAP_CHARS: |
| diff = abs(len(line1) - len(line2)) |
| valid.append((i, line1, line2, diff)) |
|
|
| if not valid: |
| return None |
|
|
| def is_semantically_ok(i: int, line1: str) -> bool: |
| |
| |
| |
| prev_word = words[i - 1].strip() |
| if prev_word and prev_word[-1] in {',', '.', '!', '?', ';', ':'}: |
| return True |
| next_word_bare = words[i].strip(".,!?;:").lower() |
| return next_word_bare in LINE_START_OK |
|
|
| semantically_ok = [v for v in valid if is_semantically_ok(v[0], v[1])] |
| pool = semantically_ok if semantically_ok else valid |
|
|
| best = min(pool, key=lambda v: v[3]) |
| return f"{best[1]}\n{best[2]}" |
|
|
|
|
| def wrap_text(text: str) -> str: |
| """متن رو طبقِ قانونِ واقعیِ ۴۲کاراکتری به ۱ یا ۲ خط می‌شکنه. |
| اگه متن از قبل یه بلوک دیالوگِ دو-گوینده‌ای (با - و خط جدید) بود، دست‌نخورده برگردون. |
| اگه (که نباید پیش بیاد، چون split_long_segment از قبل تضمین کرده) هیچ شکستنِ معتبری |
| پیدا نشد، به‌جای گم‌کردنِ متن، همون شکستنِ نزدیک‌به‌وسطِ قدیمی رو به‌عنوانِ آخرین راه‌حل می‌زنیم.""" |
| if text.startswith("- ") and "\n" in text: |
| return text |
|
|
| fitted = fits_in_two_lines(text) |
| if fitted is not None: |
| return fitted |
|
|
| |
| words = text.split() |
| best_break = -1 |
| best_diff = float('inf') |
| current_len = 0 |
| for i, word in enumerate(words): |
| current_len += len(word) + (1 if i > 0 else 0) |
| diff = abs(current_len - len(text) / 2) |
| if diff < best_diff: |
| best_diff = diff |
| best_break = i + 1 |
| if best_break <= 0 or best_break >= len(words): |
| return text |
| line1 = " ".join(words[:best_break]) |
| line2 = " ".join(words[best_break:]) |
| return f"{line1}\n{line2}" |
|
|
|
|
| def build_srt(segments: list) -> str: |
| lines = [] |
| index = 1 |
| for seg in segments: |
| text = seg["text"].strip() |
| if not text: |
| continue |
| start_tc = format_timestamp(seg["start"]) |
| end_tc = format_timestamp(seg["end"]) |
| wrapped = wrap_text(text) |
| lines.append(f"{index}\n{start_tc} --> {end_tc}\n{wrapped}") |
| index += 1 |
| return "\n\n".join(lines) + "\n" |
|
|
|
|
| def process(segments: list) -> str: |
| if not segments: |
| return "" |
|
|
| segments = rebalance_boundaries(segments) |
| merged = merge_segments(segments) |
| merged = remove_duplicates(merged) |
| fixed = fix_overlaps(merged) |
|
|
| split = [] |
| for seg in fixed: |
| split.extend(split_long_segment(seg)) |
|
|
| split = fix_overlaps(split) |
| split = merge_short_fragments(split) |
| split = fix_overlaps(split) |
| final = apply_min_display(split) |
|
|
| return build_srt(final) |
|
|
