ja_sentence_segmenter.concatenate.simple_concatenator

  1"""Simple sentence concatenator for japanese text."""
  2
  3import re
  4from collections.abc import Generator, Iterator
  5from typing import Optional, Union, overload
  6
  7
  8def __concatenate_matching_iter(
  9    texts: Iterator[str],
 10    former_matching_rule: Optional[str],
 11    latter_matching_rule: Optional[str],
 12    remove_former_matched: bool,
 13    remove_latter_matched: bool,
 14    max_concatenate_length: Optional[int],
 15) -> Generator[str, None, None]:
 16    # 累積は入力1行ごとに再走査されるので、パターンのコンパイルはループの外で1度だけ行う。
 17    former_pattern = re.compile(former_matching_rule) if former_matching_rule else None
 18    latter_pattern = re.compile(latter_matching_rule) if latter_matching_rule else None
 19
 20    try:
 21        former = next(texts)
 22        # 上限は「現在の累積で1回以上結合が起きた後」にのみ確認する。
 23        # そうしないと、1行目が単独で上限を超えている場合に
 24        # former_matching_rule が一度も評価されないまま素通りしてしまう。
 25        concatenated = False
 26
 27        for latter in texts:
 28            if concatenated and max_concatenate_length is not None and len(former) >= max_concatenate_length:
 29                yield former
 30                former = latter
 31                concatenated = False
 32                continue
 33
 34            former_match_obj = former_pattern.match(former) if former_pattern else None
 35            latter_match_obj = latter_pattern.match(latter) if latter_pattern else None
 36
 37            if former_matching_rule and latter_matching_rule and former_match_obj and latter_match_obj:
 38                tmp_former = former_match_obj.group("result") if remove_former_matched else former
 39                tmp_latter = latter_match_obj.group("result") if remove_latter_matched else latter
 40                former = tmp_former + tmp_latter
 41                concatenated = True
 42            elif former_matching_rule and not latter_matching_rule and former_match_obj:
 43                tmp_former = former_match_obj.group("result") if remove_former_matched else former
 44                former = tmp_former + latter
 45                concatenated = True
 46            elif not former_matching_rule and latter_matching_rule and latter_match_obj:
 47                tmp_latter = latter_match_obj.group("result") if remove_latter_matched else latter
 48                former += tmp_latter
 49                concatenated = True
 50            else:
 51                yield former
 52                former = latter
 53                concatenated = False
 54
 55        yield former
 56    except StopIteration:
 57        pass
 58
 59
 60@overload
 61def concatenate_matching(
 62    arg: str,
 63    former_matching_rule: Optional[str] = None,
 64    latter_matching_rule: Optional[str] = None,
 65    remove_former_matched: bool = True,
 66    remove_latter_matched: bool = True,
 67    max_concatenate_length: Optional[int] = None,
 68) -> Generator[str, None, None]: ...
 69
 70
 71@overload
 72def concatenate_matching(
 73    arg: list[str],
 74    former_matching_rule: Optional[str] = None,
 75    latter_matching_rule: Optional[str] = None,
 76    remove_former_matched: bool = True,
 77    remove_latter_matched: bool = True,
 78    max_concatenate_length: Optional[int] = None,
 79) -> Generator[str, None, None]: ...
 80
 81
 82@overload
 83def concatenate_matching(
 84    arg: Iterator[str],
 85    former_matching_rule: Optional[str] = None,
 86    latter_matching_rule: Optional[str] = None,
 87    remove_former_matched: bool = True,
 88    remove_latter_matched: bool = True,
 89    max_concatenate_length: Optional[int] = None,
 90) -> Generator[str, None, None]: ...
 91
 92
 93def concatenate_matching(
 94    arg: Union[str, list[str], Iterator[str]],
 95    former_matching_rule: Optional[str] = None,
 96    latter_matching_rule: Optional[str] = None,
 97    remove_former_matched: bool = True,
 98    remove_latter_matched: bool = True,
 99    max_concatenate_length: Optional[int] = None,
100) -> Generator[str, None, None]:
101    r"""Concatenate two lines with regular expression rule.
102
103    Parameters
104    ----------
105    arg : Union[str, list[str], Iterator[str]]
106        texts you want to concatenate.
107        a str is treated as one indivisible line and is never split, so it has
108        nothing to be concatenated with and comes back unchanged. split it into
109        lines yourself first -- `split_newline` does this -- and pass the
110        result here.
111    former_matching_rule : Optional[str], optional
112        regular expression for former line, by default None
113    latter_matching_rule : Optional[str], optional
114        regular expression for latter line, by default None
115    remove_former_matched : bool, optional
116        whether to remove matched place of former line, by default True.
117        if this is True, former_matching_rule must contain named group 'result',
118        only that group remains.
119        e.g. r"^(\s*[>]+\s*)(?P<result>.+)$"
120    remove_latter_matched : bool, optional
121        whether to remove matched place of latter line, by default True.
122        if this is True, latter_matching_rule must contain named group 'result',
123        only that group remains.
124        e.g. r"^(\s*[>]+\s*)(?P<result>.+)$"
125    max_concatenate_length : Optional[int], optional
126        soft upper bound on the length of an accumulation, by default None,
127        meaning no bound. must be positive if not None.
128        set it when segmenting untrusted text, and see the notes below for
129        what it costs. leaving it None keeps the unbounded behaviour.
130
131    Yields
132    ------
133    Generator[str, None, None]
134        concatenated texts.
135
136    Raises
137    ------
138    TypeError
139        if arg is not a str, a list or an Iterator.
140    ValueError
141        if max_concatenate_length is not None and not positive.
142        both are raised on the first iteration rather than at call time,
143        because this is a generator function.
144
145    Notes
146    -----
147    Why max_concatenate_length exists: without a bound, former_matching_rule
148    is re-applied to an ever growing accumulation and the accumulation itself
149    is rebuilt on every input line. both costs grow with the accumulated
150    length, so the total work is quadratic in the size of the input.
151
152    What setting it costs: once an accumulation reaches the bound it is
153    yielded as is and a new accumulation starts. no text is lost and no
154    exception is raised, but the output differs from an unbounded run by more
155    than where it is split.
156
157    - the bound is soft. it is only checked after an accumulation has been
158      concatenated at least once, so former_matching_rule is applied at least
159      once per accumulation even when the first line already exceeds the
160      bound. an accumulation may therefore exceed the bound by the length of
161      the line that started it plus one more line.
162    - neither matching rule is evaluated at a bound boundary, so text that
163      remove_former_matched or remove_latter_matched would have stripped
164      survives into the output there.
165    - an accumulation may be broken between an opening bracket and its
166      closing one, which stops split_punctuation from protecting the
167      punctuation inside it.
168    """
169    if max_concatenate_length is not None and max_concatenate_length <= 0:
170        msg = f"max_concatenate_length must be positive or None, got {max_concatenate_length}"
171        raise ValueError(msg)
172
173    if isinstance(arg, str):
174        yield from __concatenate_matching_iter(
175            iter([arg]), former_matching_rule, latter_matching_rule, remove_former_matched, remove_latter_matched, max_concatenate_length
176        )
177    elif isinstance(arg, list):
178        yield from __concatenate_matching_iter(
179            iter(arg), former_matching_rule, latter_matching_rule, remove_former_matched, remove_latter_matched, max_concatenate_length
180        )
181    elif isinstance(arg, Iterator):
182        yield from __concatenate_matching_iter(
183            arg, former_matching_rule, latter_matching_rule, remove_former_matched, remove_latter_matched, max_concatenate_length
184        )
185    else:
186        # 静的には到達しないが、型注釈のない呼び出しが tuple や set を渡したときに
187        # 黙って空を返さないための実行時ガード。
188        msg = f"arg must be a str, a list or an Iterator, got {type(arg).__name__}"  # type: ignore[unreachable]
189        raise TypeError(msg)