#!/usr/bin/env python3 """Re-run the vowel-length check for the Phayan post thai-vowel-length-exceptions. Reads the 3,000 most frequent words of the Thai National Corpus (as PyThaiNLP packages it) against English Wiktionary's respellings (kaikki.org extract), counts the words whose spoken vowel length differs from the written one, sizes the spelling groups the post's rules rest on, derives each exception's tone from its spelling at both lengths and compares it with Wiktionary's IPA, and re-reads every page the post quotes. Exits 0 when every recorded value still reproduces and every recorded quote is still on its page, 1 otherwise. Stdlib only. python3 check.py full run python3 check.py --recheck same; no step is slower than a minute apart from the two downloads, which it still needs python3 check.py --cache DIR keep the two downloads in DIR between runs python3 check.py --write rewrite dataset.json and dataset.csv from this run """ import csv import html import json import os import re import sys import urllib.parse import urllib.request HERE = os.path.dirname(os.path.abspath(__file__)) UA = "PhayanBlogEvidence/1.0 (https://www.phayan.com/blog)" FREQ_URL = "https://raw.githubusercontent.com/PyThaiNLP/pythainlp/dev/pythainlp/corpus/tnc_freq.txt" KAIKKI_URL = "https://kaikki.org/dictionary/Thai/kaikki.org-dictionary-Thai.jsonl" WIKT = "https://en.wiktionary.org/wiki/" PAGES = [ ("wikt-nam", WIKT + urllib.parse.quote("น้ำ"), "น้าม", "Wiktionary respells น้ำ as น้าม, with a long vowel"), ("wikt-nam-bound", WIKT + urllib.parse.quote("น้ำ"), "นั้ม-", "Wiktionary lists a short bound form of น้ำ, นั้ม-, used inside compounds"), ("wikt-len", WIKT + urllib.parse.quote("เล่น"), "เล็่น", "Wiktionary respells เล่น as เล็่น, with the shortening mark"), ("wikt-ngoen", WIKT + urllib.parse.quote("เงิน"), "เงิ็น", "Wiktionary respells เงิน as เงิ็น, with the shortening mark"), ("rtgs", "https://en.wikipedia.org/wiki/Royal_Thai_General_System_of_Transcription", "It does not differentiate between short and long vowels", "RTGS writes long and short vowels alike"), ("thai-dead-syllables", "https://en.wikipedia.org/wiki/Thai_language", 'For "dead syllables", those that end in a plosive ( /p/, /t/, /k/ ) or in a short vowel, ' "only three tonal distinctions are possible", "a dead syllable ends in a stop or a short vowel"), ("thaipod101", "https://www.thaipod101.com/", "Short Audio & Video Lessons for Fast and Easy Learning", "ThaiPod101 describes itself as short audio and video lessons"), ("anki", "https://apps.ankiweb.net/", "Anki is a flashcard program that helps you spend more time on challenging material", "Anki describes itself as a flashcard program"), ("pythainlp-tnc", "https://pythainlp.org/docs/5.4.0/api/corpus.html", "The Thai National Corpus (TNC) is a collection of text data in the Thai language", "PyThaiNLP ships the Thai National Corpus word frequencies"), ] # ---- Thai spelling ----------------------------------------------------------- TONE = '่้๊๋' MARKS = 'ฺ์' CONS = 'กขฃคฅฆงจฉชซฌญฎฏฐฑฒณดตถทธนบปผฝพฟภมยรลวศษสหฬอฮ' MID = 'กจฎฏดตบปอ' HIGH = 'ขฃฉฐถผฝศษสห' LOW_SONORANT = 'งญณนมยรลวฬ' C = '[' + CONS + ']+' # (pattern, written length, spelling group) UNITS = [(re.compile(p), l, g) for p, l, g in [ ('เ' + C + 'าะ', 'S', 'other'), ('เ' + C + 'อะ', 'S', 'other'), ('เ' + C + 'า', 'S', 'เ-า'), ('เ' + C + 'ือ', 'L', 'other'), ('เ' + C + 'ีย', 'L', 'other'), ('เ' + C + 'อ', 'L', 'other'), ('เ' + C + 'ิ', 'L', 'other'), ('เ' + C + 'ี', 'L', 'other'), ('เ' + C, 'L', 'เ-'), ('แ' + C + 'าะ', 'S', 'other'), ('แ' + C + 'ะ', 'S', 'other'), ('แ' + C, 'L', 'แ-'), ('โ' + C + 'ะ', 'S', 'other'), ('โ' + C, 'L', 'โ-'), ('ไ', 'S', 'ไ-/ใ-'), ('ใ', 'S', 'ไ-/ใ-'), ('ำ', 'S', '◌ำ'), ('ั', 'S', 'other'), ('า', 'L', 'other'), ('ิ', 'S', 'other'), ('ี', 'L', 'other'), ('ุ', 'S', 'other'), ('ู', 'L', 'other'), ('ึ', 'S', 'other'), ('ื', 'L', 'other'), ('ะ', 'S', 'other'), ]] def clean(s): return ''.join(c for c in s if c not in TONE and c not in MARKS) def nuclei(s): """[(length, group)] for every vowel the string writes.""" t = clean(s).replace('็', '') out = [] pos = 0 while pos < len(t): for pat, length, group in UNITS: m = pat.match(t, pos) if m: out.append((length, group)) pos = m.end() break else: if t[pos] == 'อ' and pos > 0 and t[pos - 1] in CONS: out.append(('L', '-อ')) pos += 1 return out def length_if_monosyllabic(word): """'L' or 'S' when the string writes exactly one vowel, else None.""" ns = nuclei(word) if len(ns) != 1: return None return 'S' if '็' in clean(word) else ns[0][0] def group(word): return nuclei(word)[0][1] def initial_class(word): t = [c for c in clean(word) if c in CONS] first = t[0] if first == 'ห' and len(t) > 1 and t[1] in LOW_SONORANT: return 'high' if first in MID: return 'mid' if first in HIGH: return 'high' return 'low' def derive_tone(cls, mark, dead, length): if mark == '่': return 'L' if cls in ('mid', 'high') else 'F' if mark == '้': return 'F' if cls in ('mid', 'high') else 'H' if mark == '๊': return 'H' if mark == '๋': return 'R' if not dead: return {'mid': 'M', 'high': 'R', 'low': 'M'}[cls] if cls in ('mid', 'high'): return 'L' return 'H' if length == 'S' else 'F' IPA_TONES = {'˧': 'M', '˨˩': 'L', '˥˩': 'F', '˦˥': 'H', '˩˩˦': 'R', '˩˦': 'R', '˨˦': 'R'} STOPS = 'ptkʔ' SONORANTS = 'mnŋjw' def ipa_parts(ipa): """(tone, final) from a one-syllable IPA string such as /naːm˦˥/.""" core = ipa.strip('/[] ').rstrip('.') tone = IPA_TONES.get(''.join(c for c in core if c in '˥˦˧˨˩')) seg = ''.join(c for c in core if c not in '˥˦˧˨˩̚ .') last = seg[-1] if seg else '' if last in STOPS: final = 'stop' elif last in SONORANTS: final = 'sonorant' else: final = 'vowel' return tone, final # ---- fetch ------------------------------------------------------------------- CACHE = None def fetch(url, binary=False): if CACHE and url in (FREQ_URL, KAIKKI_URL): path = os.path.join(CACHE, url.rsplit('/', 1)[1]) if os.path.exists(path): data = open(path, 'rb').read() return data if binary else data.decode('utf-8', 'replace') req = urllib.request.Request(url, headers={"User-Agent": UA}) for attempt in range(4): try: with urllib.request.urlopen(req, timeout=600) as r: data = r.read() break except Exception as e: if attempt == 3: raise RuntimeError("%s: %s" % (url, e)) if CACHE and url in (FREQ_URL, KAIKKI_URL): os.makedirs(CACHE, exist_ok=True) open(os.path.join(CACHE, url.rsplit('/', 1)[1]), 'wb').write(data) return data if binary else data.decode('utf-8', 'replace') def normalise(text): text = re.sub(r'|', ' ', text, flags=re.S) text = re.sub(r'<[^>]+>', ' ', text) text = html.unescape(text) text = (text.replace('’', "'").replace('‘', "'") .replace('“', '"').replace('”', '"').replace('\xa0', ' ')) return re.sub(r'\s+', ' ', text) # ---- the corpus check -------------------------------------------------------- def read_entries(wanted): body = fetch(KAIKKI_URL, binary=True).decode('utf-8', 'replace') rec, ipa = {}, {} for line in body.splitlines(): if '"word"' not in line: continue try: d = json.loads(line) except json.JSONDecodeError: continue w = d.get('word') if w not in wanted or w in rec: continue phon, first_ipa = None, None for s in d.get('sounds') or []: tags = (s.get('raw_tags') or []) + (s.get('tags') or []) if 'Orthographic/Phonemic' in tags and phon is None: phon = s.get('other') if 'Phonemic' in tags and 'bound form' not in tags and phon is None: phon = s.get('other') if s.get('ipa') and first_ipa is None and 'bound form' not in tags: first_ipa = s['ipa'] if phon is not None: rec[w] = phon ipa[w] = first_ipa return rec, ipa def compute(): freq = fetch(FREQ_URL) rows = [l.split('\t') for l in freq.splitlines() if '\t' in l] top = [(w, int(c)) for w, c in rows[:3000]] total = sum(int(c) for w, c in rows) rec, ipa = read_entries({w for w, _ in top}) mono, words = [], [] for rank, (w, c) in enumerate(top, 1): if w not in rec: continue spoken_src = rec[w] or w if '-' in spoken_src: continue wl = length_if_monosyllabic(w) pl = length_if_monosyllabic(spoken_src) if not (wl and pl): continue tone_ipa, final = ipa_parts(ipa[w]) if ipa[w] else (None, None) mark = next((ch for ch in w if ch in TONE), '') item = {"rank": rank, "word": w, "count": c, "written": wl, "spoken": pl, "phonemic": spoken_src, "group": group(w), "tone_mark": mark, "ipa": ipa[w], "final": final} mono.append(item) if wl != pl: cls = initial_class(w) def dead(length): return final == 'stop' or (final == 'vowel' and length == 'S') item.update({ "share_pct": round(c / total * 100, 3), "class": cls, "tone_from_written_length": derive_tone(cls, mark, dead(wl), wl), "tone_from_spoken_length": derive_tone(cls, mark, dead(pl), pl), "tone_ipa": tone_ipa, }) words.append(item) pools = {} for g in ('เ-', 'แ-', '-อ', 'โ-'): members = [x for x in mono if x['group'] == g and x['tone_mark'] in '่้' and x['tone_mark'] and x['final'] in ('stop', 'sonorant')] pools[g] = {"rule": "closed, carries ่ or ้", "total": len(members), "said_short": sum(1 for x in members if x['spoken'] == 'S'), "long": [x['word'] for x in members if x['spoken'] == 'L'], "short": [x['word'] for x in members if x['spoken'] == 'S']} for g in ('เ-า', 'ไ-/ใ-', '◌ำ'): members = [x for x in mono if x['group'] == g and x['tone_mark']] pools[g] = {"rule": "carries a tone mark", "total": len(members), "said_long": sum(1 for x in members if x['spoken'] == 'L'), "long": [x['word'] for x in members if x['spoken'] == 'L'], "short": [x['word'] for x in members if x['spoken'] == 'S']} long_short = [x for x in words if x['written'] == 'L'] summary = { "words_in_list": len(top), "corpus_tokens": total, "with_wiktionary_respelling": len(rec), "monosyllabic": len(mono), "exceptions": len(words), "exceptions_share_of_monosyllabic_pct": round(len(words) / len(mono) * 100, 1), "short_written_long_spoken": len(words) - len(long_short), "long_written_short_spoken": len(long_short), "long_written_short_spoken_without_tone_mark": sum(1 for x in long_short if not x['tone_mark']), "long_written_short_spoken_in_plain_e_ae_or": sum( 1 for x in long_short if x['group'] in ('เ-', 'แ-', '-อ')), "exceptions_share_of_corpus_pct": round(sum(x['count'] for x in words) / total * 100, 2), "exceptions_in_top_100": sum(1 for x in words if x['rank'] <= 100), "exceptions_in_top_1000": sum(1 for x in words if x['rank'] <= 1000), "exceptions_ending_in_stop": sum(1 for x in words if x['final'] == 'stop'), "exceptions_where_length_changes_derived_tone": sum( 1 for x in words if x['tone_from_written_length'] != x['tone_from_spoken_length']), "exceptions_where_spoken_length_tone_matches_ipa": sum( 1 for x in words if x['tone_from_spoken_length'] == x['tone_ipa']), "exceptions_where_written_length_tone_matches_ipa": sum( 1 for x in words if x['tone_from_written_length'] == x['tone_ipa']), } return summary, pools, words def page_checks(): out = [] for pid, url, quote, claim in PAGES: try: text = normalise(fetch(url)) found = normalise(quote) in text out.append({"id": pid, "url": url, "quote": quote, "claim": claim, "found": found}) except Exception as e: out.append({"id": pid, "url": url, "quote": quote, "claim": claim, "found": False, "error": str(e)}) return out CSV_FIELDS = ["rank", "word", "count", "share_pct", "written", "spoken", "phonemic", "group", "tone_mark", "ipa", "final", "class", "tone_from_written_length", "tone_from_spoken_length", "tone_ipa"] def write(summary, pools, words, pages): path = os.path.join(HERE, 'dataset.json') ds = json.load(open(path, encoding='utf-8')) ds['results'] = {"summary": summary, "pools": pools, "words": words, "pages": pages} with open(path, 'w', encoding='utf-8') as f: json.dump(ds, f, ensure_ascii=False, indent=1) f.write('\n') with open(os.path.join(HERE, 'dataset.csv'), 'w', encoding='utf-8', newline='') as f: wr = csv.writer(f) wr.writerow(['kind'] + CSV_FIELDS + ['value', 'quote', 'url']) for k, v in summary.items(): wr.writerow(['summary', '', k] + [''] * (len(CSV_FIELDS) - 2) + [v, '', FREQ_URL]) for g, p in pools.items(): n = p.get('said_short', p.get('said_long')) label = 'said_short' if 'said_short' in p else 'said_long' wr.writerow(['pool', '', g] + [''] * (len(CSV_FIELDS) - 2) + ['%s %d of %d (%s)' % (label, n, p['total'], p['rule']), '', KAIKKI_URL]) for x in words: wr.writerow(['word'] + [x.get(k, '') for k in CSV_FIELDS] + ['', '', WIKT + x['word']]) for p in pages: wr.writerow(['page', '', p['id']] + [''] * (len(CSV_FIELDS) - 2) + [p['claim'], p['quote'], p['url']]) def main(args): global CACHE if '--cache' in args: CACHE = args[args.index('--cache') + 1] summary, pools, words = compute() pages = page_checks() if '--write' in args: write(summary, pools, words, pages) ds = json.load(open(os.path.join(HERE, 'dataset.json'), encoding='utf-8')) want = ds['results'] ok = True print("corpus check (%d words)" % summary['words_in_list']) for key, value in want['summary'].items(): mark = 'ok ' if summary.get(key) == value else 'FAIL' ok &= mark == 'ok ' print(" %s %-46s recorded %s now %s" % (mark, key, value, summary.get(key))) for g, p in want['pools'].items(): same = pools.get(g) == p ok &= same n = p.get('said_short', p.get('said_long')) print(" %s pool %-8s %d of %d" % ('ok ' if same else 'FAIL', g, n, p['total'])) same = words == want['words'] ok &= same print(" %s %-46s %d words" % ('ok ' if same else 'FAIL', 'word list', len(words))) print("page checks") for p in pages: ok &= p['found'] status = 'ok ' if p['found'] else 'FAIL' print(" %s %-20s %s%s" % (status, p['id'], p['url'], '' if p['found'] else ' ' + p.get('error', 'quote gone'))) print("PASS" if ok else "FAIL") return 0 if ok else 1 if __name__ == '__main__': argv = sys.argv[1:] allowed = {'--recheck', '--write', '--cache'} if any(a.startswith('--') and a not in allowed for a in argv): print(__doc__, file=sys.stderr) sys.exit(2) sys.exit(main(argv))