rules.py

"""Rule-based preprocessing: Korean number→English unit conversion + article date injection."""

import re
from datetime import datetime

# Korean magnitude markers (cumulative within a compound number)
KO_MAGNITUDE = [("조", 10**12), ("억", 10**8), ("만", 10**4)]

# Recognized currency / counter units. Currency names follow international
# naming conventions ("Korean won", "Japanese yen", "Chinese yuan"). Counter
# units stay simple.
UNIT_MAP = {
    "원": "Korean won",
    "달러": "dollars",
    "유로": "euros",
    "엔": "Japanese yen",
    "위안": "Chinese yuan",
    "명": "people",
    "개": "units",
    "건": "cases",
    "대": "vehicles",
    "톤": "tons",
}

# Compound Korean number followed by unit. The number portion may contain any
# combination of 조 / 억 / 만 plus a tail digit run, optionally followed by an
# approximation marker (여 / 약 / 가량 / 여 등) before the unit. We require at
# least one digit-magnitude pair OR a digit-only tail to avoid matching bare
# units.
APPROX = r"(?:여|여\s+|가량|가량\s+|쯤|쯤\s+|약\s+|약)"
NUM_UNIT_PATTERN = re.compile(
    r"(?P<num>"
    r"(?:\d[\d,]*(?:\.\d+)?\s*조\s*)?"
    r"(?:\d[\d,]*(?:\.\d+)?(?:\s*" + APPROX + r")?\s*억\s*)?"
    r"(?:\d[\d,]*(?:\.\d+)?(?:\s*" + APPROX + r")?\s*만\s*)?"
    r"(?:\d[\d,]*(?:\.\d+)?)?"
    r")\s*(?P<unit>원|달러|유로|엔|위안|명|개|건|대|톤)"
)

# Percentage points. Matches all common Korean variants of "X%p / X%포인트 /
# X퍼센트 포인트". Must be detected and replaced BEFORE plain numbers so that
# "10%" doesn't slip through unattended (we want "10 percentage points", never
# just "10%").
PERCENT_POINT_PATTERN = re.compile(
    r"(?P<num>\d[\d,]*(?:\.\d+)?)\s*"
    r"(?:%\s*p|%\s*포인트|퍼센트\s*포인트|％\s*포인트)",
    re.IGNORECASE,
)


def parse_korean_number(text: str) -> float | None:
    """Parse a compound Korean magnitude string (e.g., '1조4000억', '120억', '5000만',
    '7800여억') → numeric value. Approximation markers (여 / 가량 / 쯤 / 약) are stripped."""
    s = text.replace(",", "").replace(" ", "")
    # Strip approximation markers anywhere they appear within the number string
    s = re.sub(r"(여|가량|쯤|약)", "", s)
    total = 0.0
    matched = False
    for marker, mult in KO_MAGNITUDE:
        m = re.match(r"(\d+(?:\.\d+)?)" + marker, s)
        if m:
            total += float(m.group(1)) * mult
            s = s[m.end():]
            matched = True
    if s:
        try:
            total += float(s)
            matched = True
        except ValueError:
            pass
    return total if matched else None


def format_english_amount(amount: float, unit_kor: str) -> str:
    """Format amount in English unit. Uses 'million/billion/trillion' for large numbers."""
    unit_en = UNIT_MAP.get(unit_kor, unit_kor)

    def _fmt(n: float) -> str:
        # Preserve up to 4 decimals; trim trailing zeros; thousands-separate the integer part.
        if abs(n - round(n)) < 1e-9:
            return f"{int(round(n)):,}"
        s = f"{n:.4f}".rstrip("0").rstrip(".")
        if "." in s:
            int_part, dec = s.split(".")
            return f"{int(int_part):,}.{dec}"
        return f"{int(s):,}"

    if amount >= 10**12:
        return f"{_fmt(amount / 10**12)} trillion {unit_en}"
    if amount >= 10**9:
        return f"{_fmt(amount / 10**9)} billion {unit_en}"
    if amount >= 10**6:
        return f"{_fmt(amount / 10**6)} million {unit_en}"
    # < 10^6: write as thousand-separated integer (avoid awkward "135.574 thousand")
    return f"{_fmt(amount)} {unit_en}"


def detect_korean_numbers(text: str) -> list[dict]:
    """Return list of {span, start, end, amount, unit, english} entries found in text."""
    results = []
    seen_spans = set()
    for m in NUM_UNIT_PATTERN.finditer(text):
        num_str = m.group("num").strip()
        unit = m.group("unit")
        if not re.search(r"\d", num_str):
            continue  # skip bare unit
        amount = parse_korean_number(num_str)
        if amount is None or amount == 0:
            continue
        full = f"{num_str}{unit}"
        # Dedup identical spans (e.g., the same "100억원" appearing many times)
        if full in seen_spans:
            continue
        seen_spans.add(full)
        results.append({
            "span": full,
            "start": m.start(),
            "end": m.end(),
            "amount": amount,
            "unit_ko": unit,
            "english": format_english_amount(amount, unit),
        })
    return results


