Files
live-hub-py/web/backend/services/account_value.py
T
yml2213andCursor 998998c390 feat(automation): 巡检和重登优先保住高价值账号的 CK
价值按套装和积分现算,不看标签。第 1、2 档 6 小时巡检一次,并先占重登名额。

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-10-05 17:51:04 +08:00

76 lines
2.5 KiB
Python

"""账号价值分档。
列表上的「高价值待救」和巡检 / 重登共用这里的口径,避免两套规则各说各的。
分档只读账号上已经有的字段(兑换记录、积分、鱼翅、宝典),不看标签。
数字越小越值钱:
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)