feat(automation): 巡检和重登优先保住高价值账号的 CK

价值按套装和积分现算，不看标签。第 1、2 档 6 小时巡检一次，并先占重登名额。

Co-authored-by: Cursor <cursoragent@cursor.com>

tests/test_account_overview.py

@@ -1777,11 +1777,21 @@ class TestHighValueAccountFilters:
         assert set(names) == {"acc_crown", "acc_mengtu_owned", "acc_broken"}
 
     def test_high_value_broken_filter(self):
-        # 高价值但 CK 异常待抢救
+        # 高价值但 CK 异常待抢救：两套都能换、只剩一套、或积分≥300
         items = self._query(high_value_filter="high_value_broken")
         names = [it.username for it in items]
         assert names == ["acc_broken"]
 
+        self.acc_bairimeng_owned.ck_check_status = "invalid"
+        self.acc_both_owned.ck_check_status = "invalid"
+        self.session.commit()
+        names = [
+            it.username for it in self._query(high_value_filter="high_value_broken")
+        ]
+        assert "acc_bairimeng_owned" in names
+        assert "acc_broken" in names
+        assert "acc_both_owned" not in names
+
     def test_summary_returns_high_value_breakdown(self):
         from web.backend.routers.accounts import accounts_summary
 

tests/test_account_overview_refresh_job.py

@@ -84,6 +84,24 @@ class TestSelection(_Base):
         self.account("ovr-never")
         assert self.selected()[0].username == "ovr-never"
 
+    def test_assigned_and_recent_login_lead_within_the_same_group(self):
+        """没查过的仍整体优先；同一档里已分配的在前，再按最近登录。"""
+        stale = self.account("ovr-stale-assigned", queried_days_ago=30)
+        stale.assigned_to = self.admin.id
+        recent = self.account("ovr-recent")
+        recent.cookie_updated_at = datetime.now(UTC)
+        older = self.account("ovr-older")
+        older.cookie_updated_at = datetime.now(UTC) - timedelta(days=10)
+        assigned = self.account("ovr-assigned")
+        assigned.assigned_to = self.admin.id
+        assigned.cookie_updated_at = datetime.now(UTC) - timedelta(days=1)
+        self.session.commit()
+
+        names = [a.username for a in self.selected()]
+        assert names.index("ovr-assigned") < names.index("ovr-recent")
+        assert names.index("ovr-recent") < names.index("ovr-older")
+        assert names.index("ovr-older") < names.index("ovr-stale-assigned")
+
     def test_recently_queried_is_skipped(self):
         """刚查过的要跳过（`refresh_days` 之内不再查）。"""
         self.account("ovr-fresh", queried_days_ago=1)

tests/test_automation_ck_health.py

@@ -184,6 +184,29 @@ class TestCkHealthSelection(_Base):
         assert ids[0] == never.id
         assert stale.id in ids
 
+    def test_high_value_is_rechecked_sooner_and_sorted_first(self):
+        """第 1、2 档 6 小时一查，并且到点后排在普通账号前面。"""
+        job = self._job()
+        job.interval_seconds = 24 * 3600
+        now = datetime.now(UTC)
+        plain_recent = self.make_account(
+            "plain-recent", ck_checked_at=now - timedelta(hours=7)
+        )
+        plain_old = self.make_account(
+            "plain-old", ck_checked_at=now - timedelta(hours=30)
+        )
+        valuable = self.make_account("valuable", ck_checked_at=now - timedelta(hours=7))
+        valuable.overview_queried_at = now
+        valuable.high_value_owned = []
+        valuable.points = 800
+        self.session.commit()
+
+        ids = self._selected_ids(job)
+        assert valuable.id in ids
+        assert plain_recent.id not in ids
+        assert plain_old.id in ids
+        assert ids.index(valuable.id) < ids.index(plain_old.id)
+
     def test_generated_sql_is_mysql_compatible(self):
         """回归：选行 SQL 不能出现 MySQL 不认识的 NULLS FIRST。
 

tests/test_automation_ck_relogin.py

@@ -259,6 +259,21 @@ class TestSelection(_Base):
             self.make_account(f"a{index}")
         assert len(self._ids(job)) == 3
 
+    def test_high_value_is_picked_before_plain(self):
+        """名额先给第 1 档，其次第 2 档，普通失效号排最后。"""
+        job = self.job(batch_size=3)
+        plain = self.make_account("plain")
+        one_suit = self.make_account("one-suit")
+        one_suit.overview_queried_at = datetime.now(UTC)
+        one_suit.high_value_owned = ["套装-兔兔白日梦"]
+        both = self.make_account("both")
+        both.overview_queried_at = datetime.now(UTC)
+        both.high_value_owned = []
+        self.session.commit()
+
+        ids = self._ids(job)
+        assert ids == [both.id, one_suit.id, plain.id]
+
 
 class TestSettlement(_Base):
     """结算：成功清零；失败按分型决定重试还是转人工。

