app/fp/normalize.py

"""Contract v2 normalizer: evidence spans -> emitted record, deterministically.

Implements docs/LABELING-CONTRACT.md sections 3-7. This is the only permitted
path from labels to an emitted record: annotators mark spans, this code derives
the record. Hand-editing an output string is a contract violation, so the
derivation has to be inspectable and testable on its own.

Span dicts: `{'kind', 'start', 'end'}` plus optional `text` (must match the
query bytes), `options` and `value`. Kinds: TITLE, EPISODE_TITLE, GROUP, YEAR, SEASON,
EPISODE, RANGE, LIST, RESOLUTION, DIMENSIONS, DISCARD. `options` carries
label-side decisions the contract refuses to guess from bytes: `is_title`,
`fold_case`, `items`. `value` carries a normalized season number (None means
an explicit-but-unbounded ordinal, which is omitted, never 0).

`covered` may list extra [start, end) ranges (for example a path component
carrying no field) so that the debris check is a fact about the label, not a
guess about punctuation. Debris checking is opt-in because real corpora are
partially labeled; a full-label row should turn it on.
"""

import re
import unicodedata

# A group name is copied verbatim from one bracket, so a collab credit such as
# `ACB字幕组&天空字幕组` stays one string; two separate group spans conflict.
MULTI_GROUP = re.compile(r"\]\s*\[")
EXTENSION = re.compile(r"\.[A-Za-z0-9]{1,8}$")

# These punctuation marks are used as structural delimiters by the controlled
# format suites. They are not title characters at a predicted span edge; the
# gold derivation still uses the original spans and never calls this trimmer.
# The Unicode members were added after the consumed fresh-6 audit exposed
# full-width/vertical delimiters bleeding into TITLE, SEASON, and EPISODE spans,
# then extended for the fresh-7 delimiter inventory *before* fresh-7 was
# generated, so no suite can be accused of picking delimiters after seeing the
# trimmer. The list is duplicated verbatim in goport/normalize.go; both sides
# are pinned by tests. ASCII "+" is deliberately absent: it occurs in real
# titles ("C++") and would corrupt evidence to trim it.
STRUCTURAL_EDGE_CHARS = frozenset(
    "#<>{}@|／∥｜⁘❖【】『』⟦⟧‖‣⁓⧸⁄∕⋯»〔〕❪❫｟｠〘〙〖〗«》"
)

# The roster above is only the manual half of the edge test. Any non-ASCII rune in the
# dash, bracket or punctuation categories is an edge rune too, even when the roster
# predates it: that is what stops the set from measuring which delimiters happened to be
# enumerated before a suite shipped. ASCII runes are never widened this way -- ASCII
# punctuation legitimately terminates role text ("Title...", "(Recut)", "Title '05'"), so
# ASCII stays roster-only.
_EDGE_CATEGORIES = frozenset({"Pd", "Po", "Ps", "Pe", "Pi", "Pf", "Pc"})


def is_structural_edge(ch: str) -> bool:
    """Whether `ch` may be trimmed from a *predicted* span edge. Gold spans never see this."""
    if ord(ch) < 0x80:
        return ch in STRUCTURAL_EDGE_CHARS
    return ch in STRUCTURAL_EDGE_CHARS or unicodedata.category(ch) in _EDGE_CATEGORIES

PATH_HASH_GROUP = re.compile(
    r"(?i)(?:[- ](?:subs?|raws?|team|works?|encode|unit|signal|depot|delta|plow|"
    r"cellar|marrow|nest|harbor|lantern|rip|release|group)|"
    r"(?:subs|raws|team|works|unit|signal|depot|delta|plow|cellar|marrow|nest|"
    r"harbor|lantern))$"
)

# Contract §10: sidecars and documents are not media, even with a perfect
# title. Containers (.rar/.zip/.7z/.torrent) are NOT listed: §10 parses them.
NON_MEDIA_EXTENSIONS = frozenset({
    "srt", "ass", "ssa", "vtt", "sub", "idx", "sup", "smi",          # subtitles
    "jpg", "jpeg", "png", "gif", "webp", "bmp", "tbn",               # images
    "nfo", "txt", "pdf", "epub", "docx", "doc", "md", "html", "htm", # documents
    "xml", "json", "log",
    "md5", "sfv", "sha1", "sha256", "par2",                         # checksums
    "gz", "tar", "bz2", "xz",                                       # non-media archives
})


