rht

README.md

@@ -34,15 +34,3 @@ Each vote row stores:
 - `dataset`
 - `group`
 - `session_id`
-- `pair_reason` (which sampler branch served the pair)
-- `rater_name` (optional self-reported handle, empty when anonymous)
-
-## Maintenance
-
-Scripts under `maintenance/` import from the repo root, so run them as modules from there:
-
-```
-python -m maintenance.compact_logs
-```
-
-Compaction rewrites every shard through `VOTE_COLUMNS`, so columns added to the schema are backfilled as null on older rows.

app.py

@@ -5,36 +5,22 @@ import time
 import uuid
 import os
 import html
-import sys
-
-from typing import Callable
 
 import pandas as pd
 from huggingface_hub import hf_hub_download
-from huggingface_hub.utils import EntryNotFoundError
 
 from storage import VoteStorage
-from collections import Counter, defaultdict
-
-from stats_from_logs import load_vote_logs, stats_by_md5
-from explorer import ALLOWED_CLASSIFIER_FILTERS, DISAGREEMENT_SORT_MODE, SORT_MODES, add_results_tab, build_results_data, load_more_results, on_gallery_select
+from stats_from_logs import load_stats_by_md5
+from explorer import ALLOWED_CLASSIFIER_FILTERS, add_results_tab, build_results_data, load_more_results, on_gallery_select
 
 DEBUG_MODE = os.getenv("DEBUG", "0").lower() in ("1", "true", "yes", "on")
 RATINGS_APP_TOKEN = os.getenv("RATINGS_APP_TOKEN")
 SUBMIT_KEY = os.getenv("RATINGS_SUBMIT_KEY")
 assert SUBMIT_KEY, "Missing required env var: RATINGS_SUBMIT_KEY"
 POOL_REPO_ID = "taigasan/e6-visual-ratings"
-RATER_NAME_MAX_LEN = 64
 VOTE_STORAGE = VoteStorage(mode="void" if DEBUG_MODE else "hf", token=RATINGS_APP_TOKEN)
 STATS_RELOAD_S = 30 * 60
 
