kairee
d971e718e9
Implement WBI video search with legacy fallback, classification scoring, candidate pool, permanent seen-aid dedupe, blacklist filtering, and batched Telegram notifications. Add write client for fav deal/sort, target-folder selection, insertion position calculation, read-back verification, fav_action locking, retry on failure, ignore and blacklist actions. Wire discovery, candidate cleanup, Telegram long polling, and new discover/candidates CLI commands into the scheduler. Add migration v3 for candidate duration and index stats.
83 lines
2.4 KiB
Python
83 lines
2.4 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""新视频在收藏夹中的插入位置计算（纯函数）。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Mapping, Sequence
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class VideoMeta:
|
|
up_mid: int | None
|
|
pubdate: int
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class InsertionPlan:
|
|
target_position: int
|
|
moves: tuple[tuple[int, int], ...] # (pred_aid, moved_aid)
|
|
|
|
@property
|
|
def needs_head_move(self) -> bool:
|
|
return len(self.moves) == 2
|
|
|
|
|
|
def plan_insertion(
|
|
current_ids: Sequence[int],
|
|
new_aid: int,
|
|
metadata: Mapping[int, VideoMeta],
|
|
new_meta: VideoMeta,
|
|
) -> InsertionPlan:
|
|
"""按“同 UP 连续、组间按最早投稿倒序”计算插入点。
|
|
|
|
``current_ids`` 可以包含 ``new_aid``（通常在末尾），函数会先排除它。
|
|
"""
|
|
ordered = [int(aid) for aid in current_ids if int(aid) != int(new_aid)]
|
|
if not ordered:
|
|
return InsertionPlan(target_position=1, moves=())
|
|
|
|
groups: list[dict] = []
|
|
for index, aid in enumerate(ordered):
|
|
meta = metadata.get(aid)
|
|
up_mid = meta.up_mid if meta else None
|
|
if up_mid is not None and groups and groups[-1]["up_mid"] == up_mid:
|
|
groups[-1]["aids"].append(aid)
|
|
else:
|
|
groups.append({"up_mid": up_mid, "aids": [aid], "start": index})
|
|
|
|
for group in groups:
|
|
keys = [
|
|
metadata[aid].pubdate
|
|
for aid in group["aids"]
|
|
if aid in metadata and metadata[aid].pubdate > 0
|
|
]
|
|
group["key"] = min(keys) if keys else 0
|
|
|
|
insert_index: int | None = None
|
|
if new_meta.up_mid is not None:
|
|
for group in groups:
|
|
if group["up_mid"] == new_meta.up_mid:
|
|
insert_index = group["start"]
|
|
break
|
|
if insert_index is None:
|
|
for group in groups:
|
|
if group["key"] < new_meta.pubdate:
|
|
insert_index = group["start"]
|
|
break
|
|
if insert_index is None:
|
|
insert_index = len(ordered)
|
|
|
|
target_position = insert_index + 1
|
|
if insert_index == 0:
|
|
first = ordered[0]
|
|
return InsertionPlan(
|
|
target_position=1,
|
|
moves=((first, int(new_aid)), (int(new_aid), first)),
|
|
)
|
|
predecessor = ordered[insert_index - 1]
|
|
return InsertionPlan(
|
|
target_position=target_position,
|
|
moves=((predecessor, int(new_aid)),),
|
|
)
|