def non_media_extension(query):
    """The final dot-suffix, lower-cased, when it names a non-media sidecar;
    otherwise None. Position-anchored like `EXTENSION`: only the true suffix
    counts, a `.srt` in the middle of a path is not a decision here."""
    match = EXTENSION.search(query)
    if not match:
        return None
    ext = match.group(0)[1:].lower()
    return ext if ext in NON_MEDIA_EXTENSIONS else None


def leading_hash_stem(query):
    """Whether a query has the titleless CRC-prefixed release-stem shape.

    This is intentionally narrower than ``query.startswith('[')``: the stem
    must have a hex token, two dash-separated fields, and a closed resolution
    class at the end. It is a decision-layer rule for rows whose only apparent
    title is a group-shaped token between a checksum and codec junk; ordinary
    bracket-prefixed titles and two-field releases remain untouched.
    """
    if not re.match(r"^\[[0-9a-fA-F]{7,12}\]\s+", query):
        return False
    rest = query[query.find("]") + 1:].lstrip()
    fields = rest.split(" - ")
    if len(fields) != 3 or not all(fields[:2]):
        return False
    tail = fields[2]
    return bool(re.fullmatch(r"(?:480p|576p|720p|1080p|1080i|2160p)\.[A-Za-z0-9]{1,8}",
                             tail, flags=re.IGNORECASE))


UNICODE_SPACE = re.compile("[   -   　]")
DIGITS = re.compile(r"\d+(?:\.\d+)*")
YEAR_LIKE = re.compile(r"\b\d{4}\b")
OPEN_BRACKETS = "([｛「『〈《【“｢"
CLOSE_BRACKETS = ")]｝」』〉》】”｣"
BRACKETS = OPEN_BRACKETS + CLOSE_BRACKETS
# Brackets are deliberately NOT edge-stripped: an unmatched bracket is either
# lexical title content or a label error, and silently deleting one side of a
# pair corrupts evidence. Only balanced pairs are removed (see normalize_text).
STRIP_CHARS = " \t\r\n-–—~～·•:：/\\."


def _balanced(text):
    return all(text.count(open_) >= text.count(close)
               for open_, close in zip(OPEN_BRACKETS, CLOSE_BRACKETS))


def _strip_bracket_pairs(text):
    """Contract 3: brackets are removed only as balanced pairs enclosing the
    whole fragment. `ポコニャン(叮噹貓)` keeps its parentheses (they are inside
    the title), `｢title」` and `「title」` lose them (they delimit it)."""
    while (len(text) >= 2 and text[0] in OPEN_BRACKETS and text[-1] in CLOSE_BRACKETS
           and _balanced(text[1:-1])):
        text = _fold(text[1:-1]).strip(STRIP_CHARS)
    return text

SCHEMA_VALUES = {"480p", "576p", "720p", "1080p", "1080i", "2160p"}
# Contract 7: a bare height is only HD evidence. SD heights are shared with
# computer displays (640x480 VGA, 800x480) and non-square-pixel DVD captures,
# so they never establish a broadcast class.
HEIGHT_CLASS = {720: "720p", 1080: "1080p", 2160: "2160p"}
MIN_ASPECT, MAX_ASPECT = 1.7, 1.95  # widescreen video frames only
MAX_EPISODES = 64
YEAR_MIN, YEAR_MAX = 1870, 2099

# Contract 5 value resolution. Deliberately a *closed* vocabulary: if a season
# marker is not in here, the row reports `season-unparsable` instead of guessing.
# A model that emits the right span never has to know numbers, and this table is
# the auditable place where text becomes a value.
DIGITS = re.compile(r"\d+")
FINAL_WORDS = ("final", "última", "ultimas", "last", "fin", "おしまい", "最終", "最后", "最终",
               "最後", "끝", "последн")
