|
| 1 | +# SPDX-FileCopyrightText: 2016-2026 PyThaiNLP Project |
| 2 | +# SPDX-FileType: SOURCE |
| 3 | +# SPDX-License-Identifier: Apache-2.0 |
| 4 | +"""CRF-based Thai phrase structure (chunk) parser.""" |
| 5 | +from __future__ import annotations |
| 6 | + |
| 7 | +from importlib.resources import as_file, files |
| 8 | +from typing import TYPE_CHECKING, Any, Optional, Union |
| 9 | + |
| 10 | +if TYPE_CHECKING: |
| 11 | + import types |
| 12 | + from contextlib import AbstractContextManager |
| 13 | + |
| 14 | + from pycrfsuite import ( |
| 15 | + Tagger as CRFTagger, # pyright: ignore[reportAttributeAccessIssue] # pyrefly: ignore[missing-module-attribute] |
| 16 | + ) |
| 17 | + |
| 18 | +from pythainlp.corpus import thai_stopwords |
| 19 | + |
| 20 | + |
| 21 | +def _is_stopword(word: str) -> bool: |
| 22 | + return word in thai_stopwords() |
| 23 | + |
| 24 | + |
| 25 | +def _doc2features( |
| 26 | + tokens: list[tuple[str, str]], index: int |
| 27 | +) -> dict[str, Union[str, bool]]: |
| 28 | + """Extract features for a single token in a POS-tagged sentence. |
| 29 | +
|
| 30 | + :param list[tuple[str, str]] tokens: POS-tagged sentence, |
| 31 | + a list of (word, POS-tag) pairs. |
| 32 | + :param int index: index of the token to extract features for. |
| 33 | + :return: feature dictionary for the token. |
| 34 | + :rtype: dict[str, Union[str, bool]] |
| 35 | + """ |
| 36 | + word, pos = tokens[index] |
| 37 | + f: dict[str, Union[str, bool]] = { |
| 38 | + "word": word, |
| 39 | + "word_is_stopword": _is_stopword(word), |
| 40 | + "pos": pos, |
| 41 | + } |
| 42 | + if index > 1: |
| 43 | + prevprevword, prevprevpos = tokens[index - 2] |
| 44 | + f["prev-prev-word"] = prevprevword |
| 45 | + f["prev-prevz-word_is_stopword"] = _is_stopword(prevprevword) |
| 46 | + f["prev-prevz-pos"] = prevprevpos |
| 47 | + if index > 0: |
| 48 | + prevword, prevpos = tokens[index - 1] |
| 49 | + f["prev-word"] = prevword |
| 50 | + f["prev-word_is_stopword"] = _is_stopword(prevword) |
| 51 | + f["prev-pos"] = prevpos |
| 52 | + else: |
| 53 | + f["BOS"] = True |
| 54 | + if index < len(tokens) - 2: |
| 55 | + nextnextword, nextnextpos = tokens[index + 2] |
| 56 | + f["nextnext-word"] = nextnextword |
| 57 | + f["nextnext-word_is_stopword"] = _is_stopword(nextnextword) |
| 58 | + f["nextnext-pos"] = nextnextpos |
| 59 | + if index < len(tokens) - 1: |
| 60 | + nextword, nextpos = tokens[index + 1] |
| 61 | + f["next-word"] = nextword |
| 62 | + f["next-word_is_stopword"] = _is_stopword(nextword) |
| 63 | + f["next-pos"] = nextpos |
| 64 | + else: |
| 65 | + f["EOS"] = True |
| 66 | + |
| 67 | + return f |
| 68 | + |
| 69 | + |
| 70 | +def _extract_features( |
| 71 | + doc: list[tuple[str, str]], |
| 72 | +) -> list[dict[str, Union[str, bool]]]: |
| 73 | + return [_doc2features(doc, i) for i in range(len(doc))] |
| 74 | + |
| 75 | + |
| 76 | +class CRFChunkParser: |
| 77 | + """CRF-based chunk parser for Thai text. |
| 78 | +
|
| 79 | + Parses a POS-tagged sentence into phrase-structure chunks |
| 80 | + (IOB format), following the NLTK :class:`nltk.chunk.ChunkParserI` |
| 81 | + convention. |
| 82 | +
|
| 83 | + This class supports the context manager protocol for deterministic |
| 84 | + resource cleanup: |
| 85 | +
|
| 86 | + .. code-block:: python |
| 87 | +
|
| 88 | + from pythainlp.chunk import CRFChunkParser |
| 89 | +
|
| 90 | + with CRFChunkParser() as parser: |
| 91 | + result = parser.parse(tokens_pos) |
| 92 | +
|
| 93 | + :param str corpus: corpus name for the CRF model |
| 94 | + (default: ``"orchidpp"``). |
| 95 | + """ |
| 96 | + |
| 97 | + corpus: str |
| 98 | + _model_file_ctx: Optional[AbstractContextManager[Any]] |
| 99 | + tagger: CRFTagger |
| 100 | + xseq: list[dict[str, Union[str, bool]]] |
| 101 | + |
| 102 | + def __init__(self, corpus: str = "orchidpp") -> None: |
| 103 | + self.corpus = corpus |
| 104 | + self._model_file_ctx = None |
| 105 | + self.load_model(self.corpus) |
| 106 | + |
| 107 | + def load_model(self, corpus: str) -> None: |
| 108 | + """Load the CRF model for the given corpus. |
| 109 | +
|
| 110 | + :param str corpus: corpus name. |
| 111 | + """ |
| 112 | + from pycrfsuite import ( |
| 113 | + Tagger as CRFTagger, # noqa: PLC0415 # pyright: ignore[reportAttributeAccessIssue] # pyrefly: ignore[missing-module-attribute] |
| 114 | + ) |
| 115 | + |
| 116 | + self.tagger = CRFTagger() |
| 117 | + if corpus == "orchidpp": |
| 118 | + corpus_files = files("pythainlp.corpus") |
| 119 | + model_file = corpus_files.joinpath("crfchunk_orchidpp.model") |
| 120 | + self._model_file_ctx = as_file(model_file) |
| 121 | + model_path = self._model_file_ctx.__enter__() |
| 122 | + self.tagger.open(str(model_path)) |
| 123 | + |
| 124 | + def parse(self, token_pos: list[tuple[str, str]]) -> list[str]: |
| 125 | + """Parse a POS-tagged sentence into IOB chunk labels. |
| 126 | +
|
| 127 | + :param list[tuple[str, str]] token_pos: list of (word, POS-tag) |
| 128 | + pairs. |
| 129 | + :return: list of IOB chunk labels, one per token. |
| 130 | + :rtype: list[str] |
| 131 | + """ |
| 132 | + self.xseq = _extract_features(token_pos) |
| 133 | + return self.tagger.tag(self.xseq) # type: ignore[no-any-return] |
| 134 | + |
| 135 | + def __enter__(self) -> CRFChunkParser: |
| 136 | + """Context manager entry.""" |
| 137 | + return self |
| 138 | + |
| 139 | + def __exit__( |
| 140 | + self, |
| 141 | + exc_type: Optional[type[BaseException]], |
| 142 | + exc_val: Optional[BaseException], |
| 143 | + exc_tb: Optional[types.TracebackType], |
| 144 | + ) -> None: |
| 145 | + """Context manager exit — clean up resources.""" |
| 146 | + if self._model_file_ctx is not None: |
| 147 | + try: |
| 148 | + self._model_file_ctx.__exit__(exc_type, exc_val, exc_tb) |
| 149 | + self._model_file_ctx = None |
| 150 | + except Exception: # noqa: S110 |
| 151 | + pass |
| 152 | + |
| 153 | + def __del__(self) -> None: |
| 154 | + """Attempt resource cleanup on garbage collection. |
| 155 | +
|
| 156 | + .. note:: |
| 157 | + :meth:`__del__` is not guaranteed to be called. |
| 158 | + Use the context manager protocol for reliable cleanup. |
| 159 | + """ |
| 160 | + if self._model_file_ctx is not None: |
| 161 | + try: |
| 162 | + self._model_file_ctx.__exit__(None, None, None) |
| 163 | + except Exception: # noqa: S110 |
| 164 | + pass |
0 commit comments