DenozordecandClaude Opus 5.5
e3af778068
tools/audit_tracks.py и reports/tracks_audit.md — аудит всех дорожек техданных по data/raw с планом. Пункт 1 плана: harvest_core.codec_line_rows добавляет студии из строк «Аудио 2: AC3 … (MVO, LostFilm)», «Аудио Rus: … | TVShows» (studio_source=codec), которых нет среди строк релиза; тот же шаг в tools/relabel_studios.py (повторный запуск ничего не добавляет). labels.csv: +6 428 строк в 2 862 релизах (kinozal 3 444, nnm 1 609, rutracker 1 375), со студией 29 331 -> 35 759. tests/e2e_segmenter.py: модель через E2E_MODEL. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
261 lines
14 KiB
Python
261 lines
14 KiB
Python
"""Аудит дорожек в описаниях релизов: data/raw/* против data/labels.csv.
|
||
|
||
Читает текст каждой сохранённой страницы (kinozal: «Техданные» + «Релиз»;
|
||
nnm / rutracker: пост), находит строки русских/украинских дорожек
|
||
(«Перевод N:», «Озвучка:», «Аудио N: … (MVO, X)» и т. п., без субтитров) и считает:
|
||
|
||
* какие форматы строк встречаются и какие из них не ловят правила разметки
|
||
(segmenter.RE_AUDIO_TRACK) и запасной путь сервиса (release.codec_segments);
|
||
* студии из словаря меток, названные в строках дорожек, но отсутствующие
|
||
в метках релиза (сравнение через release.same_studio — «А. Гаврилов» = «Андрей Гаврилов»);
|
||
* мусор в колонке studio (каналы, языки, тип-слова, «Лицензия», «по заказу»);
|
||
* релизы без меток: legacy / ok / есть ли в них строки дорожек.
|
||
|
||
Результат — reports/tracks_audit_numbers.md (цифры и примеры); выводы и план —
|
||
reports/tracks_audit.md. Чтение raw ~2–3 мин; --cache сохраняет тексты в pickle.
|
||
|
||
.venv\\Scripts\\python tools\\audit_tracks.py [--cache путь.pkl]
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import csv
|
||
import json
|
||
import os
|
||
import pickle
|
||
import random
|
||
import re
|
||
import sys
|
||
from collections import Counter, defaultdict
|
||
from pathlib import Path
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
sys.path.insert(0, str(ROOT))
|
||
|
||
from src.release import RE_CODEC_TRACK, codec_ok, same_studio # noqa: E402
|
||
from src.segmenter import RE_AUDIO_TRACK # noqa: E402
|
||
|
||
RAW = ROOT / "data" / "raw"
|
||
LABELS = ROOT / "data" / "labels.csv"
|
||
OUT = ROOT / "reports" / "tracks_audit_numbers.md"
|
||
SOURCES = (("kinozal", "kz-"), ("nnm", "nnm-"), ("rutracker", "rt-"))
|
||
NO_STUDIO = "(не указана)"
|
||
|
||
# строка дорожки: перевод/озвучка или кодек-строка
|
||
RE_TRANS_LINE = re.compile(
|
||
r"(?i)^\s*[#№*•\-\s]*(перевод\w*|озвуч\w*|дубляж\w*|автор перевода)\s*(№|#)?\s*[\d\-–,\s]*(\([^)]*\))?\s*[:.]")
|
||
RE_AUDIO_LINE = re.compile(
|
||
r"(?i)^\s*[#№*•\-\s]*(аудио\w*|audio\w*|звук\w*|аудиодорожк\w*|дорожк\w*|track\w*)\b")
|
||
RE_CODEC = re.compile(
|
||
r"(?i)\b(e-?ac-?3|ac-?3|ас3|dts|aac|truehd|flac|mp3|opus|l?pcm|mp2|dolby)\b|\d\s*(kbps|кбит|kb/s|khz|кгц)")
|
||
RE_RU = re.compile(r"(?i)русск|russian|\brus\b|украин|ukrain|\bukr\b|україн")
|
||
RE_NON_RU = re.compile(
|
||
r"(?i)english|\beng\b|англ|япон|japan|\bjap\b|казах|kazakh|корей|китай|немец|француз|итальян|испан|оригинал|original")
|
||
RE_TYPE = re.compile(
|
||
r"(?i)\b(MVO|DVO|AVO|DUB|VO)\b|дубл|многоголос|двухголос|двуголос|одноголос|закадр|авторск|любительск|профессиональн")
|
||
RE_SUBS = re.compile(r"(?i)субтитр")
|
||
|
||
# не студия: тех. параметры, языки, тип-слова, источник дорожки
|
||
JUNK = re.compile(
|
||
r"(?i)^(\d+\s*ch|.*\b(ac-?3|dts|aac|kbps|kb/s|ch)\b.*|l,.*|.*русск\w*.*|английск\w*|украинск\w*|французск\w*"
|
||
r"|испанск\w*|немецк\w*|итальянск\w*|японск\w*|казахск\w*|субтитр\w*|ukr|rus|eng|eur|cee|vhs|tv|ts"
|
||
r"|дубляж|полное дублирование|дублированный|дву(х)?голосый|многоголосый|одноголосый|закадровый"
|
||
r"|профессиональный|любительский|авторский|студийный|неизвестн\w*|оригинал\w*|original"
|
||
r"|.*лицензи\w*.*|звук.*|blu-?ray.*|bd.*|itunes.*|netflix|amazon|web-?dl|в составе контейнера"
|
||
r"|.*ненормативн\w*.*|18\+)$")
|
||
# виды мусора для отчёта (первое совпадение)
|
||
NOISE_KINDS = (
|
||
("каналы/кодек", r"(?i)\b(\d+\s*ch|ac-?3|dts|aac|kbps|kb/s)\b|^l,"),
|
||
("язык вместо студии", r"(?i)^(русск|английск|украинск|французск|испанск|казахск)\w*$"),
|
||
("язык-префикс «Русский, - X»", r"(?i)^(русск|украинск|english)\w*\s*,"),
|
||
("тип-слово", r"(?i)^(дубляж|полное дублирование|дву(х)?голосый|многоголосый|одноголосый|закадровый)$"),
|
||
("источник: лицензия/Blu-ray/iTunes/Netflix", r"(?i)лицензи|blu-?ray|^bd\b|itunes|netflix|amazon"),
|
||
("«X по заказу Y»", r"(?i)по заказу"),
|
||
("субтитры", r"(?i)субтитр"),
|
||
("неру дорожка", r"(?i)казах|qazaq|kazakh|английск|english"),
|
||
)
|
||
|
||
|
||
def nid(rid: str) -> str:
|
||
return re.sub(r"^(kz|nnm|rt)-", "", rid)
|
||
|
||
|
||
def norm(s: str) -> str:
|
||
return re.sub(r"\s+", " ", s).strip().lower().replace("ё", "е")
|
||
|
||
|
||
def load_texts(cache: Path | None) -> dict:
|
||
"""{(source, id): (title, text, legacy)}; kinozal — h1 и вкладки, nnm/rt — пост."""
|
||
if cache and cache.exists():
|
||
return pickle.loads(cache.read_bytes())
|
||
out = {}
|
||
for src, _ in SOURCES:
|
||
d = RAW / src
|
||
for n in sorted(os.listdir(d), key=lambda n: "legacy" in n):
|
||
if not n.endswith(".json"):
|
||
continue
|
||
rid = n.split(".")[0]
|
||
if (src, rid) in out:
|
||
continue
|
||
try:
|
||
rec = json.loads((d / n).read_text(encoding="utf-8"))
|
||
except (OSError, ValueError):
|
||
continue
|
||
if src == "kinozal":
|
||
title = rec.get("h1") or ""
|
||
text = "\n".join(x for x in (rec.get("tech_tab"), rec.get("release_tab")) if x)
|
||
else:
|
||
title, text = rec.get("title") or "", rec.get("post") or ""
|
||
out[(src, rid)] = (title, text, "legacy" in n, rec.get("ok", True))
|
||
if cache:
|
||
cache.write_bytes(pickle.dumps(out))
|
||
return out
|
||
|
||
|
||
def track_lines(text: str) -> list[tuple[str, str]]:
|
||
"""[(вид, строка)] строк русских/украинских дорожек; вид: trans | audio."""
|
||
out = []
|
||
for line in text.splitlines():
|
||
line = line.strip()
|
||
if not line or len(line) > 400 or RE_SUBS.search(line) and not RE_TYPE.search(line):
|
||
continue
|
||
if RE_TRANS_LINE.match(line):
|
||
if RE_NON_RU.search(line.split(":", 1)[0]) and not RE_RU.search(line):
|
||
continue
|
||
out.append(("trans", line))
|
||
elif (RE_AUDIO_LINE.match(line) or RE_RU.match(line)) and RE_CODEC.search(line):
|
||
if RE_RU.search(line) and (RE_TYPE.search(line) or "|" in line or "(" in line):
|
||
out.append(("audio", line))
|
||
return out
|
||
|
||
|
||
def line_prefix(line: str) -> str:
|
||
p = re.match(r"\s*\W*\w*[^:]{0,14}:?", line).group(0).strip().lower()
|
||
return re.sub(r"\s+", " ", re.sub(r"\d+", "N", p))
|
||
|
||
|
||
def studio_vocab(rows: list[dict]) -> tuple[re.Pattern, dict]:
|
||
cnt = Counter(r["studio"].strip() for r in rows if r["studio"] not in ("", NO_STUDIO))
|
||
vocab = [s for s, c in cnt.items() if c >= 5 and len(s) >= 3 and not JUNK.match(s)]
|
||
vocab.sort(key=len, reverse=True)
|
||
rx = re.compile(r"(?i)(?<![\w])(" + "|".join(map(re.escape, vocab)) + r")(?![\w])")
|
||
return rx, {norm(s): s for s in vocab}
|
||
|
||
|
||
def audit(texts: dict, rows: list[dict]) -> dict:
|
||
lab = defaultdict(list)
|
||
for r in rows:
|
||
lab[(r["source"], nid(r["release_id"]))].append(r)
|
||
voc_rx, canon = studio_vocab(rows)
|
||
R = defaultdict(lambda: defaultdict(Counter))
|
||
ex = defaultdict(list)
|
||
|
||
for (src, rid), (title, text, legacy, ok) in texts.items():
|
||
L = lab.get((src, rid), [])
|
||
lines = track_lines(text)
|
||
rel = R[src]["releases"]
|
||
rel["всего"] += 1
|
||
if not L:
|
||
rel["без меток"] += 1
|
||
key = ("legacy" if legacy else "обычный") + (", ok=False" if not ok else "") + \
|
||
(", есть строки дорожек" if lines else ", нет строк дорожек")
|
||
R[src]["без меток"][key] += 1
|
||
if lines and len(ex[(src, "unlab")]) < 3000:
|
||
ex[(src, "unlab")].append((rid, title[:80], lines[0][1][:140]))
|
||
labs = [r["studio"] for r in L if r["studio"] not in ("", NO_STUDIO) and not JUNK.match(r["studio"].strip())]
|
||
missing: dict[str, tuple[str, str]] = {}
|
||
for kind, line in lines:
|
||
R[src]["строки"][kind] += 1
|
||
if kind == "audio":
|
||
a = bool(RE_AUDIO_TRACK.match(line))
|
||
b = bool(RE_CODEC_TRACK.match(line) and codec_ok(line))
|
||
R[src]["кодек-строки"][("RE_AUDIO_TRACK " if a else "мимо RE_AUDIO_TRACK ") +
|
||
("| codec_segments" if b else "| мимо codec_segments")] += 1
|
||
if not a or not b:
|
||
miss = ("seg " if not a else "") + ("codec" if not b else "")
|
||
R[src]["форматы мимо: " + miss.strip()][line_prefix(line)] += 1
|
||
if len(ex[(src, "fmt:" + miss.strip())]) < 3000:
|
||
ex[(src, "fmt:" + miss.strip())].append((rid, line[:160]))
|
||
if not L:
|
||
continue
|
||
for m in voc_rx.finditer(line):
|
||
st = canon.get(norm(m.group(1)), m.group(1))
|
||
if any(same_studio(st, x) for x in labs) or any(same_studio(st, x) for x, _ in missing.values()):
|
||
continue
|
||
missing[norm(st)] = (st, kind)
|
||
if len(ex[(src, "miss:" + kind)]) < 5000:
|
||
ex[(src, "miss:" + kind)].append((rid, st, line[:160]))
|
||
if L:
|
||
rel["с метками"] += 1
|
||
rel["с метками, есть недостающая студия"] += bool(missing)
|
||
for st, kind in missing.values():
|
||
R[src]["недостающие студии: вид строки"][kind] += 1
|
||
R[src]["недостающие студии: топ"][st] += 1
|
||
# недостающие студии в релизе, где у меток вообще нет студий
|
||
if missing and not labs:
|
||
rel["с метками без студий, студия есть в кодек-строке"] += 1
|
||
full = norm(title + "\n" + text)
|
||
for r in L:
|
||
st = r["studio"].strip()
|
||
if st in ("", NO_STUDIO) or JUNK.match(st):
|
||
continue
|
||
R[src]["студии меток"]["всего"] += 1
|
||
words = [w for w in re.findall(r"\w{4,}", norm(st))]
|
||
if norm(st) not in full and not any(w in full for w in words):
|
||
R[src]["студии меток"]["нет в тексте/заголовке"] += 1
|
||
R[src]["студии меток: нет в тексте, studio_source"][r["studio_source"]] += 1
|
||
if len(ex[(src, "ghost")]) < 3000:
|
||
ex[(src, "ghost")].append((rid, st, r["raw"][:140]))
|
||
for r in rows:
|
||
st = r["studio"].strip()
|
||
for k, rx in NOISE_KINDS:
|
||
if re.search(rx, st):
|
||
R[r["source"]]["мусор в studio меток"][k] += 1
|
||
if len(ex[(r["source"], "noise:" + k)]) < 500:
|
||
ex[(r["source"], "noise:" + k)].append((r["release_id"], st, r["raw"][:120]))
|
||
break
|
||
return {"stats": R, "examples": ex}
|
||
|
||
|
||
def write_md(res: dict, path: Path) -> None:
|
||
R, ex = res["stats"], res["examples"]
|
||
rnd = random.Random(7)
|
||
out = ["# Аудит дорожек в raw-страницах (цифры)", "",
|
||
"Сгенерировано `tools/audit_tracks.py`; выводы и план — `reports/tracks_audit.md`.", ""]
|
||
for src, _ in SOURCES:
|
||
out += [f"## {src}", ""]
|
||
for block, cnt in R[src].items():
|
||
top = cnt.most_common(25)
|
||
out.append(f"**{block}**: " + "; ".join(f"{k} — {v}" for k, v in top))
|
||
out.append("")
|
||
for key in sorted(k for k in ex if k[0] == src):
|
||
v = ex[key]
|
||
out.append(f"<details><summary>примеры {key[1]} ({len(v)})</summary>\n")
|
||
for t in rnd.sample(v, min(15, len(v))):
|
||
out.append("- " + " · ".join(str(x).replace("|", "\\|") for x in t))
|
||
out += ["", "</details>", ""]
|
||
path.write_text("\n".join(out), encoding="utf-8")
|
||
|
||
|
||
def main() -> None:
|
||
ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
||
ap.add_argument("--cache", type=Path, help="pickle с текстами raw (создаётся, если нет)")
|
||
ap.add_argument("--out", type=Path, default=OUT)
|
||
args = ap.parse_args()
|
||
sys.stdout.reconfigure(encoding="utf-8")
|
||
with LABELS.open(encoding="utf-8-sig", newline="") as f:
|
||
rows = list(csv.DictReader(f))
|
||
res = audit(load_texts(args.cache), rows)
|
||
args.out.parent.mkdir(exist_ok=True)
|
||
write_md(res, args.out)
|
||
for src, _ in SOURCES:
|
||
print(f"== {src}")
|
||
for block, cnt in res["stats"][src].items():
|
||
print(f" {block}: " + "; ".join(f"{k} — {v}" for k, v in cnt.most_common(12)))
|
||
print(f"записано: {args.out.relative_to(ROOT)}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|