ORDINAL_WORDS = {
    "first": 1, "second": 2, "third": 3, "fourth": 4, "fifth": 5, "sixth": 6,
    "seventh": 7, "eighth": 8, "ninth": 9, "tenth": 10, "primeira": 1, "segunda": 2,
    "terceira": 3, "quarta": 4, "quinta": 5, "primera": 1, "tercera": 3,
    "premiere": 1, "deuxieme": 2, "deuxième": 2, "troisieme": 3, "troisième": 3,
    "erste": 1, "zweite": 2, "dritte": 3, "primo": 1, "secundo": 2, "seconda": 2,
    "prima": 1, "terzo": 3, "uma": 1, "dois": 2, "tres": 3, "três": 3, "cuarta": 4,
    "quinta": 5, "sexta": 6, "fifth": 5,
    "первый": 1, "второй": 2, "третий": 3, "четвёртый": 4, "пятый": 5, "шестой": 6,
    "седьмой": 7, "восьмой": 8, "девятый": 9, "десятый": 10,
}
CJK_NUMERALS = {"一": 1, "二": 2, "三": 3, "四": 4, "五": 5, "六": 6, "七": 7,
                "八": 8, "九": 9, "十": 10, "壹": 1, "貳": 2, "叁": 3}


def season_value(text):
    """int, None (=explicit season with no number), or 'unparsable'.

    Digits win over words: `2nd Season` and `第2期` are the same value. A
    multi-number marker (`Season 2 of 12`, `S01E02` mistagged) is unparsable
    rather than a pick-one, because either choice is a guess about the work.
    """
    folded = normalize_text(text, fold_case=True)
    if any(word in folded for word in FINAL_WORDS):
        return None
    numbers = DIGITS.findall(folded)
    if len(numbers) == 1:
        value = int(numbers[0])
        return value if 0 <= value <= 99 else "unparsable"
    if len(numbers) > 1:
        return "unparsable"
    for word in folded.split(" "):
        if word in ORDINAL_WORDS:
            return ORDINAL_WORDS[word]
    runs = [c for c in folded if c in CJK_NUMERALS]
    if len(runs) == 1:
        return CJK_NUMERALS[runs[0]]
    return "unparsable"


def _fold(text):
    """Contract 3: underscores and exotic whitespace become spaces, runs
    collapse, nothing is transliterated. Dots are handled separately, because
    whether one is a delimiter depends on its neighbours (see
    `fold_dot_delimiters`)."""
    text = UNICODE_SPACE.sub(" ", text)
    text = text.replace("_", " ")
    text = re.sub(r"[-–—]{2,}", "-", text)
    return re.sub(r" {2,}", " ", text).strip()


ALPHA = re.compile(r"[^\W\d_]+\Z", re.UNICODE)
NUMERIC = re.compile(r"\d+\Z")


def _segment_kind(unit):
    if not unit:
        return "empty"
    if ALPHA.match(unit):
        return "word"
    if NUMERIC.match(unit):
        return "number"
    return "mixed"


