feat: M3 — automatic channel tagging (language + topic system tags)
- Tag and ChannelTag models + migration 0003 (partial unique indexes split system vs per-user tag names) - Offline language detection (py3langid) constrained to a curated language set, with the channel's declared default language as a strong prior - Topic tags mapped from YouTube topicDetails + dominant video category; the generic "Lifestyle" catch-all is intentionally dropped - System (auto) tags are regenerated idempotently and never touch user tags; orphaned system tags are cleaned up - GET /api/tags and admin POST /api/tags/recompute; scheduled autotag job
This commit is contained in:
npeter83
parent
64b18b3cc1
commit
68dad91e8a
9 changed files with 482 additions and 1 deletions
279
backend/app/sync/autotag.py
Normal file
279
backend/app/sync/autotag.py
Normal file
|
|
@ -0,0 +1,279 @@
|
|||
"""Compute system (auto) tags for channels: a language tag and one or more topic tags.
|
||||
|
||||
Language comes from the channel's declared default language or, failing that, offline
|
||||
detection over a sample of recent video titles. Topics are mapped from YouTube's
|
||||
topicDetails categories and the channel's dominant video category. System tags are
|
||||
regenerated freely; user tags are never touched here.
|
||||
"""
|
||||
from sqlalchemy import and_, exists, func, select
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.config import settings
|
||||
from app.models import Channel, ChannelTag, Tag, Video
|
||||
|
||||
# --- language id (offline, deterministic, small) ---
|
||||
try:
|
||||
from py3langid.langid import MODEL_FILE, LanguageIdentifier
|
||||
|
||||
_identifier = LanguageIdentifier.from_pickled_model(MODEL_FILE, norm_probs=True)
|
||||
|
||||
def _classify(text: str):
|
||||
return _identifier.classify(text)
|
||||
|
||||
def _restrict(codes: list[str]) -> None:
|
||||
_identifier.set_languages(codes)
|
||||
|
||||
except Exception: # pragma: no cover - fallback to module-level API
|
||||
import py3langid as _langid
|
||||
|
||||
def _classify(text: str):
|
||||
return _langid.classify(text)
|
||||
|
||||
def _restrict(codes: list[str]) -> None:
|
||||
_langid.set_languages(codes)
|
||||
|
||||
|
||||
LANG_NAMES = {
|
||||
"en": "English",
|
||||
"hu": "Hungarian",
|
||||
"de": "German",
|
||||
"fr": "French",
|
||||
"es": "Spanish",
|
||||
"it": "Italian",
|
||||
"pt": "Portuguese",
|
||||
"nl": "Dutch",
|
||||
"ru": "Russian",
|
||||
"pl": "Polish",
|
||||
"ro": "Romanian",
|
||||
"cs": "Czech",
|
||||
"sk": "Slovak",
|
||||
"tr": "Turkish",
|
||||
"sv": "Swedish",
|
||||
"da": "Danish",
|
||||
"no": "Norwegian",
|
||||
"fi": "Finnish",
|
||||
"ja": "Japanese",
|
||||
"ko": "Korean",
|
||||
"zh": "Chinese",
|
||||
"hi": "Hindi",
|
||||
"ar": "Arabic",
|
||||
"uk": "Ukrainian",
|
||||
"el": "Greek",
|
||||
}
|
||||
# Constrain detection to plausible languages so short/branded titles can't trip the
|
||||
# detector into obscure false positives (e.g. Latin, Luxembourgish).
|
||||
_restrict(list(LANG_NAMES))
|
||||
|
||||
LANG_COLOR = "#3b82f6"
|
||||
DEFAULT_TOPIC_COLOR = "#6b7280"
|
||||
TOPIC_COLORS = {
|
||||
"Gaming": "#8b5cf6",
|
||||
"Music": "#ec4899",
|
||||
"Cooking": "#f59e0b",
|
||||
"Tech": "#06b6d4",
|
||||
"Education": "#10b981",
|
||||
"Sports": "#ef4444",
|
||||
"Film": "#a855f7",
|
||||
"TV": "#a855f7",
|
||||
"Politics": "#64748b",
|
||||
"Health & Fitness": "#22c55e",
|
||||
"Vehicles": "#eab308",
|
||||
"Animals": "#84cc16",
|
||||
"Fashion & Beauty": "#f472b6",
|
||||
"Travel": "#14b8a6",
|
||||
"Comedy": "#fb923c",
|
||||
"Entertainment": "#c084fc",
|
||||
"Business": "#0ea5e9",
|
||||
"Science": "#06b6d4",
|
||||
"News": "#94a3b8",
|
||||
}
|
||||
|
||||
# Exact topic slug -> friendly tag (for slugs the heuristics below don't catch).
|
||||
_TOPIC_SLUGS = {
|
||||
"Knowledge": "Education",
|
||||
"Food": "Cooking",
|
||||
"Technology": "Tech",
|
||||
# "Lifestyle_(sociology)" is YouTube's catch-all (on ~65% of channels) — too generic
|
||||
# to be a useful filter, so it is intentionally not mapped.
|
||||
"Society": "Society",
|
||||
"Politics": "Politics",
|
||||
"Health": "Health & Fitness",
|
||||
"Physical_fitness": "Health & Fitness",
|
||||
"Vehicle": "Vehicles",
|
||||
"Pet": "Animals",
|
||||
"Tourism": "Travel",
|
||||
"Fashion": "Fashion & Beauty",
|
||||
"Physical_attractiveness": "Fashion & Beauty",
|
||||
"Humor": "Comedy",
|
||||
"Entertainment": "Entertainment",
|
||||
"Film": "Film",
|
||||
"Television_program": "TV",
|
||||
"Performing_arts": "Performing Arts",
|
||||
"Business": "Business",
|
||||
"Military": "Military",
|
||||
"Religion": "Religion",
|
||||
"Hobby": "Hobbies",
|
||||
}
|
||||
|
||||
# Dominant video category id -> friendly tag (overly generic ones are skipped).
|
||||
_CATEGORY_NAMES = {
|
||||
1: "Film",
|
||||
2: "Vehicles",
|
||||
10: "Music",
|
||||
15: "Animals",
|
||||
17: "Sports",
|
||||
19: "Travel",
|
||||
20: "Gaming",
|
||||
23: "Comedy",
|
||||
25: "News",
|
||||
26: "Fashion & Beauty",
|
||||
27: "Education",
|
||||
28: "Tech",
|
||||
}
|
||||
|
||||
|
||||
def map_topic_slug(slug: str) -> str | None:
|
||||
low = slug.lower()
|
||||
if "game" in low:
|
||||
return "Gaming"
|
||||
if "music" in low:
|
||||
return "Music"
|
||||
if any(s in low for s in ("sport", "football", "basketball", "tennis", "boxing")):
|
||||
return "Sports"
|
||||
return _TOPIC_SLUGS.get(slug)
|
||||
|
||||
|
||||
def detect_channel_language(db: Session, channel: Channel) -> tuple[str | None, float]:
|
||||
if channel.default_language:
|
||||
return channel.default_language.split("-")[0].lower(), 0.99
|
||||
titles = (
|
||||
db.execute(
|
||||
select(Video.title)
|
||||
.where(Video.channel_id == channel.id, Video.title.is_not(None))
|
||||
.order_by(Video.published_at.desc())
|
||||
.limit(settings.autotag_title_sample)
|
||||
)
|
||||
.scalars()
|
||||
.all()
|
||||
)
|
||||
text = " . ".join(t for t in titles if t).strip()
|
||||
if len(text) < 15:
|
||||
return None, 0.0
|
||||
lang, confidence = _classify(text)
|
||||
return lang, float(confidence)
|
||||
|
||||
|
||||
def compute_channel_topics(db: Session, channel: Channel) -> set[str]:
|
||||
topics: set[str] = set()
|
||||
for url in channel.topic_categories or []:
|
||||
slug = url.rstrip("/").split("/")[-1]
|
||||
name = map_topic_slug(slug)
|
||||
if name:
|
||||
topics.add(name)
|
||||
dominant = db.execute(
|
||||
select(Video.category_id)
|
||||
.where(Video.channel_id == channel.id, Video.category_id.is_not(None))
|
||||
.group_by(Video.category_id)
|
||||
.order_by(func.count().desc())
|
||||
.limit(1)
|
||||
).scalar_one_or_none()
|
||||
if dominant is not None and dominant in _CATEGORY_NAMES:
|
||||
topics.add(_CATEGORY_NAMES[dominant])
|
||||
return topics
|
||||
|
||||
|
||||
def ensure_system_tag(db: Session, name: str, category: str, color: str | None) -> Tag:
|
||||
tag = db.execute(
|
||||
select(Tag).where(Tag.user_id.is_(None), Tag.name == name)
|
||||
).scalar_one_or_none()
|
||||
if tag is None:
|
||||
tag = Tag(user_id=None, name=name, category=category, color=color)
|
||||
db.add(tag)
|
||||
db.flush()
|
||||
return tag
|
||||
|
||||
|
||||
def apply_channel_autotags(db: Session, channel: Channel) -> None:
|
||||
desired: dict[int, float | None] = {}
|
||||
|
||||
lang, confidence = detect_channel_language(db, channel)
|
||||
if lang:
|
||||
tag = ensure_system_tag(db, LANG_NAMES.get(lang, lang.upper()), "language", LANG_COLOR)
|
||||
desired[tag.id] = confidence
|
||||
|
||||
for topic in compute_channel_topics(db, channel):
|
||||
tag = ensure_system_tag(
|
||||
db, topic, "topic", TOPIC_COLORS.get(topic, DEFAULT_TOPIC_COLOR)
|
||||
)
|
||||
desired.setdefault(tag.id, None)
|
||||
|
||||
existing = {
|
||||
ct.tag_id: ct
|
||||
for ct in db.execute(
|
||||
select(ChannelTag).where(
|
||||
ChannelTag.channel_id == channel.id,
|
||||
ChannelTag.user_id.is_(None),
|
||||
ChannelTag.source == "auto",
|
||||
)
|
||||
).scalars()
|
||||
}
|
||||
for tag_id, ct in existing.items():
|
||||
if tag_id not in desired:
|
||||
db.delete(ct)
|
||||
for tag_id, conf in desired.items():
|
||||
ct = existing.get(tag_id)
|
||||
if ct is None:
|
||||
db.add(
|
||||
ChannelTag(
|
||||
channel_id=channel.id,
|
||||
tag_id=tag_id,
|
||||
user_id=None,
|
||||
source="auto",
|
||||
confidence=conf,
|
||||
)
|
||||
)
|
||||
else:
|
||||
ct.confidence = conf
|
||||
db.commit()
|
||||
|
||||
|
||||
def run_autotag_all(db: Session, only_missing: bool = False) -> dict:
|
||||
query = select(Channel)
|
||||
if only_missing:
|
||||
query = query.where(
|
||||
~exists().where(
|
||||
and_(
|
||||
ChannelTag.channel_id == Channel.id,
|
||||
ChannelTag.source == "auto",
|
||||
ChannelTag.user_id.is_(None),
|
||||
)
|
||||
)
|
||||
)
|
||||
channels = db.execute(query).scalars().all()
|
||||
tagged = 0
|
||||
for channel in channels:
|
||||
try:
|
||||
apply_channel_autotags(db, channel)
|
||||
tagged += 1
|
||||
except Exception:
|
||||
db.rollback()
|
||||
removed = _cleanup_orphan_system_tags(db)
|
||||
return {"channels_tagged": tagged, "orphan_tags_removed": removed}
|
||||
|
||||
|
||||
def _cleanup_orphan_system_tags(db: Session) -> int:
|
||||
"""Delete system tags that no longer link to any channel (e.g. after a mapping change)."""
|
||||
orphans = (
|
||||
db.execute(
|
||||
select(Tag).where(
|
||||
Tag.user_id.is_(None),
|
||||
~exists().where(ChannelTag.tag_id == Tag.id),
|
||||
)
|
||||
)
|
||||
.scalars()
|
||||
.all()
|
||||
)
|
||||
for tag in orphans:
|
||||
db.delete(tag)
|
||||
db.commit()
|
||||
return len(orphans)
|
||||
Loading…
Add table
Add a link
Reference in a new issue