live-hub-py/web/backend/services/automation/jobs/account_overview_refresh.py

"""作业：**定时分批**补全账号基础资料（一键查询的内容）。

## 为什么不能一次性跑全量

生产有约 8,000 个「CK 有效但没查过」的账号。一次性发起等于
**~8 万次斗鱼请求**，很容易触发风控 —— 而风控的代价是所有账号一起受影响。

所以做成**定时分批**：每轮只取 `batch_size` 个（默认 30），
跑完就结束，下一轮再来。慢一点没关系，**稳**最重要。

口径与「自动巡检」（`ck_health`）一致，并复用它的两道保护：
- `cookie_operable_filter()`：账号有 CK 且没有在途尝试
- **错误率熔断**：一轮里失败比例过高（通常是风控或接口抖动）就
  **自己把自己关掉**，不再继续污染账号

## 两类账号分开

| 类型 | 条件 | 优先级 |
| --- | --- | --- |
| **没查过的** | `overview_queried_at IS NULL` | **优先**（补历史欠账）|
| 查过但过期了 | `overview_queried_at < now - interval_seconds` | 其次（周期性刷新）|

同一档里面，已分配的排在未分配前面，最近登录成功（`cookie_updated_at`）的再靠前。
价值分要先有查询结果才算得出来，所以新登录、已经分给客服的号先补。

用户明确要求这两类分开 —— 否则"周期性刷新老账号"会永远挤掉"从没查过的新账号"。

## 查询记录

每次都会建真实的 `douyu_tasks` 行（`query_account_overview`），
所以「哪个账号、什么时候、查出来什么」都有据可查；
账号上的 `overview_queried_at` 则是"下次该不该再查"的依据。

## 只查有效 CK

`ck_check_status == 'valid'`。CK 失效的账号查了也是白查（只会多一条失败记录），
交给「CK 检测与重登」去处理。
"""

from __future__ import annotations

from datetime import UTC, datetime, timedelta
from typing import Any

from loguru import logger
from sqlalchemy import or_
from sqlalchemy.orm import Session

from ....models import Account, AutomationJob
from ...account_lifecycle import MANUAL_STATES
from ...douyu_runner import DouyuBatchRunner
from ...douyu_service import cookie_operable_filter, create_douyu_planned_tasks

JOB_KEY = "account_overview_refresh"

#: `interval_seconds` 缺失时的兜底（天）—— 正常都会走界面上那个「同一账号多久重检」
DEFAULT_RECHECK_SECONDS = 7 * 24 * 3600

#: 触发熔断的最小样本量：样本太少时比例没有意义（1 个失败就是 100%）
ERROR_PAUSE_MIN_SCANNED = 10

#: 失败比例超过这个值就熔断（与 ck_health 同口径）
ERROR_PAUSE_RATIO = 0.5


def _positive_int(value: Any, default: int) -> int:
    try:
        parsed = int(value)
    except (TypeError, ValueError):
        return default
    return parsed if parsed > 0 else default


def build_select_query(db: Session, job: AutomationJob, now: datetime):
    """选出本轮要查的账号。

    - 只要 **CK 有效**（`ck_check_status == 'valid'`）—— 失效的查了也是白查
    - **没查过的优先**：`overview_queried_at IS NULL` 排在最前
      （MySQL 没有 `NULLS FIRST`，用 `IS NULL` 的布尔值降序等价实现）
    - 查过的只在超过 `interval_seconds` 后才重新入选
      （就是界面上那个「同一账号多久重检」—— 与其它作业**同一套口径**）
    """
    # ⚠️ 用 `interval_seconds`（界面上那个「同一账号多久重检」），
    # 与其它批量作业**同一套口径**。
    #
    # 之前这里读的是 `job.config["refresh_days"]` —— 于是界面上那个字段
    # 对本作业是**死的**（改了没反应）。那正是用户刚批评过的
    # 「看起来能配、其实配的是不相干的东西」。
    batch = _positive_int(job.batch_size, 30)
    return (
        _base_query(db, job, now)
        # 没查过的排最前。同一档里：已分配的优先，再是最近登录成功的
        # （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)
    )