def fold_dot_delimiters(text):
    """Contract 3.2, title fragments only, decided dot by dot inside each
    whitespace token, where a segment is a *word* (letters of any script), a
    *number* (digits only), *mixed* (letters and digits together: `1080p`,
    `2a`, `S01`, `x264`) or *empty* (a doubled or trailing dot).

    A dot splits when either neighbour is a word (`CITY.HUNTER` ->
    `CITY HUNTER`, `Akane.1988` -> `Akane 1988`, `Movie.2.Innocence` ->
    `Movie 2 Innocence`), when the right side is mixed (`Ghost.1080p` ->
    `Ghost 1080p`: a technical tag is never title text, and a split that only
    touches title-adjacent junk cannot lose evidence), or when a side is empty
    (`Dr.`, `Ghost.` lose the delimiter dot). The one shape that is never cut
    is `number . number` before any word has appeared in the token:
    `1917.2019` and `2.43.121` survive as single tokens, because a run that
    opens with digits reads as an encoded value and splitting it would be a
    guess about which part is the title. Single letters are words, so
    `A.K.A.` folds to `A K A`.

    Consequence: a title that genuinely contains a digit-dot-digit sequence
    must be labeled as two fragments. That is the honest cost of a rule an
    annotator can apply without knowing the work."""
    pieces = []
    for token in text.split(" "):
        segments = token.split(".")
        if len(segments) == 1:
            if token:
                pieces.append(token)
            continue
        kinds = [_segment_kind(s) for s in segments]
        cut, word_seen = [0], False
        for index in range(1, len(segments)):
            previous, current = kinds[index - 1], kinds[index]
            touches_word = previous == "word" or current == "word"
            if current in ("mixed", "empty") or touches_word:
                cut.append(index)  # a word or junk boundary: split here
            elif previous == "number" and current == "number" and word_seen:
                cut.append(index)  # `Akane.1988.2020` splits; `1917.2019` does not
            word_seen = word_seen or touches_word
        for boundary, edge in zip(cut, cut[1:] + [len(segments)]):
            unit = ".".join(segments[boundary:edge])
            if unit:
                pieces.append(unit)
    text = " ".join(pieces)
    return re.sub(r" {2,}", " ", text).strip()


def trim_structural_edges(query, spans):
    """Trim structural delimiter runes from predicted span edges.

    This is deliberately prediction-only. Gold labels are never rewritten:
    a label span that owns `#`, `<`, `>`, `{`, `}`, `@`, or `|` remains
    lossless. The trimmer handles a decoder that includes a neighboring
    delimiter in a role span, and drops a span made only of delimiters.
    """
    out = []
    changed = 0
    for span in spans:
        start, end = span["start"], span["end"]
        first, last = start, end
        while first < last and is_structural_edge(query[first]):
            first += 1
        while last > first and is_structural_edge(query[last - 1]):
            last -= 1
        if first != start or last != end:
            changed += 1
        if first < last:
            out.append({**span, "start": first, "end": last})
    return out, changed


def titleless_path_group(query, spans):
    """Narrow prediction gate for ``(hash)/junk/Group/res`` stems.

    The parenthesized hash, slash path, and group-like title suffix must all be
    present, with no episode, season, year, or episode-title evidence. A
    positive row with any additional title evidence is left untouched.
    """
    if not re.match(r"^\([0-9a-fA-F]{7,12}\)", query) or "/" not in query:
        return False
    titles = [span for span in spans if span["kind"] == "TITLE"]
    evidence = [span for span in spans if span["kind"] in
                ("EPISODE", "EPISODE_TITLE", "SEASON", "YEAR")]
    if len(titles) != 1 or evidence:
        return False
    return bool(PATH_HASH_GROUP.search(query[titles[0]["start"]:titles[0]["end"]].strip()))


def normalize_text(text, is_title=False, fold_case=False, keep_edges=False):
    """`keep_edges=True` is for an already-assembled title: an edge bracket or
    hyphen there belongs to an adjacent fragment, so edge stripping is skipped
    and only byte-internal rules (separators, whitespace, delimiter dots)
    apply. Fragments must not use it - that is where delimiters are removed."""
    text = _fold(text)
    if not keep_edges:
        text = text.strip(STRIP_CHARS)
        text = _strip_bracket_pairs(text)
        text = text.strip(STRIP_CHARS)
    if fold_case:
        text = text.lower()
    if is_title:
        text = fold_dot_delimiters(text)
    return _fold(text)


def _range_endpoints(text):
    """Return (None, (lo, hi), issues). Repeated season prefixes
    (`S01E01-S01E03`) and embedded technical values (`[01 ~ 24][1080p]`,
    `1280x720`) are removed first, so the endpoints are the only numbers left.
    Endpoints are 1-3 digits: a 4-digit number is a year or a checksum, not an
    episode index."""
    stripped = re.sub(r"(?i)s\d+", " ", text)
    stripped = re.sub(r"(?i)\d{3,5}\s*[x×]\s*\d{3,5}|\d{3,4}[pi]\b", " ", stripped)
    nums = [int(n) for n in re.findall(r"\d{1,3}", stripped)]
    if not nums:
        return None, None, ["range-no-digits"]
    if len(nums) > 2:
        return None, None, ["range-too-complex"]
    if len(nums) == 1:
        return None, None, ["range-single-value"]
    if not re.search(r"[-~–—至到]", text):
        return None, None, ["range-no-connector"]
    return None, (nums[0], nums[-1]), []


