# -*- coding: utf-8 -*- """候选收藏、忽略、拉黑动作。""" from __future__ import annotations import time from dataclasses import dataclass from typing import Any from ..bilibili.models import FavResource from ..config import ConfigStore from ..fav_position import VideoMeta, plan_insertion from ..repository import Repository from ..timeutil import shanghai_datetime @dataclass class ActionResult: ok: bool aid: int action: str state: str position: int | None = None already_in_folder: bool = False message: str = "" def as_dict(self) -> dict[str, Any]: return { "ok": self.ok, "aid": self.aid, "action": self.action, "state": self.state, "position": self.position, "already_in_folder": self.already_in_folder, "message": self.message, } class FavoriteActionService: def __init__( self, config_store: ConfigStore, repository: Repository, write_provider, *, clock=time.time, logger=None, ) -> None: self.config_store = config_store self.repo = repository self.write_provider = write_provider self._clock = clock self.logger = logger def _now(self) -> int: return int(self._clock()) def _log(self, level: str, event: str, **fields) -> None: if self.logger is not None: getattr(self.logger, level)(event, **fields) def _target_folder(self, candidate: dict[str, Any]): cfg = self.config_store.get() up_mid = candidate.get("up_mid") if isinstance(up_mid, int) and not isinstance(up_mid, bool): folder_id = self.repo.folder_for_up(up_mid) if folder_id is not None: row = self.repo.get_folder(folder_id) if row is not None and row.enabled: return row year = shanghai_datetime(self._clock()).year folder = cfg.folder_for_year(year) if folder is not None: return folder latest = cfg.latest_folder() if latest is not None: return latest raise RuntimeError("没有可用的目标收藏夹，请检查 folders 配置") def favorite(self, aid: int) -> ActionResult: aid = int(aid) candidate = self.repo.get_candidate(aid) if candidate is None: return ActionResult( ok=False, aid=aid, action="favorite", state="missing", message="候选不存在或已处理", ) try: folder = self._target_folder(candidate) except Exception as exc: # noqa: BLE001 return ActionResult( ok=False, aid=aid, action="favorite", state="failed", message=str(exc), ) token, state = self.repo.begin_fav_action( aid, folder.media_id, now=self._now() ) if state == "done": action = self.repo.get_fav_action(token) or {} return ActionResult( ok=True, aid=aid, action="favorite", state="done", position=action.get("position"), message="已在收藏夹", ) if state == "busy": return ActionResult( ok=False, aid=aid, action="favorite", state="busy", message="该候选正在处理中，请稍后重试", ) already = False try: client = self.write_provider.get() ids = client.fav_ids(folder.media_id) if aid in {item.aid for item in ids}: already = True position = [item.aid for item in ids].index(aid) + 1 else: client.deal(aid, folder.media_id) ids = client.fav_ids(folder.media_id) aid_order = [item.aid for item in ids] if aid not in aid_order: raise RuntimeError("deal 成功但回读收藏夹未找到该视频") metadata = { item_aid: VideoMeta( up_mid=(info.get("up_mid") if info else None), pubdate=int((info or {}).get("pubdate") or 0), ) for item_aid, info in self.repo.video_metadata(aid_order).items() } plan = plan_insertion( aid_order, aid, metadata, VideoMeta( up_mid=candidate.get("up_mid"), pubdate=int(candidate.get("pubdate") or 0), ), ) type_map = {item.aid: item.type for item in ids} for predecessor, moved in plan.moves: client.sort( folder.media_id, f"{predecessor}:{type_map.get(predecessor, 2)}:" f"{moved}:{type_map.get(moved, 2)}", ) final_order = [item.aid for item in client.fav_ids(folder.media_id)] if aid not in final_order: raise RuntimeError("排序后回读收藏夹未找到该视频") position = final_order.index(aid) + 1 if position != plan.target_position: raise RuntimeError( f"回读位置不符：期望 #{plan.target_position}，实际 #{position}" ) resource = FavResource( aid=aid, bvid=candidate.get("bvid"), title=str(candidate.get("title") or "(无标题)"), up_mid=candidate.get("up_mid"), up_name=candidate.get("up_name"), pubdate=int(candidate.get("pubdate") or 0), duration=None, cover=candidate.get("cover"), attr=0, fav_time=None, ) self.repo.upsert_from_fav( resource, folder.media_id, was_added=True, now=self._now() ) self.repo.delete_candidate(aid) self.repo.finish_fav_action(token, position=position, now=self._now()) self._log( "info", "favorite.done", aid=aid, folder_id=folder.media_id, position=position, already=already, ) return ActionResult( ok=True, aid=aid, action="favorite", state="done", position=position, already_in_folder=already, message="已在收藏夹" if already else "已加入收藏夹", ) except Exception as exc: # noqa: BLE001 失败保留候选，允许重试 self.repo.fail_fav_action(token, error=str(exc), now=self._now()) self._log( "error", "favorite.failed", aid=aid, folder_id=folder.media_id, error=str(exc), ) return ActionResult( ok=False, aid=aid, action="favorite", state="failed", message=str(exc), ) def ignore(self, aid: int) -> ActionResult: aid = int(aid) candidate = self.repo.get_candidate(aid) if candidate is None: return ActionResult( ok=False, aid=aid, action="ignore", state="missing", message="候选不存在或已处理", ) self.repo.mark_candidate_decision(aid, "rejected", now=self._now()) return ActionResult( ok=True, aid=aid, action="ignore", state="rejected", message="已忽略", ) def blacklist(self, aid: int) -> ActionResult: aid = int(aid) candidate = self.repo.get_candidate(aid) if candidate is None: return ActionResult( ok=False, aid=aid, action="blacklist", state="missing", message="候选不存在或已处理", ) up_mid = candidate.get("up_mid") if not isinstance(up_mid, int) or isinstance(up_mid, bool): return ActionResult( ok=False, aid=aid, action="blacklist", state="failed", message="候选缺少 UP mid，无法拉黑", ) self.repo.add_blacklist( up_mid, name=candidate.get("up_name"), reason="telegram", now=self._now(), ) rejected = self.repo.reject_pending_for_up(up_mid, now=self._now()) self.repo.mark_candidate_decision(aid, "rejected", now=self._now()) return ActionResult( ok=True, aid=aid, action="blacklist", state="rejected", message=f"已拉黑 UP，并忽略该 UP 的 {rejected} 条待决策候选", )