diff --git a/pythainlp/corpus/common.py b/pythainlp/corpus/common.py index 7acc36e27..075b1e62a 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,31 @@ 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] - + # Skip completely empty or whitespace-only lines without warning. + if not line.strip(): + continue + 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 (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, + ) + continue provs.add(prov["name_th"]) prov_details.append(prov) @@ -294,12 +312,23 @@ 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"]) + 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 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 @@ -317,17 +346,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 (SyntaxError, TypeError, ValueError): + 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 +397,26 @@ 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] - + 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 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"]] + )