web/backend/routers/accounts.py

@@ -3,7 +3,7 @@
 from typing import Any
 
 from fastapi import APIRouter, Depends, HTTPException, Query
-from sqlalchemy import String, and_, cast, func, or_, select, text
+from sqlalchemy import and_, func, or_, select, text
 from sqlalchemy.orm import Session, defer, joinedload
 
 from ..database import get_db
@@ -53,6 +53,12 @@ from ..services.account_service import (
     cookie_account_ids_query,
     parse_and_build_accounts,
 )
+from ..services.account_value import (
+    SUIT_BAIRIMENG,
+    SUIT_MENGTU,
+    is_high_value_expr,
+    suit_owned_expr,
+)
 from ..services.audit_service import record_account_action, record_audit
 from ..services.douyu_service import cookie_holding_filter
 
@@ -79,17 +85,6 @@ def _query_str(value) -> str:
     return value.strip() if isinstance(value, str) else ""
 
 
-def _high_value_contains_expr(column, text: str):
-    """匹配 JSON 字段中是否包含特定商品名（同时兼容 MySQL 与 SQLite 字符存储）。"""
-    import json
-
-    escaped = json.dumps(text).strip('"')
-    return or_(
-        cast(column, String).like(f"%{text}%"),
-        cast(column, String).like(f"%{escaped}%"),
-    )
-
-
 def _filter_accounts_query(
     db: Session,
     query,
@@ -146,10 +141,8 @@ def _filter_accounts_query(
     # 高价值套装筛选
     hv = _query_str(high_value_filter)
     if hv:
-        bairimeng_expr = _high_value_contains_expr(
-            Account.high_value_owned, "兔兔白日梦"
-        )
-        mengtu_expr = _high_value_contains_expr(Account.high_value_owned, "狂战萌兔")
+        bairimeng_expr = suit_owned_expr(Account.high_value_owned, SUIT_BAIRIMENG)
+        mengtu_expr = suit_owned_expr(Account.high_value_owned, SUIT_MENGTU)
         if hv == "both_available":
             query = query.filter(
                 Account.overview_queried_at.isnot(None),
@@ -176,14 +169,7 @@ def _filter_accounts_query(
             query = query.filter(Account.overview_queried_at.is_(None))
         elif hv == "high_value_broken":
             query = query.filter(
-                or_(
-                    and_(
-                        Account.overview_queried_at.isnot(None),
-                        ~bairimeng_expr,
-                        ~mengtu_expr,
-                    ),
-                    Account.points >= 300,
-                ),
+                is_high_value_expr(),
                 Account.ck_check_status.in_(["invalid", "error"]),
             )
 
@@ -494,17 +480,10 @@ def accounts_summary(
     by_status = {str(row[0] or ""): int(row[1] or 0) for row in status_rows}
     visible_total = sum(by_status.values())
     # 高价值及核心资产分布（单次聚合，供前端快捷胶囊展示）
-    bairimeng_expr = _high_value_contains_expr(Account.high_value_owned, "兔兔白日梦")
-    mengtu_expr = _high_value_contains_expr(Account.high_value_owned, "狂战萌兔")
+    bairimeng_expr = suit_owned_expr(Account.high_value_owned, SUIT_BAIRIMENG)
+    mengtu_expr = suit_owned_expr(Account.high_value_owned, SUIT_MENGTU)
     high_value_broken_expr = and_(
-        or_(
-            and_(
-                Account.overview_queried_at.isnot(None),
-                ~bairimeng_expr,
-                ~mengtu_expr,
-            ),
-            Account.points >= 300,
-        ),
+        is_high_value_expr(),
         Account.ck_check_status.in_(["invalid", "error"]),
     )
 

web/backend/services/account_value.py

@@ -0,0 +1,75 @@
+"""账号价值分档。
+
+列表上的「高价值待救」和巡检 / 重登共用这里的口径，避免两套规则各说各的。
+
+分档只读账号上已经有的字段（兑换记录、积分、鱼翅、宝典），不看标签。
+数字越小越值钱：
+
+1. 查过资料，兔兔白日梦和狂战萌兔都还没换
+2. 查过资料、还剩一套能换，或者积分 ≥ 300（没查过但积分已经 ≥ 300 也算）
+3. 有鱼翅，或宝典已开通
+4. 查过资料，但上面都不沾
+5. 还没查过资料
+
+「高价值」= 第 1、2 档。巡检对这两档用更短的重检间隔，重登也先拿这两档。
+"""
+
+from __future__ import annotations
+
+import json
+
+from sqlalchemy import String, and_, case, cast, or_
+
+from ..models import Account
+
+SUIT_BAIRIMENG = "兔兔白日梦"
+SUIT_MENGTU = "狂战萌兔"
+HIGH_VALUE_POINTS = 300
+
+# 第 1、2 档都是「还值得抢救」的号。
+HIGH_VALUE_TIER_MAX = 2
+
+# 高价值账号的巡检间隔。作业上配置的间隔更短时，跟那个更短的走。
+HIGH_VALUE_RECHECK_SECONDS = 6 * 3600
+
+
+def suit_owned_expr(column, text: str):
+    """JSON 列里是否包含这个套装名。MySQL 与 SQLite 都按字符串匹配。"""
+    escaped = json.dumps(text).strip('"')
+    return or_(
+        cast(column, String).like(f"%{text}%"),
+        cast(column, String).like(f"%{escaped}%"),
+    )
+
+
+def value_tier_expr():
+    """1–5 档。没查过、也没有任何资产信号时落在第 5 档。"""
+    queried = Account.overview_queried_at.isnot(None)
+    owns_bairimeng = suit_owned_expr(Account.high_value_owned, SUIT_BAIRIMENG)
+    owns_mengtu = suit_owned_expr(Account.high_value_owned, SUIT_MENGTU)
+    both_available = and_(queried, ~owns_bairimeng, ~owns_mengtu)
+    one_available = and_(
+        queried,
+        or_(
+            and_(~owns_bairimeng, owns_mengtu),
+            and_(owns_bairimeng, ~owns_mengtu),
+        ),
+    )
+    return case(
+        (both_available, 1),
+        (or_(one_available, Account.points >= HIGH_VALUE_POINTS), 2),
+        (or_(Account.gold_balance > 0, Account.elite_opened.is_(True)), 3),
+        (queried, 4),
+        else_=5,
+    )
+
+
+def is_high_value_expr():
+    """第 1、2 档：巡检加密、重登插队、列表「高价值待救」都用这一条。"""
+    return value_tier_expr() <= HIGH_VALUE_TIER_MAX
+
+
+def high_value_recheck_seconds(interval_seconds: int) -> int:
+    """高价值账号多久重检一次。不会比作业配置的间隔更长。"""
+    interval = max(int(interval_seconds or 0), 60)
+    return min(HIGH_VALUE_RECHECK_SECONDS, interval)

web/backend/services/automation/jobs/account_overview_refresh.py

@@ -20,6 +20,9 @@
 | **没查过的** | `overview_queried_at IS NULL` | **优先**（补历史欠账）|
 | 查过但过期了 | `overview_queried_at < now - interval_seconds` | 其次（周期性刷新）|
 
+同一档里面，已分配的排在未分配前面，最近登录成功（`cookie_updated_at`）的再靠前。
+价值分要先有查询结果才算得出来，所以新登录、已经分给客服的号先补。
+
 用户明确要求这两类分开 —— 否则"周期性刷新老账号"会永远挤掉"从没查过的新账号"。
 
 ## 查询记录
@@ -86,9 +89,16 @@ def build_select_query(db: Session, job: AutomationJob, now: datetime):
     batch = _positive_int(job.batch_size, 30)
     return (
         _base_query(db, job, now)
-        # 没查过的（IS NULL = 1）排最前 —— MySQL 没有 NULLS FIRST，
-        # 用 `IS NULL` 的布尔值降序等价实现
-        .order_by(Account.overview_queried_at.is_(None).desc(), Account.id.asc())
+        # 没查过的排最前。同一档里：已分配的优先，再是最近登录成功的
+        # （cookie_updated_at 越新越靠前；空的排后面）。
+        # MySQL 没有 NULLS FIRST，用 IS NULL 的布尔值表达。
+        .order_by(
+            Account.overview_queried_at.is_(None).desc(),
+            Account.assigned_to.isnot(None).desc(),
+            Account.cookie_updated_at.is_(None).asc(),
+            Account.cookie_updated_at.desc(),
+            Account.id.asc(),
+        )
         .limit(batch)
     )
 

web/backend/services/automation/jobs/ck_health.py

@@ -13,11 +13,16 @@ from concurrent.futures import ThreadPoolExecutor, as_completed
 from datetime import UTC, datetime, timedelta
 from typing import Any
 
-from sqlalchemy import func, or_
+from sqlalchemy import and_, func, or_
 from sqlalchemy.orm import Session
 
 from ....models import Account, AutomationJob
 from ...account_lifecycle import MANUAL_STATES, on_ck_verdict
+from ...account_value import (
+    HIGH_VALUE_TIER_MAX,
+    high_value_recheck_seconds,
+    value_tier_expr,
+)
 from ...cookie_check_service import check_douyu_cookie
 from ...douyu_service import cookie_holding_filter, cookie_operable_filter
 
@@ -64,13 +69,19 @@ def _candidates_query(db: Session, job: AutomationJob, now: datetime):
     """本轮「待检测账号」的完整候选集（**不带 LIMIT**，供计数复用）。
 
     口径与人工点检测一致（``cookie_operable_filter``：账号有 CK 且没有在途尝试），
-    再加三条：距上次检测超过作业间隔、账号允许自动管理、**未被停用/归档**。
+    再加三条：距上次检测超过间隔、账号允许自动管理、**未被停用/归档**。
+    第 1、2 档用 6 小时间隔，其余用作业上配置的间隔。
 
     排序刻意不用 ``NULLS FIRST``：那是 PostgreSQL/SQLite 语法，**MySQL 会报
     1064 语法错误**（生产踩过一次）。改用 ``ck_checked_at IS NULL DESC``，
     三个方言下语义一致且都是合法 SQL。
     """
-    cutoff = now - timedelta(seconds=max(int(job.interval_seconds or 0), 60))
+    interval = max(int(job.interval_seconds or 0), 60)
+    cutoff = now - timedelta(seconds=interval)
+    # 第 1、2 档 6 小时查一次（作业间隔更短时跟着更短的走），其余用作业间隔。
+    # 到点的高价值号先占本轮名额，剩下的再按「最久没查」填。
+    hv_cutoff = now - timedelta(seconds=high_value_recheck_seconds(interval))
+    tier = value_tier_expr()
     return (
         db.query(Account)
         .filter(cookie_operable_filter())
@@ -80,10 +91,12 @@ def _candidates_query(db: Session, job: AutomationJob, now: datetime):
         .filter(
             or_(
                 Account.ck_checked_at.is_(None),
-                Account.ck_checked_at < cutoff,
+                and_(tier <= HIGH_VALUE_TIER_MAX, Account.ck_checked_at < hv_cutoff),
+                and_(tier > HIGH_VALUE_TIER_MAX, Account.ck_checked_at < cutoff),
             )
         )
         .order_by(
+            tier.asc(),
             Account.ck_checked_at.is_(None).desc(),
             Account.ck_checked_at.asc(),
             Account.id.asc(),

web/backend/services/automation/jobs/ck_relogin.py

@@ -30,6 +30,7 @@ from ...account_lifecycle import (
     on_login_success,
     sync_status,
 )
+from ...account_value import value_tier_expr
 from ...douyu_service import cookie_holding_filter, cookie_operable_filter
 from ...login_service import LoginBatchRunner, get_relogin_limits
 
@@ -121,7 +122,10 @@ def build_select_query(
     )
     if system_user_id is not None:
         pass  # 预留：将来若需要限制到某个归属人，在这里加过滤
+    # 名额先给第 1、2 档（两套还能换 / 还剩一套 / 积分≥300）。
+    # 同一档里仍按退避时间，避免把还没到点的高价值号提前拉起来。
     return query.order_by(
+        value_tier_expr().asc(),
         Account.relogin_next_at.is_(None).desc(),
         Account.relogin_next_at.asc(),
         Account.id.asc(),

web/frontend/src/pages/AccountsPage.tsx

@@ -858,7 +858,7 @@ export default function AccountsPage() {
       icon: '⚠️',
       count: summary.high_value?.high_value_broken,
       color: '#f5222d',
-      tooltip: '可换高价值套装但 CK 失效，需重登抢救',
+      tooltip: '两套还能换、只剩一套，或积分≥300，且 CK 已失效',
       active: highValueFilter === 'high_value_broken' && minPoints === undefined && maxPoints === undefined && minGold === undefined && maxGold === undefined && eliteOpened === undefined && esportsOpened === undefined && !roleBindStatus,
       onClick: () => {
         if (highValueFilter === 'high_value_broken') {