def expand_range(text, cap=MAX_EPISODES):
    _, endpoints, issues = _range_endpoints(text)
    if issues:
        return [], issues
    a, b = endpoints
    if a == b:
        issues.append("range-single-value")
    lo, hi = min(a, b), max(a, b)
    if hi - lo + 1 > cap:
        return [], ["list-too-long"]
    if hi - lo + 1 > 200:
        issues.append("range-suspiciously-long")
    return list(range(lo, hi + 1)), issues


def expand_list(span, text=None):
    """Item values come from the label side when there is one: re-reading them
    from bytes would reintroduce exactly the ambiguity the contract says to
    quarantine. A *predicted* span has no label side, so for those the digits
    inside the claimed boundary are read - the model's claim is the boundary.
    """
    values = span.get("options", {}).get("items")
    if values is None:
        values = re.findall(r"\d+", text if text is not None else span["text"])
    values = sorted({int(v) for v in values})
    if not values:
        return [], ["list-empty"]
    if len(values) > MAX_EPISODES:
        return [], ["list-too-long"]
    return values, []


def debris(query, claims):
    """Unclaimed, non-punctuation text: what a full-label row cannot explain.

    Only the container suffix at the very end of the query is excused, and it
    is excused by *position*, not by appearance: a `.720p` left unclaimed in
    the middle of a name is a missing decision and must be reported even
    though it looks like an extension. Stripping dot-suffixes from the
    leftovers instead silently forgives exactly the technical tags that strict
    rows exist to catch."""
    mask = bytearray(len(query))
    suffix = EXTENSION.search(query)  # anchored at `$`, so this is the final suffix
    for start, end in list(claims) + ([(suffix.start(), suffix.end())] if suffix else []):
        for i in range(max(0, min(start, len(query))), min(max(0, end), len(query))):
            mask[i] = 1
    text = "".join(ch for i, ch in enumerate(query) if not mask[i])
    # Contract 3.3: a run of underscores or hyphens that only delimits fields is
    # debris the pipeline removes, not text that needs a recorded decision. `_`
    # counted as a word character here, which made every underscored layout look
    # unexplainable.
    text = text.replace("_", " ")
    return [part for part in re.split(r"[^\w]+", text, flags=re.UNICODE) if part]


