d3713d64b9
source-stats liczyl flagi tag/obsada dla 2,5 mln par serwis-scena, czyli ok. 5 mln losowych odczytow indeksow; trwal 22-32 min i przekroczyl limit (GOON-V). Teraz flagi z probki 3000 scen na serwis (stala, po md5), liczba scen, swiezosc i miniatury dalej dokladnie. 1 min 51 s, gwiazdki bez zmian, procenty +-2 pkt. Autovacuum na domyslnych 20% nie ruszal duzych tabel od tygodni (scene_performers 11% stron widocznych, statystyki scenes z 25.08). Migracja 0045 daje 2% dla szesciu najwiekszych tabel. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
317 lines
13 KiB
Python
317 lines
13 KiB
Python
"""run_source_stats, policz ocenę 0-5★ per origin (tube źródło) do source_stats.
|
|
|
|
Trzy osie (user request):
|
|
1. freshness , częstotliwość odświeżania (wiek najnowszej sceny + wolumen 7d),
|
|
2. richness bogactwo metadanych (% scen z thumb/tag/perf/desc/studio/dur),
|
|
3. health czy realnie gra. Z TELEMETRII (playback_events 7d: success rate
|
|
+ ttff) gdy jest dość prób; inaczej PROXY z typu ekstraktora
|
|
(natywny mp4 > hoster/WebView), z adnotacją basis.
|
|
|
|
stars = ważona średnia osi, ale health==0 (offline) GATE'uje stars do 0, źródło
|
|
świeże i bogate, które nie gra, ma być na dnie (to cały sens, casus hqfap/4k69).
|
|
|
|
Liczone w workerze (scheduler), bo richness to agregat po ~2M live playback_sources.
|
|
Retencja telemetrii: kasuje playback_events starsze niż _EVENT_RETENTION_DAYS.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from sqlalchemy import text
|
|
|
|
from app.db import session_scope
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
_TELEMETRY_WINDOW_DAYS = 7
|
|
# Poniżej tego telemetria niemiarodajna → proxy. Próg podniesiony 10→25 (2026-06-29):
|
|
# freshporno (5★ fresh+rich, realnie gra, zweryfikowane 206/507MB) dostało health=0 →
|
|
# "OFFLINE" na podstawie 10 prób z jednego pechowego okna (CDN-node blip), user-report
|
|
# cb526949. 25 prób to sensowniejszy próg pewności zanim ogłosimy źródło offline; przy
|
|
# mniejszej próbce lecimy na proxy/heurystykę (nie zerujemy gwiazdek znanemu-dobremu).
|
|
_TELEMETRY_MIN_ATTEMPTS = 25
|
|
_EVENT_RETENTION_DAYS = 30
|
|
# Flagi tag/perf/desc/studio/dur liczone z probki scen na origin, nie ze wszystkich.
|
|
# Pelny przebieg to ~5 mln losowych odczytow indeksow scene_tags/scene_performers
|
|
# (2,5 mln par origin-scena) i 22-32 min; 2026-09-21 przekroczyl 30 min limitu
|
|
# (Sentry GOON-V). Probka po md5(scene_id) jest stala miedzy przebiegami, wiec
|
|
# gwiazdki nie skacza; blad procentu przy 3000 scen to ok. 1 pkt.
|
|
_RICHNESS_SAMPLE = 3000
|
|
# Richness: wagi składowych (suma=1.0). thumb to minimum higieny, canonical-bogactwo
|
|
# (desc/studio/tag/perf) waży więcej. dur średnio.
|
|
_RICHNESS_WEIGHTS = {
|
|
"thumb": 0.12,
|
|
"tag": 0.20,
|
|
"perf": 0.20,
|
|
"desc": 0.16,
|
|
"studio": 0.16,
|
|
"dur": 0.16,
|
|
}
|
|
|
|
|
|
def _health_proxy_for(sitetag: str) -> tuple[int, str]:
|
|
"""Proxy health gdy brak telemetrii: z mechanizmu resolve. Nie może dać 5★,
|
|
bez realnego sygnału nie wiemy że gra, max 4. Zwraca (score, mechanizm)."""
|
|
try:
|
|
from app.extractors import _REGISTRY # type: ignore
|
|
|
|
fn = _REGISTRY.get(sitetag)
|
|
except Exception:
|
|
fn = None
|
|
if fn is None:
|
|
return 2, "unknown" # nie ma ekstraktora → niepewna grywalność
|
|
mod = getattr(fn, "__module__", "") or ""
|
|
if mod.endswith("_vps_blocked_fallback"):
|
|
return 3, "webview" # gra w WebView (residential IP), ale wolniej/ciężej
|
|
if mod.endswith("_embed_iframe"):
|
|
return 3, "hoster" # phone-side resolve hostera (dood/luluvid), średnio
|
|
if mod.endswith("_ytdlp"):
|
|
return 4, "ytdlp" # server-side, zwykle szybki/pewny
|
|
return 4, "native" # dedykowany natywny ekstraktor (direct mp4/HLS)
|
|
|
|
|
|
def _freshness_score(newest_at: datetime | None, new_7d: int, scenes: int) -> int:
|
|
if not scenes or newest_at is None:
|
|
return 0
|
|
now = datetime.now(timezone.utc)
|
|
age_days = (now - newest_at).total_seconds() / 86400.0
|
|
if age_days <= 2:
|
|
base = 5
|
|
elif age_days <= 4:
|
|
base = 4
|
|
elif age_days <= 10:
|
|
base = 3
|
|
elif age_days <= 30:
|
|
base = 2
|
|
else:
|
|
base = 1
|
|
# Zamrożone źródło (świeżość tylko z dawnego importu), brak nowych w 7d ścina do 1.
|
|
if new_7d == 0 and age_days > 4:
|
|
base = 1
|
|
return base
|
|
|
|
|
|
def _richness_score(pcts: dict[str, float]) -> int:
|
|
weighted = sum(pcts.get(k, 0.0) * w for k, w in _RICHNESS_WEIGHTS.items())
|
|
stars = round(weighted / 20.0) # 0..100% → 0..5
|
|
return max(1, min(5, stars))
|
|
|
|
|
|
def _health_score_from_telemetry(attempts: int, successes: int, p50_ttff_ms: int | None) -> int:
|
|
if attempts <= 0:
|
|
return 0
|
|
rate = 100.0 * successes / attempts
|
|
if rate >= 90:
|
|
s = 5
|
|
elif rate >= 75:
|
|
s = 4
|
|
elif rate >= 50:
|
|
s = 3
|
|
elif rate >= 25:
|
|
s = 2
|
|
elif rate > 0:
|
|
s = 1
|
|
else:
|
|
return 0 # ~0% sukcesów = offline
|
|
# Składowa "szybkość": wolny start (>8s mediana) ścina o 1 (nie poniżej 1).
|
|
if p50_ttff_ms is not None and p50_ttff_ms > 8000 and s > 1:
|
|
s -= 1
|
|
return s
|
|
|
|
|
|
def _overall_stars(freshness: int, richness: int, health: int) -> int:
|
|
if health == 0:
|
|
# Offline gate: health==0 (~0% udanych odtworzeń w telemetrii) zwykle = offline.
|
|
# ALE telemetria bywa NIEPEŁNA, część udanych odtworzeń nie wysyła eventu (zweryfikowane
|
|
# 2026-07-20: freshporno gra natywnie na realnym telefonie, PlaybackState=PLAYING, ale
|
|
# sukces się nie raportuje → sam błąd → fałszywe 0% → fałszywe offline). Więc gdy źródło
|
|
# jest EWIDENTNIE żywe (świeże ORAZ bogate ORAZ aktywnie ingestowane), nie chowamy go
|
|
# całkiem, tylko floor 2★ "działa, ale telemetria kuleje". Martwe źródła i tak mają
|
|
# niski freshness → gate zadziała. Patrz ticket: luka w emisji success eventu (native fallback).
|
|
if freshness >= 4 and richness >= 4:
|
|
return 2
|
|
return 0
|
|
raw = 0.40 * freshness + 0.30 * richness + 0.30 * health
|
|
return max(1, min(5, round(raw)))
|
|
|
|
|
|
def run_source_stats() -> dict:
|
|
"""Przelicz source_stats dla wszystkich live tube origins. Zwraca podsumowanie."""
|
|
now = datetime.now(timezone.utc)
|
|
win_start = now - timedelta(days=_TELEMETRY_WINDOW_DAYS)
|
|
|
|
with session_scope() as session:
|
|
# 1a) Freshness + miniatury: dokladnie, po DISTINCT (origin, scene) z live sources.
|
|
rows = session.execute(
|
|
text(
|
|
"""
|
|
WITH live AS (
|
|
SELECT DISTINCT ON (origin, scene_id) origin, scene_id,
|
|
(thumbnail_url IS NOT NULL) AS has_thumb, created_at
|
|
FROM playback_sources
|
|
WHERE dead_at IS NULL AND origin LIKE 'tube:%'
|
|
)
|
|
SELECT origin,
|
|
count(*) AS scenes,
|
|
count(*) FILTER (WHERE created_at > :win) AS new_7d,
|
|
max(created_at) AS newest_at,
|
|
100.0*avg(has_thumb::int) AS thumb
|
|
FROM live
|
|
GROUP BY origin
|
|
"""
|
|
),
|
|
{"win": win_start},
|
|
).all()
|
|
|
|
# 1b) Reszta richness z probki scen (patrz _RICHNESS_SAMPLE).
|
|
flags = {
|
|
r.origin: r
|
|
for r in session.execute(
|
|
text(
|
|
"""
|
|
WITH live AS (
|
|
SELECT DISTINCT origin, scene_id
|
|
FROM playback_sources
|
|
WHERE dead_at IS NULL AND origin LIKE 'tube:%'
|
|
),
|
|
sample AS (
|
|
SELECT origin, scene_id FROM (
|
|
SELECT origin, scene_id, row_number() OVER (
|
|
PARTITION BY origin ORDER BY md5(scene_id::text)
|
|
) AS rn
|
|
FROM live
|
|
) ranked
|
|
WHERE rn <= :n
|
|
)
|
|
SELECT sm.origin,
|
|
100.0*avg((EXISTS(SELECT 1 FROM scene_tags st
|
|
WHERE st.scene_id=s.id))::int) AS tag,
|
|
100.0*avg((EXISTS(SELECT 1 FROM scene_performers sp
|
|
WHERE sp.scene_id=s.id))::int) AS perf,
|
|
100.0*avg((s.description IS NOT NULL
|
|
AND length(btrim(s.description))>0)::int) AS descr,
|
|
100.0*avg((s.studio_id IS NOT NULL)::int) AS studio,
|
|
100.0*avg((s.duration_sec IS NOT NULL)::int) AS dur
|
|
FROM sample sm JOIN scenes s ON s.id = sm.scene_id
|
|
GROUP BY sm.origin
|
|
"""
|
|
),
|
|
{"n": _RICHNESS_SAMPLE},
|
|
).all()
|
|
}
|
|
|
|
# 2) Telemetria health per origin (okno 7d).
|
|
tele = {
|
|
r.origin: r
|
|
for r in session.execute(
|
|
text(
|
|
"""
|
|
SELECT origin,
|
|
count(*) AS attempts,
|
|
count(*) FILTER (WHERE status='success') AS successes,
|
|
percentile_disc(0.5) WITHIN GROUP (ORDER BY ttff_ms)
|
|
FILTER (WHERE status='success' AND ttff_ms IS NOT NULL) AS p50_ttff
|
|
FROM playback_events
|
|
WHERE created_at > :win
|
|
GROUP BY origin
|
|
"""
|
|
),
|
|
{"win": win_start},
|
|
).all()
|
|
}
|
|
|
|
computed = 0
|
|
for r in rows:
|
|
origin = r.origin
|
|
sitetag = origin.split(":", 1)[1] if ":" in origin else origin
|
|
f = flags.get(origin)
|
|
pcts = {
|
|
"thumb": float(r.thumb or 0),
|
|
"tag": float(f.tag or 0) if f else 0.0,
|
|
"perf": float(f.perf or 0) if f else 0.0,
|
|
"desc": float(f.descr or 0) if f else 0.0,
|
|
"studio": float(f.studio or 0) if f else 0.0,
|
|
"dur": float(f.dur or 0) if f else 0.0,
|
|
}
|
|
freshness = _freshness_score(r.newest_at, int(r.new_7d), int(r.scenes))
|
|
richness = _richness_score(pcts)
|
|
|
|
t = tele.get(origin)
|
|
if t and int(t.attempts) >= _TELEMETRY_MIN_ATTEMPTS:
|
|
health = _health_score_from_telemetry(
|
|
int(t.attempts), int(t.successes),
|
|
int(t.p50_ttff) if t.p50_ttff is not None else None,
|
|
)
|
|
health_basis = "telemetry"
|
|
pb_attempts = int(t.attempts)
|
|
pb_success_rate = round(100.0 * int(t.successes) / int(t.attempts))
|
|
p50_ttff = int(t.p50_ttff) if t.p50_ttff is not None else None
|
|
mechanism = None
|
|
else:
|
|
health, mechanism = _health_proxy_for(sitetag)
|
|
health_basis = "proxy"
|
|
pb_attempts = int(t.attempts) if t else 0
|
|
pb_success_rate = (
|
|
round(100.0 * int(t.successes) / int(t.attempts)) if t and int(t.attempts) else None
|
|
)
|
|
p50_ttff = None
|
|
|
|
stars = _overall_stars(freshness, richness, health)
|
|
components = {
|
|
"pct": {k: round(v) for k, v in pcts.items()},
|
|
"health_basis": health_basis,
|
|
"mechanism": mechanism,
|
|
"pb_attempts_7d": pb_attempts,
|
|
"pb_success_rate": pb_success_rate,
|
|
"p50_ttff_ms": p50_ttff,
|
|
}
|
|
|
|
session.execute(
|
|
text(
|
|
"""
|
|
INSERT INTO source_stats
|
|
(origin, stars, freshness, richness, health, scenes, new_7d,
|
|
newest_at, components, computed_at)
|
|
VALUES
|
|
(:origin, :stars, :freshness, :richness, :health, :scenes, :new_7d,
|
|
:newest_at, CAST(:components AS jsonb), now())
|
|
ON CONFLICT (origin) DO UPDATE SET
|
|
stars=EXCLUDED.stars, freshness=EXCLUDED.freshness,
|
|
richness=EXCLUDED.richness, health=EXCLUDED.health,
|
|
scenes=EXCLUDED.scenes, new_7d=EXCLUDED.new_7d,
|
|
newest_at=EXCLUDED.newest_at, components=EXCLUDED.components,
|
|
computed_at=now()
|
|
"""
|
|
),
|
|
{
|
|
"origin": origin,
|
|
"stars": stars,
|
|
"freshness": freshness,
|
|
"richness": richness,
|
|
"health": health,
|
|
"scenes": int(r.scenes),
|
|
"new_7d": int(r.new_7d),
|
|
"newest_at": r.newest_at,
|
|
"components": json.dumps(components),
|
|
},
|
|
)
|
|
computed += 1
|
|
|
|
# 3) Sprzątanie: usuń source_stats dla origins bez już-live sources (np. hqfap/4k69
|
|
# po ukryciu) + retencja telemetrii.
|
|
live_origins = {r.origin for r in rows}
|
|
stale = session.execute(text("SELECT origin FROM source_stats")).scalars().all()
|
|
for o in stale:
|
|
if o not in live_origins:
|
|
session.execute(
|
|
text("DELETE FROM source_stats WHERE origin=:o"), {"o": o}
|
|
)
|
|
session.execute(
|
|
text("DELETE FROM playback_events WHERE created_at < :cut"),
|
|
{"cut": now - timedelta(days=_EVENT_RETENTION_DAYS)},
|
|
)
|
|
|
|
log.info("[source-stats] computed %d origins", computed)
|
|
return {"computed": computed}
|