Compare commits

..

3 commits

Author SHA1 Message Date
goon-foss
77679b26bb perf(scenes): lista serwisu od najnowszych prosto z indeksu
Some checks failed
Backend tests / test (push) Has been cancelled
Ekran serwisu (/scenes?origin=tube:X&sort=release_date) dla serwisow,
ktorych scen nie ma wsrod najnowszych w katalogu (perverzija, hdporn.gg,
tubepornclassic, yesporn), sortowal wszystkie ich sceny: 9-19 s na
zimnym cache, 1-4 s na cieplym, 39 wejsc w 3 h.

Wzorem scene_tags.sort_date (GOON-64): data sortu powielona w
playback_sources i pilnowana wyzwalaczami (0048), wypelniona skryptem,
indeks (origin, sort_date DESC, scene_id DESC) WHERE dead_at IS NULL
(0049). API dolacza DISTINCT ON zgodny z indeksem, bo 190 tys. scen ma
dwa zywe zrodla z tego samego serwisu. Po zmianie 0,03-0,24 s, takze
dalsze strony.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-09-27 20:18:59 +02:00
goon-foss
559c3bf01c fix(performers): skrypt rozbijajacy zlepionych wykonawcow "A, B"
Scraper siski do 494b15a zapisywal kilka osob z alt jako jedna. Skrypt
przepina sceny, filmy i ulubione na prawdziwe osoby (resolve_performer),
czarnych list nie przenosi. "Woods, Monique" rozpoznaje jako jedna osobe
tylko gdy tytul sceny to potwierdza. Na prodzie: 660 zlepkow, 1194
powiazania, 4 nowe osoby, 4 niejasne zostaly.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-09-27 19:37:23 +02:00
goon-foss
c97af57979 perf(tags): liczniki tagow JAV i animacji liczone w tle
/tags?vertical=jav liczyl agregat po zrodlach i scene_tags na kazde
wejscie, dwa razy: 1 s na rozgrzanym cache, 17-24 s na zimnym. Teraz
czyta tabele tag_vertical_counts (migracja 0047), ktora odswieza
refresh_taxonomy_counts co 3 h. Zakladka JAV: 0,35 s. Gdy tabela jest
pusta dla wertykalu, zostaje stary agregat.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-09-27 19:37:23 +02:00
11 changed files with 523 additions and 22 deletions

View file

@ -0,0 +1,41 @@
"""Liczniki tagow per wertykal (JAV, animacja), liczone w tle zamiast na kazde wejscie.
/tags?vertical=jav liczyl na zywo: sceny z zywym zrodlem JAV, ich scene_tags, GROUP BY,
i to dwa razy (licznik total i strona). Na rozgrzanym cache 1 s, na zimnym 17-24 s
(log API 2026-09-27, zakladka JAV w apce). Tabela (vertical, tag_id) zamiast kolumny
na `tags` na kazdy wertykal, zeby nowy wertykal nie wymagal migracji. Wypelnia ja
refresh_taxonomy_counts (co 3 h); dopoki dla wertykalu nie ma wierszy, API liczy po staremu.
Revision ID: 0047_tag_vertical_counts
Revises: 0046_scenes_title_stats
Create Date: 2026-09-27
"""
from __future__ import annotations
import sqlalchemy as sa
from alembic import op
from sqlalchemy.dialects.postgresql import UUID
revision = "0047_tag_vertical_counts"
down_revision: str | None = "0046_scenes_title_stats"
branch_labels = None
depends_on = None
def upgrade() -> None:
op.create_table(
"tag_vertical_counts",
sa.Column("vertical", sa.String(32), nullable=False),
sa.Column("tag_id", UUID(as_uuid=True), nullable=False),
sa.Column("scene_count", sa.Integer, nullable=False),
sa.PrimaryKeyConstraint("vertical", "tag_id", name="pk_tag_vertical_counts"),
sa.ForeignKeyConstraint(
["tag_id"], ["tags.id"],
name="fk_tag_vertical_counts_tag_id_tags",
ondelete="CASCADE",
),
)
def downgrade() -> None:
op.drop_table("tag_vertical_counts")

View file

