From 63f24e3f56694c0401c69ce4bff49ad8284a2f7f Mon Sep 17 00:00:00 2001 From: Arthit Suriyawongkul Date: Sat, 14 Mar 2026 02:43:20 +0700 Subject: [PATCH 1/3] Defensive file loading Check row completeness before append to dictionary Check types before evaluate Signed-off-by: Arthit Suriyawongkul --- pythainlp/corpus/common.py | 100 +++++++++++++++++++++++++++---------- 1 file changed, 73 insertions(+), 27 deletions(-) diff --git a/pythainlp/corpus/common.py b/pythainlp/corpus/common.py index 7acc36e27..c0a933646 100644 --- a/pythainlp/corpus/common.py +++ b/pythainlp/corpus/common.py @@ -7,6 +7,7 @@ from __future__ import annotations import ast +import warnings from typing import TYPE_CHECKING if TYPE_CHECKING: @@ -109,14 +110,21 @@ def provinces( prov_details = [] for line in get_corpus_as_is(_THAI_THAILAND_PROVINCES_FILENAME): - p = line.split(",") - - prov = {} - prov["name_th"] = p[0] - prov["abbr_th"] = p[1] - prov["name_en"] = p[2] - prov["abbr_en"] = p[3] - + parts = line.split(",") + try: + prov = { + "name_th": parts[0], + "abbr_th": parts[1], + "name_en": parts[2], + "abbr_en": parts[3], + } + except IndexError: + warnings.warn( + f"Skipping malformed province entry: {line!r}", + UserWarning, + stacklevel=2, + ) + continue provs.add(prov["name_th"]) prov_details.append(prov) @@ -294,12 +302,22 @@ def thai_dict() -> dict[str, list[str]]: return _THAI_DICT path = str(path) - _THAI_DICT = {"word": [], "meaning": []} + words: list[str] = [] + meanings: list[str] = [] with open(path, newline="\n", encoding="utf-8") as csvfile: reader = csv.DictReader(csvfile, delimiter=",") for row in reader: - _THAI_DICT["word"].append(row["word"]) - _THAI_DICT["meaning"].append(row["meaning"]) + try: + words.append(row["word"]) + meanings.append(row["meaning"]) + except KeyError as e: + warnings.warn( + f"Skipping malformed thai_dict entry, missing column {e}: {dict(row)!r}", + UserWarning, + stacklevel=2, + ) + continue + _THAI_DICT = {"word": words, "meaning": meanings} return _THAI_DICT @@ -317,17 +335,35 @@ def thai_wsd_dict() -> dict[str, Union[list[str], list[list[str]]]]: return _THAI_WSD_DICT thai_wsd = thai_dict() - _THAI_WSD_DICT = {"word": [], "meaning": []} - for i, j in zip(thai_wsd["word"], thai_wsd["meaning"]): - all_value = list(ast.literal_eval(j).values()) - use = [] - for k in all_value: - use.extend(k) - use = list(set(use)) - if len(use) > 1: - _THAI_WSD_DICT["word"].append(i) # type: ignore[arg-type] - _THAI_WSD_DICT["meaning"].append(use) # type: ignore[arg-type] - + words: list[str] = [] + meanings: list[list[str]] = [] + for word, meaning in zip(thai_wsd["word"], thai_wsd["meaning"]): + try: + parsed = ast.literal_eval(meaning) + except (ValueError, SyntaxError): + warnings.warn( + f"Skipping thai_wsd_dict entry for word {word!r}: " + f"meaning could not be parsed: {meaning!r}", + UserWarning, + stacklevel=2, + ) + continue + if not isinstance(parsed, dict): + warnings.warn( + f"Skipping thai_wsd_dict entry for word {word!r}: " + f"expected dict after parsing, got {type(parsed).__name__!r}", + UserWarning, + stacklevel=2, + ) + continue + senses: list[str] = [] + for sense_list in parsed.values(): + senses.extend(sense_list) + senses = list(set(senses)) + if len(senses) > 1: + words.append(word) + meanings.append(senses) + _THAI_WSD_DICT = {"word": words, "meaning": meanings} return _THAI_WSD_DICT @@ -350,14 +386,24 @@ def thai_synonyms() -> dict[str, Union[list[str], list[list[str]]]]: return _THAI_SYNONYMS path = str(path) - _THAI_SYNONYMS = {"word": [], "pos": [], "synonym": []} + words: list[str] = [] + pos_tags: list[str] = [] + synonym_groups: list[list[str]] = [] with open(path, newline="\n", encoding="utf-8") as csvfile: reader = csv.DictReader(csvfile, delimiter=",") for row in reader: - _THAI_SYNONYMS["word"].append(row["word"]) # type: ignore[arg-type] - _THAI_SYNONYMS["pos"].append(row["pos"]) # type: ignore[arg-type] - _THAI_SYNONYMS["synonym"].append(row["synonym"].split("|")) # type: ignore[arg-type] - + try: + words.append(row["word"]) + pos_tags.append(row["pos"]) + synonym_groups.append(row["synonym"].split("|")) + except KeyError as e: + warnings.warn( + f"Skipping malformed thai_synonyms entry, missing column {e}: {dict(row)!r}", + UserWarning, + stacklevel=2, + ) + continue + _THAI_SYNONYMS = {"word": words, "pos": pos_tags, "synonym": synonym_groups} return _THAI_SYNONYMS From 9f9ec05601d6e052767ccd562c9f5b516935799e Mon Sep 17 00:00:00 2001 From: Arthit Suriyawongkul Date: Sat, 14 Mar 2026 03:34:45 +0700 Subject: [PATCH 2/3] Add more validation and tests Signed-off-by: Arthit Suriyawongkul --- pythainlp/corpus/common.py | 36 +++-- tests/core/test_corpus.py | 305 +++++++++++++++++++++++++++++++++++++ 2 files changed, 328 insertions(+), 13 deletions(-) diff --git a/pythainlp/corpus/common.py b/pythainlp/corpus/common.py index c0a933646..ca1b0ec44 100644 --- a/pythainlp/corpus/common.py +++ b/pythainlp/corpus/common.py @@ -120,7 +120,14 @@ def provinces( } except IndexError: warnings.warn( - f"Skipping malformed province entry: {line!r}", + f"Skipping malformed province entry (too few fields): {line!r}", + UserWarning, + stacklevel=2, + ) + continue + if not all(v.strip() for v in prov.values()): + warnings.warn( + f"Skipping province entry with blank or empty field(s): {line!r}", UserWarning, stacklevel=2, ) @@ -307,16 +314,17 @@ def thai_dict() -> dict[str, list[str]]: with open(path, newline="\n", encoding="utf-8") as csvfile: reader = csv.DictReader(csvfile, delimiter=",") for row in reader: - try: - words.append(row["word"]) - meanings.append(row["meaning"]) - except KeyError as e: + word = row.get("word") + meaning = row.get("meaning") + if not word or not word.strip() or not meaning or not meaning.strip(): warnings.warn( - f"Skipping malformed thai_dict entry, missing column {e}: {dict(row)!r}", + f"Skipping thai_dict entry with missing or empty field(s): {dict(row)!r}", UserWarning, stacklevel=2, ) continue + words.append(word) + meanings.append(meaning) _THAI_DICT = {"word": words, "meaning": meanings} return _THAI_DICT @@ -340,7 +348,7 @@ def thai_wsd_dict() -> dict[str, Union[list[str], list[list[str]]]]: for word, meaning in zip(thai_wsd["word"], thai_wsd["meaning"]): try: parsed = ast.literal_eval(meaning) - except (ValueError, SyntaxError): + except (SyntaxError, TypeError, ValueError): warnings.warn( f"Skipping thai_wsd_dict entry for word {word!r}: " f"meaning could not be parsed: {meaning!r}", @@ -392,17 +400,19 @@ def thai_synonyms() -> dict[str, Union[list[str], list[list[str]]]]: with open(path, newline="\n", encoding="utf-8") as csvfile: reader = csv.DictReader(csvfile, delimiter=",") for row in reader: - try: - words.append(row["word"]) - pos_tags.append(row["pos"]) - synonym_groups.append(row["synonym"].split("|")) - except KeyError as e: + word = row.get("word") + pos = row.get("pos") + synonym = row.get("synonym") + if not word or not word.strip() or not pos or not pos.strip() or not synonym or not synonym.strip(): warnings.warn( - f"Skipping malformed thai_synonyms entry, missing column {e}: {dict(row)!r}", + f"Skipping thai_synonyms entry with missing or empty field(s): {dict(row)!r}", UserWarning, stacklevel=2, ) continue + words.append(word) + pos_tags.append(pos) + synonym_groups.append(synonym.split("|")) _THAI_SYNONYMS = {"word": words, "pos": pos_tags, "synonym": synonym_groups} return _THAI_SYNONYMS diff --git a/tests/core/test_corpus.py b/tests/core/test_corpus.py index a7c4d3e38..58ea2ec34 100644 --- a/tests/core/test_corpus.py +++ b/tests/core/test_corpus.py @@ -316,3 +316,308 @@ def test_revise_wordset(self): ["ที่", "ถูก", "สังหาร", "เมื่อ", "ปี", " ", "พ.ศ.", " ", "2492"], ] self.assertIsInstance(revise_newmm_default_wordset(training_data), set) + + +class DefensiveLoadingTestCase(unittest.TestCase): + """Tests for warning-based defensive loading in corpus/common.py.""" + + # ------------------------------------------------------------------ + # provinces() + # ------------------------------------------------------------------ + + def test_provinces_skips_short_line_with_warning(self): + import pythainlp.corpus.common as m + + with patch.object(m, "_THAI_THAILAND_PROVINCES", frozenset()): + with patch.object(m, "_THAI_THAILAND_PROVINCES_DETAILS", []): + with patch( + "pythainlp.corpus.common.get_corpus_as_is", + return_value=["กรุงเทพ,กทม"], # only 2 fields, needs 4 + ): + with self.assertWarns(UserWarning) as cm: + result = m.provinces() + self.assertEqual(len(result), 0) + self.assertIn("too few fields", str(cm.warning)) + + def test_provinces_skips_blank_field_with_warning(self): + import pythainlp.corpus.common as m + + with patch.object(m, "_THAI_THAILAND_PROVINCES", frozenset()): + with patch.object(m, "_THAI_THAILAND_PROVINCES_DETAILS", []): + with patch( + "pythainlp.corpus.common.get_corpus_as_is", + return_value=["กรุงเทพ,,Bangkok,BKK"], # blank abbr_th + ): + with self.assertWarns(UserWarning) as cm: + result = m.provinces() + self.assertEqual(len(result), 0) + self.assertIn("blank or empty", str(cm.warning)) + + def test_provinces_skips_whitespace_only_field_with_warning(self): + import pythainlp.corpus.common as m + + with patch.object(m, "_THAI_THAILAND_PROVINCES", frozenset()): + with patch.object(m, "_THAI_THAILAND_PROVINCES_DETAILS", []): + with patch( + "pythainlp.corpus.common.get_corpus_as_is", + return_value=["กรุงเทพ, ,Bangkok,BKK"], # whitespace-only abbr_th + ): + with self.assertWarns(UserWarning): + result = m.provinces() + self.assertEqual(len(result), 0) + + def test_provinces_loads_valid_entry(self): + import pythainlp.corpus.common as m + + with patch.object(m, "_THAI_THAILAND_PROVINCES", frozenset()): + with patch.object(m, "_THAI_THAILAND_PROVINCES_DETAILS", []): + with patch( + "pythainlp.corpus.common.get_corpus_as_is", + return_value=["กรุงเทพมหานคร,กทม,Bangkok,BKK"], + ): + result = m.provinces() + self.assertIn("กรุงเทพมหานคร", result) + + # ------------------------------------------------------------------ + # thai_dict() + # ------------------------------------------------------------------ + + def test_thai_dict_skips_none_word_with_warning(self): + import pythainlp.corpus.common as m + + mock_rows = [ + {"word": "แมว", "meaning": "cat"}, + {"word": None, "meaning": "some meaning"}, # None word + ] + with patch.object(m, "_THAI_DICT", {}): + with patch( + "pythainlp.corpus.common.get_corpus_path", + return_value="/mock/path", + ): + with patch("builtins.open", mock_open()): + with patch("csv.DictReader", return_value=mock_rows): + with self.assertWarns(UserWarning) as cm: + result = m.thai_dict() + self.assertEqual(result["word"], ["แมว"]) + self.assertIn("missing or empty", str(cm.warning)) + + def test_thai_dict_skips_empty_word_with_warning(self): + import pythainlp.corpus.common as m + + mock_rows = [ + {"word": "แมว", "meaning": "cat"}, + {"word": "", "meaning": "empty word field"}, # empty word + ] + with patch.object(m, "_THAI_DICT", {}): + with patch( + "pythainlp.corpus.common.get_corpus_path", + return_value="/mock/path", + ): + with patch("builtins.open", mock_open()): + with patch("csv.DictReader", return_value=mock_rows): + with self.assertWarns(UserWarning): + result = m.thai_dict() + self.assertEqual(result["word"], ["แมว"]) + + def test_thai_dict_skips_whitespace_only_word_with_warning(self): + import pythainlp.corpus.common as m + + mock_rows = [ + {"word": "แมว", "meaning": "cat"}, + {"word": " ", "meaning": "whitespace word"}, + ] + with patch.object(m, "_THAI_DICT", {}): + with patch( + "pythainlp.corpus.common.get_corpus_path", + return_value="/mock/path", + ): + with patch("builtins.open", mock_open()): + with patch("csv.DictReader", return_value=mock_rows): + with self.assertWarns(UserWarning): + result = m.thai_dict() + self.assertEqual(result["word"], ["แมว"]) + + def test_thai_dict_skips_none_meaning_with_warning(self): + import pythainlp.corpus.common as m + + mock_rows = [ + {"word": "แมว", "meaning": "cat"}, + {"word": "หมา", "meaning": None}, # None meaning (short CSV row) + ] + with patch.object(m, "_THAI_DICT", {}): + with patch( + "pythainlp.corpus.common.get_corpus_path", + return_value="/mock/path", + ): + with patch("builtins.open", mock_open()): + with patch("csv.DictReader", return_value=mock_rows): + with self.assertWarns(UserWarning): + result = m.thai_dict() + self.assertEqual(result["word"], ["แมว"]) + + def test_thai_dict_returns_empty_when_no_path(self): + import pythainlp.corpus.common as m + + with patch.object(m, "_THAI_DICT", {}): + with patch( + "pythainlp.corpus.common.get_corpus_path", return_value=None + ): + result = m.thai_dict() + self.assertEqual(result, {}) + + # ------------------------------------------------------------------ + # thai_wsd_dict() + # ------------------------------------------------------------------ + + def test_thai_wsd_dict_skips_none_meaning_with_warning(self): + import pythainlp.corpus.common as m + + # None meaning causes TypeError in ast.literal_eval + mock_source = {"word": ["แมว"], "meaning": [None]} + with patch.object(m, "_THAI_WSD_DICT", {}): + with patch("pythainlp.corpus.common.thai_dict", return_value=mock_source): + with self.assertWarns(UserWarning) as cm: + result = m.thai_wsd_dict() + self.assertEqual(result["word"], []) + self.assertIn("could not be parsed", str(cm.warning)) + + def test_thai_wsd_dict_skips_unparseable_meaning_with_warning(self): + import pythainlp.corpus.common as m + + mock_source = {"word": ["แมว"], "meaning": ["not valid python literal!!!"]} + with patch.object(m, "_THAI_WSD_DICT", {}): + with patch("pythainlp.corpus.common.thai_dict", return_value=mock_source): + with self.assertWarns(UserWarning) as cm: + result = m.thai_wsd_dict() + self.assertEqual(result["word"], []) + self.assertIn("could not be parsed", str(cm.warning)) + + def test_thai_wsd_dict_skips_non_dict_meaning_with_warning(self): + import pythainlp.corpus.common as m + + # Parses OK but yields a list, not a dict + mock_source = {"word": ["แมว"], "meaning": ["['cat', 'kitty']"]} + with patch.object(m, "_THAI_WSD_DICT", {}): + with patch("pythainlp.corpus.common.thai_dict", return_value=mock_source): + with self.assertWarns(UserWarning) as cm: + result = m.thai_wsd_dict() + self.assertEqual(result["word"], []) + self.assertIn("expected dict", str(cm.warning)) + + # ------------------------------------------------------------------ + # thai_synonyms() + # ------------------------------------------------------------------ + + def test_thai_synonyms_skips_none_synonym_with_warning(self): + import pythainlp.corpus.common as m + + mock_rows = [ + {"word": "แมว", "pos": "n", "synonym": "cat|kitty"}, + {"word": "หมา", "pos": "n", "synonym": None}, # None synonym + ] + with patch.object(m, "_THAI_SYNONYMS", {}): + with patch( + "pythainlp.corpus.common.get_corpus_path", + return_value="/mock/path", + ): + with patch("builtins.open", mock_open()): + with patch("csv.DictReader", return_value=mock_rows): + with self.assertWarns(UserWarning): + result = m.thai_synonyms() + self.assertEqual(result["word"], ["แมว"]) + self.assertEqual(result["synonym"], [["cat", "kitty"]]) + + def test_thai_synonyms_skips_none_word_with_warning(self): + import pythainlp.corpus.common as m + + mock_rows = [ + {"word": "แมว", "pos": "n", "synonym": "cat|kitty"}, + {"word": None, "pos": "n", "synonym": "dog"}, # None word + ] + with patch.object(m, "_THAI_SYNONYMS", {}): + with patch( + "pythainlp.corpus.common.get_corpus_path", + return_value="/mock/path", + ): + with patch("builtins.open", mock_open()): + with patch("csv.DictReader", return_value=mock_rows): + with self.assertWarns(UserWarning): + result = m.thai_synonyms() + self.assertEqual(result["word"], ["แมว"]) + + def test_thai_synonyms_skips_none_pos_with_warning(self): + import pythainlp.corpus.common as m + + mock_rows = [ + {"word": "แมว", "pos": "n", "synonym": "cat|kitty"}, + {"word": "หมา", "pos": None, "synonym": "dog"}, # None pos + ] + with patch.object(m, "_THAI_SYNONYMS", {}): + with patch( + "pythainlp.corpus.common.get_corpus_path", + return_value="/mock/path", + ): + with patch("builtins.open", mock_open()): + with patch("csv.DictReader", return_value=mock_rows): + with self.assertWarns(UserWarning): + result = m.thai_synonyms() + self.assertEqual(result["word"], ["แมว"]) + + def test_thai_synonyms_skips_empty_pos_with_warning(self): + import pythainlp.corpus.common as m + + mock_rows = [ + {"word": "แมว", "pos": "n", "synonym": "cat|kitty"}, + {"word": "หมา", "pos": "", "synonym": "dog"}, # empty pos + ] + with patch.object(m, "_THAI_SYNONYMS", {}): + with patch( + "pythainlp.corpus.common.get_corpus_path", + return_value="/mock/path", + ): + with patch("builtins.open", mock_open()): + with patch("csv.DictReader", return_value=mock_rows): + with self.assertWarns(UserWarning): + result = m.thai_synonyms() + self.assertEqual(result["word"], ["แมว"]) + + def test_thai_synonyms_skips_whitespace_only_fields_with_warning(self): + import pythainlp.corpus.common as m + + mock_rows = [ + {"word": "แมว", "pos": "n", "synonym": "cat|kitty"}, + {"word": " ", "pos": "n", "synonym": "dog"}, # whitespace word + {"word": "หมา", "pos": " ", "synonym": "dog"}, # whitespace pos + {"word": "ปลา", "pos": "n", "synonym": " "}, # whitespace synonym + ] + with patch.object(m, "_THAI_SYNONYMS", {}): + with patch( + "pythainlp.corpus.common.get_corpus_path", + return_value="/mock/path", + ): + with patch("builtins.open", mock_open()): + with patch("csv.DictReader", return_value=mock_rows): + with self.assertWarns(UserWarning): + result = m.thai_synonyms() + self.assertEqual(result["word"], ["แมว"]) + + def test_thai_synonyms_loads_valid_rows(self): + import pythainlp.corpus.common as m + + mock_rows = [ + {"word": "แมว", "pos": "n", "synonym": "cat|kitty"}, + {"word": "หมา", "pos": "n", "synonym": "dog"}, + ] + with patch.object(m, "_THAI_SYNONYMS", {}): + with patch( + "pythainlp.corpus.common.get_corpus_path", + return_value="/mock/path", + ): + with patch("builtins.open", mock_open()): + with patch("csv.DictReader", return_value=mock_rows): + result = m.thai_synonyms() + self.assertEqual(result["word"], ["แมว", "หมา"]) + self.assertEqual(result["pos"], ["n", "n"]) + self.assertEqual( + result["synonym"], [["cat", "kitty"], ["dog"]] + ) From 7e4b3fe649eb6cacf24ce94749e1a03a438cf670 Mon Sep 17 00:00:00 2001 From: Arthit Suriyawongkul Date: Sat, 14 Mar 2026 03:43:42 +0700 Subject: [PATCH 3/3] Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- pythainlp/corpus/common.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/pythainlp/corpus/common.py b/pythainlp/corpus/common.py index ca1b0ec44..075b1e62a 100644 --- a/pythainlp/corpus/common.py +++ b/pythainlp/corpus/common.py @@ -110,6 +110,9 @@ def provinces( prov_details = [] for line in get_corpus_as_is(_THAI_THAILAND_PROVINCES_FILENAME): + # Skip completely empty or whitespace-only lines without warning. + if not line.strip(): + continue parts = line.split(",") try: prov = {