"""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