-def _drop_cdn_unavailable(df: pd.DataFrame, label: str) -> pd.DataFrame:
-    keep = df["cdn_available"].astype("boolean").fillna(True).astype(bool)
-    if dropped := int((~keep).sum()):
-        print(f"{label}: dropping {dropped} items unavailable on the e621 CDN")
-
-    return df[keep].reset_index(drop=True)
-
 # -- Pool dataset -----------------------------------------------------------
 _pool_path = hf_hub_download(
     repo_id=POOL_REPO_ID,
@@ -42,221 +28,111 @@ _pool_path = hf_hub_download(
     repo_type="dataset",
     token=RATINGS_APP_TOKEN
 )
-_pool_df = _drop_cdn_unavailable(pd.read_parquet(_pool_path), "pool")
-_pool_df[["wins", "losses", "ties", "votes", "winrate"]] = (0, 0, 0, 0, 0.0)
-
-WINS_LOC = _pool_df.columns.get_loc("wins")
-LOSSES_LOC = _pool_df.columns.get_loc("losses")
-TIES_LOC = _pool_df.columns.get_loc("ties")
-VOTES_LOC = _pool_df.columns.get_loc("votes")
-WINRATE_LOC = _pool_df.columns.get_loc("winrate")
-
-_md5_to_idx = { md5: idx for idx, md5 in enumerate(_pool_df["md5"]) }
-
-_pool_lock = threading.Lock()
-
+_pool_df = pd.read_parquet(_pool_path)
+_pool_group_dfs = {g: gdf for g, gdf in _pool_df.groupby("group")}
+_stats_lock = threading.Lock()
 _stats_last_loaded_at = 0.0
+_stats_by_key: dict[str, tuple[int, int]] = {}
 _explorer_df = pd.DataFrame(columns=["group", "id", "md5", "rating", "sample_url", "image_url", "classifier", "classifier_score", "percentile"])
 
-_disputed_queues: dict[str, list[tuple[int, int, float]]] = {}
-
-def _ordered_pair(pair: tuple[str, str]) -> tuple[str, str]:
-    if pair[0] <= pair[1]:
-        return pair
-
-    return (pair[1], pair[0])
-
-def _load_stats() -> None:
-    global _disputed_queues
-
-    VOTE_STORAGE.sync()
-    load_vote_logs(repo_id=POOL_REPO_ID, token=RATINGS_APP_TOKEN)
 
-    n_missing = 0
-    with _pool_lock:
-        VOTE_STORAGE.sync()
-
-        votes_df = load_vote_logs(repo_id=POOL_REPO_ID, token=RATINGS_APP_TOKEN)
-        for md5, stats in stats_by_md5(votes_df).items():
-            if (idx := _md5_to_idx.get(md5)) is not None:
-                _pool_df.iloc[idx, [WINS_LOC, LOSSES_LOC, TIES_LOC, VOTES_LOC, WINRATE_LOC]] = (
-                    stats.wins, stats.losses, stats.ties, stats.votes, stats.winrate
-                )
-            else:
-                n_missing += 1
-
-        pair_votes = Counter(
-            _ordered_pair(pair)
-            for pair in zip(votes_df["md5a"], votes_df["md5b"])
+def _reload_stats_if_due(force: bool = False):
+    global _stats_last_loaded_at, _stats_by_key, _explorer_df
+    now = time.time()
+    if not force and (now - _stats_last_loaded_at) < STATS_RELOAD_S:
+        return
+    with _stats_lock:
+        now = time.time()
+        if not force and (now - _stats_last_loaded_at) < STATS_RELOAD_S:
+            return
+        _stats_by_key = load_stats_by_md5(
+            repo_id=POOL_REPO_ID,
+            token=RATINGS_APP_TOKEN,
         )
-
-    if n_missing:
-        print(f"{n_missing} md5s have stats but are not in the pool!", file=sys.stderr)
-
-    try:
-        disputed_df = pd.read_parquet(hf_hub_download(
+        classifier_scores_path = hf_hub_download(
             repo_id=POOL_REPO_ID,
-            filename="disputed_pairs.parquet",
+            filename="classifier_scores.parquet",
             repo_type="dataset",
             token=RATINGS_APP_TOKEN,
-        ))
-    except EntryNotFoundError:
-        _disputed_queues.clear()
-    else:
-        new_disputed: dict[str, list[tuple[int, int, float]]] = defaultdict(list)
-
-        for md5a, md5b, group, margin in disputed_df.itertuples(index=False):
-            if md5b < md5a:
-                md5a, md5b = md5b, md5a
-
-            if (
-                (idx_a := _md5_to_idx.get(md5a)) is not None
-                and (idx_b := _md5_to_idx.get(md5b)) is not None
-                and pair_votes[(md5a, md5b)] == 1
-            ):
-                new_disputed[group].append((idx_a, idx_b, margin))
-
-        n_disputed = 0
-        for disputed in new_disputed.values():
-            n_disputed += len(disputed)
-            disputed.sort(key=lambda item: item[2])
-
-        print(f"disputed queue: {n_disputed} pairs across {len(new_disputed)} groups")
+        )
+        validation_set_path = hf_hub_download(
+            repo_id=POOL_REPO_ID,
+            filename="validation_set.parquet",
+            repo_type="dataset",
+            token=RATINGS_APP_TOKEN,
+        )
+        validation_df = pd.read_parquet(
+            validation_set_path,
+            columns=["group", "id", "md5", "rating", "sample_url", "image_url"],
+        )
+        classifier_scores_df = pd.read_parquet(classifier_scores_path)
+        assert {"classifier", "md5", "classifier_score", "percentile"}.issubset(classifier_scores_df.columns), "classifier_scores.parquet missing expected columns"
+        classifier_scores_df = classifier_scores_df[["classifier", "md5", "classifier_score", "percentile"]]
+        classifier_scores_df["classifier"] = classifier_scores_df["classifier"].astype(str)
+        classifier_scores_df["md5"] = classifier_scores_df["md5"].astype(str)
+        validation_df["md5"] = validation_df["md5"].astype(str)
+        _explorer_df = validation_df.merge(classifier_scores_df, on="md5", how="left", validate="one_to_many")
+        _stats_last_loaded_at = now
+
+
+_reload_stats_if_due(force=True)
+
+def _pool_fetch_pair(group_name: str) -> tuple:
+    gdf = _pool_group_dfs[group_name]
+    assert len(gdf) >= 2, f"Not enough rows for group: {group_name}"
+    md5_keys = gdf["md5"].astype(str)
+    wins = md5_keys.map(lambda k: _stats_by_key.get(k, (0, 0))[0])
+    losses = md5_keys.map(lambda k: _stats_by_key.get(k, (0, 0))[1])
+
+    def _pick_from_mask(mask: pd.Series):
+        candidate_df = gdf[mask]
+        if len(candidate_df) < 2:
+            return None
+        sample = candidate_df.sample(2, replace=False)
+        return sample.iloc[0], sample.iloc[1]
+    
+    
+    # 1) Repeat the lowest-margin edge participating in a cycle. (To prevent deadlock, stop if all margins are 4+.)
+    #   a) If deadlocked on a cycle with 4+ images and no inner cycles, sample a random missing edge inside the cycle.
+    # 2) Pair images that both have wins only . (One of them will lose/tie. Stop when there is only one left.)
+    # 3) Pair images that both have losses only. (One of them will win/tie. Stop when there is only one left.)
+    # 4) Pair images with only 2 edges.
+    # 5) X% chance, re-sample an existing edge, inversely proportional to existing number of samples.
+    # 6) Y% chance, sample a random missing edge between images already sampled.
+    # 7) Pair an unsampled image with a random sampled image. 
+
+    # 2) Pair images that currently have wins-only records.
+    picked = _pick_from_mask((wins > 0) & (losses == 0))
+    if picked is not None:
+        return picked[0], picked[1], "wins-only"
 
-        _disputed_queues = new_disputed
+    # 3) Pair images that currently have losses-only records.
+    picked = _pick_from_mask((wins == 0) & (losses > 0))
+    if picked is not None:
+        return picked[0], picked[1], "losses-only"
 
-    classifier_scores_path = hf_hub_download(
-        repo_id=POOL_REPO_ID,
-        filename="classifier_scores.parquet",
-        repo_type="dataset",
-        token=RATINGS_APP_TOKEN,
-    )
-    test_set_path = hf_hub_download(
-        repo_id=POOL_REPO_ID,
-        filename="test_set.parquet",
-        repo_type="dataset",
-        token=RATINGS_APP_TOKEN,
-    )
-    test_df = _drop_cdn_unavailable(
-        pd.read_parquet(
-            test_set_path,
-            columns=["group", "id", "md5", "rating", "sample_url", "image_url", "cdn_available"],
-        ),
-        "test set",
-    ).drop(columns="cdn_available")
-
-    classifier_scores_df = pd.read_parquet(classifier_scores_path)
-    assert {"classifier", "md5", "classifier_score", "percentile"}.issubset(classifier_scores_df.columns), "classifier_scores.parquet missing expected columns"
-    classifier_scores_df = classifier_scores_df[["classifier", "md5", "classifier_score", "percentile"]]
-    classifier_scores_df["classifier"] = classifier_scores_df["classifier"].astype(str)
-    classifier_scores_df["md5"] = classifier_scores_df["md5"].astype(str)
-    test_df["md5"] = test_df["md5"].astype(str)
-
-    global _explorer_df
-    _explorer_df = test_df.merge(classifier_scores_df, on="md5", how="left", validate="one_to_many")
-
-def _stats_reloader() -> None:
-    while True:
-        time.sleep(STATS_RELOAD_S)
-        _load_stats()
-
-_load_stats()
-threading.Thread(target=_stats_reloader, daemon=True).start()
-
-def _pick_from(df: pd.DataFrame, *, weights: pd.Series | None = None) -> tuple[pd.Series, pd.Series, int] | None:
-    if len(df) < 2:
-        return None
-
-    sample = df.sample(2, weights=weights, replace=False)
-    return sample.iloc[0], sample.iloc[1], len(df)
-
-def _pick_similar(
-    df: pd.DataFrame,
-    distance: Callable[[pd.DataFrame, pd.Series], pd.Series],
-    *,
-    weights: Callable[[pd.DataFrame], pd.Series] | None = None,
-    other_df: pd.DataFrame | None = None,
-) -> tuple[pd.Series, pd.Series, int] | None:
-    if len(df) < 2:
-        return None
-
-    if other_df is None:
-        other_df = df
-    elif len(other_df) < 2:
-        return None
-
-    weight_vals: pd.Series | None = None
-    if weights is not None:
-        weight_vals = weights(df)
-
-    first = df.sample(weights=weight_vals).iloc[0]
-    weight_vals = 1.0 / (1.0 + distance(other_df, first))
-
-    while True:
-        other = other_df.sample(weights=weight_vals).iloc[0]
-        if other["md5"] != first["md5"]:
-            return first, other, len(df)
-
-def _pool_fetch_pair(group: str) -> tuple[pd.Series, pd.Series, int, str]:
-    queue = _disputed_queues.get(group)
-    if queue and random.random() < 0.33:
-        try:
-            idx_a, idx_b, _ = queue.pop()
-        except IndexError:
-            pass
+    # 4) Pair images that currently have exactly 2 total edges.
+    vote_totals = wins + losses
+    picked = _pick_from_mask(vote_totals == 2)
+    if picked is not None:
+        return picked[0], picked[1], "total_votes=2"
+
+    # 7) Prefer pairing an unsampled image with a random previously sampled image.
+    unsampled_mask = vote_totals == 0
+    if unsampled_mask.any():
+        unsampled_row = gdf[unsampled_mask].sample(1).iloc[0]
+        sampled_df = gdf[~unsampled_mask]
+        if len(sampled_df) >= 1:
+            sampled_row = sampled_df.sample(1).iloc[0]
         else:
-            return _pool_df.iloc[idx_a], _pool_df.iloc[idx_b], len(queue), "disputed"
+            sampled_row = gdf.drop(index=unsampled_row.name).sample(1).iloc[0]
+        return unsampled_row, sampled_row, "unsampled+sampled"
 
-    gdf = _pool_df[_pool_df["group"] == group]
-    voted = gdf[gdf["votes"] > 0]
-    votes = voted["votes"]
+    # 8) Safety fall back to low-vote weighted sampling.
+    sample_weights = 1.0 / (vote_totals + 1.0)
+    sample = gdf.sample(2, weights=sample_weights, replace=False)
+    return sample.iloc[0], sample.iloc[1], "low-vote"
 
-    # Pair first-time winners.
-    picked = _pick_from(voted[(votes == 1) & (voted["wins"] == 1)])
-    if picked is not None:
-        return *picked, "new-winners"
-
-    # Pair first-time losers.
-    picked = _pick_from(voted[(votes == 1) & (voted["losses"] == 1)])
-    if picked is not None:
-        return *picked, "new-losers"
-
-    def record_distance(df: pd.DataFrame, pivot: pd.Series) -> pd.Series:
-        return (
-            (df["wins"] - pivot["wins"])**2 +
-            (df["losses"] - pivot["losses"])**2
-        )**0.75 # L2 is a bit too loose
-
-    # Link cliques to main network and break ties.
-    nonties = votes - voted["ties"]
-    picked = _pick_similar(
-        voted[(nonties == 0) | (votes == 2)],
-        record_distance,
-        other_df=voted[nonties > 3],
-    )
-    if picked is not None:
-        return *picked, "sparse"
-
-    # Introduce new images.
-    if len(voted) < 8 or random.random() < 0.2:
-        unvoted = gdf[gdf["votes"] == 0]
-        match len(unvoted):
-            case 0:
-                pass
-            case 1:
-                return unvoted.iloc[0], voted.iloc[0], 1, "new"
-            case _:
-                picked = _pick_from(unvoted)
-                assert picked is not None
-                return *picked, "new"
-
-    # Vote-weighted random sampling between similar winrates, slighlty biased against picking losers.
-    picked = _pick_similar(
-        voted, record_distance,
-        weights=lambda df: 1.0 / (df["votes"]**1.25 + 0.1 * df["losses"]),
-    )
-    assert picked is not None
-    return *picked, "fair-probe"
 
 def _row_image_url(row) -> str:
     sample_url = row.get("sample_url")
@@ -272,56 +148,89 @@ DATASETS: dict[str, dict] = {
         "fetch_pair": _pool_fetch_pair,
         "get_id": lambda row: row["md5"],
         "get_image": _row_image_url,
-        "groups": sorted(_pool_df["group"].unique()),
+        "groups": {g: g for g in sorted(_pool_df["group"].unique())},
     },
 }
 DEFAULT_DATASET = list(DATASETS.keys())[0]
 
-def _format_rating_post_title(post_id: int, votes: int, label: str) -> str:
-    return f"<strong>{label}</strong>: <a href=\"https://e621.net/posts/{post_id}\" target=\"_blank\" rel=\"noreferrer\">Post #{post_id}</a> | {votes} {'Vote' if votes == 1 else 'Votes'}"
+def _select_groups(cfg: dict, rating_pref: str) -> list[str]:
+    groups = list(cfg["groups"].keys())
+    if rating_pref == "all":
+        return groups
+    return [g for g in groups if g.endswith(f"_{rating_pref}")]
+
+def _commit_oldest_pending(state: dict):
+    pending = state.setdefault("pending", [])
+    if len(pending) <= 1:
+        return
+    oldest = pending.pop(0)
+    if oldest.get("winner") in ("A", "B"):
+        _apply_local_stats_update(oldest["winner"], oldest["key_a"], oldest["key_b"])
+    threading.Thread(target=VOTE_STORAGE.append_vote_row, args=(oldest.copy(), oldest.get("winner")), daemon=True).start()
+
+
+def _apply_local_stats_update(winner: str, key_a: str, key_b: str):
+    assert winner in ("A", "B")
+    with _stats_lock:
+        wins_a, losses_a = _stats_by_key.get(str(key_a), (0, 0))
+        wins_b, losses_b = _stats_by_key.get(str(key_b), (0, 0))
+        if winner == "A":
+            _stats_by_key[str(key_a)] = (wins_a + 1, losses_a)
+            _stats_by_key[str(key_b)] = (wins_b, losses_b + 1)
+        else:
+            _stats_by_key[str(key_a)] = (wins_a, losses_a + 1)
+            _stats_by_key[str(key_b)] = (wins_b + 1, losses_b)
+
+
+def _format_rating_post_row(post_id: int, wins: int, losses: int, label: str | None = None) -> str:
+    total_votes = wins + losses
+    url = f"https://e621.net/posts/{post_id}"
+    row = f"{url} | Times rated: {total_votes}"
+    return f"{label}: {row}" if label else row
 
 def _render_current(state: dict, submit_status: str = "") -> tuple:
-    votes_a = _pool_df.iloc[_md5_to_idx[state["key_a"]], VOTES_LOC]
-    votes_b = _pool_df.iloc[_md5_to_idx[state["key_b"]], VOTES_LOC]
-    title_a = _format_rating_post_title(state["id_a"], votes_a, "Image A")
-    title_b = _format_rating_post_title(state["id_b"], votes_b, "Image B")
-    img_a_html = f"<div class=\"rating-card\"><div class=\"rating-card-title\">{title_a}</div><div class=\"rating-image-frame\"><img src=\"{html.escape(state['url_a'])}\" class=\"rating-image\" loading=\"eager\"></div></div>"
-    img_b_html = f"<div class=\"rating-card\"><div class=\"rating-card-title\">{title_b}</div><div class=\"rating-image-frame\"><img src=\"{html.escape(state['url_b'])}\" class=\"rating-image\" loading=\"eager\"></div></div>"
+    _reload_stats_if_due()
+    wins_a, losses_a = _stats_by_key.get(str(state["key_a"]), (0, 0))
+    wins_b, losses_b = _stats_by_key.get(str(state["key_b"]), (0, 0))
+    title_a = "Image A"
+    title_b = "Image B"
+    img_a_html = f"<div class=\"rating-card\"><div class=\"rating-card-title\"><strong>{html.escape(title_a)}</strong></div><div class=\"rating-image-frame\"><img src=\"{html.escape(state['url_a'])}\" class=\"rating-image\" loading=\"eager\" referrerpolicy=\"no-referrer\"></div></div>"
+    img_b_html = f"<div class=\"rating-card\"><div class=\"rating-card-title\"><strong>{html.escape(title_b)}</strong></div><div class=\"rating-image-frame\"><img src=\"{html.escape(state['url_b'])}\" class=\"rating-image\" loading=\"eager\" referrerpolicy=\"no-referrer\"></div></div>"
+    link_a = _format_rating_post_row(state["id_a"], wins_a, losses_a, label="Image A")
+    link_b = _format_rating_post_row(state["id_b"], wins_b, losses_b, label="Image B")
+    can_go_back = bool(state.get("can_go_back"))
+    back_md = "[Undo Rating (Ctrl+z)](#back)" if can_go_back else "<span class='subtle-back-link-disabled'>Undo Rating (Ctrl+z)</span>"
+    group_md = f"<span class='subtle-note'>Group: {state['group']}</span>"
+    pair_reason = state.get("pair_reason", "")
+    pair_reason_md = f"<span class='subtle-note'>Pair: {html.escape(pair_reason)}</span>" if pair_reason else ""
+    status_md = f"<span class='submit-status-msg'>{html.escape(submit_status)}</span>" if submit_status else ""
+    return img_a_html, img_b_html, link_a, link_b, back_md, group_md, pair_reason_md, status_md, state
 
-    can_go_back = bool(state.get("pending", ()))
-    pair_details = f"/ {state['group']} / {state.get('pair_reason', 'unknown')} ({state.get('n_candidates', '?')})"
 
-    return img_a_html, img_b_html, gr.Button(interactive=can_go_back), html.escape(pair_details), _submit_status(submit_status), state
 
 def _normalize_rating_pref(pref: str | None) -> str:
     return pref if pref in ("safe", "all") else "safe"
 
-def _initial_load(state: dict, rating_pref: str | None, submit_key: str | None, rater_name: str | None, image_height: str, groups: list[str]):
-    rating_pref = _normalize_rating_pref(rating_pref)
+
+def _initial_load(state: dict, pref: str | None, submit_key: str | None):
+    rating_pref = _normalize_rating_pref(pref)
     submit_key = _normalize_submit_key(submit_key)
-    rater_name = _normalize_rater_name(rater_name)
-    return rating_pref, submit_key, rater_name, image_height, image_height, groups, *new_round(DEFAULT_DATASET, groups, state)
+    return rating_pref, submit_key, *new_round(DEFAULT_DATASET, rating_pref, state)
 
-def _on_groups_change(groups: list[str], state: dict):
-    return *new_round(DEFAULT_DATASET, groups, state), groups
 
-def _on_image_height_change(image_height: str) -> tuple[str, str]:
-    return image_height, image_height
+def _on_rating_change(rating_pref: str, state: dict):
+    rating_pref = _normalize_rating_pref(rating_pref)
+    return *new_round(DEFAULT_DATASET, rating_pref, state), rating_pref
 
-def _normalize_submit_key(submit_key: str | None) -> str:
-    return (submit_key or "").strip()
 
-def _normalize_rater_name(rater_name: str | None) -> str:
-    return (rater_name or "").strip()[:RATER_NAME_MAX_LEN]
+def _normalize_submit_key(submit_key: str | None) -> str:
+    return submit_key or ""
 
-def _submit_status(message: str = "") -> gr.HTML:
-    # "hidden" takes up no space but keeps the block mounted, so the first message
-    # to arrive renders immediately instead of on the next update.
-    return gr.HTML(value=message, visible=True if message else "hidden")
 
 def _filtered_explorer_df(rating_pref: str) -> pd.DataFrame:
     return _filtered_explorer_df_by_classifier(rating_pref, ALLOWED_CLASSIFIER_FILTERS[0])
 
+
 def _filtered_explorer_df_by_classifier(rating_pref: str, classifier_name: str) -> pd.DataFrame:
     if rating_pref == "all":
         rating_filtered = _explorer_df
@@ -331,10 +240,12 @@ def _filtered_explorer_df_by_classifier(rating_pref: str, classifier_name: str)
     assert classifier_name in ALLOWED_CLASSIFIER_FILTERS, f"Unsupported classifier filter: {classifier_name}"
     return rating_filtered[rating_filtered["classifier"] == classifier_name]
 
-def _load_results(rating_pref: str, sort_mode: str, classifier_filter: str):
-    rating_pref = _normalize_rating_pref(rating_pref)
-    sort_mode = _normalize_sort_mode(sort_mode)
-    classifier_name = _normalize_classifier_filter(classifier_filter)
+
+def _load_results(rating_pref_value: str, sort_mode_value: str, classifier_filter_value: str):
+    rating_pref = _normalize_rating_pref(rating_pref_value)
+    sort_mode = _normalize_sort_mode(sort_mode_value)
+    classifier_name = _normalize_classifier_filter(classifier_filter_value)
+    _reload_stats_if_due()
     filtered_explorer_df = _filtered_explorer_df_by_classifier(rating_pref, classifier_name)
     summary, score_distribution_plot, distribution_data, gallery_items, page_meta, next_offset, btn_update = build_results_data(
         filtered_explorer_df,
@@ -345,10 +256,12 @@ def _load_results(rating_pref: str, sort_mode: str, classifier_filter: str):
     )
     return summary, score_distribution_plot, distribution_data, gallery_items, btn_update, "Click an image to reveal its ID and link.", page_meta, next_offset
 
+
 def _normalize_sort_mode(sort_mode: str | None) -> str:
-    if sort_mode in SORT_MODES:
+    if sort_mode in ("Default", "Rating: Low to High", "Rating: High to Low"):
         return sort_mode
-    return DISAGREEMENT_SORT_MODE
+    return "Default"
+
 
 def _normalize_classifier_filter(classifier_name: str | None) -> str:
     if classifier_name in ALLOWED_CLASSIFIER_FILTERS:
@@ -357,36 +270,33 @@ def _normalize_classifier_filter(classifier_name: str | None) -> str:
 
 # -- Gradio callbacks -------------------------------------------------------
 
-def new_round(dataset_name: str, groups: list[str], state: dict) -> tuple:
-    if not groups:
-        return "", "", gr.skip(), "", _submit_status("Please select at least one group."), state
-
+def new_round(dataset_name: str, rating_pref: str, state: dict) -> tuple:
     cfg = DATASETS[dataset_name]
+    groups = _select_groups(cfg, rating_pref)
+    assert groups, f"No groups for rating preference: {rating_pref}"
     group = random.choice(groups)
-    # n_candidates: how many candidates the sampler branch had to draw from
-    # (images for the cascade branches, pairs left for the disputed queue).
-    # Display-only — shown next to pair_reason in the UI, never logged.
-    row_a, row_b, n_candidates, pair_reason = cfg["fetch_pair"](group)
-
+    pair_data = cfg["fetch_pair"](cfg["groups"][group])
+    if len(pair_data) == 3:
+        row_a, row_b, pair_reason = pair_data
+    else:
+        row_a, row_b = pair_data
+        pair_reason = ""
     state.setdefault("session_id", uuid.uuid4().hex)
-    state["n_candidates"] = n_candidates
     key_a = cfg["get_id"](row_a)
     key_b = cfg["get_id"](row_b)
     id_a = int(row_a["id"])
     id_b = int(row_b["id"])
-    state.update(dataset=dataset_name, key_a=key_a, key_b=key_b, id_a=id_a, id_b=id_b, group=group, pair_reason=pair_reason)
+    state.update(dataset=dataset_name, rating_pref=rating_pref, key_a=key_a, key_b=key_b, id_a=id_a, id_b=id_b, group=group, pair_reason=pair_reason)
     url_a = cfg["get_image"](row_a)
     url_b = cfg["get_image"](row_b)
     state["url_a"] = url_a
     state["url_b"] = url_b
     return _render_current(state)
 
-def _queue_decision(winner: str | None, state: dict, rater_name: str):
+def _queue_decision(winner: str | None, state: dict):
     assert state.get("session_id"), "Missing session_id: refusing to record vote"
-
-    pending = state.setdefault("pending", [])
-    pending.append({
-        "rater_name": rater_name,
+    state.setdefault("pending", [])
+    decision = {
         "winner": winner,
         "key_a": state["key_a"],
         "key_b": state["key_b"],
@@ -395,92 +305,53 @@ def _queue_decision(winner: str | None, state: dict, rater_name: str):
         "url_a": state["url_a"],
         "url_b": state["url_b"],
         "dataset": state["dataset"],
+        "rating_pref": state["rating_pref"],
         "group": state["group"],
         "pair_reason": state.get("pair_reason", ""),
         "session_id": state["session_id"],
-    })
-
-    if len(pending) > 1:
-        VOTE_STORAGE.queue_row(pending.pop(0))
-
-def _add_vote(idx: int, col_loc: int, delta: int = 1) -> None:
-    _pool_df.iloc[idx, [col_loc, VOTES_LOC]] += delta
-
-    wins, ties, votes = _pool_df.iloc[idx, [WINS_LOC, TIES_LOC, VOTES_LOC]]
-    _pool_df.iloc[idx, WINRATE_LOC] = (wins + 0.5 * ties) / max(votes, 1)
-
-def vote(winner: str | None, state: dict, groups: list[str], submit_key: str | None, rater_name: str | None) -> tuple:
-    submit_key = _normalize_submit_key(submit_key)
-    if submit_key != SUBMIT_KEY:
-        headline = "That submit key isn't valid." if submit_key else "A submit key is required to record votes."
-        return _render_current(state, (
-            f"<strong>{headline}</strong> Nothing you vote on is being saved. "
-            "Enter a working key under <em>Settings → Submit Key</em> — ask for one on "
-            "<a href='https://discord.com/channels/1019133813105905664/1531769333196259499/1531769333196259499' target='_blank' rel='noopener noreferrer'>Discord</a>."
-        ))
-
-    if not groups:
-        return "", "", gr.skip(), "", _submit_status("Please select at least one group."), state
-
-    _queue_decision(winner, state, _normalize_rater_name(rater_name))
-
-    a_idx = _md5_to_idx[state["key_a"]]
-    b_idx = _md5_to_idx[state["key_b"]]
-    with _pool_lock:
-        match winner:
-            case "A":
-                _add_vote(a_idx, WINS_LOC)
-                _add_vote(b_idx, LOSSES_LOC)
-            case "B":
-                _add_vote(a_idx, LOSSES_LOC)
-                _add_vote(b_idx, WINS_LOC)
-            case None:
-                _add_vote(a_idx, TIES_LOC)
-                _add_vote(b_idx, TIES_LOC)
-            case _:
-                raise AssertionError
-
-    return new_round(state["dataset"], groups, state)
+    }
+    state["pending"].append(decision)
+    state["last_decision"] = decision
+    state["can_go_back"] = True
+    _commit_oldest_pending(state)
+
+def vote(winner: str | None, state: dict, submit_key: str | None) -> tuple:
+    assert winner in ("A", "B", None)
+    if _normalize_submit_key(submit_key) != SUBMIT_KEY:
+        return _render_current(state, "Wrong submission key.")
+    _queue_decision(winner, state)
+    return new_round(state["dataset"], state["rating_pref"], state)
 
 def go_back(state: dict) -> tuple:
     pending = state.setdefault("pending", [])
-
-    if pending:
-        last = pending.pop()
-        state.update(
-            dataset=last["dataset"],
-            key_a=last["key_a"],
-            key_b=last["key_b"],
-            id_a=last["id_a"],
-            id_b=last["id_b"],
-            url_a=last["url_a"],
-            url_b=last["url_b"],
-            group=last["group"],
-            pair_reason=last.get("pair_reason", ""),
-        )
-
-        a_idx = _md5_to_idx[state["key_a"]]
-        b_idx = _md5_to_idx[state["key_b"]]
-        with _pool_lock:
-            match last["winner"]:
-                case "A":
-                    _add_vote(a_idx, WINS_LOC, -1)
-                    _add_vote(b_idx, LOSSES_LOC, -1)
-                case "B":
-                    _add_vote(a_idx, LOSSES_LOC, -1)
-                    _add_vote(b_idx, WINS_LOC, -1)
-                case None:
-                    _add_vote(a_idx, TIES_LOC, -1)
-                    _add_vote(b_idx, TIES_LOC, -1)
-                case _:
-                    raise AssertionError
-
+    if not state.get("can_go_back"):
+        return _render_current(state)
+    last = state.get("last_decision")
+    if not last:
+        state["can_go_back"] = False
+        return _render_current(state)
+    if pending and pending[-1] == last:
+        pending.pop()
+    state["can_go_back"] = False
+    state["last_decision"] = None
+    state.update(
+        dataset=last["dataset"],
+        rating_pref=last["rating_pref"],
+        key_a=last["key_a"],
+        key_b=last["key_b"],
+        id_a=last["id_a"],
+        id_b=last["id_b"],
+        url_a=last["url_a"],
+        url_b=last["url_b"],
+        group=last["group"],
+        pair_reason=last.get("pair_reason", ""),
+    )
     return _render_current(state)
 
 # -- UI ---------------------------------------------------------------------
 
 with gr.Blocks(
-    title="e621 Visual Ratings",
+    title="Image Rater",
     head="""
     <script>
     const VOTE_COOLDOWN_MS = 1500;
@@ -563,15 +434,9 @@ with gr.Blocks(
         } else if ((e.key === 'z' || e.key === 'Z') && (e.ctrlKey || e.metaKey) && ratingTabActive) {
             e.preventDefault();
             backButton?.click();
-        } else if (e.key === 'ArrowDown') {
-            if (ratingTabActive) {
-                e.preventDefault();
-                backButton?.click();
-            }
-            if (resultsTabActive) {
-                e.preventDefault();
-                resultsLoadMoreButton?.click();
-            }
+        } else if (e.key === 'ArrowDown' && resultsTabActive) {
+            e.preventDefault();
+            resultsLoadMoreButton?.click();
         }
     });
     document.addEventListener('click', function (e) {
@@ -631,6 +496,7 @@ with gr.Blocks(
     }
     .rating-image-frame {
         width: 100%;
+        height: 512px;
         border: 1px solid #e6e6e6;
         border-radius: 8px;
         background: #333;
@@ -663,18 +529,21 @@ with gr.Blocks(
     .hidden-action-btn {
         display: none !important;
     }
+    #submit-status {
+        position: fixed;
+        left: 50%;
+        bottom: 20px;
+        transform: translateX(-50%);
+        z-index: 9998;
+        pointer-events: none;
+        min-height: 1.2em;
+    }
     .submit-status-msg {
-        border: 1px solid #f0b429;
-        border-left-width: 4px;
-        background: rgba(240, 180, 41, 0.12);
+        background: rgba(20, 20, 20, 0.92);
+        color: #fff;
+        padding: 8px 12px;
         border-radius: 8px;
-        padding: 10px 14px;
-        font-size: 0.95rem;
-        line-height: 1.45;
-    }
-    .submit-status-msg a {
-        text-decoration: underline;
-        font-weight: 600;
+        font-size: 0.92rem;
     }
     #results-gallery {
         --explorer-thumb-ratio: 1 / 1;
@@ -689,77 +558,54 @@ with gr.Blocks(
         object-fit: contain !important;
         background: #1f2937;
     }
-    a {
-        padding: 0 !important;
-    }
     """,
-    fill_width=True,
 ) as demo:
     state = gr.State({})
     rating_pref_store = gr.BrowserState(default_value="safe", storage_key="rating_pref")
     submit_key_store = gr.BrowserState(default_value="", storage_key="submit_key")
-    rater_name_store = gr.BrowserState(default_value="", storage_key="rater_name")
-    results_sort_store = gr.BrowserState(default_value=DISAGREEMENT_SORT_MODE, storage_key="results_sort_mode")
+    results_sort_store = gr.BrowserState(default_value="Default", storage_key="results_sort_mode")
     results_classifier_store = gr.BrowserState(default_value=ALLOWED_CLASSIFIER_FILTERS[0], storage_key="results_classifier")
-    image_height_store = gr.BrowserState(default_value=768, storage_key="image_height")
-    groups_store = gr.BrowserState(default_value=[
-        group
-        for group in DATASETS[DEFAULT_DATASET]["groups"]
-        if group.endswith("_safe")
-    ], storage_key="groups")
 
     with gr.Tabs():
         with gr.Tab("Image Quality Rater"):
             gr.Markdown("Rate relative image quality. Choose the image with better quality, or select same quality if they are comparable. Both images are drawn from the same group to avoid cross-group bias.")
-            submit_status = gr.HTML(html_template="<div class='submit-status-msg'>${value}</div>", elem_id="submit-status", visible="hidden")
 
             with gr.Row():
                 img_a = gr.HTML(elem_id="img-a")
                 img_b = gr.HTML(elem_id="img-b")
 
-            with gr.Row(equal_height=True):
-                btn_a = gr.Button("⬅️ Prefer A", variant="primary", elem_id="btn-vote-a")
-
-                with gr.Column(scale=0), gr.Group():
-                    btn_skip = gr.Button("⬆️ Same Quality", elem_id="btn-skip")
-                    btn_back_action = gr.Button("⬇️ Undo", elem_id="btn-back-action")
-
-                btn_b = gr.Button("➡️ Prefer B", variant="primary", elem_id="btn-vote-b")
+            with gr.Row():
+                btn_a = gr.Button("👍 Prefer A", variant="primary", elem_id="btn-vote-a")
+                btn_skip = gr.Button("Same quality", elem_id="btn-skip")
+                btn_b = gr.Button("👍 Prefer B", variant="primary", elem_id="btn-vote-b")
 
             with gr.Accordion("Settings", open=False):
-                groups_select = gr.CheckboxGroup(
-                    choices=DATASETS[DEFAULT_DATASET]["groups"],
-                    label="Categories",
-                    show_label=True,
-                    show_select_all=True
-                )
-                image_height_slider = gr.Slider(
-                    minimum=512, maximum=2048, step=16, precision=0,
-                    label="Image Size",
+                gr.Markdown("<span style='color:#888;font-size:0.9em;'>Advanced options</span>")
+                rating_dd = gr.Dropdown(
+                    choices=["safe", "all"],
+                    value="safe",
+                    label="Rating",
+                    elem_id="rating-pref",
                 )
                 submit_key_tb = gr.Textbox(
                     value="",
                     type="password",
-                    label="Submit Key",
+                    label="Submit key",
                     elem_id="submit-key",
                 )
-                rater_name_tb = gr.Textbox(
-                    value="",
-                    label="Rater Name",
-                    elem_id="rater-name",
-                    max_length=RATER_NAME_MAX_LEN,
-                    info=(
-                        "Entirely optional. Leave this blank to rate anonymously"
-                    ),
-                )
 
-            pair_details = gr.HTML(html_template="Dataset: <a href='https://huggingface.co/datasets/taigasan/e6-visual-ratings' target='_blank' rel='noopener noreferrer'>taigasan/e6-visual-ratings</a> ${value}")
-            gr.HTML("<span class='subtle-note'>Keyboard Shortcuts: ⬅️ Vote A, ⬆️ Same Quality, ➡️ Vote B, ⬇️ or Ctrl+Z Undo</span>")
-            image_height = gr.HTML(html_template="<style>.rating-image-frame { height:${value}px; }</style>", apply_default_css=False)
+            link_a = gr.Markdown(label="Image A link")
+            link_b = gr.Markdown(label="Image B link")
+            back_link = gr.Markdown(elem_classes=["subtle-back-link-wrap"])
+            btn_back_action = gr.Button("Undo Rating (Ctrl+z)", elem_id="btn-back-action", elem_classes=["hidden-action-btn"])
+            details_md = gr.Markdown()
+            pair_reason_md = gr.Markdown()
+            submit_status_md = gr.Markdown(elem_id="submit-status")
+            gr.Markdown("<span class='subtle-note'>Dataset: <a href='https://huggingface.co/datasets/taigasan/e6-visual-ratings' target='_blank' rel='noopener noreferrer'>taigasan/e6-visual-ratings</a></span>")
+            gr.Markdown("<span class='subtle-note'>Keyboard Shortcuts: ⬅️ vote A, ⬆️ same quality, ➡️ vote B, Ctrl+z undo rating</span>")
 
         (
             results_summary_md,
-            results_rating_dd,
             results_sort_dd,
             results_classifier_dd,
             results_score_distribution_plot,
@@ -771,7 +617,7 @@ with gr.Blocks(
             results_page_offset_state,
         ) = add_results_tab(_pool_df)
 
-    outputs = [img_a, img_b, btn_back_action, pair_details, submit_status, state]
+    outputs = [img_a, img_b, link_a, link_b, back_link, details_md, pair_reason_md, submit_status_md, state]
     results_outputs = [
         results_summary_md,
         results_score_distribution_plot,
@@ -783,29 +629,22 @@ with gr.Blocks(
         results_page_offset_state,
     ]
 
-    vote_inputs = [state, groups_store, submit_key_store, rater_name_store]
-    btn_a.click(fn=lambda s, g, k, n: vote("A", s, g, k, n), inputs=vote_inputs, outputs=outputs, queue=False, show_progress="hidden")
-    btn_b.click(fn=lambda s, g, k, n: vote("B", s, g, k, n), inputs=vote_inputs, outputs=outputs, queue=False, show_progress="hidden")
-    btn_skip.click(fn=lambda s, g, k, n: vote(None, s, g, k, n), inputs=vote_inputs, outputs=outputs, queue=False, show_progress="hidden")
+    btn_a.click(fn=lambda s, k: vote("A", s, k), inputs=[state, submit_key_store], outputs=outputs, queue=False, show_progress="hidden")
+    btn_b.click(fn=lambda s, k: vote("B", s, k), inputs=[state, submit_key_store], outputs=outputs, queue=False, show_progress="hidden")
+    btn_skip.click(fn=lambda s, k: vote(None, s, k), inputs=[state, submit_key_store], outputs=outputs, queue=False, show_progress="hidden")
     btn_back_action.click(fn=go_back, inputs=[state], outputs=outputs, queue=False, show_progress="hidden")
-    # Clear the warning once the key is right; leave it alone mid-typing.
-    submit_key_tb.change(fn=lambda k: (_normalize_submit_key(k), _submit_status() if _normalize_submit_key(k) == SUBMIT_KEY else gr.skip()), inputs=[submit_key_tb], outputs=[submit_key_store, submit_status], queue=False, show_progress="hidden")
-    rater_name_tb.change(fn=_normalize_rater_name, inputs=[rater_name_tb], outputs=[rater_name_store], queue=False, show_progress="hidden")
-    groups_select.change(fn=_on_groups_change, inputs=[groups_select, state], outputs=[*outputs, groups_store], queue=False, show_progress="hidden")
-    image_height_slider.change(fn=_on_image_height_change, inputs=[image_height_slider], outputs=[image_height_store, image_height], queue=False, show_progress="hidden")
-
-    results_rating_dd.change(fn=_normalize_rating_pref, inputs=[results_rating_dd], outputs=[rating_pref_store], queue=False, show_progress="hidden")
-    results_rating_dd.change(fn=_load_results, inputs=[results_rating_dd, results_sort_store, results_classifier_store], outputs=results_outputs, queue=False, show_progress="hidden")
+    rating_dd.change(fn=_on_rating_change, inputs=[rating_dd, state], outputs=[*outputs, rating_pref_store], queue=False, show_progress="hidden")
+    submit_key_tb.input(fn=_normalize_submit_key, inputs=[submit_key_tb], outputs=[submit_key_store], queue=False, show_progress="hidden")
+    submit_key_tb.change(fn=_normalize_submit_key, inputs=[submit_key_tb], outputs=[submit_key_store], queue=False, show_progress="hidden")
+    rating_dd.change(fn=_load_results, inputs=[rating_dd, results_sort_store, results_classifier_store], outputs=results_outputs, queue=False, show_progress="hidden")
     results_sort_dd.change(fn=_normalize_sort_mode, inputs=[results_sort_dd], outputs=[results_sort_store], queue=False, show_progress="hidden")
     results_sort_dd.change(fn=_load_results, inputs=[rating_pref_store, results_sort_dd, results_classifier_store], outputs=results_outputs, queue=False, show_progress="hidden")
     results_classifier_dd.change(fn=_normalize_classifier_filter, inputs=[results_classifier_dd], outputs=[results_classifier_store], queue=False, show_progress="hidden")
     results_classifier_dd.change(fn=_load_results, inputs=[rating_pref_store, results_sort_store, results_classifier_dd], outputs=results_outputs, queue=False, show_progress="hidden")
-
-    demo.load(fn=_initial_load, inputs=[state, rating_pref_store, submit_key_store, rater_name_store, image_height_store, groups_store], outputs=[results_rating_dd, submit_key_tb, rater_name_tb, image_height_slider, image_height, groups_select, *outputs], queue=False, show_progress="hidden")
+    demo.load(fn=_initial_load, inputs=[state, rating_pref_store, submit_key_store], outputs=[rating_dd, submit_key_tb, *outputs], queue=False, show_progress="hidden")
     demo.load(fn=_load_results, inputs=[rating_pref_store, results_sort_store, results_classifier_store], outputs=results_outputs, queue=False, show_progress="hidden")
     demo.load(fn=_normalize_sort_mode, inputs=[results_sort_store], outputs=[results_sort_dd], queue=False, show_progress="hidden")
     demo.load(fn=_normalize_classifier_filter, inputs=[results_classifier_store], outputs=[results_classifier_dd], queue=False, show_progress="hidden")
-
     results_load_more_btn.click(
         fn=lambda r, s, c, o: load_more_results(_filtered_explorer_df_by_classifier(_normalize_rating_pref(r), _normalize_classifier_filter(c)), _explorer_df, s, o),
         inputs=[rating_pref_store, results_sort_store, results_classifier_store, results_page_offset_state],

maintenance/compact_logs.py → compact_logs.py

@@ -1,20 +1,21 @@
-"""Merge the append-only vote shards into one. Run: python -m maintenance.compact_logs"""
+#!/usr/bin/env python3
 from __future__ import annotations
 
 import os
 import time
 import uuid
+from pathlib import Path
+from tempfile import NamedTemporaryFile
 
 import pandas as pd
 from huggingface_hub import CommitOperationAdd, CommitOperationDelete, HfApi, hf_hub_download
 
-from storage import VOTE_COLUMNS
-
-VOTES_REPO_ID = "RedRocket/e6-visual-ratings"
+VOTES_REPO_ID = "taigasan/e6-visual-ratings"
 VOTES_REPO_TYPE = "dataset"
 VOTES_LOG_SUBDIR = "ratings_log"
 RATINGS_APP_TOKEN_ENV = "RATINGS_APP_TOKEN"
 
+
 def _list_vote_shards(api: HfApi) -> list[str]:
     files = api.list_repo_files(repo_id=VOTES_REPO_ID, repo_type=VOTES_REPO_TYPE)
     shard_prefix = f"{VOTES_LOG_SUBDIR}/votes_"
@@ -25,19 +26,21 @@ def _list_vote_shards(api: HfApi) -> list[str]:
         and f.endswith(".parquet")
     )
 
+
 def _new_compacted_shard_path() -> str:
     ts = int(time.time())
     return f"{VOTES_LOG_SUBDIR}/votes_{ts}_{uuid.uuid4().hex}.parquet"
 
-def compact_votes() -> tuple[int, int, str] | None:
+
+def compact_votes() -> tuple[int, int, str]:
     token = os.getenv(RATINGS_APP_TOKEN_ENV)
     api = HfApi(token=token)
 
     shards = _list_vote_shards(api)
-    if len(shards) < 2:
-        return None
+    if not shards:
+        raise FileNotFoundError(f"No vote shards found in {VOTES_REPO_ID}/{VOTES_LOG_SUBDIR}")
 
-    frames: list[pd.DataFrame] = []
+    frames = []
     for shard in shards:
         shard_local = hf_hub_download(
             repo_id=VOTES_REPO_ID,
@@ -45,47 +48,47 @@ def compact_votes() -> tuple[int, int, str] | None:
             repo_type=VOTES_REPO_TYPE,
             token=token,
         )
-        frames.append(pd.read_parquet(shard_local))
+        frame = pd.read_parquet(shard_local)
+        frames.append(frame)
 
+    input_row_count = sum(len(frame) for frame in frames)
     combined = pd.concat(frames, ignore_index=True, sort=False)
+    output_row_count = int(len(combined))
+    if output_row_count != input_row_count:
+        raise RuntimeError(
+            f"Refusing to commit: row mismatch during compaction "
+            f"({input_row_count} -> {output_row_count})."
+        )
 
-    # Older shards may predate newly added vote columns; normalize the same
-    # way VoteStorage does on upload.
-    for col in VOTE_COLUMNS:
-        if col not in combined.columns:
-            combined[col] = None
-
-    combined = combined[VOTE_COLUMNS]
-
-    output_row_count = len(combined)
-    assert len(combined) == sum(len(frame) for frame in frames)
-
-    compacted_data = combined.to_parquet(index=False)
-    compacted_path = _new_compacted_shard_path()
-    api.create_commit(
-        repo_id=VOTES_REPO_ID,
-        repo_type=VOTES_REPO_TYPE,
-        commit_message=f"compact {len(shards)} vote shards",
-        operations=[
-            CommitOperationAdd(path_or_fileobj=compacted_data, path_in_repo=compacted_path),
-            *(CommitOperationDelete(path_in_repo=shard) for shard in shards),
-        ],
-    )
+    with NamedTemporaryFile(suffix=".parquet", delete=False) as tmp:
+        tmp_path = Path(tmp.name)
+    try:
+        combined.to_parquet(tmp_path, index=False)
+        compacted_path = _new_compacted_shard_path()
+        operations = [
+            CommitOperationAdd(path_or_fileobj=str(tmp_path), path_in_repo=compacted_path),
+            *[CommitOperationDelete(path_in_repo=shard) for shard in shards],
+        ]
+        api.create_commit(
+            repo_id=VOTES_REPO_ID,
+            repo_type=VOTES_REPO_TYPE,
+            commit_message=f"compact {len(shards)} vote shard(s)",
+            operations=operations,
+        )
+    finally:
+        tmp_path.unlink(missing_ok=True)
 
-    return len(shards), len(combined), compacted_path
+    return len(shards), output_row_count, compacted_path
 
-def _main() -> None:
-    result = compact_votes()
-    if result is None:
-        print(f"Nothing to compact.")
-        return
 
-    shard_count, row_count, compacted_path = result
+def main() -> None:
+    shard_count, row_count, compacted_path = compact_votes()
     print(
-        f"Compacted {shard_count} shards into "
+        f"Compacted {shard_count} shard(s) into "
         f"{VOTES_REPO_ID}/{compacted_path} "
         f"with {row_count} rows."
     )
 
+
 if __name__ == "__main__":
-    _main()
+    main()

explorer.py

@@ -3,13 +3,11 @@ import matplotlib.pyplot as plt
 import numpy as np
 import pandas as pd
 import html
-from matplotlib.colors import LinearSegmentedColormap
 from matplotlib.figure import Figure
 
 PAGE_SIZE = 30
-DISAGREEMENT_SORT_MODE = "Classifier Disagreement: High to Low"
-SORT_MODES = [DISAGREEMENT_SORT_MODE, "Rating: Low to High", "Rating: High to Low"]
-ALLOWED_CLASSIFIER_FILTERS = ("Siglip + Hydra head", "TaggerExperiment", "PairsExperiment backbone + Hydra head")
+SORT_MODES = ["Default", "Rating: Low to High", "Rating: High to Low"]
+ALLOWED_CLASSIFIER_FILTERS = ("SiglipFinetune", "ScoringHead-v5", "TaggerExperiment")
 
 
 def _row_image_url(row) -> str | None:
@@ -34,14 +32,10 @@ def _gallery_items(meta: list[dict]) -> list[tuple[str, str]]:
 
 def _distribution_data(
     explorer_df: pd.DataFrame,
-    classifier_name: str | None = None,
 ) -> dict[str, list[float] | int]:
-    scores_df = explorer_df
-    if classifier_name is not None:
-        scores_df = scores_df[scores_df["classifier"].astype(str) == classifier_name]
     valid_scores = [
         float(score)
-        for score in scores_df["classifier_score"].tolist()
+        for score in explorer_df["classifier_score"].tolist()
         if pd.notna(score)
     ]
     if not valid_scores:
@@ -54,65 +48,30 @@ def _distribution_data(
     }
 
 
-PLOT_BG = "#0b0e14"
-PLOT_TEXT = "#e6edf7"
-PLOT_MUTED = "#8b98ad"
-PLOT_GRID = "#1c2230"
-PLOT_ACCENT = "#fbbf24"
-_BAR_CMAP = LinearSegmentedColormap.from_list("explorer_bars", ["#22d3ee", "#4f8cf7", "#8b5cf6"])
-
-
 def _classifier_score_distribution_plot(
     distribution_data: dict[str, list[float] | int],
     selected_score: float | None = None,
 ) -> Figure:
-    fig, ax = plt.subplots(figsize=(6.4, 2.4))
-    fig.patch.set_facecolor(PLOT_BG)
-    ax.set_facecolor(PLOT_BG)
+    fig, ax = plt.subplots(figsize=(6, 2.2))
+    fig.patch.set_facecolor("#0f1117")
+    ax.set_facecolor("#151922")
     bin_edges = np.asarray(distribution_data.get("bin_edges", []), dtype=float)
     counts = np.asarray(distribution_data.get("counts", []), dtype=float)
     if counts.size > 0 and bin_edges.size == counts.size + 1:
         widths = np.diff(bin_edges)
-        centers = bin_edges[:-1] + widths / 2.0
-        span = float(centers[-1] - centers[0]) or 1.0
-        colors = _BAR_CMAP((centers - centers[0]) / span)
-        ax.bar(bin_edges[:-1], counts, width=widths * 0.92, align="edge", color=colors, linewidth=0.0, zorder=3)
-        ax.fill_between(centers, counts, step="mid", color="#4f8cf7", alpha=0.10, linewidth=0.0, zorder=2)
-        ax.set_ylabel("Count", color=PLOT_MUTED, fontsize=8.5, labelpad=6)
-        ax.set_ylim(0, float(counts.max()) * 1.18)
+        ax.bar(bin_edges[:-1], counts, width=widths, align="edge", color="#3b82f6", alpha=0.9, edgecolor="#93c5fd", linewidth=0.35)
+        ax.set_ylabel("Count", color="#e5e7eb")
         if selected_score is not None:
-            score = float(selected_score)
-            for width, alpha in ((7.0, 0.07), (4.0, 0.12), (1.6, 0.95)):
-                ax.axvline(score, color=PLOT_ACCENT, linewidth=width, alpha=alpha, zorder=4)
-            at_right_edge = score > (bin_edges[0] + bin_edges[-1]) / 2.0
-            ax.annotate(
-                f"{score:.4f}",
-                xy=(score, ax.get_ylim()[1]),
-                xytext=(-6 if at_right_edge else 6, -4),
-                textcoords="offset points",
-                ha="right" if at_right_edge else "left",
-                va="top",
-                fontsize=8.0,
-                color=PLOT_ACCENT,
-                zorder=5,
-                bbox=dict(boxstyle="round,pad=0.28", facecolor=PLOT_BG, edgecolor=PLOT_ACCENT, alpha=0.85, linewidth=0.6),
-            )
+            ax.axvline(float(selected_score), color="#f97316", linewidth=2.0)
     else:
-        ax.text(0.5, 0.5, "No classifier scores available.", ha="center", va="center", transform=ax.transAxes, color=PLOT_MUTED, fontsize=9)
+        ax.text(0.5, 0.5, "No classifier scores available.", ha="center", va="center", transform=ax.transAxes, color="#e5e7eb")
         ax.set_yticks([])
-    total = int(distribution_data.get("total", 0) or 0)
-    title = "Classifier score distribution"
-    if total:
-        title = f"{title}  ·  {total:,} images"
-    ax.set_title(title, color=PLOT_TEXT, fontsize=10.5, fontweight="semibold", loc="left", pad=10)
-    ax.set_xlabel("Classifier score", color=PLOT_MUTED, fontsize=8.5, labelpad=6)
-    ax.tick_params(colors=PLOT_MUTED, labelsize=8, length=0, pad=4)
-    for name, spine in ax.spines.items():
-        spine.set_visible(name == "bottom")
-        spine.set_color(PLOT_GRID)
-        spine.set_linewidth(0.8)
-    ax.grid(axis="y", color=PLOT_GRID, linewidth=0.7, alpha=0.9, zorder=0)
-    ax.set_axisbelow(True)
+    ax.set_title("Classifier Score Distribution", color="#f3f4f6")
+    ax.set_xlabel("Classifier score", color="#e5e7eb")
+    ax.tick_params(colors="#d1d5db")
+    for spine in ax.spines.values():
+        spine.set_color("#4b5563")
+    ax.grid(axis="y", color="#374151", alpha=0.4)
     fig.tight_layout()
     return fig
 
@@ -123,20 +82,6 @@ def _build_page_meta(
     sort_mode: str,
     offset: int,
 ) -> tuple[list[dict[str, str | int]], int, bool, int]:
-    all_scores = all_explorer_df[["md5", "classifier", "classifier_score", "percentile"]].copy()
-    all_scores = all_scores[
-        all_scores["classifier"].notna()
-        & all_scores["classifier_score"].notna()
-        & all_scores["percentile"].notna()
-    ]
-    all_scores["md5"] = all_scores["md5"].astype(str)
-    all_scores["classifier"] = all_scores["classifier"].astype(str)
-    all_scores = all_scores[all_scores["classifier"].isin(ALLOWED_CLASSIFIER_FILTERS)]
-    all_scores = all_scores.drop_duplicates(subset=["md5", "classifier"], keep="last")
-    classifier_order = {name: i for i, name in enumerate(ALLOWED_CLASSIFIER_FILTERS)}
-    all_scores["classifier_order"] = all_scores["classifier"].map(classifier_order)
-    all_scores = all_scores.sort_values(["md5", "classifier_order"], kind="mergesort")
-
     filtered = explorer_df[["id", "md5", "sample_url", "image_url", "classifier_score", "percentile"]].copy()
     has_sample = filtered["sample_url"].notna() & (filtered["sample_url"] != "")
     has_image = filtered["image_url"].notna() & (filtered["image_url"] != "")
@@ -146,12 +91,17 @@ def _build_page_meta(
         filtered = filtered.sort_values("classifier_score", ascending=True, kind="mergesort")
     elif sort_mode == "Rating: High to Low":
         filtered = filtered.sort_values("classifier_score", ascending=False, kind="mergesort")
-    else:
-        disagreement_by_md5 = all_scores.groupby("md5")["percentile"].agg(lambda s: s.max() - s.min())
-        filtered = filtered.assign(_disagreement=filtered["md5"].astype(str).map(disagreement_by_md5).fillna(0.0))
-        filtered = filtered.sort_values("_disagreement", ascending=False, kind="mergesort")
-        filtered = filtered.drop(columns="_disagreement")
     page_df = filtered.iloc[offset:offset + PAGE_SIZE]
+    all_scores = all_explorer_df[["md5", "classifier", "classifier_score", "percentile"]].copy()
+    all_scores = all_scores[
+        all_scores["classifier"].notna()
+        & all_scores["classifier_score"].notna()
+        & all_scores["percentile"].notna()
+    ]
+    all_scores["md5"] = all_scores["md5"].astype(str)
+    all_scores["classifier"] = all_scores["classifier"].astype(str)
+    all_scores = all_scores.drop_duplicates(subset=["md5", "classifier"], keep="last")
+    all_scores = all_scores.sort_values(["md5", "classifier"], kind="mergesort")
     all_scores_by_md5 = {
         str(md5): [
             {
@@ -187,8 +137,8 @@ def build_results_data(
     classifier_name: str,
 ) -> tuple[str, Figure, dict[str, list[float] | int], list[tuple[str, str]], list[dict[str, str | int]], int, dict]:
     page_meta, next_offset, has_more, total = _build_page_meta(explorer_df, all_explorer_df, sort_mode, offset=0)
-    summary = f"Showing {total} images from test_set.parquet joined with classifier_scores.parquet (rating: {rating_pref}, classifier: {classifier_name}, sort: {sort_mode})."
-    distribution_data = _distribution_data(all_explorer_df, classifier_name)
+    summary = f"Showing {total} images from validation_set.parquet joined with classifier_scores.parquet (rating: {rating_pref}, classifier: {classifier_name}, sort: {sort_mode})."
+    distribution_data = _distribution_data(explorer_df)
     score_distribution_plot = _classifier_score_distribution_plot(distribution_data)
     return summary, score_distribution_plot, distribution_data, _gallery_items(page_meta), page_meta, next_offset, gr.update(visible=has_more)
 
@@ -258,16 +208,10 @@ def add_results_tab(pool_df: pd.DataFrame):
         results_load_more_btn = gr.Button("Load more (ArrowDown)", elem_id="btn-results-load-more")
         selected_image_md = gr.Markdown("Click an image to reveal its ID and link.")
         results_score_distribution_plot = gr.Plot(label="Classifier score distribution")
-        results_rating_dd = gr.Dropdown(
-            choices=["safe", "all"],
-            value="safe",
-            label="Rating",
-        )
         results_sort_dd = gr.Dropdown(
             choices=SORT_MODES,
-            value=DISAGREEMENT_SORT_MODE,
+            value="Default",
             label="Sort",
-            info="Disagreement surfaces images the classifiers rate most differently — useful for spotting which classifier is off.",
             elem_id="results-sort-mode",
         )
         results_classifier_dd = gr.Dropdown(
@@ -281,7 +225,6 @@ def add_results_tab(pool_df: pd.DataFrame):
         results_page_offset_state = gr.State(0)
     return (
         results_summary_md,
-        results_rating_dd,
         results_sort_dd,
         results_classifier_dd,
         results_score_distribution_plot,

maintenance/refresh_cdn_availability.py

@@ -1,159 +0,0 @@
-#!/usr/bin/env python3
-from __future__ import annotations
-
-import csv
-import gzip
-import io
-import os
-import sys
-from collections.abc import Iterator
-
-import httpx
-import pandas as pd
-from huggingface_hub import CommitOperationAdd, HfApi, hf_hub_download
-from tqdm import tqdm
-
-POOL_REPO_ID = "RedRocket/e6-visual-ratings"
-POOL_REPO_TYPE = "dataset"
-POOL_FILES = ("pool.parquet", "test_set.parquet")
-RATINGS_APP_TOKEN_ENV = "RATINGS_APP_TOKEN"
-
-# Nullable boolean: True = the file is still served, False = it is gone.
-CDN_AVAILABLE_COLUMN = "cdn_available"
-
-# https://e621.net/help/api asks that bulk lookups go through the daily export
-# rather than the site ("If you are looking for bulk data, or are expecting to
-# do tens of thousands of lookups, please use the daily data exports"), so this
-# whole refresh is one request. Sampled posts confirm the export is enough on
-# its own: every is_deleted post 404s on the CDN, as does every md5 the export
-# does not list at all (its post was replaced, so nothing serves that file).
-POSTS_EXPORT_URL = "https://static1.e621.net/data/db_export/posts.csv.gz"
-USER_AGENT = f"e6-visual-ratings-maintenance/1.0 (https://huggingface.co/datasets/{POOL_REPO_ID})"
-REQUEST_TIMEOUT_S = 60.0
-
-class _ChunkReader(io.RawIOBase):
-    """File-like view over a chunk iterator, so gzip can stream the download."""
-
-    def __init__(self, chunks: Iterator[bytes], progress: tqdm) -> None:
-        self._chunks = chunks
-        self._progress = progress
-        self._buffer = b""
-
-    def readable(self) -> bool:
-        return True
-
-    def readinto(self, view) -> int:
-        while not self._buffer:
-            chunk = next(self._chunks, b"")
-            if not chunk:
-                return 0
-
-            self._buffer = chunk
-            self._progress.update(len(chunk))
-
-        taken = min(len(view), len(self._buffer))
-        view[:taken] = self._buffer[:taken]
-        self._buffer = self._buffer[taken:]
-        return taken
-
-def _exported_availability(wanted: set[str]) -> dict[str, bool]:
-    """Maps each wanted md5 the daily export knows about to `not is_deleted`."""
-    # Post descriptions blow past the default csv field limit.
-    csv.field_size_limit(sys.maxsize)
-
-    availability: dict[str, bool] = {}
-    with httpx.stream(
-        "GET",
-        POSTS_EXPORT_URL,
-        headers={"User-Agent": USER_AGENT},
-        timeout=REQUEST_TIMEOUT_S,
-        follow_redirects=True,
-    ) as response:
-        response.raise_for_status()
-        total = int(response.headers.get("content-length", 0)) or None
-
-        with tqdm(total=total, unit="B", unit_scale=True, desc="posts.csv.gz") as progress:
-            chunks = io.BufferedReader(_ChunkReader(response.iter_bytes(), progress))
-            rows = csv.reader(io.TextIOWrapper(gzip.GzipFile(fileobj=chunks), encoding="utf-8", newline=""))
-
-            header = next(rows)
-            md5_at = header.index("md5")
-            is_deleted_at = header.index("is_deleted")
-
-            for row in rows:
-                md5 = row[md5_at]
-                if md5 in wanted:
-                    availability[md5] = row[is_deleted_at] == "f"
-
-    return availability
-
-def _labels(availability: pd.Series) -> pd.Series:
-    return availability.map({True: "available", False: "missing"}).fillna("unknown")
-
-def refresh_cdn_availability() -> None:
-    token = os.getenv(RATINGS_APP_TOKEN_ENV)
-    api = HfApi(token=token)
-
-    frames: dict[str, pd.DataFrame] = {}
-    for filename in POOL_FILES:
-        local_path = hf_hub_download(
-            repo_id=POOL_REPO_ID,
-            filename=filename,
-            repo_type=POOL_REPO_TYPE,
-            token=token,
-        )
-        frames[filename] = pd.read_parquet(local_path)
-
-    md5s = {md5 for df in frames.values() for md5 in df["md5"]}
-    print(f"looking up {len(md5s)} unique md5s from {', '.join(frames)}")
-
-    exported = _exported_availability(md5s)
-    # An md5 the export does not list belongs to a post that was replaced, and
-    # its file is gone from the CDN just like a deleted one's.
-    availability = {md5: exported.get(md5, False) for md5 in md5s}
-    available = sum(availability.values())
-    print(f"export: {available} available, {len(md5s) - available} missing")
-
-    operations: list[CommitOperationAdd] = []
-    for filename, df in frames.items():
-        refreshed = pd.Series([availability.get(md5) for md5 in df["md5"]], index=df.index, dtype="boolean")
-        counts = _labels(refreshed).value_counts()
-
-        if CDN_AVAILABLE_COLUMN in df.columns:
-            previous = _labels(df[CDN_AVAILABLE_COLUMN].astype("boolean"))
-            changed = int((previous != _labels(refreshed)).sum())
-        else:
-            changed = len(df)
-
-        print(
-            f"{filename}: {len(df)} rows, "
-            f"{counts.get('available', 0)} available, "
-            f"{counts.get('missing', 0)} missing, "
-            f"{counts.get('unknown', 0)} unknown "
-            f"({changed} changed)"
-        )
-
-        if not changed:
-            continue
-
-        df[CDN_AVAILABLE_COLUMN] = refreshed
-        operations.append(CommitOperationAdd(
-            path_or_fileobj=df.to_parquet(index=False),
-            path_in_repo=filename,
-        ))
-
-    if not operations:
-        print("Nothing changed.")
-        return
-
-    updated = [operation.path_in_repo for operation in operations]
-    api.create_commit(
-        repo_id=POOL_REPO_ID,
-        repo_type=POOL_REPO_TYPE,
-        commit_message=f"refresh {CDN_AVAILABLE_COLUMN} in {', '.join(updated)}",
-        operations=operations,
-    )
-    print(f"Updated {', '.join(updated)} in {POOL_REPO_ID}.")
-
-if __name__ == "__main__":
-    refresh_cdn_availability()

stats_from_logs.py

@@ -1,24 +1,11 @@
-from collections import defaultdict
-from dataclasses import dataclass
+from collections import Counter
 
 import pandas as pd
 from huggingface_hub import HfApi, hf_hub_download
 
-VOTES_LOG_SUBDIR = "ratings_log"
-
-@dataclass(slots=True)
-class Stats:
-    wins: int = 0
-    losses: int = 0
-    ties: int = 0
 
-    @property
-    def votes(self) -> int:
-        return self.wins + self.losses + self.ties
+VOTES_LOG_SUBDIR = "ratings_log"
 
-    @property
-    def winrate(self) -> float:
-        return (self.wins + self.ties * 0.5) / max(self.votes, 1)
 
 def _list_remote_log_files(repo_id: str, token: str | None) -> list[str]:
     api = HfApi(token=token)
@@ -29,36 +16,42 @@ def _list_remote_log_files(repo_id: str, token: str | None) -> list[str]:
         if f.startswith(f"{VOTES_LOG_SUBDIR}/") and f.endswith(".parquet")
     )
 
-def load_vote_logs(*, repo_id: str, token: str | None) -> pd.DataFrame:
-    """All logged votes as one DataFrame with columns md5a, md5b, winner_md5."""
-    frames = [
-        pd.read_parquet(
-            hf_hub_download(
-                repo_id=repo_id,
-                filename=path_in_repo,
-                repo_type="dataset",
-                token=token,
-            ),
-            columns=["md5a", "md5b", "winner_md5"],
-        )
-        for path_in_repo in _list_remote_log_files(repo_id, token)
-    ]
-    if not frames:
-        return pd.DataFrame(columns=["md5a", "md5b", "winner_md5"])
-    return pd.concat(frames, ignore_index=True)
 
-def stats_by_md5(votes_df: pd.DataFrame) -> dict[str, Stats]:
-    stats: dict[str, Stats] = defaultdict(Stats)
+def _accumulate_stats_from_df(df: pd.DataFrame, wins_counter: Counter, losses_counter: Counter):
+    if df.empty:
+        return
+    valid = df[df["winner_md5"].notna()].copy()
+    if valid.empty:
+        return
+    valid["md5a"] = valid["md5a"].astype(str)
+    valid["md5b"] = valid["md5b"].astype(str)
+    valid["winner_md5"] = valid["winner_md5"].astype(str)
+
+    a_won_mask = valid["winner_md5"] == valid["md5a"]
+    b_won_mask = valid["winner_md5"] == valid["md5b"]
 
-    for md5a, md5b, winner_md5 in votes_df.itertuples(index=False):
-        if winner_md5 == md5a:
-            stats[md5a].wins += 1
-            stats[md5b].losses += 1
-        elif winner_md5 == md5b:
-            stats[md5b].wins += 1
-            stats[md5a].losses += 1
-        else:
-            stats[md5a].ties += 1
-            stats[md5b].ties += 1
+    winner_keys = pd.concat([valid.loc[a_won_mask, "md5a"], valid.loc[b_won_mask, "md5b"]], ignore_index=True)
+    loser_keys = pd.concat([valid.loc[a_won_mask, "md5b"], valid.loc[b_won_mask, "md5a"]], ignore_index=True)
+
+    for key, count in winner_keys.value_counts().items():
+        wins_counter[str(key)] += int(count)
+    for key, count in loser_keys.value_counts().items():
+        losses_counter[str(key)] += int(count)
+
+
+def load_stats_by_md5(*, repo_id: str, token: str | None) -> dict[str, tuple[int, int]]:
+    wins_counter: Counter[str] = Counter()
+    losses_counter: Counter[str] = Counter()
+
+    for path_in_repo in _list_remote_log_files(repo_id, token):
+        local_path = hf_hub_download(
+            repo_id=repo_id,
+            filename=path_in_repo,
+            repo_type="dataset",
+            token=token,
+        )
+        df = pd.read_parquet(local_path, columns=["md5a", "md5b", "winner_md5"])
+        _accumulate_stats_from_df(df, wins_counter, losses_counter)
 
-    return stats
+    all_keys = set(wins_counter) | set(losses_counter)
+    return {k: (int(wins_counter.get(k, 0)), int(losses_counter.get(k, 0))) for k in all_keys}

storage.py

@@ -23,126 +23,86 @@ VOTE_COLUMNS = [
     "dataset",
     "group",
     "session_id",
-    "pair_reason",  # sampler channel; lets dataset builds treat model-selected pairs specially
-    "rater_name",  # optional self-reported handle; empty when the rater stays anonymous
 ]
 
 
 class VoteStorage:
-    def __init__(self, mode: str, token: str | None = None) -> None:
+    def __init__(self, mode: str, token: str | None = None):
         assert mode in ("hf", "void"), f"Unsupported storage mode: {mode}"
         self.mode = mode
+        self._token = token
         is_debug_mode = self.mode == "void"
-
         self._flush_every = 3 if is_debug_mode else 50
         self._flush_interval_sec = 15.0 if is_debug_mode else 300.0
-        self._hf_api = HfApi(token=token)
-
-        self._flush_condition = threading.Condition(threading.Lock())
-        self._sync_event = threading.Event()
-        self._sync_lock = threading.Lock()
+        self._votes_lock = threading.Lock()
         self._votes_buffer: list[dict] = []
-        self._shutdown = False
-
+        self._stop_event = threading.Event()
         self._flush_thread = threading.Thread(target=self._flush_loop, daemon=True)
         self._flush_thread.start()
-
         atexit.register(self.close)
 
-    def _upload_votes_batch(self, batch: list[dict]) -> None:
-        assert batch
-        if self.mode == "void":
-            return
+    def _hf_token(self) -> str | None:
+        return self._token
 
-        df = pd.DataFrame(batch)
-
-        for col in VOTE_COLUMNS:
-            if col not in df.columns:
-                df[col] = None
-
-        df = df[VOTE_COLUMNS]
+    def _empty_votes_df(self) -> pd.DataFrame:
+        return pd.DataFrame(columns=VOTE_COLUMNS)
 
+    def _upload_votes_batch(self, df: pd.DataFrame, commit_message: str):
+        assert set(VOTE_COLUMNS).issubset(df.columns), "Missing vote columns in upload batch"
+        if self.mode == "void":
+            _ = commit_message
+            return
         ts = int(time.time())
         shard = f"votes_{ts}_{uuid.uuid4().hex}.parquet"
+        api = HfApi(token=self._hf_token())
+        with NamedTemporaryFile(suffix=".parquet", delete=False) as tmp:
+            tmp_path = tmp.name
+        try:
+            df[VOTE_COLUMNS].to_parquet(tmp_path, index=False)
+            api.upload_file(
+                path_or_fileobj=tmp_path,
+                path_in_repo=f"{VOTES_LOG_SUBDIR}/{shard}",
+                repo_id=VOTES_REPO_ID,
+                repo_type=VOTES_REPO_TYPE,
+                commit_message=commit_message,
+            )
+        finally:
+            if os.path.exists(tmp_path):
+                os.remove(tmp_path)
+
+    def _flush_votes(self, force: bool = False):
+        with self._votes_lock:
+            if not self._votes_buffer:
+                return
+            if not force and len(self._votes_buffer) < self._flush_every:
+                return
+            batch = list(self._votes_buffer)
+            self._votes_buffer.clear()
+        incoming = pd.DataFrame(batch)
+        for col in VOTE_COLUMNS:
+            if col not in incoming.columns:
+                incoming[col] = None
+        self._upload_votes_batch(incoming[VOTE_COLUMNS], commit_message=f"append {len(batch)} vote rows")
 
-        self._hf_api.upload_file(
-            path_or_fileobj=df.to_parquet(index=False),
-            path_in_repo=f"{VOTES_LOG_SUBDIR}/{shard}",
-            repo_id=VOTES_REPO_ID,
-            repo_type=VOTES_REPO_TYPE,
-            commit_message=f"upload {len(df)} vote rows",
-        )
-
-    def _flush_loop(self) -> None:
-        while True:
-            with self._flush_condition:
-                while True:
-                    # Forced sync.
-                    if not self._sync_event.is_set():
-                        if self._votes_buffer:
-                            break
-
-                        self._sync_event.set()
-
-                    # Shutdown wanted.
-                    if self._shutdown:
-                        if self._votes_buffer:
-                            break
-
-                        return
-
-                    # Have enough votes to flush now.
-                    if len(self._votes_buffer) >= self._flush_every:
-                        break
-
-                    # Wait for a notify to flush early or shutdown.
-                    if not self._flush_condition.wait(self._flush_interval_sec):
-                        # Interval elapsed. Flush if there is at least one vote.
-                        if self._votes_buffer:
-                            break
-
-                # Atomically take the batch of votes.
-                batch = self._votes_buffer
-                self._votes_buffer = []
-
-            self._upload_votes_batch(batch)
-
-    def sync(self) -> None:
-        with self._sync_lock:
-            with self._flush_condition:
-                is_shutdown = self._shutdown
-                if not is_shutdown:
-                    self._sync_event.clear()
-                    self._flush_condition.notify()
-
-            if not is_shutdown:
-                self._sync_event.wait()
-
-        if is_shutdown:
-            self._flush_thread.join()
-
-    def close(self) -> None:
-        with self._flush_condition:
-            self._shutdown = True
-            self._flush_condition.notify()
+    def _flush_loop(self):
+        while not self._stop_event.wait(self._flush_interval_sec):
+            self._flush_votes(force=True)
 
-        self._flush_thread.join()
+    def close(self):
+        if self._stop_event.is_set():
+            return
+        self._stop_event.set()
+        self._flush_thread.join(timeout=1.0)
+        self._flush_votes(force=True)
 
-    def queue_row(self, state: dict) -> None:
+    def append_vote_row(self, state: dict, winner: str | None):
         id_a = int(state["id_a"])
         id_b = int(state["id_b"])
-
-        winner_md5: str | None
-        match state["winner"]:
-            case "A":
-                winner_md5 = state["key_a"]
-            case "B":
-                winner_md5 = state["key_b"]
-            case None:
-                winner_md5 = None
-            case _:
-                raise AssertionError
-
+        winner_md5 = None
+        if winner == "A":
+            winner_md5 = state["key_a"]
+        elif winner == "B":
+            winner_md5 = state["key_b"]
         vote_row = {
             "vote_id": uuid.uuid4().hex,
             "timestamp": datetime.now(timezone.utc).isoformat(timespec="seconds"),
@@ -154,12 +114,7 @@ class VoteStorage:
             "dataset": state["dataset"],
             "group": state["group"],
             "session_id": state["session_id"],
-            "pair_reason": state.get("pair_reason", ""),
-            "rater_name": state.get("rater_name", ""),
         }
-
-        with self._flush_condition:
+        with self._votes_lock:
             self._votes_buffer.append(vote_row)
-
-            if len(self._votes_buffer) == self._flush_every:
-                self._flush_condition.notify()
+        self._flush_votes()