| """Surface string -> typed value, for predictors that can only point at text. |
| |
| A span extractor finds ``March 3rd`` or ``$1,204.50``; the field wants |
| ``2026-03-03`` or ``1204.5``. These parsers are what lets a span-only system be |
| scored on typed fields at all -- and what it cannot do (arithmetic) stays visible |
| as misses on ``computed`` gold. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import datetime as dt |
| import re |
| from typing import Any, Optional |
|
|
| from .schema import DATE, DATETIME, DURATION, NUMBER, TEXT, TIME, Field |
|
|
| |
| _NUM = re.compile(r"(?:(?<![A-Za-z0-9])-)?\d[\d,]*(?:\.\d+)?") |
| _WORDS = {"zero": 0, "one": 1, "two": 2, "three": 3, "four": 4, "five": 5, "six": 6, "seven": 7, |
| "eight": 8, "nine": 9, "ten": 10, "eleven": 11, "twelve": 12, "fifteen": 15, "twenty": 20, |
| "thirty": 30, "forty": 40, "fifty": 50, "hundred": 100} |
| _TIME = re.compile(r"\b(\d{1,2}):(\d{2})(?::(\d{2}))?\s*([ap]\.?m\.?)?", re.I) |
| _UNIT_DAYS = {"day": 1, "days": 1, "week": 7, "weeks": 7} |
| _UNIT_MIN = {"minute": 1, "minutes": 1, "min": 1, "mins": 1, "hour": 60, "hours": 60, "hr": 60, "hrs": 60, "h": 60} |
| _UNIT_MON = {"month": 1, "months": 1, "year": 12, "years": 12} |
|
|
|
|
| def number(s: str, integer: bool = False) -> Optional[float]: |
| m = _NUM.search(s) |
| if m: |
| try: |
| v = float(m.group().replace(",", "")) |
| except ValueError: |
| return None |
| else: |
| w = s.strip().lower() |
| if w not in _WORDS: |
| return None |
| v = float(_WORDS[w]) |
| return int(round(v)) if integer else v |
|
|
|
|
| def date(s: str, ref: Optional[str]) -> Optional[str]: |
| d = _datetime(s, ref) |
| return d.date().isoformat() if d else None |
|
|
|
|
| def datetime_(s: str, ref: Optional[str]) -> Optional[str]: |
| d = _datetime(s, ref) |
| if d is None or (d.hour, d.minute) == (0, 0) and not _TIME.search(s): |
| return None |
| return d.strftime("%Y-%m-%dT%H:%M") |
|
|
|
|
| def time(s: str) -> Optional[str]: |
| m = _TIME.search(s) |
| if not m: |
| return None |
| h, mi = int(m.group(1)), int(m.group(2)) |
| ampm = (m.group(4) or "").lower().replace(".", "") |
| if ampm == "pm" and h < 12: |
| h += 12 |
| if ampm == "am" and h == 12: |
| h = 0 |
| if h > 23 or mi > 59: |
| return None |
| return f"{h:02d}:{mi:02d}" |
|
|
|
|
| def duration(s: str, unit: str) -> Optional[float]: |
| v = number(s) |
| if v is None: |
| return None |
| words = re.findall(r"[a-z]+", s.lower()) |
| table = (_UNIT_DAYS if unit == "days" else _UNIT_MIN if unit == "minutes" else |
| _UNIT_MON if unit == "months" else {unit: 1}) |
| for w in words: |
| if w in table: |
| return v * table[w] |
| return v |
|
|
|
|
| def _datetime(s: str, ref: Optional[str]) -> Optional[dt.datetime]: |
| import dateparser |
| base = dt.datetime.fromisoformat(ref) if ref else dt.datetime(2026, 1, 1) |
| settings = {"RELATIVE_BASE": base, "PREFER_DATES_FROM": "past", "RETURN_AS_TIMEZONE_AWARE": False, |
| "STRICT_PARSING": False, "PREFER_DAY_OF_MONTH": "first"} |
| text = re.sub(r"^\s*(on|last)\s+", lambda m: "last " if m.group(1) == "last" else "", s.strip(), flags=re.I) |
| try: |
| return dateparser.parse(text, languages=["en"], settings=settings) |
| except Exception: |
| return None |
|
|
|
|
| def to_type(f: Field, s: str, ref: Optional[str]) -> Any: |
| """Parse a candidate string as the field's (item) type; None if it is not one.""" |
| t = f.item_type |
| if t == TEXT: |
| return s.strip() or None |
| if t == NUMBER: |
| return number(s, bool(f.options.get("integer"))) |
| if t == DATE: |
| return date(s, ref) |
| if f.type == DATETIME: |
| return datetime_(s, ref) |
| if f.type == TIME: |
| return time(s) |
| if f.type == DURATION: |
| v = duration(s, f.options.get("unit", "days")) |
| return int(round(v)) if v is not None and f.options.get("integer") else v |
| return None |
|
|
