| """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 |
|
|
| |
| |
| MULTI_GROUP = re.compile(r"\]\s*\[") |
| EXTENSION = re.compile(r"\.[A-Za-z0-9]{1,8}$") |
|
|
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| STRUCTURAL_EDGE_CHARS = frozenset( |
| "#<>{}@|／∥｜⁘❖【】『』⟦⟧‖‣⁓⧸⁄∕⋯»〔〕❪❫｟｠〘〙〖〗«》" |
| ) |
|
|
| |
| |
| |
| |
| |
| |
| _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))$" |
| ) |
|
|
| |
| |
| NON_MEDIA_EXTENSIONS = frozenset({ |
| "srt", "ass", "ssa", "vtt", "sub", "idx", "sup", "smi", |
| "jpg", "jpeg", "png", "gif", "webp", "bmp", "tbn", |
| "nfo", "txt", "pdf", "epub", "docx", "doc", "md", "html", "htm", |
| "xml", "json", "log", |
| "md5", "sfv", "sha1", "sha256", "par2", |
| "gz", "tar", "bz2", "xz", |
| }) |
|
|
|
|
| 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 |
| |
| |
| |
| 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"} |
| |
| |
| |
| HEIGHT_CLASS = {720: "720p", 1080: "1080p", 2160: "2160p"} |
| MIN_ASPECT, MAX_ASPECT = 1.7, 1.95 |
| MAX_EPISODES = 64 |
| YEAR_MIN, YEAR_MAX = 1870, 2099 |
|
|
| |
| |
| |
| |
| 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) |
| elif previous == "number" and current == "number" and word_seen: |
| cut.append(index) |
| 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) |
| 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]) |
| |
| |
| |
| |
| 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"): |
| |
| |
| |
| |
| |
| 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"] |
| else: |
| value = season_value(text) |
| 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: |
| |
| |
| 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: |
| |
| issues.append("list-too-long") |
| out["episodes"] = values |
| if group: |
| out["release_group"] = group |
| if resolution: |
| out["resolution"] = resolution |
| |
| 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 |
|
|