@ -0,0 +1,86 @@
"""Data sortowania sceny powielona w playback_sources (lista scen jednego serwisu).
Ekran serwisu w apce (`/scenes?origin=tube:X&sort=release_date`) sortuje po dacie z
`scenes`, a filtruje po `playback_sources`. Dla serwisow, ktorych scen nie ma wsrod
najnowszych w katalogu (perverzija, hdporn.gg, tubepornclassic, yesporn), przejscie po
ix_scenes_effective_date_desc nie znajduje strony w 1 s, wiec API bierze wszystkie id
scen serwisu (20-115 tys.) i sortuje je po dacie. Na rozgrzanym cache 1-4 s, na zimnym
9-19 s (log API 2026-09-27, 39 wejsc w 3 h). Ten sam problem co przy tagach (0042).
`sort_date` = to samo wyrazenie co w 0042 i w `_effective_date_desc()`. Indeks
(origin, sort_date DESC, scene_id DESC) WHERE dead_at IS NULL z migracji 0049 daje od
razu najnowsze sceny serwisu.
Utrzymanie w bazie, bo do playback_sources pisze ingest, merge (przenosi zrodla miedzy
scenami) i skrypty:
- BEFORE INSERT / UPDATE OF scene_id na playback_sources: data z wiersza sceny,
- wyzwalacz scen z 0042 poprawia teraz tez zrodla tej sceny.
Kolumna nullable i bez DEFAULT, wiec ALTER jest natychmiastowy. Istniejace 4,4 mln
wierszy wypelnia scripts/backfill_playback_sort_date.py, dopiero potem indeks (0049)
i przelaczenie zapytania.
Revision ID: 0048_playback_sort_date
Revises: 0047_tag_vertical_counts
Create Date: 2026-09-27
"""
from __future__ import annotations
from alembic import op
revision = "0048_playback_sort_date"
down_revision: str | None = "0047_tag_vertical_counts"
branch_labels = None
depends_on = None
_SORT_DATE = "COALESCE({t}.release_date, ({t}.created_at AT TIME ZONE 'UTC')::date)"
def upgrade() -> None:
op.execute("SET LOCAL lock_timeout = '5s'")
op.execute("ALTER TABLE playback_sources ADD COLUMN IF NOT EXISTS sort_date date")
op.execute(f"""
CREATE OR REPLACE FUNCTION playback_sources_fill_sort_date() RETURNS trigger
LANGUAGE plpgsql AS $$
BEGIN
SELECT {_SORT_DATE.format(t='s')} INTO NEW.sort_date
FROM scenes s WHERE s.id = NEW.scene_id;
RETURN NEW;
END $$
""")
op.execute("""
CREATE TRIGGER trg_playback_sources_fill_sort_date
BEFORE INSERT OR UPDATE OF scene_id ON playback_sources
FOR EACH ROW EXECUTE FUNCTION playback_sources_fill_sort_date()
""")
# Ta sama funkcja co w 0042, z dopisanym playback_sources. Wyzwalacz na scenes
# zostaje bez zmian.
op.execute(f"""
CREATE OR REPLACE FUNCTION scenes_propagate_sort_date() RETURNS trigger
LANGUAGE plpgsql AS $$
DECLARE d date := {_SORT_DATE.format(t='NEW')};
BEGIN
UPDATE scene_tags SET sort_date = d
WHERE scene_id = NEW.id AND sort_date IS DISTINCT FROM d;
UPDATE playback_sources SET sort_date = d
WHERE scene_id = NEW.id AND sort_date IS DISTINCT FROM d;
RETURN NULL;
END $$
""")
def downgrade() -> None:
op.execute("SET LOCAL lock_timeout = '5s'")
op.execute(f"""
CREATE OR REPLACE FUNCTION scenes_propagate_sort_date() RETURNS trigger
LANGUAGE plpgsql AS $$
DECLARE d date := {_SORT_DATE.format(t='NEW')};
BEGIN
UPDATE scene_tags SET sort_date = d
WHERE scene_id = NEW.id AND sort_date IS DISTINCT FROM d;
RETURN NULL;
END $$
""")
op.execute("DROP TRIGGER IF EXISTS trg_playback_sources_fill_sort_date ON playback_sources")
op.execute("DROP FUNCTION IF EXISTS playback_sources_fill_sort_date()")
op.execute("ALTER TABLE playback_sources DROP COLUMN IF EXISTS sort_date")

