From 7ad2e563e1aaad8b8fe449478127ef39d7bf89ee Mon Sep 17 00:00:00 2001 From: Nitsud Yarg Date: Sat, 2 May 2026 10:56:50 -0700 Subject: [PATCH] progress subs to notes --- fugashi_parser.py | 206 ++++++++++++++++++++++++++++++++++++++++++ jp_readings.py | 175 +---------------------------------- subtitle_processor.py | 53 +++++++++++ 3 files changed, 260 insertions(+), 174 deletions(-) create mode 100644 fugashi_parser.py create mode 100644 subtitle_processor.py diff --git a/fugashi_parser.py b/fugashi_parser.py new file mode 100644 index 0000000..a153b69 --- /dev/null +++ b/fugashi_parser.py @@ -0,0 +1,206 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +from __future__ import annotations + +import argparse +import html +import json +import os +import re +import sys +from dataclasses import dataclass, asdict +from typing import Dict, List, Any + +import requests + +request_url = "http://127.0.0.1:19633" +request_timeout = 10 + +from fugashi import Tagger + + +# Ensure UTF-8 output +if hasattr(sys.stdout, "reconfigure"): + sys.stdout.reconfigure(encoding="utf-8", errors="backslashreplace") + + +KATAKANA_START = ord("ァ") +KATAKANA_END = ord("ヶ") +KATAKANA_TO_HIRAGANA_OFFSET = ord("ぁ") - ord("ァ") + +KANJI_RE = re.compile(r"[\u4E00-\u9FFF]") + +def katakana_to_hiragana(text: str) -> str: + return "".join( + chr(ord(ch) + KATAKANA_TO_HIRAGANA_OFFSET) if KATAKANA_START <= ord(ch) <= KATAKANA_END else ch + for ch in text + ) + +def has_kanji(text: str) -> bool: + return bool(KANJI_RE.search(text)) + +def has_japanese(text: str) -> bool: + return any( + 0x3040 <= ord(ch) <= 0x30FF or + 0x3400 <= ord(ch) <= 0x9FFF + for ch in text + ) + + +def is_content_word(pos1: str) -> bool: + return pos1 in {"名詞", "動詞", "形容詞", "副詞"} + + +def should_skip(surface: str, pos1: str) -> bool: + return not surface.strip() or pos1 == "補助記号" + + +def clean_lemma(lemma: str) -> str: + return re.split(r"-", lemma, maxsplit=1)[0] if lemma else lemma + + +def get_feature(feature, name: str, default=""): + return getattr(feature, name, default) or default + + +def build_ruby(surface: str, reading: str) -> str: + if not reading or surface == reading or not has_japanese(surface): + return html.escape(surface) + return f"{html.escape(surface)}{html.escape(reading)}" + +def build_kana(surface: str, reading: str) -> str: + if not reading or surface == reading or not has_japanese(surface): + return surface + return f"「{reading}」" + +def build_vocab_link(surface: str, reading: str, pos1: str) -> str: + if should_skip(surface, pos1) or not is_content_word(pos1): + return surface + if surface == reading: + return f"[[{surface}]]" + return f"[[{reading}|{surface}]]" + + +@dataclass +class TokenReading: + surface: str + reading_hiragana: str + lemma: str + lemma_reading_hiragana: str + pos: List[str] + + +# ---------- Persistent vocab handling ---------- + +def load_vocab(path: str) -> Dict[str, Dict]: + if not os.path.exists(path): + return {} + + with open(path, "r", encoding="utf-8") as f: + raw = json.load(f) + + # normalize + convert surfaces_seen → set + out = {} + for lemma, entry in raw.items(): + out[lemma] = { + "lemma": entry.get("lemma", lemma), + "lemma_reading_hiragana": entry.get("lemma_reading_hiragana", ""), + "pos": entry.get("pos", []), + "mention_count": int(entry.get("mention_count", 0)), + "surfaces_seen": set(entry.get("surfaces_seen", [])), # ← key change + } + return out + + +def save_vocab(path: str, vocab: Dict[str, Dict]): + os.makedirs(os.path.dirname(path) or ".", exist_ok=True) + + # convert sets → sorted lists + serializable = { + lemma: { + **entry, + "surfaces_seen": sorted(entry.get("surfaces_seen", [])) + } + for lemma, entry in vocab.items() + } + + with open(path, "w", encoding="utf-8") as f: + json.dump(serializable, f, ensure_ascii=False, indent=2) + + +def update_vocab(vocab: Dict[str, Dict], tokens: List[TokenReading]): + for t in tokens: + pos1 = t.pos[0] if t.pos else "" + + if should_skip(t.surface, pos1) or not is_content_word(pos1): + continue + + entry = vocab.get(t.lemma) + + if entry is None: + vocab[t.lemma] = { + "lemma": t.lemma, + "lemma_reading_hiragana": t.lemma_reading_hiragana, + "pos": t.pos, + "mention_count": 1, + "surfaces_seen": {t.surface}, # ← set + } + else: + entry["mention_count"] += 1 + entry.setdefault("surfaces_seen", set()).add(t.surface) + + +# ---------- Main analysis ---------- + +def analyze(text: str, vocab: Dict[str, Dict]): + tagger = Tagger() + + tokens: List[TokenReading] = [] + ruby_parts = [] + kana_parts = [] + vocab_link_parts = [] + + for word in tagger(text): + f = word.feature + surface = word.surface + + pos = [ + get_feature(f, "pos1"), + get_feature(f, "pos2"), + get_feature(f, "pos3"), + get_feature(f, "pos4"), + ] + + kana = get_feature(f, "kana") + #reading = katakana_to_hiragana(kana) if kana else "" + if kana: + if kana == surface: + reading = "" + else: + reading = katakana_to_hiragana(kana) + else: + reading = "" + + lemma = clean_lemma(get_feature(f, "lemma") or surface) + vocab_link_parts.append(build_vocab_link(surface, lemma, pos[0])) + + lform = get_feature(f, "lForm") + kana_base = get_feature(f, "kanaBase") + lemma_reading = katakana_to_hiragana(lform or kana_base or kana or lemma) + + token = TokenReading(surface, reading, lemma, lemma_reading, pos) + tokens.append(token) + + ruby_parts.append(build_ruby(surface, reading)) + kana_parts.append(build_kana(surface, reading)) + + update_vocab(vocab, tokens) + + return { + "original_text": text, + "kana_reading": "".join(kana_parts), + "ruby_html": "".join(ruby_parts), + "notes_link":"".join(vocab_link_parts), + "token_readings": [asdict(t) for t in tokens], + } diff --git a/jp_readings.py b/jp_readings.py index e0c658e..345e6ff 100644 --- a/jp_readings.py +++ b/jp_readings.py @@ -1,177 +1,4 @@ -#!/usr/bin/env python3 -# -*- coding: utf-8 -*- - -from __future__ import annotations - -import argparse -import html -import json -import os -import re -import sys -from dataclasses import dataclass, asdict -from typing import Dict, List, Any - -from fugashi import Tagger - - -# Ensure UTF-8 output -if hasattr(sys.stdout, "reconfigure"): - sys.stdout.reconfigure(encoding="utf-8", errors="backslashreplace") - - -KATAKANA_START = ord("ァ") -KATAKANA_END = ord("ヶ") -KATAKANA_TO_HIRAGANA_OFFSET = ord("ぁ") - ord("ァ") - - -def katakana_to_hiragana(text: str) -> str: - return "".join( - chr(ord(ch) + KATAKANA_TO_HIRAGANA_OFFSET) if KATAKANA_START <= ord(ch) <= KATAKANA_END else ch - for ch in text - ) - - -def has_japanese(text: str) -> bool: - return any( - 0x3040 <= ord(ch) <= 0x30FF or - 0x3400 <= ord(ch) <= 0x9FFF - for ch in text - ) - - -def is_content_word(pos1: str) -> bool: - return pos1 in {"名詞", "動詞", "形容詞", "副詞"} - - -def should_skip(surface: str, pos1: str) -> bool: - return not surface.strip() or pos1 == "補助記号" - - -def clean_lemma(lemma: str) -> str: - return re.split(r"-", lemma, maxsplit=1)[0] if lemma else lemma - - -def get_feature(feature, name: str, default=""): - return getattr(feature, name, default) or default - - -def build_ruby(surface: str, reading: str) -> str: - if not reading or surface == reading or not has_japanese(surface): - return html.escape(surface) - return f"{html.escape(surface)}{html.escape(reading)}" - - -@dataclass -class TokenReading: - surface: str - reading_hiragana: str - lemma: str - lemma_reading_hiragana: str - pos: List[str] - - -# ---------- Persistent vocab handling ---------- - -def load_vocab(path: str) -> Dict[str, Dict]: - if not os.path.exists(path): - return {} - - with open(path, "r", encoding="utf-8") as f: - raw = json.load(f) - - # normalize + convert surfaces_seen → set - out = {} - for lemma, entry in raw.items(): - out[lemma] = { - "lemma": entry.get("lemma", lemma), - "lemma_reading_hiragana": entry.get("lemma_reading_hiragana", ""), - "pos": entry.get("pos", []), - "mention_count": int(entry.get("mention_count", 0)), - "surfaces_seen": set(entry.get("surfaces_seen", [])), # ← key change - } - return out - - -def save_vocab(path: str, vocab: Dict[str, Dict]): - os.makedirs(os.path.dirname(path) or ".", exist_ok=True) - - # convert sets → sorted lists - serializable = { - lemma: { - **entry, - "surfaces_seen": sorted(entry.get("surfaces_seen", [])) - } - for lemma, entry in vocab.items() - } - - with open(path, "w", encoding="utf-8") as f: - json.dump(serializable, f, ensure_ascii=False, indent=2) - - -def update_vocab(vocab: Dict[str, Dict], tokens: List[TokenReading]): - for t in tokens: - pos1 = t.pos[0] if t.pos else "" - - if should_skip(t.surface, pos1) or not is_content_word(pos1): - continue - - entry = vocab.get(t.lemma) - - if entry is None: - vocab[t.lemma] = { - "lemma": t.lemma, - "lemma_reading_hiragana": t.lemma_reading_hiragana, - "pos": t.pos, - "mention_count": 1, - "surfaces_seen": {t.surface}, # ← set - } - else: - entry["mention_count"] += 1 - entry.setdefault("surfaces_seen", set()).add(t.surface) - - -# ---------- Main analysis ---------- - -def analyze(text: str, vocab: Dict[str, Dict]): - tagger = Tagger() - - tokens: List[TokenReading] = [] - ruby_parts = [] - - for word in tagger(text): - f = word.feature - surface = word.surface - - pos = [ - get_feature(f, "pos1"), - get_feature(f, "pos2"), - get_feature(f, "pos3"), - get_feature(f, "pos4"), - ] - - kana = get_feature(f, "kana") - reading = katakana_to_hiragana(kana) if kana else "" - - lemma = clean_lemma(get_feature(f, "lemma") or surface) - - lform = get_feature(f, "lForm") - kana_base = get_feature(f, "kanaBase") - lemma_reading = katakana_to_hiragana(lform or kana_base or kana or lemma) - - token = TokenReading(surface, reading, lemma, lemma_reading, pos) - tokens.append(token) - - ruby_parts.append(build_ruby(surface, reading)) - - update_vocab(vocab, tokens) - - return { - "original_text": text, - "ruby_html": "".join(ruby_parts), - "token_readings": [asdict(t) for t in tokens], - } - +from fugashi_parser import * # ---------- CLI ---------- diff --git a/subtitle_processor.py b/subtitle_processor.py new file mode 100644 index 0000000..ae0a3cb --- /dev/null +++ b/subtitle_processor.py @@ -0,0 +1,53 @@ +from fugashi_parser import * + +# ---------- CLI ---------- + +def main(): +# parser = argparse.ArgumentParser() +# parser.add_argument("--vocab-file", required=True) +# parser.add_argument("text", nargs="*") +# args = parser.parse_args() + + vocab = load_vocab('foo.json') + + with open("/run/media/deck/YF8SD/Video/[Retimed and Corrected] Yu-Gi-Oh! Duel Monsters - 001.srt", "r", encoding="utf-8") as f: + lines = f.readlines() + + + notes_lines = [] + + # Example: Modify the 3rd line (index 2) + #lines[2] = "This is the new subtitle text\n" + ## First lien has some unicode bom bs + for i in range(len(lines)): + if re.fullmatch(r'\d+', lines[i].strip()) or i == 0: + notes_lines.append('##### ' + lines[i].strip() + " `" + lines[i+1].strip() + "`" + "\n") + i+=1 + continue + if has_japanese(lines[i]): + print(lines[i]) + result = analyze(lines[i],vocab) + print(result['kana_reading']) + notes_lines.append(result["ruby_html"] + "\n" ) + notes_lines.append (result['notes_link'] + "\n") + if lines[i] != result['kana_reading']: + lines[i] = result['kana_reading'] + "\n" # lost the new line + else: + notes_lines.append("`" + lines[i].strip() + "`" + "\n") + + with open("subtitle.srt", "w", encoding="utf-8") as f: + f.writelines(lines) + + with open("notestest.md", "w", encoding="utf-8") as f: + f.writelines(notes_lines) + + #vocab = load_vocab(args.vocab_file) + #result = analyze(text, vocab) + #save_vocab('foo.json', vocab) + + #result["vocab_file"] = args.vocab_file + #print(json.dumps(result, ensure_ascii=False, indent=2)) + + +if __name__ == "__main__": + main() \ No newline at end of file