1827dd6c4e
Self-hosted personal YouTube management app. FastAPI + SQLite backend, React + Vite + Tailwind frontend. Dockerfiles and compose included for Portainer deployment. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
87 lines
2.9 KiB
Python
87 lines
2.9 KiB
Python
"""Surprise Me scoring logic."""
|
|
import random
|
|
from datetime import datetime, time
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import text
|
|
|
|
|
|
SURPRISE_SQL = """
|
|
WITH candidate_scores AS (
|
|
SELECT
|
|
v.id AS video_id,
|
|
v.youtube_video_id,
|
|
v.title,
|
|
v.thumbnail_url,
|
|
v.duration_seconds,
|
|
v.channel_id,
|
|
c.name AS channel_name,
|
|
c.thumbnail_url AS channel_thumbnail_url,
|
|
uv.watched,
|
|
uv.watch_progress_seconds,
|
|
uv.downloaded,
|
|
uv.last_watched_at,
|
|
-- Unplayed download bonus
|
|
CASE WHEN uv.downloaded = 1 AND (uv.watched IS NULL OR uv.watched = 0) THEN 40 ELSE 0 END
|
|
-- Recency penalty
|
|
+ CASE
|
|
WHEN uv.last_watched_at IS NOT NULL
|
|
AND uv.last_watched_at > datetime('now', '-7 days') THEN -50
|
|
WHEN uv.last_watched_at IS NOT NULL
|
|
AND uv.last_watched_at > datetime('now', '-30 days') THEN -20
|
|
ELSE 0
|
|
END
|
|
-- Late evening duration bonus (applied in Python)
|
|
+ :duration_bonus_active * CASE WHEN v.duration_seconds > 2700 THEN 10 ELSE 0 END
|
|
-- Random jitter
|
|
+ (ABS(RANDOM()) % 11 - 5) AS base_score
|
|
FROM videos v
|
|
JOIN user_videos uv ON v.id = uv.video_id AND uv.user_id = :user_id
|
|
JOIN channels c ON v.channel_id = c.id
|
|
WHERE uv.downloaded = 1
|
|
)
|
|
SELECT * FROM candidate_scores
|
|
ORDER BY base_score DESC
|
|
LIMIT 50
|
|
"""
|
|
|
|
|
|
def get_surprise_videos(db: Session, user_id: int, limit: int = 10) -> list[dict]:
|
|
now = datetime.now()
|
|
late_evening = now.time() >= time(21, 0)
|
|
|
|
rows = db.execute(
|
|
text(SURPRISE_SQL),
|
|
{"user_id": user_id, "duration_bonus_active": 1 if late_evening else 0},
|
|
).mappings().all()
|
|
|
|
# Apply channel diversity penalty in Python
|
|
seen_channels: dict[int, int] = {}
|
|
results = []
|
|
for row in rows:
|
|
row = dict(row)
|
|
channel_id = row["channel_id"]
|
|
penalty = seen_channels.get(channel_id, 0) * 30
|
|
row["final_score"] = row["base_score"] - penalty
|
|
seen_channels[channel_id] = seen_channels.get(channel_id, 0) + 1
|
|
results.append(row)
|
|
|
|
results.sort(key=lambda r: r["final_score"], reverse=True)
|
|
return results[:limit]
|
|
|
|
|
|
def get_discovery_injection(db: Session, user_id: int) -> dict | None:
|
|
"""Return one unseen discovery queue item to inject into Surprise Me."""
|
|
row = db.execute(
|
|
text("""
|
|
SELECT dq.id, c.id AS channel_id, c.name, c.thumbnail_url,
|
|
dq.source, dq.score
|
|
FROM discovery_queue dq
|
|
JOIN channels c ON dq.channel_id = c.id
|
|
WHERE dq.user_id = :user_id AND dq.seen = 0
|
|
ORDER BY dq.score DESC
|
|
LIMIT 1
|
|
"""),
|
|
{"user_id": user_id},
|
|
).mappings().first()
|
|
return dict(row) if row else None
|