Compare commits
3 commits
494b15aedd
...
77679b26bb
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
77679b26bb | ||
|
|
559c3bf01c | ||
|
|
c97af57979 |
11 changed files with 523 additions and 22 deletions
41
alembic/versions/20260927_0047_tag_vertical_counts.py
Normal file
41
alembic/versions/20260927_0047_tag_vertical_counts.py
Normal 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")
|
||||||
86
alembic/versions/20260927_0048_playback_sources_sort_date.py
Normal file
86
alembic/versions/20260927_0048_playback_sources_sort_date.py
Normal 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")
|
||||||
35
alembic/versions/20260927_0049_playback_origin_sort_idx.py
Normal file
35
alembic/versions/20260927_0049_playback_origin_sort_idx.py
Normal 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")
|
||||||
|
|
@ -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)
|
||||||
|
|
|
||||||
|
|
@ -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(
|
||||||
|
|
|
||||||
|
|
@ -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",
|
||||||
]
|
]
|
||||||
|
|
|
||||||
|
|
@ -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)
|
||||||
|
|
|
||||||
25
app/models/tag_vertical_count.py
Normal file
25
app/models/tag_vertical_count.py
Normal 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)
|
||||||
|
|
@ -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),
|
||||||
|
|
|
||||||
76
scripts/backfill_playback_sort_date.py
Normal file
76
scripts/backfill_playback_sort_date.py
Normal 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)
|
||||||
172
scripts/split_comma_performers.py
Normal file
172
scripts/split_comma_performers.py
Normal 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)
|
||||||
Loading…
Add table
Reference in a new issue