View file

@ -0,0 +1,35 @@
"""Indeks "najnowsze sceny serwisu" (kolumna z 0048).
(origin, sort_date DESC, scene_id DESC) WHERE dead_at IS NULL: lista scen serwisu,
sort=release_date, idzie po tym indeksie od najnowszej i konczy po LIMIT, niezaleznie od
tego, gdzie w calym katalogu leza sceny serwisu. Uzasadnienie i pomiary w 0048.
Zakladac DOPIERO po wypelnieniu `sort_date` (scripts/backfill_playback_sort_date.py),
z tego samego powodu co 0043. CONCURRENTLY w autocommit_block; nieudany zostawia indeks
INVALID, wtedy DROP INDEX i ponownie.
Revision ID: 0049_playback_origin_sort_idx
Revises: 0048_playback_sort_date
Create Date: 2026-09-27
"""
from __future__ import annotations
from alembic import op
revision = "0049_playback_origin_sort_idx"
down_revision: str | None = "0048_playback_sort_date"
branch_labels = None
depends_on = None
def upgrade() -> None:
with op.get_context().autocommit_block():
op.execute(
"CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_playback_sources_origin_sort "
"ON playback_sources (origin, sort_date DESC, scene_id DESC) WHERE dead_at IS NULL"
)
def downgrade() -> None:
with op.get_context().autocommit_block():
op.execute("DROP INDEX CONCURRENTLY IF EXISTS ix_playback_sources_origin_sort")

View file