def parse_article_date(date_str: str) -> str | None:
    """Normalize an article date to a YYYY-MM-DD string, return None if unparseable."""
    if not date_str:
        return None
    s = date_str.strip()
    # Try a few common formats
    for fmt in ("%Y-%m-%d", "%Y.%m.%d", "%Y/%m/%d", "%Y%m%d"):
        try:
            return datetime.strptime(s[:10], fmt).strftime("%Y-%m-%d")
        except ValueError:
            continue
    return None


def _format_simple_number(num_str: str) -> str:
    """Format a plain digit string (with possible commas/decimals) keeping the value as-is."""
    s = num_str.replace(",", "")
    try:
        n = float(s)
    except ValueError:
        return num_str
    if abs(n - round(n)) < 1e-9:
        return f"{int(round(n)):,}"
    return f"{n:,g}"


def replace_numbers_inline(text: str) -> tuple[str, list[dict]]:
    """Replace Korean number+unit spans in `text` with their English equivalents.

    Two passes (each non-overlapping internally), applied in priority order so
    that longer / more specific patterns win:
      1. Percentage points  ("10%포인트" → "10 percentage points")
      2. Korean magnitude+unit ("120억원" → "12 billion Korean won")

    Returns (rewritten_text, list_of_replacements_with_offsets_in_original).
    """
    matches: list[dict] = []
    used_ranges: list[tuple[int, int]] = []

    def overlaps(s: int, e: int) -> bool:
        return any(not (e <= us or s >= ue) for us, ue in used_ranges)

    # Pass 1: percentage points (highest priority)
    for m in PERCENT_POINT_PATTERN.finditer(text):
        s, e = m.start(), m.end()
        if overlaps(s, e):
            continue
        num_str = m.group("num")
        english = f"{_format_simple_number(num_str)} percentage points"
        matches.append({
            "start": s,
            "end": e,
            "span": text[s:e],
            "amount": None,
            "unit_ko": "%포인트",
            "english": english,
        })
        used_ranges.append((s, e))

    # Pass 2: Korean magnitude+currency/counter
    for m in NUM_UNIT_PATTERN.finditer(text):
        s, e = m.start(), m.end()
        if overlaps(s, e):
            continue
        num_str = m.group("num").strip()
        unit = m.group("unit")
        if not re.search(r"\d", num_str):
            continue
        amount = parse_korean_number(num_str)
        if amount is None or amount == 0:
            continue
        english = format_english_amount(amount, unit)
        matches.append({
            "start": s,
            "end": e,
            "span": text[s:e],
            "amount": amount,
            "unit_ko": unit,
            "english": english,
        })
        used_ranges.append((s, e))

    # Apply replacements right-to-left to preserve earlier offsets
    matches.sort(key=lambda r: r["start"])
    out = text
    for r in sorted(matches, key=lambda r: r["start"], reverse=True):
        out = out[:r["start"]] + r["english"] + out[r["end"]:]

    return out, matches


def preprocess(text: str, article_date: str | None) -> tuple[str, dict]:
    """Inline-replace Korean numbers with English units inside the user text.

    The article date is NOT injected into the user message; callers should append
    it to the system prompt via `system_prompt_date_suffix()` instead.

    Returns (rewritten_text, debug_info).
    """
    norm_date = parse_article_date(article_date) if article_date else None
    rewritten, replacements = replace_numbers_inline(text)
    return rewritten, {"date": norm_date, "conversions": replacements}


def system_prompt_date_suffix(article_date: str | None) -> str:
    """Return the line to append to the system prompt for date anchoring.

    Empty string if no date is available.
    """
    norm_date = parse_article_date(article_date) if article_date else None
    if not norm_date:
        return ""
    return f"\n\n[Article published date: {norm_date}]"


# Small system-prompt addendum: explains both the inline numbers and the date header.
RULE_PROMPT_ADDENDUM = """

# RULE-PRECOMPUTED METADATA
The source text has been preprocessed before reaching you:
1. Korean monetary/number expressions (with 조 / 억 / 만 markers) have been REPLACED INLINE with their English-unit equivalents (e.g., "12 billion won", "5.4 trillion won", "89.7 billion dollars"). When you see such an English-unit expression in the source, output it VERBATIM in the translation — do not modify, recompute, or reformat the magnitude. Treat these expressions as pre-translated tokens that must pass through unchanged.
2. If a "[Article published date: YYYY-MM-DD]" line is provided at the top, use it to resolve relative date expressions. "지난 X일" means the Xth of the same month as the published date (or the most recent past Xth if X > current day-of-month). Do NOT translate "지난 X일" as "the Xth of last month" unless the source explicitly contains "지난달".
"""


if __name__ == "__main__":
    # Smoke test
    samples = [
        ("1023억원과 5조4000억원, 그리고 300만원", "2025-03-15"),
        ("897억달러 매출", "2025-04-02"),
        ("지난 19일 회담이 열렸다", "2025-01-23"),
        ("100억원대 분쟁이 발생", "2024-11-01"),
        ("5000만원의 보너스를 받았다", "2025-02-10"),
    ]
    for text, date in samples:
        out, info = preprocess(text, date)
        print(f"\n--- date={date} ---")
        print(f"input:  {text}")
        print(f"detected: {info['conversions']}")
        print(f"resolved date: {info['date']}")
        print(f"augmented:\n{out}")