Compare commits

..

No commits in common. "77679b26bbb611677b4d5b0f8c42defd93faddca" and "494b15aedd98b5b1be64bebdb0a1441c122c042c" have entirely different histories.

11 changed files with 22 additions and 523 deletions

View file

@ -1,41 +0,0 @@
"""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

@ -1,86 +0,0 @@
"""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

@ -1,35 +0,0 @@
"""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,14 +759,7 @@ 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())
# Ekran serwisu od najnowszych: sceny serwisu bierzemy wprost z indeksu if origin:
# 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
@ -918,21 +911,6 @@ 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()
@ -1076,7 +1054,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 not origin_sort and sort in ("created_at", "release_date"): if rows is None and origin and not q 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,7 +27,6 @@ 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__)
@ -159,21 +158,10 @@ 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:
# Liczniki wertykału czytamy z tag_vertical_counts (refresh_taxonomy_counts co 3 h, # Wertykał liczymy NA ŻYWO, bez denormalizacji: to małe zbiory (animacja ~9,5 tys.
# migracja 0047). Agregat na żywo trwał 1 s na rozgrzanym cache i 17-24 s na # scen wobec 3,8 mln całości), więc agregat jest tani, a dokładanie kolumny
# zimnym (zakładka JAV, 2026-09-27). Zostaje jako zapas, gdy tabela jest jeszcze # `scene_count_<wertykał>` do `tags` kosztowałoby migrację przy każdym nowym
# pusta dla tego wertykału, np. zaraz po wdrożeniu. # wertykale. Zdenormalizowany `Tag.scene_count` opisuje WYŁĄCZNIE główny feed.
# 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,7 +38,6 @@ 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",
@ -85,5 +84,4 @@ __all__ = [
"StudioAlias", "StudioAlias",
"StudioExternalRef", "StudioExternalRef",
"Tag", "Tag",
"TagVerticalCount",
] ]

View file

@ -8,10 +8,9 @@ nie tworzył duplikatów.
from __future__ import annotations from __future__ import annotations
import uuid import uuid
from datetime import date, datetime from datetime import datetime
from sqlalchemy import ( from sqlalchemy import (
Date,
DateTime, DateTime,
ForeignKey, ForeignKey,
Integer, Integer,
@ -67,8 +66,3 @@ 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

@ -1,25 +0,0 @@
"""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, VERTICALS from app.verticals import ALL_VERTICAL_ORIGINS
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
@ -129,35 +129,11 @@ _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

@ -1,76 +0,0 @@
"""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

@ -1,172 +0,0 @@
"""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)