@ -759,7 +759,14 @@ def list_scenes(
# stwierdzic: 148 s na produkcji (pomiar 2026-09-16). Zwieramy od razu. # stwierdzic: 148 s na produkcji (pomiar 2026-09-16). Zwieramy od razu.
base = base.where(false() if origin else ~live_playback_exists()) base = base.where(false() if origin else ~live_playback_exists())
if origin: # Ekran serwisu od najnowszych: sceny serwisu bierzemy wprost z indeksu
# ix_playback_sources_origin_sort (data sortu powielona w zrodle, migracje 0048/0049)
# zamiast szukac ich w calym katalogu po dacie. Wtedy JOIN w galezi sortu zastepuje
# ponizszy EXISTS. Z fraza albo tagiem zostaja dotychczasowe sciezki.
origin_sort = bool(
sort == "release_date" and origin and ":" in origin and tag_sort_id is None and not q
)
if origin and not origin_sort:
# Pelny origin z ekranu strony ('tube:hdporngg') porownujemy wprost: rownosc idzie # Pelny origin z ekranu strony ('tube:hdporngg') porownujemy wprost: rownosc idzie
# po ix_playback_sources_origin_live, a ILIKE '%...%' przeszukiwal kazde zrodlo # po ix_playback_sources_origin_live, a ILIKE '%...%' przeszukiwal kazde zrodlo
# kazdej kandydujacej sceny (wyszukiwanie w obrebie strony 4,6 s, Sentry Slow DB # kazdej kandydujacej sceny (wyszukiwanie w obrebie strony 4,6 s, Sentry Slow DB
@ -911,6 +918,21 @@ def list_scenes(
ordered = base.join( ordered = base.join(
st, and_(st.scene_id == Scene.id, st.tag_id == tag_sort_id) st, and_(st.scene_id == Scene.id, st.tag_id == tag_sort_id)
).order_by(st.sort_date.desc(), st.scene_id.desc()) ).order_by(st.sort_date.desc(), st.scene_id.desc())
elif origin_sort:
# Scena moze miec dwa zywe zrodla z tego samego serwisu (190 tys. par), a oba
# niosa te sama date, wiec DISTINCT ON zgodny z porzadkiem indeksu zbija je w
# jeden wiersz bez sortowania: skan dalej konczy sie po LIMIT. Pomiar i
# uzasadnienie w migracji 0048 (perverzija 13,7 s na zimnym cache).
ps_sort = (
select(PlaybackSource.scene_id, PlaybackSource.sort_date)
.where(PlaybackSource.origin == origin, PlaybackSource.dead_at.is_(None))
.distinct(PlaybackSource.sort_date, PlaybackSource.scene_id)
.order_by(PlaybackSource.sort_date.desc(), PlaybackSource.scene_id.desc())
.subquery("origin_sort")
)
ordered = base.join(ps_sort, ps_sort.c.scene_id == Scene.id).order_by(
ps_sort.c.sort_date.desc(), ps_sort.c.scene_id.desc()
)
else: else:
ordered = base.order_by( ordered = base.order_by(
_effective_date_desc(), Scene.created_at.desc(), Scene.id.desc() _effective_date_desc(), Scene.created_at.desc(), Scene.id.desc()
@ -1054,7 +1076,7 @@ def list_scenes(
Scene.id Scene.id
== any_(bindparam("origin_scene_ids", origin_ids, type_=ARRAY(PG_UUID(as_uuid=True)))) == any_(bindparam("origin_scene_ids", origin_ids, type_=ARRAY(PG_UUID(as_uuid=True))))
) )
if rows is None and origin and not q and sort in ("created_at", "release_date"): if rows is None and origin and not q and not origin_sort and sort in ("created_at", "release_date"):
walked = _search_walk_ids(session, ordered, _off, _want, _ORIGIN_LIST_WALK_BUDGET_MS) walked = _search_walk_ids(session, ordered, _off, _want, _ORIGIN_LIST_WALK_BUDGET_MS)
if walked is not None: if walked is not None:
rows = _scenes_in_order(session, walked) rows = _scenes_in_order(session, walked)

View file

@ -27,6 +27,7 @@ from app.models.playback_source import PlaybackSource
from app.models.scene import ScenePerformer, SceneTag from app.models.scene import ScenePerformer, SceneTag
from app.models.studio import Studio from app.models.studio import Studio
from app.models.tag import Tag from app.models.tag import Tag
from app.models.tag_vertical_count import TagVerticalCount
from app.verticals import VERTICALS from app.verticals import VERTICALS
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
@ -158,10 +159,21 @@ def list_tags(
return TagListOut(items=items, total=total, page=page, per_page=per_page) return TagListOut(items=items, total=total, page=page, per_page=per_page)
if vertical in VERTICALS: if vertical in VERTICALS:
# Wertykał liczymy NA ŻYWO, bez denormalizacji: to małe zbiory (animacja ~9,5 tys. # Liczniki wertykału czytamy z tag_vertical_counts (refresh_taxonomy_counts co 3 h,
# scen wobec 3,8 mln całości), więc agregat jest tani, a dokładanie kolumny # migracja 0047). Agregat na żywo trwał 1 s na rozgrzanym cache i 17-24 s na
# `scene_count_<wertykał>` do `tags` kosztowałoby migrację przy każdym nowym # zimnym (zakładka JAV, 2026-09-27). Zostaje jako zapas, gdy tabela jest jeszcze
# wertykale. Zdenormalizowany `Tag.scene_count` opisuje WYŁĄCZNIE główny feed. # pusta dla tego wertykału, np. zaraz po wdrożeniu.
# Zdenormalizowany `Tag.scene_count` opisuje WYŁĄCZNIE główny feed.
has_counts = session.execute(
select(TagVerticalCount.tag_id).where(TagVerticalCount.vertical == vertical).limit(1)
).first()
if has_counts:
v_count = (
select(TagVerticalCount.tag_id, TagVerticalCount.scene_count.label("c"))
.where(TagVerticalCount.vertical == vertical)
.subquery()
)
else:
v_scenes = ( v_scenes = (
select(PlaybackSource.scene_id) select(PlaybackSource.scene_id)
.where( .where(

View file

@ -38,6 +38,7 @@ from app.models.scene import (
from app.models.source import Source from app.models.source import Source
from app.models.studio import Studio, StudioAlias, StudioExternalRef from app.models.studio import Studio, StudioAlias, StudioExternalRef
from app.models.tag import Tag from app.models.tag import Tag
from app.models.tag_vertical_count import TagVerticalCount
__all__ = [ __all__ = [
"Base", "Base",
@ -84,4 +85,5 @@ __all__ = [
"StudioAlias", "StudioAlias",
"StudioExternalRef", "StudioExternalRef",
"Tag", "Tag",
"TagVerticalCount",
] ]

View file

@ -8,9 +8,10 @@ nie tworzył duplikatów.
from __future__ import annotations from __future__ import annotations
import uuid import uuid
from datetime import datetime from datetime import date, datetime
from sqlalchemy import ( from sqlalchemy import (
Date,
DateTime, DateTime,
ForeignKey, ForeignKey,
Integer, Integer,
@ -66,3 +67,8 @@ class PlaybackSource(UUIDPKMixin, TimestampMixin, Base):
API filtruje takie sources z listy. None = aktywny.""" API filtruje takie sources z listy. None = aktywny."""
dead_reason: Mapped[str | None] = mapped_column(String(512), nullable=True) dead_reason: Mapped[str | None] = mapped_column(String(512), nullable=True)
# Data sortowania sceny (premiera albo dzien dodania), powielona tu pod indeks
# "najnowsze sceny serwisu" (migracje 0048 i 0049). Pisze ja WYZWALACZ w bazie,
# nie aplikacja: wartosc podana przy INSERT i tak zostanie nadpisana.
sort_date: Mapped[date | None] = mapped_column(Date)

View file

@ -0,0 +1,25 @@
"""Liczba scen tagu w wertykale (JAV, animacja), odswiezana w tle.
Wypelnia `refresh_taxonomy_counts`, czyta /tags?vertical=. Ta sama definicja co dawny
agregat na zywo: sceny z >= 1 zywym zrodlem z originow wertykalu, bez progu dlugosci.
Patrz migracja 0047.
"""
from __future__ import annotations
import uuid
from sqlalchemy import ForeignKey, Integer, String
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import Mapped, mapped_column
from app.models.base import Base
class TagVerticalCount(Base):
__tablename__ = "tag_vertical_counts"
vertical: Mapped[str] = mapped_column(String(32), primary_key=True)
tag_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True), ForeignKey("tags.id", ondelete="CASCADE"), primary_key=True
)
scene_count: Mapped[int] = mapped_column(Integer, nullable=False)

View file

@ -23,7 +23,7 @@ import logging
from sqlalchemy import text from sqlalchemy import text
from app.db import session_scope from app.db import session_scope
from app.verticals import ALL_VERTICAL_ORIGINS from app.verticals import ALL_VERTICAL_ORIGINS, VERTICALS
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
@ -129,11 +129,35 @@ _STUDIOS_SQL = text(
) )
# Tagi wertykalu (lista w zakladce JAV/animacja): sceny z zywym zrodlem z jego originow,
# bez progu dlugosci, ta sama definicja co dawny agregat w /tags?vertical=. Podmiana
# calego kompletu w jednej transakcji, to kilka tysiecy wierszy (migracja 0047).
_VERTICAL_TAGS_DELETE_SQL = text("DELETE FROM tag_vertical_counts WHERE vertical = :vertical")
_VERTICAL_TAGS_INSERT_SQL = text(
"""
INSERT INTO tag_vertical_counts (vertical, tag_id, scene_count)
SELECT :vertical, st.tag_id, count(*)
FROM scene_tags st
JOIN (
SELECT DISTINCT ps.scene_id FROM playback_sources ps
WHERE ps.dead_at IS NULL AND ps.origin = ANY(:origins)
) v ON v.scene_id = st.scene_id
GROUP BY st.tag_id
"""
)
def refresh_taxonomy_counts() -> dict[str, int]: def refresh_taxonomy_counts() -> dict[str, int]:
"""Przelicza scene_count dla tags/performers/studios. Zwraca rowcount per tabela """Przelicza scene_count dla tags/performers/studios. Zwraca rowcount per tabela
(ile wierszy faktycznie się zmieniło).""" (ile wierszy faktycznie się zmieniło)."""
out: dict[str, int] = {} out: dict[str, int] = {}
with session_scope() as session: with session_scope() as session:
for vertical, origins in VERTICALS.items():
session.execute(_VERTICAL_TAGS_DELETE_SQL, {"vertical": vertical})
out[f"tags_{vertical}"] = session.execute(
_VERTICAL_TAGS_INSERT_SQL, {"vertical": vertical, "origins": list(origins)}
).rowcount or 0
session.commit()
for name, stmt in ( for name, stmt in (
("tags", _TAGS_SQL), ("tags", _TAGS_SQL),
("performers", _PERFORMERS_SQL), ("performers", _PERFORMERS_SQL),

View file

@ -0,0 +1,76 @@
"""Wypelnia playback_sources.sort_date w istniejacych wierszach (migracja 0048).
Kopia scripts/backfill_scene_tags_sort_date.py dla playback_sources: partiami po zakresach
stron heapu, commit po kazdej partii, VACUUM co kilka partii, pauza oddaje dysk apce.
Nowe wiersze w trakcie wypelnia juz wyzwalacz z 0048. Wznawialny: rusza tylko wiersze z
sort_date IS NULL, na koncu dobiera resztki.
docker compose run -d --name backfill_ps_sort worker \
python3 -u /srv/scripts/backfill_playback_sort_date.py [stron_na_partie] [pauza_s]
Po nim: alembic upgrade (0049, indeks) i dopiero wtedy kod, ktory z indeksu korzysta.
"""
from __future__ import annotations
import sys
import time
from sqlalchemy import text
from app.db import engine
PAGES = int(sys.argv[1]) if len(sys.argv) > 1 else 1000
PAUSE = float(sys.argv[2]) if len(sys.argv) > 2 else 10.0
VACUUM_EVERY = 8
SORT_DATE = "COALESCE(s.release_date, (s.created_at AT TIME ZONE 'UTC')::date)"
def rel_pages() -> int:
with engine.connect() as c:
return int(c.execute(text(
"SELECT pg_relation_size('playback_sources') / current_setting('block_size')::int"
)).scalar())
def vacuum(analyze: bool = False) -> None:
with engine.connect().execution_options(isolation_level="AUTOCOMMIT") as c:
c.execute(text("VACUUM (ANALYZE) playback_sources" if analyze else "VACUUM playback_sources"))
t0 = time.time()
total = 0
page = 0
batch = 0
pages = rel_pages()
print(f"playback_sources: {pages} stron, partia {PAGES} stron, pauza {PAUSE} s", flush=True)
while page < pages:
tb = time.time()
with engine.begin() as c:
c.execute(text("SET LOCAL statement_timeout = '600s'"))
n = c.execute(text(f"""
UPDATE playback_sources ps SET sort_date = {SORT_DATE}
FROM scenes s
WHERE s.id = ps.scene_id AND ps.sort_date IS NULL
AND ps.ctid >= format('(%s,0)', CAST(:p0 AS int))::tid
AND ps.ctid < format('(%s,0)', CAST(:p1 AS int))::tid
"""), {"p0": page, "p1": page + PAGES}).rowcount
total += n
page += PAGES
batch += 1
if batch % VACUUM_EVERY == 0:
vacuum()
pages = rel_pages() # tabela moze urosnac, zanim VACUUM zwolni miejsce
print(f"partia {batch}: strony do {page}/{pages}, {n} wierszy w {time.time() - tb:.1f} s, "
f"razem {total}, {time.time() - t0:.0f} s", flush=True)
if n:
time.sleep(PAUSE)
with engine.begin() as c:
rest = c.execute(text(f"""
UPDATE playback_sources ps SET sort_date = {SORT_DATE}
FROM scenes s WHERE s.id = ps.scene_id AND ps.sort_date IS NULL
""")).rowcount
print(f"resztki: {rest}", flush=True)
vacuum(analyze=True)
with engine.connect() as c:
left = c.execute(text("SELECT count(*) FROM playback_sources WHERE sort_date IS NULL")).scalar()
print(f"KONIEC: {total + rest} wierszy, puste po wszystkim: {left}, {time.time() - t0:.0f} s", flush=True)

View file

@ -0,0 +1,172 @@
"""Rozbija zlepionych wykonawcow typu "Cory Chase, Gigi Sweets" na prawdziwe osoby.
Scraper siski do 2026-09-27 (494b15a) bral pierwszy czlon alt='A, B - Tytul' jako jedno
nazwisko, wiec powstalo ~660 takich "osob". Kazda czesc przechodzi przez
resolve_performer (ta sama sciezka co ingest: nazwa, potem alias, na koncu nowa osoba),
powiazania ze scenami, filmami i ulubionymi przechodza na czesci, zlepek znika.
Czarne listy NIE przechodza: blokada "A, B" to nie blokada wszystkiego z A i z B.
Pomijamy nazwiska z ref do TPDB/StashDB (kanon nie ma przecinkow, wiec to bylby
swiadomy wyjatek) i takie, ktorych czesci nie wygladaja na nazwiska.
docker compose run --rm worker python3 -u /srv/scripts/split_comma_performers.py # na sucho
docker compose run --rm worker python3 -u /srv/scripts/split_comma_performers.py --commit
--commit najpierw kopiuje zlepki i ich powiazania do tabel *_comma_20260927.
"""
from __future__ import annotations
import collections
import sys
from sqlalchemy import text
from app.connectors.base import RawPerformer
from app.db import session_scope
from app.normalize.scenes import normalize_performer
from app.resolve.performer_resolver import resolve_performer
COMMIT = "--commit" in sys.argv
BACKUP_SUFFIX = "comma_20260927"
# Tabele z (..., performer_id) w kluczu glownym, powiazania przenosimy na kazda czesc.
MOVE_TABLES = ("scene_performers", "movie_performers", "favorite_performers")
DROP_TABLES = ("blacklisted_performers",)
def parts_of(name: str) -> list[str] | None:
parts = [p.strip() for p in name.split(",")]
if len(parts) < 2 or len(parts) > 10 or any(not p for p in parts):
return None
for p in parts:
words = p.split()
# Cyfry tylko w jednowyrazowym nicku ("Subgirl0831"), nie w "Scene 2".
if len(p) < 2 or len(words) > 4 or (len(words) > 1 and any(ch.isdigit() for ch in p)):
return None
# "Merce Palau aka Merce", "Ruby Reign. Yoya Grey": inny separator, zostawiamy.
if "." in p or " aka " in f" {p.lower()} ":
return None
return parts
def two_single_words(parts: list[str]) -> bool:
return len(parts) == 2 and all(len(p.split()) == 1 for p in parts)
def reversed_person(session, pid, parts: list[str]) -> str | None:
""""Woods, Monique" to jedna osoba w zapisie nazwisko, imie, gdy "Monique Woods" jest
w bazie i stoi w tytule ktorejs z jej scen. Samo istnienie nazwiska nie wystarcza:
"Jenna, Chloe" przy "Wild Brunette Threesome" to raczej dwie osoby."""
if not two_single_words(parts):
return None
name = f"{parts[1]} {parts[0]}"
norm = normalize_performer(RawPerformer(name=name)).name_normalized
hit = session.execute(text("""
SELECT 1 FROM performers WHERE name_normalized = :n
AND EXISTS (SELECT 1 FROM scene_performers sp JOIN scenes sc ON sc.id = sp.scene_id
WHERE sp.performer_id = :pid AND sc.title ILIKE :pat)
LIMIT 1
"""), {"n": norm, "pid": pid, "pat": f"%{name}%"}).first()
return name if hit else None
def columns(session, table: str) -> list[str]:
return list(session.execute(text(
"SELECT column_name FROM information_schema.columns "
"WHERE table_schema = 'public' AND table_name = :t ORDER BY ordinal_position"
), {"t": table}).scalars())
with session_scope() as s:
tube_source = s.execute(text("SELECT id FROM sources WHERE name = 'tube-scraper'")).scalar_one()
canon_sources = list(s.execute(text(
"SELECT id FROM sources WHERE kind IN ('tpdb', 'stashdb')"
)).scalars())
junk = s.execute(text("""
SELECT p.id, p.canonical_name FROM performers p
WHERE p.canonical_name LIKE '%,%'
AND NOT EXISTS (
SELECT 1 FROM performer_external_refs r
WHERE r.performer_id = p.id AND r.source_id = ANY(:canon))
ORDER BY p.canonical_name
"""), {"canon": canon_sources}).all()
print(f"zlepkow z przecinkiem bez refu kanonu: {len(junk)}", flush=True)
usage = collections.Counter()
for t in MOVE_TABLES + DROP_TABLES:
usage[t] = s.execute(text(
f"SELECT count(*) FROM {t} WHERE performer_id = ANY(:ids)"
), {"ids": [j.id for j in junk]}).scalar()
print("powiazania:", dict(usage), flush=True)
plan: list[tuple] = []
skipped: list[str] = []
for pid, name in junk:
parts = parts_of(name)
if parts is None:
skipped.append(name)
continue
whole = reversed_person(s, pid, parts)
if whole:
print(f" nazwisko, imie: {name!r} -> {whole!r}")
parts = [whole]
elif two_single_words(parts) and s.execute(
text("SELECT 1 FROM performers WHERE name_normalized = :n LIMIT 1"),
{"n": normalize_performer(RawPerformer(name=f"{parts[1]} {parts[0]}")).name_normalized},
).first():
# Odwrocone nazwisko istnieje, ale tytul go nie potwierdza: nie zgadujemy.
skipped.append(name)
continue
plan.append((pid, name, parts))
print(f"do rozbicia: {len(plan)}, pominiete: {len(skipped)}", flush=True)
for name in skipped[:30]:
print(" pomijam:", name)
existing = new = 0
samples = []
targets: dict = {}
for pid, name, parts in plan:
ids = []
for part in parts:
norm = normalize_performer(RawPerformer(name=part))
hit = s.execute(text(
"SELECT id FROM performers WHERE name_normalized = :n "
"UNION ALL SELECT performer_id FROM performer_aliases WHERE alias_normalized = :n LIMIT 1"
), {"n": norm.name_normalized}).scalar()
if hit is not None and hit != pid:
existing += 1
else:
new += 1
if COMMIT:
ids.append(resolve_performer(s, norm=norm, source_id=tube_source).id)
samples.append((name, part, "jest" if hit is not None and hit != pid else "NOWA"))
targets[pid] = ids
print(f"czesci: {existing} juz w bazie, {new} nowych osob", flush=True)
for name, part, st in samples[:10] + [x for x in samples if x[2] == "NOWA"]:
print(f" {name!r} -> {part!r} ({st})")
if not COMMIT:
print("NA SUCHO, nic nie zmienione. --commit zeby wykonac.")
sys.exit(0)
ids = [pid for pid, _, _ in plan]
s.execute(text(f"DROP TABLE IF EXISTS performers_{BACKUP_SUFFIX}"))
s.execute(text(f"CREATE TABLE performers_{BACKUP_SUFFIX} AS SELECT * FROM performers WHERE id = ANY(:ids)"), {"ids": ids})
for t in MOVE_TABLES + DROP_TABLES:
s.execute(text(f"DROP TABLE IF EXISTS {t}_{BACKUP_SUFFIX}"))
s.execute(text(f"CREATE TABLE {t}_{BACKUP_SUFFIX} AS SELECT * FROM {t} WHERE performer_id = ANY(:ids)"), {"ids": ids})
moved = collections.Counter()
for t in MOVE_TABLES:
cols = columns(s, t)
sel = ", ".join(":new" if c == "performer_id" else c for c in cols)
for pid, _, _ in plan:
for new_id in targets[pid]:
if new_id == pid:
continue
moved[t] += s.execute(text(
f"INSERT INTO {t} ({', '.join(cols)}) SELECT {sel} FROM {t} "
f"WHERE performer_id = :old ON CONFLICT DO NOTHING"
), {"new": new_id, "old": pid}).rowcount or 0
deleted = s.execute(text("DELETE FROM performers WHERE id = ANY(:ids)"), {"ids": ids}).rowcount
s.commit()
print(f"przeniesione powiazania: {dict(moved)}, skasowane zlepki: {deleted}", flush=True)