401 lines
14 KiB
Python
401 lines
14 KiB
Python
"""Stage 1 — heuristic deterministic extractor (pure string operations, no regex).
|
||
|
||
Keyword dictionaries, positional windows, sentence scanning, and weighted
|
||
confidence — see docs/PRESCREEN_HYBRID_REFACTOR_PLAN.md §3.2. Zero marginal
|
||
cost, no hallucination risk; low-confidence results fall through to the LLM
|
||
stage (extractor_hybrid.py).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from ..core.logging import get_logger
|
||
from .extractor import PrescreenContractMeta, _parse_iso, _score_confidence
|
||
|
||
log = get_logger(__name__)
|
||
|
||
EXTRACTOR_VERSION = "heuristic-v2"
|
||
|
||
# ── dictionaries (module-level constants) ────────────────────────────────────
|
||
|
||
_CONTRACT_TYPE_PHRASES: tuple[tuple[str, tuple[str, ...]], ...] = (
|
||
("supply", ("договор поставки", "договор купли-продажи", "договор купли продажи")),
|
||
("services", ("договор возмездного оказания услуг", "договор оказания услуг")),
|
||
("contract_work", ("договор строительного подряда", "договор подряда")),
|
||
("lease", ("договор субаренды", "договор аренды", "договор лизинга")),
|
||
("nda", ("договор о неразглашении", "договор о конфиденциальности")),
|
||
)
|
||
|
||
# Legal entity forms (lowercase); longest-first so the scanner is greedy.
|
||
_ENTITY_FORMS: tuple[str, ...] = (
|
||
"общество с ограниченной ответственностью",
|
||
"общество с дополнительной ответственностью",
|
||
"публичное акционерное общество",
|
||
"закрытое акционерное общество",
|
||
"акционерное общество",
|
||
"частное унитарное предприятие",
|
||
"индивидуальный предприниматель",
|
||
"ооо",
|
||
"пао",
|
||
"зао",
|
||
"одо",
|
||
"чуп",
|
||
"ао",
|
||
"ип",
|
||
)
|
||
|
||
# Amount trigger words (lowercase substrings; a digit run must follow).
|
||
_AMOUNT_TRIGGERS: tuple[str, ...] = (
|
||
"общая стоимость",
|
||
"составляет",
|
||
"стоимостью",
|
||
"цена",
|
||
"цене",
|
||
"оплат",
|
||
"сумма",
|
||
)
|
||
|
||
_CURRENCY_TOKENS: dict[str, str] = {
|
||
"руб": "RUB",
|
||
"рублей": "RUB",
|
||
"рубля": "RUB",
|
||
"₽": "RUB",
|
||
"byn": "BYN",
|
||
"br": "BYN",
|
||
"usd": "USD",
|
||
"$": "USD",
|
||
"eur": "EUR",
|
||
"€": "EUR",
|
||
}
|
||
|
||
# Boolean-clause keyword stems (lowercase substring membership per sentence).
|
||
_PENALTY_KEYWORDS: tuple[str, ...] = ("неустойк", "штраф", "пеня", "пени")
|
||
_TERMINATION_KEYWORDS: tuple[str, ...] = ("расторг", "односторонн", "отказаться от исполнения")
|
||
_ARBITRATION_KEYWORDS: tuple[str, ...] = ("арбитражн",)
|
||
|
||
_HEADER_CAP_CHARS = 1500
|
||
_CURRENCY_TAIL_CHARS = 50
|
||
_MAX_PARTY_NAME_TOKENS = 8
|
||
_MAX_PARTIES = 2
|
||
|
||
_SPACES = " \t\u00a0\u202f"
|
||
_QUOTE_PAIRS = {"«": "»", "“": "”", '"': '"'}
|
||
|
||
|
||
# ── text helpers ─────────────────────────────────────────────────────────────
|
||
|
||
|
||
def _collapse(text: str) -> str:
|
||
"""Collapse every whitespace run (incl. newlines and NBSP) to one space."""
|
||
return " ".join(text.split())
|
||
|
||
|
||
def _skip_spaces(s: str, i: int) -> int:
|
||
while i < len(s) and s[i] in _SPACES:
|
||
i += 1
|
||
return i
|
||
|
||
|
||
def _split_sentences(text: str) -> list[str]:
|
||
"""Sentence segmentation via str.split on '. ' and newlines (plan §3.2)."""
|
||
parts: list[str] = []
|
||
for chunk in text.split("\n"):
|
||
parts.extend(chunk.split(". "))
|
||
return parts
|
||
|
||
|
||
def _find_closing_quote(s: str, open_idx: int) -> int:
|
||
close = _QUOTE_PAIRS.get(s[open_idx], s[open_idx])
|
||
return s.find(close, open_idx + 1)
|
||
|
||
|
||
# ── contract type ────────────────────────────────────────────────────────────
|
||
|
||
|
||
def _extract_contract_type(lower: str) -> str | None:
|
||
"""Exact phrase dictionary match on the normalized text."""
|
||
for name, phrases in _CONTRACT_TYPE_PHRASES:
|
||
for phrase in phrases:
|
||
if phrase in lower:
|
||
return name
|
||
return None
|
||
|
||
|
||
# ── parties (positional: header window only) ─────────────────────────────────
|
||
|
||
|
||
def _header_window(text: str) -> str:
|
||
"""Text up to the first numbered/ПРЕДМЕТ heading, capped at ~1500 chars."""
|
||
lines: list[str] = []
|
||
size = 0
|
||
for line in text.splitlines():
|
||
stripped = line.strip()
|
||
lower = stripped.lower()
|
||
if lines and (lower.startswith("1.") or lower.startswith("1 ") or "предмет" in lower):
|
||
break
|
||
lines.append(line)
|
||
size += len(line) + 1
|
||
if size >= _HEADER_CAP_CHARS:
|
||
break
|
||
return "\n".join(lines)
|
||
|
||
|
||
def _is_name_token(token: str) -> bool:
|
||
if not token:
|
||
return False
|
||
# Initials like «И.И.» / «П.»
|
||
if len(token) <= 3 and token[0].isupper() and token.endswith("."):
|
||
return True
|
||
return token[0].isupper()
|
||
|
||
|
||
def _capture_party_name(header: str, start: int, form_end: int) -> tuple[str, int] | None:
|
||
"""Capture a party name after an entity-form token at `start`.
|
||
|
||
Quoted names run from the form through the closing quote; unquoted names
|
||
consume following Capitalized words until a lowercase word, punctuation,
|
||
or line end.
|
||
"""
|
||
j = _skip_spaces(header, form_end)
|
||
if j < len(header) and header[j] in _QUOTE_PAIRS:
|
||
close = _find_closing_quote(header, j)
|
||
if close != -1:
|
||
return header[start : close + 1].strip(), close + 1
|
||
|
||
end = form_end
|
||
tokens = 0
|
||
j = _skip_spaces(header, form_end)
|
||
while j < len(header) and tokens < _MAX_PARTY_NAME_TOKENS:
|
||
line_end = header.find("\n", j)
|
||
segment_end = line_end if line_end != -1 else len(header)
|
||
k = j
|
||
while k < segment_end and header[k] not in _SPACES:
|
||
k += 1
|
||
token = header[j:k]
|
||
if not _is_name_token(token):
|
||
break
|
||
end = k
|
||
tokens += 1
|
||
j = _skip_spaces(header, k)
|
||
if j >= len(header) or header[j] == "\n":
|
||
break
|
||
name = header[start:end].strip()
|
||
if not name:
|
||
return None
|
||
return name, max(end, form_end)
|
||
|
||
|
||
def _extract_parties(header: str) -> tuple[str | None, str | None]:
|
||
"""Token-scan the header window for entity forms; capture party names."""
|
||
lower = header.lower()
|
||
n = len(header)
|
||
found: list[str] = []
|
||
i = 0
|
||
while i < n and len(found) < _MAX_PARTIES:
|
||
form: str | None = None
|
||
for candidate in _ENTITY_FORMS:
|
||
end = i + len(candidate)
|
||
if not lower.startswith(candidate, i):
|
||
continue
|
||
before_ok = i == 0 or not lower[i - 1].isalnum()
|
||
after_ok = end >= n or not lower[end].isalnum()
|
||
if before_ok and after_ok:
|
||
form = candidate
|
||
break
|
||
if form is None:
|
||
i += 1
|
||
continue
|
||
captured = _capture_party_name(header, i, i + len(form))
|
||
if captured is None:
|
||
i += len(form)
|
||
continue
|
||
name, capture_end = captured
|
||
# Drop bare forms without a name (e.g. a lone «ООО»).
|
||
if name.lower() != form:
|
||
if name not in found:
|
||
found.append(name)
|
||
i = max(capture_end, i + len(form))
|
||
if len(found) >= 2:
|
||
return found[0], found[1]
|
||
if len(found) == 1:
|
||
return found[0], None
|
||
return None, None
|
||
|
||
|
||
# ── amount + currency (trigger-word scan + manual digit parse) ───────────────
|
||
|
||
|
||
def _scan_number(s: str, i: int) -> tuple[float, int] | None:
|
||
"""Walk chars at s[i] parsing digits, space-separated thousands, decimals."""
|
||
n = len(s)
|
||
j = i
|
||
int_part: list[str] = []
|
||
dec_part: list[str] = []
|
||
while j < n and s[j].isdigit():
|
||
int_part.append(s[j])
|
||
j += 1
|
||
if not int_part:
|
||
return None
|
||
while j < n:
|
||
ch = s[j]
|
||
if ch in ".," and j + 1 < n and s[j + 1].isdigit() and not dec_part:
|
||
j += 1
|
||
while j < n and s[j].isdigit():
|
||
dec_part.append(s[j])
|
||
j += 1
|
||
break # a decimal part terminates the number
|
||
if ch in _SPACES:
|
||
# Thousands separator: space + exactly 3 digits, not part of a longer run.
|
||
k = j + 1
|
||
group = ""
|
||
while k < n and s[k].isdigit() and len(group) < 3:
|
||
group += s[k]
|
||
k += 1
|
||
if len(group) == 3 and (k >= n or not s[k].isdigit()):
|
||
int_part.append(group)
|
||
j = k
|
||
continue
|
||
break
|
||
raw = "".join(int_part) + ("." + "".join(dec_part) if dec_part else "")
|
||
return float(raw), j
|
||
|
||
|
||
def _find_currency(lower: str, start: int, end: int) -> str | None:
|
||
window = lower[start : min(end, len(lower))]
|
||
for token, code in _CURRENCY_TOKENS.items():
|
||
if token in window:
|
||
return code
|
||
return None
|
||
|
||
|
||
def _find_earliest_trigger(lower: str, start: int) -> tuple[int, int] | None:
|
||
best: tuple[int, int] | None = None
|
||
for trigger in _AMOUNT_TRIGGERS:
|
||
idx = lower.find(trigger, start)
|
||
if idx != -1 and (best is None or idx < best[0]):
|
||
best = (idx, idx + len(trigger))
|
||
return best
|
||
|
||
|
||
def _extract_amount(collapsed: str, lower: str) -> tuple[float | None, str | None]:
|
||
pos = 0
|
||
while True:
|
||
hit = _find_earliest_trigger(lower, pos)
|
||
if hit is None:
|
||
return None, None
|
||
trig_start, trig_end = hit
|
||
j = _skip_spaces(collapsed, trig_end)
|
||
if j < len(collapsed) and collapsed[j].isdigit():
|
||
parsed = _scan_number(collapsed, j)
|
||
if parsed is not None:
|
||
value, num_end = parsed
|
||
return value, _find_currency(lower, trig_start, num_end + _CURRENCY_TAIL_CHARS)
|
||
pos = trig_start + 1
|
||
|
||
|
||
# ── dates (trigger tokens + manual dd.mm.yyyy splitter) ──────────────────────
|
||
|
||
|
||
def _parse_ddmmyyyy(s: str, i: int) -> str | None:
|
||
"""Parse dd.mm.yyyy / dd/mm.yyyy at s[i]; None when not a date boundary."""
|
||
if i > 0 and s[i - 1].isdigit():
|
||
return None
|
||
if i + 10 > len(s):
|
||
return None
|
||
if not (s[i].isdigit() and s[i + 1].isdigit()):
|
||
return None
|
||
if s[i + 2] not in "./":
|
||
return None
|
||
if not (s[i + 3].isdigit() and s[i + 4].isdigit()):
|
||
return None
|
||
if s[i + 5] not in "./":
|
||
return None
|
||
if not all(s[k].isdigit() for k in (i + 6, i + 7, i + 8, i + 9)):
|
||
return None
|
||
if i + 10 < len(s) and s[i + 10].isdigit():
|
||
return None
|
||
return _parse_iso(s[i : i + 10])
|
||
|
||
|
||
def _find_date_after_token(s: str, token: str, start: int) -> tuple[str, int, int] | None:
|
||
"""Find `token` as a standalone word followed by a dd.mm.yyyy date.
|
||
|
||
Returns (iso_date, date_start, scan_end) or None.
|
||
"""
|
||
pos = start
|
||
n = len(s)
|
||
while pos < n:
|
||
idx = s.find(token, pos)
|
||
if idx == -1:
|
||
return None
|
||
before_ok = idx == 0 or not s[idx - 1].isalnum()
|
||
j = idx + len(token)
|
||
if before_ok and j < n and s[j] in _SPACES:
|
||
j = _skip_spaces(s, j)
|
||
iso = _parse_ddmmyyyy(s, j)
|
||
if iso is not None:
|
||
return iso, j, j + 10
|
||
pos = idx + 1
|
||
return None
|
||
|
||
|
||
def _extract_dates(lower_collapsed: str) -> tuple[str | None, str | None]:
|
||
start_hit = None
|
||
for token in ("с", "от"):
|
||
hit = _find_date_after_token(lower_collapsed, token, 0)
|
||
if hit is not None and (start_hit is None or hit[1] < start_hit[1]):
|
||
start_hit = hit
|
||
if start_hit is not None:
|
||
# v1 semantics: end date = «по <date>» anywhere after the start date.
|
||
end_hit = _find_date_after_token(lower_collapsed, "по", start_hit[2])
|
||
return start_hit[0], end_hit[0] if end_hit is not None else None
|
||
for token in ("по", "до"):
|
||
hit = _find_date_after_token(lower_collapsed, token, 0)
|
||
if hit is not None:
|
||
return None, hit[0]
|
||
return None, None
|
||
|
||
|
||
# ── boolean clauses (sentence scanning) ──────────────────────────────────────
|
||
|
||
|
||
def _has_clause(sentences: list[str], keywords: tuple[str, ...]) -> bool:
|
||
return any(keyword in sentence for sentence in sentences for keyword in keywords)
|
||
|
||
|
||
# ── Stage 1 extractor ────────────────────────────────────────────────────────
|
||
|
||
|
||
class HeuristicExtractor:
|
||
"""Deterministic keyword/positional extractor (Stage 1, sync, zero cost)."""
|
||
|
||
extractor_version = EXTRACTOR_VERSION
|
||
|
||
def extract(self, text: str) -> PrescreenContractMeta:
|
||
collapsed = _collapse(text)
|
||
lower = collapsed.lower()
|
||
header = _header_window(text)
|
||
sentences = [s.lower() for s in _split_sentences(text)] if text else []
|
||
|
||
party_a, party_b = _extract_parties(header)
|
||
amount, currency = _extract_amount(collapsed, lower)
|
||
start_date, end_date = _extract_dates(lower)
|
||
|
||
meta = PrescreenContractMeta(
|
||
contract_type=_extract_contract_type(lower),
|
||
party_a=party_a,
|
||
party_b=party_b,
|
||
total_amount=amount,
|
||
currency=currency,
|
||
start_date=start_date,
|
||
end_date=end_date,
|
||
has_penalty_clause=None if not text else _has_clause(sentences, _PENALTY_KEYWORDS),
|
||
has_termination_clause=(
|
||
None if not text else _has_clause(sentences, _TERMINATION_KEYWORDS)
|
||
),
|
||
has_arbitration=None if not text else _has_clause(sentences, _ARBITRATION_KEYWORDS),
|
||
)
|
||
meta.confidence_score = _score_confidence(meta)
|
||
return meta
|
||
|
||
|
||
__all__ = ["EXTRACTOR_VERSION", "HeuristicExtractor"]
|