feat(ingest): backfill brakujacej dlugosci scen
1 755 421 scen nie ma dlugosci ani na scenie, ani na zrodle, wiec wypadaja z feedu: w SQL "duration_sec >= 60" nie jest prawda dla NULL, przez co scena o nieznanej dlugosci znika tak samo jak dziesieciosekundowy smiec. Filtr zostaje, bo odcina smieci, uzupelniamy dane. Swiadomie waskie originy. Uniwersalny regex trafia w smieci: porntrex ma w HTML "duration: 5" reklamy, mypornerleak dlugosc ciastka 720, a xhamster i porntrex wypisuja dlugosci filmow z paska obok (naiwne dopasowanie dalo 198s zamiast 240s). Zla dlugosc jest gorsza niz brak, bo dedup opiera sie na niej najmocniej. - xvideos, xnxx: VideoObject LD-JSON, 5/5 na probce - xhamster: videoEntity z kontrola zgodnosci id z URL-a - porntrex pominiety: strona szczegolow nie podaje wlasnej dlugosci Kolejka jak przy phashu (plan raz, mielenie paczkami, przezywa restart): 845 899 scen, 500 stron co 5 min, czyli ok. 6 dni. Plan powtarza sie co 24h, zeby swieze sceny bez dlugosci tez trafialy do kolejki. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
goon-foss
parent
cddf60acc2
commit
544de9c5ba
8 changed files with 409 additions and 0 deletions
55
alembic/versions/20260820_0031_duration_backfill_queue.py
Normal file
55
alembic/versions/20260820_0031_duration_backfill_queue.py
Normal file
|
|
@ -0,0 +1,55 @@
|
|||
"""duration_backfill_queue: kolejka scen do doliczenia dlugosci ze strony tuba
|
||||
|
||||
Uzasadnienie i liczby: `app/models/duration_backfill_queue.py`.
|
||||
|
||||
Revision ID: 0031_duration_backfill
|
||||
Revises: 0030_scenes_quality_idx
|
||||
Create Date: 2026-08-20
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import sqlalchemy as sa
|
||||
from alembic import op
|
||||
from sqlalchemy.dialects import postgresql
|
||||
|
||||
# Uwaga: `alembic_version.version_num` to varchar(32), dluzsze id nie wejdzie.
|
||||
revision: str = "0031_duration_backfill"
|
||||
down_revision: str | None = "0030_scenes_quality_idx"
|
||||
branch_labels = None
|
||||
depends_on = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
op.create_table(
|
||||
"duration_backfill_queue",
|
||||
sa.Column("scene_id", postgresql.UUID(as_uuid=True), nullable=False),
|
||||
sa.Column("page_url", sa.String(length=1024), nullable=False),
|
||||
sa.Column("origin", sa.String(length=64), nullable=False),
|
||||
sa.Column("attempts", sa.Integer(), nullable=False, server_default="0"),
|
||||
sa.Column("done_at", sa.DateTime(timezone=True), nullable=True),
|
||||
sa.Column(
|
||||
"created_at",
|
||||
sa.DateTime(timezone=True),
|
||||
server_default=sa.func.now(),
|
||||
nullable=False,
|
||||
),
|
||||
sa.ForeignKeyConstraint(
|
||||
["scene_id"], ["scenes.id"], ondelete="CASCADE",
|
||||
name="fk_duration_backfill_queue_scene_id_scenes",
|
||||
),
|
||||
sa.PrimaryKeyConstraint("scene_id", name="pk_duration_backfill_queue"),
|
||||
)
|
||||
# Paczka bierze wiersze niezrobione, po originie (zeby nie zawalic jednego hosta).
|
||||
op.create_index(
|
||||
"ix_duration_backfill_queue_pending",
|
||||
"duration_backfill_queue",
|
||||
["origin", "attempts"],
|
||||
postgresql_where=sa.text("done_at IS NULL"),
|
||||
)
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
op.drop_index(
|
||||
"ix_duration_backfill_queue_pending", table_name="duration_backfill_queue"
|
||||
)
|
||||
op.drop_table("duration_backfill_queue")
|
||||
|
|
@ -138,6 +138,16 @@ class Settings(BaseSettings):
|
|||
sched_phash_backfill_batch: int = Field(
|
||||
default=800, validation_alias="GOON_SCHED_PHASH_BACKFILL_BATCH"
|
||||
)
|
||||
# Doliczanie brakujacej dlugosci scen (patrz app/models/duration_backfill_queue.py).
|
||||
# 500 stron co 5 min to ~144 tys. dziennie, wiec kolejka 846 tys. schodzi w ok. 6 dni.
|
||||
# To ~1,7 zapytania na sekunde rozlozone na trzy duze tuby, czyli ruch, ktorego one
|
||||
# nie zauwaza. Job ma 30-minutowy limit, a paczka schodzi w ok. 3 min, wiec z zapasem.
|
||||
sched_duration_backfill_minutes: int = Field(
|
||||
default=5, validation_alias="GOON_SCHED_DURATION_BACKFILL_MINUTES"
|
||||
)
|
||||
sched_duration_backfill_batch: int = Field(
|
||||
default=500, validation_alias="GOON_SCHED_DURATION_BACKFILL_BATCH"
|
||||
)
|
||||
# Próg ciszy dla konektorów FILMOWYCH. Wcześniej filmy nie były pokryte watchdogiem
|
||||
# w ogóle, streamporn.vip stał 23 dni i nic nie krzyknęło. Filmy ingestujemy
|
||||
# codziennie, więc 72h to bezpieczny margines na jeden nieudany cykl.
|
||||
|
|
|
|||
|
|
@ -89,3 +89,68 @@ def extract_duration_sec(html: str) -> int | None:
|
|||
return total
|
||||
|
||||
return None
|
||||
|
||||
|
||||
# --- Wzorce punktowe, per origin --------------------------------------------------
|
||||
#
|
||||
# `extract_duration_sec` jest generyczny i tym samym łatwowierny. Na stronach, które
|
||||
# wypisują długości filmów z paska "powiązane", trafia w cudzą wartość, a na porntreksie
|
||||
# w `duration: '5'` reklamy. Backfill katalogu (`app/scheduler/duration_backfill.py`)
|
||||
# używa więc wyłącznie wzorców z potwierdzoną, punktową kotwicą.
|
||||
|
||||
# xhamster trzyma prawdziwą długość w `videoEntity`, obok id zgodnego z URL-em. Sprawdzone
|
||||
# 2026-08-20: dla `.../macy-gray-i-try-pmv-10506445` videoEntity daje 240 s, a naiwne
|
||||
# szukanie najbliższego `"duration"` dawało 198 s z filmu obok.
|
||||
_XH_ENTITY_RE = re.compile(
|
||||
r'"videoEntity"\s*:\s*\{(?P<body>.{0,600}?)\}', re.DOTALL
|
||||
)
|
||||
_XH_ID_RE = re.compile(r'"id"\s*:\s*(\d+)')
|
||||
_XH_DUR_RE = re.compile(r'"duration"\s*:\s*(\d+)')
|
||||
# id filmu jest na końcu sluga: `/videos/<slug>-<id>`.
|
||||
_XH_URL_ID_RE = re.compile(r"-(\d{4,})(?:[/?#]|$)")
|
||||
|
||||
# Górna granica zdrowego rozsądku: 6 h. Wyżej to prawie na pewno zła wartość
|
||||
# (sklejony live albo pomyłka jednostek), a długość idzie potem do dedupu.
|
||||
_MAX_SANE_DURATION_SEC = 6 * 3600
|
||||
|
||||
|
||||
def _extract_xhamster(html: str, page_url: str) -> int | None:
|
||||
want = _XH_URL_ID_RE.search(page_url)
|
||||
for m in _XH_ENTITY_RE.finditer(html):
|
||||
body = m.group("body")
|
||||
vid = _XH_ID_RE.search(body)
|
||||
dur = _XH_DUR_RE.search(body)
|
||||
if not dur:
|
||||
continue
|
||||
# Bez zgodności id wolimy nie zapisać nic, niż zapisać długość obcego filmu.
|
||||
if want and vid and vid.group(1) != want.group(1):
|
||||
continue
|
||||
n = int(dur.group(1))
|
||||
if 0 < n <= _MAX_SANE_DURATION_SEC:
|
||||
return n
|
||||
return None
|
||||
|
||||
|
||||
def _extract_ldjson_only(html: str, page_url: str) -> int | None:
|
||||
"""xvideos i xnxx: `VideoObject` z ISO 8601. Sprawdzone 5/5 na losowej próbce."""
|
||||
if (m := _LD_DURATION_RE.search(html)) and (
|
||||
parsed := _parse_iso8601(m.group(1))
|
||||
) is not None:
|
||||
if 0 < parsed <= _MAX_SANE_DURATION_SEC:
|
||||
return parsed
|
||||
return None
|
||||
|
||||
|
||||
#: Origin → wzorzec. Dopisuj TYLKO to, co sprawdziłeś na próbce prawdziwych stron;
|
||||
#: porntrex i mypornerleak celowo tu nie ma, patrz `app/models/duration_backfill_queue.py`.
|
||||
ORIGIN_DURATION_RESOLVERS = {
|
||||
"tube:xhamstercom": _extract_xhamster,
|
||||
"tube:xvideoscom": _extract_ldjson_only,
|
||||
"tube:xnxxcom": _extract_ldjson_only,
|
||||
}
|
||||
|
||||
|
||||
def extract_duration_for_origin(origin: str, html: str, page_url: str) -> int | None:
|
||||
"""Długość dla originu z potwierdzonym wzorcem; None gdy origin nieobsługiwany."""
|
||||
fn = ORIGIN_DURATION_RESOLVERS.get(origin)
|
||||
return fn(html, page_url) if fn else None
|
||||
|
|
|
|||
|
|
@ -20,6 +20,7 @@ from app.models.performer import Performer, PerformerAlias, PerformerExternalRef
|
|||
from app.models.play_progress import MoviePlayProgress, ScenePlayProgress
|
||||
from app.models.playback_event import PlaybackEvent
|
||||
from app.models.playback_source import PlaybackSource
|
||||
from app.models.duration_backfill_queue import DurationBackfillQueue
|
||||
from app.models.phash_backfill_queue import PhashBackfillQueue
|
||||
from app.models.phash_blacklist import PhashBlacklist
|
||||
from app.models.saved_search import SavedSearch
|
||||
|
|
|
|||
67
app/models/duration_backfill_queue.py
Normal file
67
app/models/duration_backfill_queue.py
Normal file
|
|
@ -0,0 +1,67 @@
|
|||
"""Kolejka scen, którym trzeba doliczyć długość ze strony tuba.
|
||||
|
||||
Filtr `min_duration_sec=60` odcina śmieciowe klipy i tak ma zostać, ale w SQL-u
|
||||
`duration_sec >= 60` NIE jest prawdą dla NULL, więc scena o NIEZNANEJ długości znika
|
||||
z feedu tak samo jak dziesięciosekundowy śmieć. Audyt 2026-08-20: **1 755 421 scen**
|
||||
nie ma długości ani na scenie, ani na źródle, czyli scraper nigdy jej nie zebrał.
|
||||
Największe zaległości to porntrex (676 tys.), xhamster (361 tys.), xvideos (259 tys.)
|
||||
i xnxx (226 tys.).
|
||||
|
||||
**Dlaczego nie jeden uniwersalny regex po stronie szczegółów.** Bo trafia w śmieci.
|
||||
Na próbce: porntrex ma w HTML `duration: '5'` (długość reklamy), mypornerleak
|
||||
`disclaimer_cookie_duration":"720"` (ciastko), a xhamster i porntrex wypisują długości
|
||||
WSZYSTKICH filmów powiązanych z paska obok. Naiwne dopasowanie przy jednym xhamsterze
|
||||
dało 198 s zamiast prawdziwych 240 s. Zła długość jest gorsza niż jej brak, bo dedup
|
||||
traktuje długość jako najmocniejszy pojedynczy sygnał i skleiłby obce sceny.
|
||||
|
||||
Dlatego backfill obsługuje tylko originy z POTWIERDZONYM, punktowym wzorcem
|
||||
(`app/extractors/duration_extract.py`), a nie wszystko, co się nawinie:
|
||||
|
||||
* xvideos, xnxx: `VideoObject` LD-JSON z ISO 8601, sprawdzone 5/5
|
||||
* xhamster: `"videoEntity":{"id":<id z URL>,…,"duration":<sekundy>}`, z kontrolą
|
||||
zgodności id, bo bez niej łapie film z paska obok
|
||||
|
||||
porntrex świadomie NIE jest tu obsługiwany: jego strona szczegółów w ogóle nie podaje
|
||||
własnej długości (`video_duration` w flashvars nie występuje), a jedyne `durations`
|
||||
w HTML należą do filmów powiązanych. Dla niego drogą jest listing, bo kafelek długość
|
||||
pokazuje i `porntrex_browse.py` już ją stamtąd czyta.
|
||||
|
||||
Kolejka jest tabelą z tego samego powodu co przy phashu: wyznaczenie kandydatów to
|
||||
ciężki przegląd całego katalogu, a mielenie ma przeżyć restart workera i pokazywać
|
||||
postęp. Patrz [[phash_backfill_queue]] dla bliźniaczego wzorca.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import uuid
|
||||
from datetime import datetime
|
||||
|
||||
from sqlalchemy import DateTime, ForeignKey, Integer, String, func
|
||||
from sqlalchemy.dialects.postgresql import UUID as PGUUID
|
||||
from sqlalchemy.orm import Mapped, mapped_column
|
||||
|
||||
from app.models.base import Base
|
||||
|
||||
|
||||
class DurationBackfillQueue(Base):
|
||||
__tablename__ = "duration_backfill_queue"
|
||||
|
||||
scene_id: Mapped[uuid.UUID] = mapped_column(
|
||||
PGUUID(as_uuid=True),
|
||||
ForeignKey("scenes.id", ondelete="CASCADE"),
|
||||
primary_key=True,
|
||||
)
|
||||
#: Strona tuba, z której czytamy długość. Zapisana przy planowaniu, żeby mielenie
|
||||
#: nie musiało co paczkę wracać do `playback_sources`.
|
||||
page_url: Mapped[str] = mapped_column(String(1024), nullable=False)
|
||||
#: Decyduje, KTÓRY punktowy wzorzec zostanie użyty. Bez tego wracamy do zgadywania.
|
||||
origin: Mapped[str] = mapped_column(String(64), nullable=False)
|
||||
#: Rośnie przy każdej nieudanej próbie; po wyczerpaniu limitu wiersz jest odkładany,
|
||||
#: żeby skasowana strona nie blokowała kolejki w nieskończoność.
|
||||
attempts: Mapped[int] = mapped_column(Integer, nullable=False, server_default="0")
|
||||
#: Ustawione po sukcesie ALBO po wyczerpaniu prób (wtedy długość została pusta).
|
||||
done_at: Mapped[datetime | None] = mapped_column(
|
||||
DateTime(timezone=True), nullable=True
|
||||
)
|
||||
created_at: Mapped[datetime] = mapped_column(
|
||||
DateTime(timezone=True), server_default=func.now(), nullable=False
|
||||
)
|
||||
148
app/scheduler/duration_backfill.py
Normal file
148
app/scheduler/duration_backfill.py
Normal file
|
|
@ -0,0 +1,148 @@
|
|||
"""Dolicza brakującą długość scenom ze stron tubów, paczkami.
|
||||
|
||||
Po co i dlaczego tylko wybrane originy: `app/models/duration_backfill_queue.py`.
|
||||
W skrócie: bez długości scena wypada z feedu, bo `duration_sec >= 60` nie jest prawdą
|
||||
dla NULL, a zła długość jest gorsza niż brak, bo dedup opiera się na niej najmocniej.
|
||||
Stąd wyłącznie wzorce punktowe z `ORIGIN_DURATION_RESOLVERS`.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
|
||||
from sqlalchemy import text
|
||||
|
||||
from app.db import session_scope
|
||||
from app.extractors.duration_extract import (
|
||||
ORIGIN_DURATION_RESOLVERS,
|
||||
extract_duration_for_origin,
|
||||
)
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
# Po tylu nieudanych próbach odkładamy wiersz. Strony bywają skasowane i nie ma sensu
|
||||
# wracać do nich w kółko, blokując kolejkę.
|
||||
_MAX_ATTEMPTS = 3
|
||||
_DEFAULT_BATCH = 500
|
||||
# Sześć wątków na paczkę, a paczki są rozłożone w czasie przez scheduler. Pobrania
|
||||
# rozkładają się na trzy różne serwisy, więc żaden nie dostaje serii pod rząd.
|
||||
_CONCURRENCY = 6
|
||||
_FETCH_TIMEOUT = 20.0
|
||||
|
||||
|
||||
def plan_duration_backfill(limit: int | None = None) -> int:
|
||||
"""Wrzuca do kolejki sceny bez długości, mające źródło na obsługiwanym originie.
|
||||
|
||||
Idempotentne: `ON CONFLICT DO NOTHING`, więc można wołać wielokrotnie, także po
|
||||
dopisaniu nowego originu do `ORIGIN_DURATION_RESOLVERS`.
|
||||
"""
|
||||
origins = sorted(ORIGIN_DURATION_RESOLVERS)
|
||||
if not origins:
|
||||
return 0
|
||||
sql = """
|
||||
INSERT INTO duration_backfill_queue (scene_id, page_url, origin)
|
||||
SELECT DISTINCT ON (s.id) s.id, ps.page_url, ps.origin
|
||||
FROM scenes s
|
||||
JOIN playback_sources ps ON ps.scene_id = s.id
|
||||
WHERE s.duration_sec IS NULL
|
||||
AND ps.origin = ANY(:origins)
|
||||
AND ps.page_url IS NOT NULL
|
||||
AND ps.dead_at IS NULL
|
||||
ORDER BY s.id, ps.origin
|
||||
ON CONFLICT (scene_id) DO NOTHING
|
||||
"""
|
||||
if limit is not None:
|
||||
# LIMIT musi objąć SELECT, nie INSERT, stąd podzapytanie przy ograniczonym planie.
|
||||
sql = f"""
|
||||
INSERT INTO duration_backfill_queue (scene_id, page_url, origin)
|
||||
SELECT * FROM (
|
||||
SELECT DISTINCT ON (s.id) s.id, ps.page_url, ps.origin
|
||||
FROM scenes s
|
||||
JOIN playback_sources ps ON ps.scene_id = s.id
|
||||
WHERE s.duration_sec IS NULL
|
||||
AND ps.origin = ANY(:origins)
|
||||
AND ps.page_url IS NOT NULL
|
||||
AND ps.dead_at IS NULL
|
||||
ORDER BY s.id, ps.origin
|
||||
) q LIMIT {int(limit)}
|
||||
ON CONFLICT (scene_id) DO NOTHING
|
||||
"""
|
||||
with session_scope() as s:
|
||||
n = s.execute(text(sql), {"origins": origins}).rowcount
|
||||
log.info("duration-backfill: zaplanowano %d scen (originy: %s)", n, ",".join(origins))
|
||||
return n
|
||||
|
||||
|
||||
def _fetch_one(row: tuple) -> tuple:
|
||||
"""(scene_id, duration|None). Sieć i parsowanie, bez dotykania bazy."""
|
||||
scene_id, page_url, origin = row
|
||||
from app.extractors._fetch import browser_get
|
||||
|
||||
try:
|
||||
r = browser_get(page_url, timeout=_FETCH_TIMEOUT, follow_redirects=True)
|
||||
if r.status_code >= 400:
|
||||
return (scene_id, None)
|
||||
return (scene_id, extract_duration_for_origin(origin, r.text, page_url))
|
||||
except Exception as e: # pojedyncza strona nie może wywrócić paczki
|
||||
log.debug("duration-backfill: %s padlo (%s)", page_url, e)
|
||||
return (scene_id, None)
|
||||
|
||||
|
||||
def run_duration_backfill(*, batch: int = _DEFAULT_BATCH) -> dict[str, int]:
|
||||
"""Bierze paczkę z kolejki, dolicza długość, zapisuje. Pusta kolejka = no-op."""
|
||||
with session_scope() as s:
|
||||
rows = s.execute(
|
||||
text(
|
||||
"""
|
||||
SELECT scene_id, page_url, origin
|
||||
FROM duration_backfill_queue
|
||||
WHERE done_at IS NULL AND attempts < :max_attempts
|
||||
ORDER BY attempts, created_at
|
||||
LIMIT :batch
|
||||
"""
|
||||
),
|
||||
{"max_attempts": _MAX_ATTEMPTS, "batch": batch},
|
||||
).fetchall()
|
||||
|
||||
if not rows:
|
||||
return {"taken": 0, "filled": 0, "missed": 0}
|
||||
|
||||
with ThreadPoolExecutor(max_workers=_CONCURRENCY) as pool:
|
||||
results = list(pool.map(_fetch_one, [tuple(r) for r in rows]))
|
||||
|
||||
filled = [(sid, dur) for sid, dur in results if dur]
|
||||
missed = [sid for sid, dur in results if not dur]
|
||||
|
||||
with session_scope() as s:
|
||||
for scene_id, dur in filled:
|
||||
# Warunek na NULL, bo między planem a zapisem scena mogła zostać scalona
|
||||
# albo wzbogacona z kanonu, a ten jest wiarygodniejszy niż tube.
|
||||
s.execute(
|
||||
text(
|
||||
"UPDATE scenes SET duration_sec = :d "
|
||||
" WHERE id = :id AND duration_sec IS NULL"
|
||||
),
|
||||
{"d": dur, "id": scene_id},
|
||||
)
|
||||
s.execute(
|
||||
text(
|
||||
"UPDATE duration_backfill_queue SET done_at = now() WHERE scene_id = :id"
|
||||
),
|
||||
{"id": scene_id},
|
||||
)
|
||||
if missed:
|
||||
# Wyczerpane próby zamykamy, żeby kolejka nie kręciła się w kółko na martwych.
|
||||
s.execute(
|
||||
text(
|
||||
"""
|
||||
UPDATE duration_backfill_queue
|
||||
SET attempts = attempts + 1,
|
||||
done_at = CASE WHEN attempts + 1 >= :max_attempts
|
||||
THEN now() ELSE NULL END
|
||||
WHERE scene_id = ANY(:ids)
|
||||
"""
|
||||
),
|
||||
{"ids": missed, "max_attempts": _MAX_ATTEMPTS},
|
||||
)
|
||||
|
||||
return {"taken": len(rows), "filled": len(filled), "missed": len(missed)}
|
||||
|
|
@ -292,6 +292,38 @@ def _job_phash_backfill(batch: int) -> None:
|
|||
log.exception("[scheduler] phash backfill failed")
|
||||
|
||||
|
||||
def _job_duration_plan() -> None:
|
||||
"""Dopisuje do kolejki sceny, ktore od wczoraj przyszly bez dlugosci. Scrapery
|
||||
czesci tubow jej nie zbieraja przy ingescie, wiec bez tego zaleglosc odrastalaby
|
||||
z kazda doba. `ON CONFLICT DO NOTHING`, wiec powtorzenie nic nie psuje."""
|
||||
try:
|
||||
from app.scheduler.duration_backfill import plan_duration_backfill
|
||||
|
||||
added = _run_with_timeout(
|
||||
lambda: plan_duration_backfill(), label="duration-plan"
|
||||
)
|
||||
log.info("[scheduler] duration plan done (%s)", added)
|
||||
except Exception:
|
||||
log.exception("[scheduler] duration plan failed")
|
||||
|
||||
|
||||
def _job_duration_backfill(batch: int) -> None:
|
||||
"""Dolicza dlugosc scenom, ktore jej nie maja. Bez dlugosci scena wypada z feedu,
|
||||
bo `duration_sec >= 60` nie jest prawda dla NULL. Kolejke planujemy raz
|
||||
(`plan_duration_backfill`), tu tylko mielimy paczkami, bo to pobrania cudzych stron.
|
||||
Pusta kolejka = no-op, wiec nic nie kosztuje."""
|
||||
try:
|
||||
from app.scheduler.duration_backfill import run_duration_backfill
|
||||
|
||||
out = _run_with_timeout(
|
||||
lambda: run_duration_backfill(batch=batch), label="duration-backfill"
|
||||
)
|
||||
if out and out.get("taken"):
|
||||
log.info("[scheduler] duration backfill %s", out)
|
||||
except Exception:
|
||||
log.exception("[scheduler] duration backfill failed")
|
||||
|
||||
|
||||
def _job_phash_blacklist() -> None:
|
||||
"""Wyłapuje phashe, które przestały cokolwiek znaczyć (zaślepka / czarna klatka /
|
||||
intro studia) i kasuje je z fingerprintów. Audyt 2026-07-27: 73 takie wartości przy
|
||||
|
|
@ -547,6 +579,35 @@ def build_scheduler(cfg: dict[str, Any]) -> BlockingScheduler:
|
|||
)
|
||||
log.info("scheduler: phash blacklist every %dh", cfg["phash_blacklist_hours"])
|
||||
|
||||
if cfg.get("duration_backfill_minutes"):
|
||||
sched.add_job(
|
||||
_job_duration_plan,
|
||||
IntervalTrigger(hours=24, start_date=INTERVAL_ANCHOR),
|
||||
id="duration_plan",
|
||||
replace_existing=True,
|
||||
max_instances=1,
|
||||
coalesce=True,
|
||||
)
|
||||
log.info("scheduler: duration plan every 24h")
|
||||
if cfg.get("duration_backfill_minutes"):
|
||||
_dur_batch = int(cfg.get("duration_backfill_batch") or 300)
|
||||
sched.add_job(
|
||||
_job_duration_backfill,
|
||||
IntervalTrigger(
|
||||
minutes=cfg["duration_backfill_minutes"], start_date=INTERVAL_ANCHOR
|
||||
),
|
||||
args=[_dur_batch],
|
||||
id="duration_backfill",
|
||||
replace_existing=True,
|
||||
max_instances=1,
|
||||
coalesce=True,
|
||||
)
|
||||
log.info(
|
||||
"scheduler: duration backfill every %dmin (batch=%d)",
|
||||
cfg["duration_backfill_minutes"],
|
||||
_dur_batch,
|
||||
)
|
||||
|
||||
if cfg.get("phash_backfill_minutes"):
|
||||
_batch = int(cfg.get("phash_backfill_batch") or 400)
|
||||
sched.add_job(
|
||||
|
|
|
|||
|
|
@ -219,6 +219,8 @@ def run_forever() -> int:
|
|||
# Blacklista zdegenerowanych phashy, zaślepki dzielone przez setki scen (audyt
|
||||
# 2026-07-27). Bez tego zaślepka jest zawsze najbliższym trafieniem w Path 3.
|
||||
"phash_blacklist_hours": getattr(settings, "sched_phash_blacklist_hours", 24) or None,
|
||||
"duration_backfill_minutes": getattr(settings, "sched_duration_backfill_minutes", 5) or None,
|
||||
"duration_backfill_batch": getattr(settings, "sched_duration_backfill_batch", 500),
|
||||
"phash_backfill_minutes": getattr(settings, "sched_phash_backfill_minutes", 20) or None,
|
||||
"phash_backfill_batch": getattr(settings, "sched_phash_backfill_batch", 400),
|
||||
# Śmieciowi performerzy, kategorie/studia udające osoby (audyt 2026-07-27).
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue