171 lines
6.7 KiB
Python
171 lines
6.7 KiB
Python
#!/usr/bin/env python3
|
||
"""Conservative Chinese-number classification and normalization."""
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
from dataclasses import asdict, dataclass
|
||
|
||
DIGIT = {
|
||
"零": 0, "〇": 0, "一": 1, "幺": 1, "二": 2, "两": 2, "三": 3,
|
||
"四": 4, "五": 5, "六": 6, "七": 7, "八": 8, "九": 9,
|
||
}
|
||
SMALL = {"十": 10, "百": 100, "千": 1000}
|
||
BIG = {"万": 10_000, "亿": 100_000_000}
|
||
CN = "零〇一二两三四五六七八九十百千万亿幺"
|
||
CN_RE = rf"[{CN}]+"
|
||
PURE_DIGITS = set(DIGIT)
|
||
|
||
|
||
@dataclass
|
||
class Decision:
|
||
raw: str
|
||
normalized: str
|
||
kind: str
|
||
confidence: float
|
||
start: int
|
||
end: int
|
||
context: str
|
||
|
||
def as_dict(self) -> dict:
|
||
return asdict(self)
|
||
|
||
|
||
def chinese_to_int(raw: str) -> int | None:
|
||
raw = raw.replace("两", "二")
|
||
if not raw or any(c not in DIGIT and c not in SMALL and c not in BIG for c in raw):
|
||
return None
|
||
if set(raw) <= PURE_DIGITS:
|
||
return int("".join(str(DIGIT[c]) for c in raw))
|
||
# "两万五" is conventionally ambiguous between 20,005 and 25,000.
|
||
if any(c in BIG for c in raw) and raw[-1] in DIGIT:
|
||
return None
|
||
total = section = number = 0
|
||
for c in raw:
|
||
if c in DIGIT:
|
||
number = DIGIT[c]
|
||
elif c in SMALL:
|
||
section += (number or 1) * SMALL[c]
|
||
number = 0
|
||
else:
|
||
section += number
|
||
total += (section or 1) * BIG[c]
|
||
section = number = 0
|
||
return total + section + number
|
||
|
||
|
||
def _ctx(text: str, start: int, end: int, width: int = 18) -> str:
|
||
return text[max(0, start - width):min(len(text), end + width)]
|
||
|
||
|
||
def _add(found: list[Decision], text: str, start: int, end: int,
|
||
normalized: str, kind: str, confidence: float):
|
||
raw = text[start:end]
|
||
if raw == normalized:
|
||
return
|
||
found.append(Decision(raw, normalized, kind, confidence, start, end,
|
||
_ctx(text, start, end)))
|
||
|
||
|
||
def _overlaps(a: Decision, accepted: list[Decision]) -> bool:
|
||
return any(a.start < b.end and b.start < a.end for b in accepted)
|
||
|
||
|
||
def _money_text(raw: str, currency: str) -> str | None:
|
||
if raw and raw[-1] in BIG:
|
||
base = chinese_to_int(raw[:-1])
|
||
return f"{base}{raw[-1]}{currency}" if base is not None else None
|
||
value = chinese_to_int(raw)
|
||
return f"{value}{currency}" if value is not None else None
|
||
|
||
|
||
def normalize(text: str) -> tuple[str, list[dict]]:
|
||
"""Normalize only values whose semantic type is visible in nearby syntax."""
|
||
found: list[Decision] = []
|
||
|
||
# Percentages.
|
||
for m in re.finditer(rf"百分之(?P<n>{CN_RE})", text):
|
||
value = chinese_to_int(m.group("n"))
|
||
if value is not None:
|
||
_add(found, text, m.start(), m.end(), f"{value}%", "percentage", 0.99)
|
||
|
||
# Full dates and standalone day-of-month references.
|
||
for m in re.finditer(
|
||
rf"(?:(?P<y>{CN_RE})年)?(?P<mo>{CN_RE})月(?P<d>{CN_RE})(?P<u>日|号)", text):
|
||
y = chinese_to_int(m.group("y")) if m.group("y") else None
|
||
mo, d = chinese_to_int(m.group("mo")), chinese_to_int(m.group("d"))
|
||
if mo is not None and d is not None and 1 <= mo <= 12 and 1 <= d <= 31:
|
||
norm = f"{y}年" if y is not None else ""
|
||
norm += f"{mo}月{d}{m.group('u')}"
|
||
_add(found, text, m.start(), m.end(), norm, "date", 0.98)
|
||
for m in re.finditer(rf"(?P<n>{CN_RE})(?P<u>日|号)", text):
|
||
value = chinese_to_int(m.group("n"))
|
||
if value is not None and 1 <= value <= 31:
|
||
_add(found, text, m.start(), m.end(), f"{value}{m.group('u')}",
|
||
"day", 0.93)
|
||
|
||
# Money: preserve the stated currency and avoid silently inventing one.
|
||
for m in re.finditer(
|
||
rf"(?P<n>{CN_RE})(?P<u>元|块钱|人民币|美元|美金|港币)", text):
|
||
norm = _money_text(m.group("n"), m.group("u"))
|
||
if norm is not None:
|
||
_add(found, text, m.start(), m.end(), norm,
|
||
"money", 0.98)
|
||
for m in re.finditer(rf"(?P<n>[一二两三四五六七八九十百千万]+)个(?P<u>亿|万)", text):
|
||
value = chinese_to_int(m.group("n"))
|
||
if value is not None and re.search(r"投|融|估值|收入|销售|利润|资金|金额",
|
||
_ctx(text, m.start(), m.end())):
|
||
_add(found, text, m.start(), m.end(), f"{value}{m.group('u')}元",
|
||
"money", 0.90)
|
||
for m in re.finditer(
|
||
rf"(?P<n>[一二两三四五六七八九十百千]+)(?P<u>万|亿)", text):
|
||
context = _ctx(text, m.start(), m.end())
|
||
value = chinese_to_int(m.group("n"))
|
||
if value is not None and re.search(
|
||
r"投|融|估值|收入|销售|利润|资金|金额|打款|退出|回购|报价", context):
|
||
_add(found, text, m.start(), m.end(),
|
||
f"{value}{m.group('u')}元", "money", 0.91)
|
||
|
||
# Quantities with an explicit unit.
|
||
units = "个月|小时|分钟|公斤|毫克|毫升|人|家|份|批|次|台|套|件|天|年|吨|克|升|例|名"
|
||
for m in re.finditer(rf"(?P<n>{CN_RE})(?P<u>{units})", text):
|
||
value = chinese_to_int(m.group("n"))
|
||
if value is not None:
|
||
_add(found, text, m.start(), m.end(), f"{value}{m.group('u')}",
|
||
"quantity", 0.96)
|
||
|
||
# Codes are spoken digit-by-digit. 幺 is a particularly strong signal.
|
||
for m in re.finditer(
|
||
rf"(?<![{CN}])(?P<n>[零〇一二两三四五六七八九幺]{{2,}})(?![{CN}])",
|
||
text):
|
||
raw = m.group("n")
|
||
context = _ctx(text, m.start(), m.end())
|
||
code_context = ("幺" in raw or
|
||
re.search(r"项目|编号|代号|批号|工单|报告|BF|DF|CF|MFG",
|
||
context, re.I))
|
||
if code_context:
|
||
value = "".join(str(DIGIT[c]) for c in raw)
|
||
_add(found, text, m.start(), m.end(), value, "code", 0.97)
|
||
|
||
# Highest-confidence, longest non-overlapping decisions win.
|
||
accepted: list[Decision] = []
|
||
for d in sorted(found, key=lambda x: (-x.confidence, -(x.end - x.start), x.start)):
|
||
if not _overlaps(d, accepted):
|
||
accepted.append(d)
|
||
|
||
out = text
|
||
for d in sorted(accepted, key=lambda x: x.start, reverse=True):
|
||
out = out[:d.start] + d.normalized + out[d.end:]
|
||
|
||
# Preserve unresolved numeric speech as an explicit structured observation.
|
||
uncertain: list[Decision] = []
|
||
occupied = accepted[:]
|
||
for m in re.finditer(CN_RE, text):
|
||
if len(m.group()) < 2:
|
||
continue
|
||
d = Decision(m.group(), m.group(), "unknown", 0.30, m.start(), m.end(),
|
||
_ctx(text, m.start(), m.end()))
|
||
if not _overlaps(d, occupied):
|
||
uncertain.append(d)
|
||
all_decisions = sorted(accepted + uncertain, key=lambda x: x.start)
|
||
return out, [d.as_dict() for d in all_decisions]
|