def _base_query(db: Session, job: AutomationJob, now: datetime):
    """**不带排序和 LIMIT** 的基础查询。

    单独抽出来是因为计数不能再走 `build_select_query`：
    SQLAlchemy 不允许在已 `LIMIT` 的查询上再 `order_by(None)`
    （抛 "order_by() being called on a Query which already has LIMIT"）。
    """
    # ⚠️ 用 `interval_seconds`（界面上那个「同一账号多久重检」），
    # 与其它批量作业**同一套口径**。
    #
    # 之前这里读的是 `job.config["refresh_days"]` —— 于是界面上那个字段
    # 对本作业是**死的**（改了没反应）。那正是用户刚批评过的
    # 「看起来能配、其实配的是不相干的东西」。
    recheck_seconds = _positive_int(job.interval_seconds, DEFAULT_RECHECK_SECONDS)
    cutoff = now - timedelta(seconds=recheck_seconds)
    return db.query(Account).filter(
        cookie_operable_filter(),
        Account.auto_managed.is_(True),
        Account.status.notin_(MANUAL_STATES),
        Account.ck_check_status == "valid",
        or_(
            Account.overview_queried_at.is_(None),
            Account.overview_queried_at < cutoff,
        ),
    )


def count_remaining(db: Session, job: AutomationJob, now: datetime) -> int:
    """还剩多少待查（只用于提示，不参与调度）。"""
    return _base_query(db, job, now).count()


def _system_user_id(db: Session) -> int:
    """任务行的 created_by —— 定时任务不是某个人的操作，记在超管名下。"""
    from ....models import User

    user = (
        db.query(User)
        .filter(User.role == "super_admin")
        .order_by(User.id.asc())
        .first()
    )
    if user is None:
        raise RuntimeError("库中没有超管用户，无法为定时任务提供 created_by")
    return int(user.id)


def run_account_overview_refresh(db: Session, job: AutomationJob) -> dict[str, Any]:
    """本轮：选一批 → 建任务 → 就地跑完 → 看失败率决定要不要熔断。"""
    now = datetime.now(UTC)
    accounts = build_select_query(db, job, now).all()
    if not accounts:
        return {
            "scanned": 0,
            "valid": 0,
            "invalid": 0,
            "succeeded": 0,
            "failed": 0,
            "message": "全库暂无待补全或到期的账号",
        }

    account_ids = [a.id for a in accounts]
    batch_id, count = create_douyu_planned_tasks(
        db,
        account_ids,
        "query_account_overview",
        "elite",
        _system_user_id(db),
        # 标上来源，便于在任务列表里区分「定时巡检」与「人工点查询」
        {"source": "scheduled_refresh"},
    )
    if count == 0:
        return {
            "scanned": 0,
            "valid": 0,
            "invalid": 0,
            "succeeded": 0,
            "failed": 0,
            "message": "候选账号均不可查（可能正有在途任务）",
        }

    # 复用现有执行器：并发、代理、进度、失败处理都是现成的
    # ⚠️ 必须 close_db=False，避免 runner 结束时把调度线程的 session 关闭
    runner = DouyuBatchRunner(
        db,
        batch_id,
        "query_account_overview",
        concurrency=max(1, min(int(job.concurrency or 1), 10)),
        close_db=False,
    )
    runner.run()
    db.commit()

    from ....models import DouyuTask

    rows = db.query(DouyuTask).filter(DouyuTask.batch_id == batch_id).all()
    succeeded = sum(1 for t in rows if t.status == "success")
    failed = sum(1 for t in rows if t.status == "failed")

    error_ratio = failed / len(rows) if rows else 0.0
    paused = False
    if len(rows) >= ERROR_PAUSE_MIN_SCANNED and error_ratio > ERROR_PAUSE_RATIO:
        # 大面积失败通常是风控或斗鱼接口抖动。继续跑只会把更多账号写坏，
        # 所以**自己把自己关掉**，交给人来确认（与 ck_health 同口径）。
        job.enabled = False
        db.commit()
        paused = True
        logger.warning(
            f"[overview] 失败率 {error_ratio:.0%} 过高，已自动关闭作业"
            f"（batch={batch_id}，failed={failed}/{len(rows)}）—— 请确认是否被风控"
        )

    message = (
        f"本轮查 {len(rows)} 个：成功 {succeeded} / 失败 {failed}"
        f"（剩 {count_remaining(db, job, now)} 个待查）"
    )
    if paused:
        # 把**可能的原因**写进消息里：运维看作业历史时才知道该去查什么，
        # 否则只看到"失败率高"会一头雾水。
        message += "；失败率过高，已自动关闭作业（大面积失败通常是**被风控**或接口抖动，请先确认再重开）"
    logger.info(f"[overview] {message}")
    return {
        "scanned": len(rows),
        "valid": succeeded,
        "invalid": failed,
        "succeeded": succeeded,
        "failed": failed,
        "batch_id": batch_id,
        "paused": paused,
        "message": message,
    }