jp_text_parsing/fugashi_parser.py

427 lines
13 KiB
Python

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from __future__ import annotations
import argparse
import html
import json
import os
from pathlib import Path
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)
CONTENT_LINK_TAG = {
'名詞': 'meishi', # noun
'動詞': 'doshi', # verb
'形容詞': 'keiyoshi' # Adjective
#'副詞' : 'fukushi' #adverb
}
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:
pos1, pos2, pos3, pos4 = pos
if not is_vocab_token(pos, keep_pronouns=True,keep_numbers=False):
return surface
# if surface == reading:
# return f"[[{surface}]]"
return f"[[{reading}_{pos1}|{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,
}
YOMITAN_TO_MICAB_POS = {
'interjection': 'foo',
'noun': 'meishi',
'5-dan': '',
'1-dan': '',
'prefix': '',
'intransitive': '',
'aux-verb': '',
'suru': '',
'transitive': '',
'counter': '',
'na-adj': '',
'exp': '',
'adjective': '',
'suffix': '',
'conjunction': '',
'interjection': '',
'adverb': '',
'pronoun': '',
'unclass': '',
}
def write_word_note(word: str, destination_dir: Path, part_of_speech:str='', overwrite:bool=False):
word_file_name = word
# Not ready for different file names yet
if part_of_speech != '':
word_file_name = word + '_' + part_of_speech
if (Path(destination_dir.joinpath(word_file_name + '.md')).exists() == False) or overwrite == True:
definitions = anki_fields_term(word)
if definitions.get("fields"):
with open(destination_dir.joinpath(word_file_name + '.md'), 'w') as vocab_note:
for entry in definitions.get("fields"):
#pos = entry['part-of-speech'] ## POS is always 'Unknown' for some reason
pos = re.match(r'.*<span.*?"part-of-speech-info".*?>(.*?)<',entry['glossary-first'])
if pos:
pos = pos.group(1)
# if part_of_speech != '' and pos:
# print(pos)
#vocab_note.write("\n")
#print(entry['expression'], file=vocab_note)
#print(entry['furigana-plain'], file=vocab_note)
print(entry['furigana'], file=vocab_note)
if entry['expression'] != entry['reading']:
print(entry['reading'], file=vocab_note)
print("> ",re.sub(r'<.*?>', '', entry['glossary-plain-no-dictionary']), "\n", file=vocab_note)
#print(f"Part of Speech: {entry['part-of-speech']}", file=vocab_note)
print(f"Frequency: {entry['frequency-average-rank']}", file=vocab_note)
#print(entry['glossary-plain-no-dictionary'], file=vocab_note)
vocab_note.write(entry['glossary-first'])
#print("##### Other Definitions", file=vocab_note)
print("\n", file=vocab_note)
# if CONTENT_LINK_TAG.get(part_of_speech, '') != '':
# print(f"#{CONTENT_LINK_TAG[part_of_speech]}", file=vocab_note)