def record(query, spans, *, covered=None, check_debris=False):
    """Derive the emitted record from labeled spans. Returns (record|None,
    issues). Issues never silently change a value; conflicting or
    unrepresentable evidence is reported so the row can be quarantined."""
    issues = []
    claims = [(s["start"], s["end"]) for s in spans] + list(covered or [])
    title, episode_title, episodes = [], [], []
    group = year = season = resolution = dimensions = None
    for span in sorted(spans, key=lambda s: (s["start"], s["end"])):
        kind = span["kind"]
        text = span.get("text")
        if text is None:
            text = query[span["start"]:span["end"]]
        elif query[span["start"]:span["end"]] != text:
            issues.append("span-text-mismatch")
        opts = span.get("options", {})
        if kind in ("TITLE", "EPISODE_TITLE"):
            # Which fragment is the work title and which is the episode title is
            # the annotator's decision, recorded as a span kind. Position cannot
            # decide it: a second TITLE fragment is usually the same work title
            # in another script, while a second EPISODE_TITLE fragment is the
            # next segment of one episode's name.
            bucket = title if kind == "TITLE" else episode_title
            value = normalize_text(text, is_title=True,
                                   fold_case=opts.get("fold_case", False))
            if value:
                bucket.append(value)
            else:
                issues.append("empty-title-fragment")
        elif kind == "GROUP":
            value = normalize_text(text)
            if group is not None:
                issues.append("group-multiple-brackets")
            elif MULTI_GROUP.search(value):
                issues.append("group-multiple-names")
            group = group or value
        elif kind == "YEAR":
            digits = YEAR_LIKE.findall(text)
            if len(digits) != 1:
                issues.append("year-ambiguous")
                continue
            value = int(digits[0])
            if not YEAR_MIN <= value <= YEAR_MAX:
                issues.append("year-out-of-range")
            elif year is not None and year != value:
                issues.append("year-conflict")
            else:
                year = value
        elif kind == "SEASON":
            if "value" in span:
                value = span["value"]  # annotator-supplied (contract 5, worksheet)
            else:
                value = season_value(text)  # predicted span: text decides, not a head
                if value == "unparsable":
                    issues.append("season-unparsable")
                    continue
            if value is None:
                issues.append("season-unspecified")
            elif season is not None and season != value:
                issues.append("season-conflict")
            elif not 0 <= value <= 99:
                issues.append("season-out-of-range")
            else:
                season = value
        elif kind == "EPISODE":
            nums = re.findall(r"\d+", text)
            if len(nums) != 1:
                issues.append("episode-ambiguous")
            else:
                episodes.append(int(nums[0]))
        elif kind == "RANGE":
            values, extra = expand_range(text)
            issues += extra
            episodes += values
        elif kind == "LIST":
            values, extra = expand_list(span, text)
            issues += extra
            episodes += values
        elif kind == "RESOLUTION":
            value = normalize_text(text, fold_case=True)
            if value not in SCHEMA_VALUES:
                issues.append("resolution-unknown-class")
            elif resolution and resolution != value:
                issues.append("resolution-conflict")
            else:
                resolution = value
        elif kind == "DIMENSIONS":
            match = re.search(r"(\d{3,5})\s*[x×]\s*(\d{3,5})", text, re.I)
            if not match:
                issues.append("dimensions-unreadable")
                continue
            width, height = int(match.group(1)), int(match.group(2))
            value = HEIGHT_CLASS.get(height) if MIN_ASPECT <= width / height <= MAX_ASPECT else None
            if value is None:
                issues.append("dimensions-no-class")
            elif resolution and resolution != value:
                issues.append("resolution-conflict")
            else:
                resolution, dimensions = value, text
        elif kind != "DISCARD":
            issues.append("unhandled-kind:" + kind)

    if check_debris:
        left = debris(query, claims)
        if left:
            issues.append("unexplained-debris:" + "|".join(left[:6]))

    out = {}
    if title:
        # join=True: an edge bracket here belongs to an adjacent fragment, so
        # only byte-internal rules apply to the assembled title.
        out["title"] = normalize_text(" ".join(title), is_title=True, keep_edges=True)
    if episode_title:
        out["episode_title"] = normalize_text(" ".join(episode_title), is_title=True,
                                             keep_edges=True)
    if year is not None:
        out["year"] = year
    if season is not None:
        out["season"] = season
    if episodes:
        values = sorted(set(episodes))
        if len(values) > MAX_EPISODES:
            # Contract 9: never silently truncate; the row is quarantined.
            issues.append("list-too-long")
        out["episodes"] = values
    if group:
        out["release_group"] = group
    if resolution:
        out["resolution"] = resolution
    # `required: title` means a title-less record is the same output as no call.
    return (out if out.get("title") else None), issues


def gated_record(query, spans, **kwargs):
    """`record` followed by the decision layer applied at prediction time.

    A non-media suffix or the narrow CRC-prefixed titleless release-stem shape
    forces abstention and reports why in the issues list.
    """
    ext = non_media_extension(query)
    trimmed, trimmed_count = trim_structural_edges(query, spans)
    out, issues = record(query, trimmed, **kwargs)
    if trimmed_count:
        issues = issues + [f"trimmed-structural-edges:{trimmed_count}"]
    if out is not None and ext is not None:
        return None, issues + [f"gated-non-media-extension:{ext}"]
    if out is not None and leading_hash_stem(query):
        return None, issues + ["gated-leading-hash-stem"]
    if out is not None and titleless_path_group(query, trimmed):
        return None, issues + ["gated-titleless-path-group"]
    return out, issues