jp_text_parsing/fugashi_parser.py

363 lines
10 KiB
Python

#!/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
#import unidic
# 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("")
CONTENT_POS1 = {"名詞", "動詞", "形容詞", "形状詞", "副詞", "代名詞"}
def is_vocab_token(pos, keep_pronouns=False, keep_numbers=False):
pos1, pos2, pos3, pos4 = pos
if pos1 not in CONTENT_POS1:
return False
if pos1 == "代名詞" and not keep_pronouns:
return False
if pos1 == "名詞" and pos2 == "数詞" and not keep_numbers:
return False
return True
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
)
KANJI_RE = re.compile(r"[\u4E00-\u9FFF]")
def has_kanji(text: str) -> bool:
return bool(KANJI_RE.search(text))
# def has_kanji(text):
# return any(
# 0x3400 <= ord(ch) <= 0x4DBF or
# 0x4E00 <= ord(ch) <= 0x9FFF or
# 0xF900 <= ord(ch) <= 0xFAFF
# 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 anki_fields_term(term : str) -> dict:
params = {
"text": term,
"type": "term",
#"markers": ["audio", "cloze-body-kana", "conjugation", "expression", "furigana", "furigana-plain", "glossary", "glossary-brief", "glossary-no-dictionary", "glossary-first", "glossary-first-brief", "glossary-first-no-dictionary", "part-of-speech", "phonetic-transcriptions", "pitch-accents", "pitch-accent-graphs", "pitch-accent-graphs-jj", "pitch-accent-positions", "pitch-accent-categories", "reading", "tags", "clipboard-image", "clipboard-text", "cloze-body", "cloze-prefix", "cloze-suffix", "dictionary", "dictionary-alias", "document-title", "frequencies", "frequency-harmonic-rank", "frequency-harmonic-occurrence", "frequency-average-rank", "frequency-average-occurrence", "screenshot", "search-query", "popup-selection-text", "sentence", "sentence-furigana", "sentence-furigana-plain", "url"],
"markers": ["audio", "expression", "cloze-body" , "furigana","furigana-plain", "part-of-speech", "reading", "glossary-first", "glossary-plain-no-dictionary" , "glossary-first-no-dictionary" , "dictionary", "frequencies", "frequency-average-rank", "frequency-average-occurrence", "screenshot", "sentence", "sentence-furigana", "sentence-furigana-plain" ],
"maxEntries": 2,
"includeMedia": False,
}
response = requests.post(request_url + "/ankiFields", json = params, timeout = request_timeout)
return json.loads(response.text)
def term_entries(term :str) -> dict:
#print("Requesting termEntries:")
params = {
"term": term,
"maxEntries": 1,
"includeMedia": False,
}
response = requests.post(request_url + "/termEntries", json = params, timeout = request_timeout)
return json.loads(response.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 should_ruby(surface, reading_hiragana):
if not reading_hiragana:
return False
if surface == reading_hiragana:
return False
if not has_kanji(surface):
return False
return True
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"<ruby>{html.escape(surface)}<rt>{html.escape(reading)}</rt></ruby>"
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, pos) -> str:
if not is_vocab_token(pos, keep_pronouns=True,keep_numbers=False):
return surface
if surface == reading:
return f"[[{surface}]]"
return f"[[{reading}|{surface}]]"
## Predicate filtering Verbs
def is_predicate_start(pos):
pos1, pos2, pos3, pos4 = pos
return (
pos1 == "動詞" and pos2 == "一般"
) or (
pos1 == "形容詞"
) or (
pos1 == "形状詞"
)
def attaches_to_predicate(pos):
pos1, pos2, pos3, pos4 = pos
# ます, た, ない, れる, られる, たい, etc.
if pos1 == "助動詞":
return True
# helper verbs like いる, ある, しまう in constructions
if pos1 == "動詞" and pos2 == "非自立可能":
return True
# te-form connector in 読んでいる, 食べている
if pos1 == "助詞" and pos2 == "接続助詞":
return True
# suffixes attached to predicates
if pos1 == "接尾辞":
return True
return False
def chunk_predicates(tokens):
"""
tokens should be a list of TokenReading objects:
token.surface
token.lemma
token.pos
"""
chunks = []
i = 0
while i < len(tokens):
token = tokens[i]
if not is_predicate_start(token.pos):
i += 1
continue
chunk = [token]
j = i + 1
while j < len(tokens) and attaches_to_predicate(tokens[j].pos):
chunk.append(tokens[j])
j += 1
chunks.append({
"surface": "".join(t.surface for t in chunk),
"head_lemma": token.lemma,
"head_reading_hiragana": token.lemma_reading_hiragana,
"pos": token.pos,
"parts": [
{
"surface": t.surface,
"lemma": t.lemma,
"reading_hiragana": t.reading_hiragana,
"pos": t.pos,
}
for t in chunk
],
})
i = j
return chunks
## Vocab Filtering
def is_vocab_token(pos, keep_pronouns=False, keep_numbers=False):
pos1, pos2, pos3, pos4 = pos
if pos1 not in {"名詞", "動詞", "形容詞", "形状詞", "副詞", "代名詞"}:
return False
if pos1 == "代名詞" and not keep_pronouns:
return False
if pos1 == "名詞" and pos2 == "数詞" and not keep_numbers:
return False
return True
@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
if not is_vocab_token(t.pos, keep_pronouns=True,keep_numbers=False):
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))
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))
predicate_chunks = chunk_predicates(tokens)
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],
"predicate_chunks" : predicate_chunks,
}