# -*- coding: utf-8 -*-
# ginkgostu.py — 东北财经大学·白果云课堂 通知轮询 / 推送 / 自动签到（单文件合并模块）
#
# 本文件由以下四个模块合并而成（2026-09-18），功能不变：
#   ginkgo_errors.py      -> 「限流错误分类」一节
#   ginkgo_delivery.py    -> 「持久化投递 / 账号状态存储」一节
#   ginkgostu_signin.py   -> 「签到 URL 解析 / 签到接口」一节（AES/SSO 与轮询登录共用一份）
#   ginkgostu.py          -> 主体（常量 / AES / SSO 登录 / 轮询 / 指令 / 调度）
# 合并后自动签到固定走进程内本地 SSO + 签到，openapi 远程回退接口已移除。
#
# 指令（QQ 与 TG 通用，无需管理员）
#   .ginkgo help                    帮助
#   .ginkgo rule                    查看轮询规则
#   .ginkgo bind <账号> <密码>       绑定账号（成功后返回学生姓名）
#   .ginkgo unbind [账号]            解绑账号
#   .ginkgo status                  查看绑定账号与开关状态
#   .ginkgo sign                    强制执行最新有效签到活动
#   .ginkgo notice <int>            查看最新 N 条通知（两类合并）
#   .ginkgo notice <on|off>         开关通知推送（关闭且自动签到也关时停止轮询；
#                                   仅开自动签到时只在 10 秒档时段轮询；两个都关则
#                                   只每天 11:55 / 23:55 保活探活一次）
#   .ginkgo auto-sign-in <on|off>   开关自动签到
#
# 状态：本目录下 ginkgostu/<学号>.json（账号密码明文保存，按需求）
# 依赖：requests；AES-128 为自带纯 Python 实现（不引入第三方加密库）
# 限流：检测到 HTTP 429 或应用层限流提示时，自动关闭该账号的推送与自动签到并通知用户；
#       轮询按账号并发执行（上限 ACCOUNT_CONCURRENCY），慢账号不阻塞其它账号。
#
# ── 协议（2026-09-14 在 bj 实测跑通） ────────────────────────────────────────
# 白果云 SPA 自带登录已被关闭（GET /student/getLoginMethodSet: normalAuthLogin=0），
# 必须走 SSO OAuth 授权码流程：
#   1) GET  /student/getLoginUnifySet            -> loginUrl / clientId / rollBackUrl / responseType
#   2) GET  <loginUrl>?response_type=code&client_id=..&redirect_uri=<rollBackUrl>
#                                              -> SSO 登录页，页面含 __token__（CSRF）
#   3) POST /auth/widget?layer=image_verify&widget=Slider&action=get_verify&type=login_image_verify
#       服务端直接把滑块答案下发：data.d 用 data.token 前 16 字节作 AES-128-ECB 密钥
#       解密即得 x（配合 data.offset 容错）——无需图像识别
#   4) POST SSO 登录页: username / password(AES-128-CBC+ZeroPadding, key=iv=__token__ 前 16 字节)
#                       verify_token / verify_code(x) / fingerprint / __token__
#       -> 302，Location 形如 https://ginkgostu.dufe.edu.cn/login?code=xxx&state=...
#   5) POST /student/authLogin {code}            -> data.token / data.studentName
# 取数（请求头带 token，可附随机 fingerprint）：
#   通知类:  GET  /student/message/getMessageAll?pageSize=20&pageNo=1&readFlag=0   （0=全部）
#   互动消息: POST /student/busiMessage/list   body {}                            （空=全部，含已读）
# 自动签到：进程内直接跑 SSO 登录 + getSignInInfo + InTheSign（签到代码源自
#   hk openapi 后端 ginkgo_sign_in.py 的移植，与本文件轮询登录共用 AES/SSO 实现）。
import asyncio
import base64
import copy
import html as _html
import json
import os
import re
import tempfile
import threading
import time
import uuid
import uuid as _uuid
from contextlib import contextmanager
from datetime import datetime, timedelta, timezone, time as dtime
from email.utils import parsedate_to_datetime
from html.parser import HTMLParser
from urllib.parse import urlsplit
from functools import wraps
import weakref

import requests

import identity
from channel import get_channel, send_message, push_to_user


# ==================== 限流错误分类（原 ginkgo_errors.py） ====================
class RateLimitedError(Exception):
    """An upstream explicitly refused requests due to throttling."""


_RATE_MESSAGE = re.compile(
    r"too\s+many\s+requests|rate[\s_-]*limit(?:ed|ing|\s+exceeded)?|"
    r"request(?:s)?\s+throttl|throttl(?:ed|ing)|"
    r"(?:请求|访问|操作|登录|登陆|接口|提交).{0,12}(?:过于频繁|太频繁|频繁|频率.{0,6}(?:超|限制))|"
    r"(?:限流|请求次数.{0,8}(?:超限|超出|超过)|访问次数.{0,8}(?:超限|超出|超过))",
    re.I,
)
_RATE_CODES = {"429", "4290", "40029", "90029", "RATE_LIMIT", "RATE_LIMITED", "RATE_LIMIT_EXCEEDED",
               "TOO_MANY_REQUESTS", "REQUEST_TOO_FREQUENT", "THROTTLED",
               "THROTTLING", "FREQUENT_REQUESTS"}
_FIELDS = ("code", "msg", "message", "error")


def _status(value):
    try:
        return int(value)
    except (ValueError, TypeError):
        return 0


def _explicit(value, depth=0):
    if depth > 3:
        return False
    if isinstance(value, dict):
        for key in _FIELDS:
            item = value.get(key)
            if key == "code" and isinstance(item, (str, int)):
                if str(item).strip().upper().replace("-", "_") in _RATE_CODES:
                    return True
            elif isinstance(item, dict):
                if _explicit(item, depth + 1):
                    return True
            elif isinstance(item, str) and _RATE_MESSAGE.search(item[:1024]):
                return True
    return False


class _LoginErrors(HTMLParser):
    """Extract only explicitly marked login error elements; ignore scripts/notices."""
    def __init__(self):
        super().__init__()
        self.stack = []
        self.messages = []

    def handle_starttag(self, tag, attrs):
        attrs = dict(attrs)
        marker = (attrs.get("id", "") + " " + attrs.get("class", "")).lower()
        explicit = bool(re.search(r"(?:^|[\s_-])(?:error|errors|errormsg|errormessage)(?:$|[\s_-])", marker))
        ignored = tag in ("script", "style", "template") or bool(self.stack and self.stack[-1][2])
        active = explicit or bool(self.stack and self.stack[-1][1])
        if tag not in ("input", "br", "img", "meta", "link", "hr", "wbr", "source"):
            self.stack.append((tag, active, ignored))

    def handle_endtag(self, tag):
        for index in range(len(self.stack) - 1, -1, -1):
            if self.stack[index][0] == tag:
                del self.stack[index:]
                break

    def handle_data(self, data):
        if self.stack and self.stack[-1][1] and not self.stack[-1][2]:
            self.messages.append(data)


def is_rate_limited(status, payload=None, text=""):
    """Check HTTP status and explicit error fields, never arbitrary payload data/HTML."""
    status = _status(status)
    if status == 429 or _explicit(payload):
        return True
    # Plain response bodies are evidence only on HTTP errors, and must be short.
    if status >= 400 and isinstance(text, str):
        plain = text.strip()
        if len(plain) <= 1024 and not re.search(r"<[^>]+>", plain):
            return bool(_RATE_MESSAGE.search(plain))
    return False


def _valid_retry_after(value):
    if not isinstance(value, str) or not value.strip():
        return False
    value = value.strip()
    if re.fullmatch(r"[0-9]+", value):
        return True
    try:
        return parsedate_to_datetime(value).tzinfo is not None
    except (ValueError, TypeError, OverflowError):
        return False


def check_response(response, what="upstream"):
    """Return response unchanged, or raise RateLimitedError before caller parsing.

    `what` is deliberately not interpolated: URLs and upstream bodies may contain
    credentials. Safe to use as a requests response hook through a small lambda.
    """
    if response is None:
        return response
    status = _status(getattr(response, "status_code", 0))
    message = "上游请求过于频繁，请稍后重试"
    if status == 429:
        raise RateLimitedError(message)
    headers = getattr(response, "headers", {}) or {}
    retry_after = next((v for k, v in headers.items() if str(k).lower() == "retry-after"), None)
    if status >= 400 and _valid_retry_after(retry_after):
        raise RateLimitedError(message)
    try:
        payload = response.json()
    except (ValueError, TypeError, AttributeError):
        payload = None
    text = getattr(response, "text", "") or ""
    if is_rate_limited(status, payload, text):
        raise RateLimitedError(message)
    # Login pages can return HTTP 200 with an explicit error element. Never scan
    # the whole successful HTML document, inline scripts, or announcement text.
    if payload is None and isinstance(text, str) and "<" in text:
        parser = _LoginErrors()
        try:
            parser.feed(text[:262144])
        except (ValueError, TypeError):
            pass
        if _RATE_MESSAGE.search(" ".join(parser.messages)[:1024]):
            raise RateLimitedError(message)
    return response


def safe_error(exception):
    """Propagate explicit throttling, otherwise return a credential-free diagnostic.

    Never stringify arbitrary exceptions: requests URLs, query strings, tokens,
    passwords and JSON decode snippets can all include secrets.
    """
    if isinstance(exception, RateLimitedError):
        raise exception
    response = getattr(exception, "response", None)
    if response is not None:
        check_response(response, "request error")
    status = _status(getattr(response, "status_code", 0))
    if status:
        return "upstream HTTP %d" % status
    return "upstream request or response error (%s)" % type(exception).__name__




# ==================== 持久化投递 / 账号状态存储（原 ginkgo_delivery.py） ====================
MAX_SIGNIN_ATTEMPTS = 6
SIGNIN_BACKOFF = (3, 3, 3, 3, 3)
LEASE_SECONDS = 600
AUDIT_CAP = 2000
TERMINAL_CAP = 1000
# 单条 outbox 的投递尝试上限。达到后置终态 failed，停止无限重试
# （避免一个永远失败的目标让队列永久 pending）。
MAX_DELIVERY_ATTEMPTS = 10
# outbox 终态：sent=已投递完所有目标；failed=尝试次数用尽仍未投递完
OUTBOX_TERMINAL = ("sent", "failed")
_LOCKS = {}
_LOCKS_GUARD = threading.Lock()


def target_key(channel_name, target_id):
    """推送目标在 outbox 行内的稳定标识（用于逐目标成功集合 sent_targets）"""
    return "%s:%s" % (channel_name, target_id)


def backoff(attempts):
    return min(3600, 5 * (2 ** min(max(int(attempts) - 1, 0), 10)))


def audit(data, event, key, **fields):
    data.setdefault("delivery_audit", []).append(
        dict(at=time.time(), event=event, key=str(key), **fields))
    data["delivery_audit"] = data["delivery_audit"][-AUDIT_CAP:]


def prune_terminal(data):
    # Pending records are never evicted. seen_keys remain an exact unique ledger
    # so evicting terminal delivery history cannot turn old discoveries into new.
    for field, terminal in (("outbox", set(OUTBOX_TERMINAL)),
                            ("signin_jobs", {"succeeded", "failed", "cancelled"})):
        records = data.get(field, {})
        done = sorted((key for key, row in records.items() if row.get("status") in terminal),
                      key=lambda key: records[key].get("finished_at") or records[key].get("sent_at") or 0)
        for key in done[:-TERMINAL_CAP]:
            del records[key]


def enqueue(data, key, kind, text, now=None):
    now = time.time() if now is None else now
    outbox = data.setdefault("outbox", {})
    if key not in outbox:
        outbox[key] = {"key": key, "kind": kind, "text": str(text),
                       "status": "pending", "attempts": 0, "sent_targets": [],
                       "created_at": now, "next_attempt_at": now}
        audit(data, "outbox_enqueued", key, kind=kind)
    return outbox[key]


def signin_source(job):
    """Legacy jobs had no source and therefore retain automatic semantics."""
    return "manual" if isinstance(job, dict) and job.get("source") == "manual" else "auto"


def signin_enabled(data, job):
    return signin_source(job) == "manual" or bool(data.get("auto_sign_in"))


def _find_signin_job(jobs, activity_key, item):
    """Find new stable activity ids and legacy message-key jobs without overlap."""
    if activity_key in jobs:
        return activity_key, jobs[activity_key]
    link = str((item or {}).get("link") or "")
    for key, row in jobs.items():
        if row.get("activity_key") == activity_key:
            return key, row
        if link and str((row.get("item") or {}).get("link") or "") == link:
            return key, row
    return None, None


def enqueue_signin(data, item, source="auto", now=None, run_id=None):
    """Create/reset one activity job while preserving terminal history in audit."""
    now = time.time() if now is None else now
    source = "manual" if source == "manual" else "auto"
    item = copy.deepcopy(item if isinstance(item, dict) else {})
    activity_key = str(item.get("signin_activity_key") or item.get("key") or "")
    if not activity_key:
        return "invalid", None, None
    jobs = data.setdefault("signin_jobs", {})
    key, old = _find_signin_job(jobs, activity_key, item)
    if old:
        status = old.get("status")
        if status in ("pending", "running"):
            if source == "manual" and signin_source(old) == "auto":
                old.update(source="manual", activity_key=activity_key)
                old.setdefault("run_id", run_id or uuid.uuid4().hex)
                audit(data, "manual_signin_promoted", key, status=status,
                      attempts=old.get("attempts", 0), run_id=old.get("run_id"))
            return "active", key, old
        if status == "succeeded":
            return "succeeded", key, old
        if source != "manual" or status not in ("failed", "cancelled"):
            return "terminal", key, old
        audit(data, "manual_signin_reset", key, old_status=status,
              old_attempts=old.get("attempts", 0), old_reason=str(old.get("reason") or "")[:300],
              old_run_id=old.get("run_id"))
    else:
        key = activity_key
    job = {"key": key, "activity_key": activity_key, "item": item,
           "source": source, "run_id": run_id or uuid.uuid4().hex,
           "status": "pending", "attempts": 0,
           "created_at": now, "next_attempt_at": now}
    jobs[key] = job
    audit(data, "signin_enqueued", key, source=source, run_id=job["run_id"])
    return ("reset" if old else "created"), key, job


def _signin_run_token(job):
    return str(job.get("run_id") or "legacy")


def retry_receipt_key(key, job, attempt):
    return "signin-retry:%s:%s:%s" % (key, _signin_run_token(job), int(attempt))


def signin_type_name(item=None, payload=None):
    item = item if isinstance(item, dict) else {}
    payload = payload if isinstance(payload, dict) else {}
    name = payload.get("typeName") or item.get("signin_type_name")
    if name:
        return str(name)
    raw = payload.get("type") if payload.get("type") is not None else item.get("signin_type")
    try:
        signin_type = int(raw)
    except (TypeError, ValueError):
        return "未知签到类型"
    return {0: "二维码签到", 1: "数字码签到", 2: "普通签到", 3: "直播签到"}.get(
        signin_type, "未知签到类型(%s)" % signin_type)


def retry_receipt_text(job, reason, attempt):
    job = job if isinstance(job, dict) else {}
    item = job.get("item") if isinstance(job.get("item"), dict) else {}
    title = str(item.get("title") or "").strip()
    lines = ["【白果云·%s签到进度】" % ("手动" if signin_source(job) == "manual" else "自动")]
    if title:
        lines.append("活动：%s" % title[:120])
    lines.append("签到类型：%s" % signin_type_name(item))
    lines.append("第 %d/%d 次尝试失败" % (attempt, MAX_SIGNIN_ATTEMPTS))
    lines.append("原因：%s" % str(reason or "未返回明确提交成功结果")[:300])
    lines.append("%d秒后进行第 %d/%d 次" % (
        SIGNIN_BACKOFF[attempt - 1], attempt + 1, MAX_SIGNIN_ATTEMPTS))
    lines.append("链接：%s" % str(item.get("link") or item.get("jump") or ""))
    return "\n".join(lines)


def enqueue_retry_receipt(data, key, job, reason, attempt, now=None):
    """Idempotently persist one mirrored progress receipt for attempts 1..5."""
    attempt = int(attempt)
    if attempt < 1 or attempt >= MAX_SIGNIN_ATTEMPTS:
        return None
    return enqueue(data, retry_receipt_key(key, job, attempt), "signin",
                   retry_receipt_text(job, reason, attempt), now)


def receipt_text(item, ok, message, payload=None, attempts=None, source="auto"):
    item = item if isinstance(item, dict) else {}
    payload = payload if isinstance(payload, dict) else {}
    title = str(payload.get("title") or item.get("title") or "").strip()
    lines = ["【白果云·%s签到】" % ("手动" if source == "manual" else "自动")]
    if title:
        lines.append("活动：%s" % title[:120])
    lines.append("签到类型：%s" % signin_type_name(item, payload))
    lines.append("结果：%s" % ("签到成功" if ok else "签到失败"))
    reason = str(message or ("提交已确认成功" if ok else "未返回明确成功结果"))[:300]
    # 签到成功时附上本次使用的签到码（普通签到无码则为空，不追加）
    used_code = str(payload.get("signInCode") or "").strip()
    if ok and used_code:
        reason = "%s，用码 %s" % (reason, used_code)
    lines.append("原因：%s" % reason)
    if attempts is not None:
        lines.append("尝试次数：%s/%s（首次 + 最多 5 次重试）" % (attempts, MAX_SIGNIN_ATTEMPTS))
    lines.append("链接：%s" % str(item.get("link") or item.get("jump") or ""))
    return "\n".join(lines)


def signin_receipt_key(key, job):
    """回执键：同一活动的每次重置（run_id）各一份，避免不同轮次互相覆盖。"""
    suffix = (":" + str(job["run_id"])) if (isinstance(job, dict) and job.get("run_id")) else ""
    return "signin:" + key + suffix


def finish_job(data, key, ok, reason, payload=None, status=None):
    job = data.setdefault("signin_jobs", {}).get(key)
    if not job or job.get("status") in ("succeeded", "failed", "cancelled"):
        return
    job.update(status=status or ("succeeded" if ok else "failed"),
               reason=str(reason), finished_at=time.time(), next_attempt_at=None)
    # Every reset gets a fresh receipt key; legacy jobs retain their historical key.
    receipt_key = signin_receipt_key(key, job)
    enqueue(data, receipt_key, "signin", receipt_text(
        job.get("item"), ok, reason, payload, job.get("attempts", 0), signin_source(job)))
    audit(data, "signin_finished", key, status=job["status"],
          source=signin_source(job), run_id=job.get("run_id"),
          attempts=job.get("attempts", 0), reason=str(reason))


def cancel_signins(data, reason, source=None):
    """Cancel all jobs, or just one source (used by auto-sign-in off)."""
    for key, job in list(data.setdefault("signin_jobs", {}).items()):
        if source is None or signin_source(job) == source:
            finish_job(data, key, False, reason, status="cancelled")


def has_due(data, now=None):
    now = time.time() if now is None else now
    for entry in data.get("outbox", {}).values():
        if entry.get("status") not in OUTBOX_TERMINAL and (entry.get("kind") != "notice" or data.get("notice_push")):
            if (entry.get("next_attempt_at") or 0) <= now:
                return True
    # 限流冷却期内不下发签到任务（outbox 推送不受影响：那不请求白果云）。
    if int(data.get("rate_limited_until") or 0) > now:
        return False
    for job in data.get("signin_jobs", {}).values():
        if (job.get("status") in ("pending", "running") and signin_enabled(data, job)
                and (job.get("next_attempt_at") or 0) <= now):
            return True
    return False


def discover(data, items, notice_text, now=None):
    """Seen ledger + all eligible outbox records + jobs in ONE caller transaction."""
    now = time.time() if now is None else now
    seen_list = list(dict.fromkeys(data.get("seen_keys") or []))
    seen = set(seen_list)
    for item in sorted(items, key=lambda row: row.get("ts") or 0):
        key = str(item.get("key") or "")
        if not key or key in seen:
            continue
        # 只给本轮确实被消费的消息记 seen：通知推送关时不消费通知、
        # 自动签到关时不消费签到活动（与 keepalive 不写水位、commit 守卫
        # "don't consume disabled arrivals" 同一约定）；否则开关重新打开后，
        # 关闭期间到达的消息会因已记 seen 而永远不补推。
        consumed = False
        if data.get("notice_push"):
            enqueue(data, "notice:" + key, "notice", notice_text(item), now)
            consumed = True
        if (data.get("auto_sign_in") and item.get("kind") == "busi"
                and "签到" in str(item.get("text") or "")
                and item.get("signin_activity_key")):
            enqueue_signin(data, item, "auto", now)
            consumed = True
        if not consumed:
            continue
        seen.add(key)
        seen_list.append(key)
        audit(data, "discovered", key)
    data["seen_keys"] = seen_list
    data["last_poll"] = now


def claim_outbox(data, key, claim_id, now):
    row = data.get("outbox", {}).get(key)
    if (not row or row.get("status") in OUTBOX_TERMINAL or (row.get("next_attempt_at") or 0) > now
            or (row.get("kind") == "notice" and not data.get("notice_push"))):
        return False
    row.update(status="sending", claim_id=claim_id,
               attempts=row.get("attempts", 0) + 1, last_attempt_at=now,
               next_attempt_at=now + LEASE_SECONDS)
    audit(data, "outbox_attempt", key, attempts=row["attempts"], claim_id=claim_id)


def ack_outbox(data, key, claim_id, acknowledged, reason="", now=None, succeeded=None):
    """确认一次 outbox 投递尝试。

    succeeded: 本次**新确认成功**的目标 [(channel_name, target_id), ...]（可选）。
               累积进 row["sent_targets"]，重试时据此跳过已成功目标，
               绝不重复推送（避免 TG 重复轰炸）。
    终态:
      acknowledged is True                 -> sent（全部目标都已投递）
      未确认 且 attempts >= 上限            -> failed（终态，停止重试）
      未确认 且 未到上限                    -> pending + backoff（保留该行继续重试）
    """
    now = time.time() if now is None else now
    row = data.get("outbox", {}).get(key)
    if not row or row.get("claim_id") != claim_id or row.get("status") != "sending":
        return False
    if succeeded:
        have = list(row.get("sent_targets") or [])
        for name, target_id in succeeded:
            token = target_key(name, target_id)
            if token not in have:
                have.append(token)
        row["sent_targets"] = have
    # bool True only; truthy strings/objects/integers are NOT channel acknowledgments.
    if acknowledged is True:
        row.update(status="sent", sent_at=now, next_attempt_at=None, last_error="")
    else:
        last_error = str(reason or "channel did not acknowledge True")
        if row.get("attempts", 0) >= MAX_DELIVERY_ATTEMPTS:
            row.update(status="failed", last_error=last_error, finished_at=now,
                       next_attempt_at=None)
        else:
            row.update(status="pending", last_error=last_error,
                       next_attempt_at=now + backoff(row["attempts"]))
    if acknowledged is True:
        event = "outbox_ack"
    elif row.get("status") == "failed":
        event = "outbox_failed"
    else:
        event = "outbox_retry"
    audit(data, event, key, attempts=row["attempts"], reason=row.get("last_error", ""),
          sent_targets=list(row.get("sent_targets") or []))
    return True


def claim_signin(data, key, claim_id, now):
    row = data.get("signin_jobs", {}).get(key)
    if (not row or not signin_enabled(data, row)
            or row.get("status") not in ("pending", "running")
            or int(data.get("rate_limited_until") or 0) > now   # 冷却期内不碰上游
            or (row.get("next_attempt_at") or 0) > now):
        return False
    if row.get("attempts", 0) >= MAX_SIGNIN_ATTEMPTS:
        finish_job(data, key, False, row.get("reason") or "最终尝试未得到确认（进程中断或超时）")
        return
    row.update(status="running", claim_id=claim_id,
               attempts=row.get("attempts", 0) + 1, last_attempt_at=now,
               next_attempt_at=now + LEASE_SECONDS)
    audit(data, "signin_attempt", key, attempts=row["attempts"], claim_id=claim_id)


def abandon_signin_claim(data, key, claim_id):
    """A switch changed after reservation but before transport invocation."""
    row = data.get("signin_jobs", {}).get(key)
    if not row or row.get("claim_id") != claim_id:
        return False
    row["attempts"] = max(0, row.get("attempts", 0) - 1)
    row.pop("claim_id", None)
    if row.get("status") in ("pending", "running"):
        if signin_source(row) == "auto":
            finish_job(data, key, False, "自动签到已关闭，取消剩余重试", status="cancelled")
        else:
            row.update(status="pending", next_attempt_at=time.time())
    audit(data, "signin_reservation_cancelled", key, attempts=row["attempts"],
          source=signin_source(row))


def complete_signin(data, key, claim_id, message, payload, now=None):
    now = time.time() if now is None else now
    row = data.get("signin_jobs", {}).get(key)
    if not row or row.get("claim_id") != claim_id:
        return False
    commit = payload.get("commit") if isinstance(payload, dict) else None
    ok = isinstance(commit, dict) and commit.get("success") is True
    if row.get("status") != "running":
        # 竞态：网络请求在网期间用户关闭了自动签到 → 任务已被取消，但上游其实已经
        # 提交成功。必须如实记为成功，并纠正已入队的"失败"回执（未投递则改写正文，
        # 已投递则补发一条成功回执），否则用户会收到与事实相反的失败通知。
        if row.get("status") != "cancelled" or not ok:
            return False
        text = receipt_text(row.get("item"), True, message, payload,
                            row.get("attempts", 0), signin_source(row))
        stale = (data.get("outbox") or {}).get(signin_receipt_key(key, row))
        if isinstance(stale, dict) and stale.get("status") not in OUTBOX_TERMINAL:
            stale["text"] = text                      # 尚未投递：直接改正文案
        elif isinstance(stale, dict):
            enqueue(data, signin_receipt_key(key, row) + ":after-cancel", "signin", text)
        row["status"] = "running"                     # 让 finish_job 接受这次真实成功
        finish_job(data, key, True, message, payload)
        audit(data, "signin_succeeded_after_cancel", key,
              attempts=row.get("attempts", 0), reason=str(message))
        return True
    reason = str(message or ("签到成功" if ok else "未返回明确提交成功结果"))
    if ok or row["attempts"] >= MAX_SIGNIN_ATTEMPTS:
        finish_job(data, key, ok, reason, payload)
    else:
        row.update(status="pending", reason=reason,
                   next_attempt_at=now + SIGNIN_BACKOFF[row["attempts"] - 1])
        enqueue_retry_receipt(data, key, row, reason, row["attempts"], now)
        audit(data, "signin_retry", key, attempts=row["attempts"],
              next_attempt_at=row["next_attempt_at"], reason=reason)


class AccountStore:
    def __init__(self, directory):
        self.directory = os.path.abspath(directory)

    def path(self, account):
        account = str(account)
        if not account or not account.isdigit():
            raise ValueError("account must contain digits only")
        return os.path.join(self.directory, account + ".json")

    @contextmanager
    def locked(self, account):
        path = self.path(account)
        os.makedirs(self.directory, exist_ok=True)
        with _LOCKS_GUARD:
            lock = _LOCKS.setdefault(path, threading.RLock())
        with lock:
            with open(path + ".lock", "a+b") as handle:
                if os.name == "nt":
                    import msvcrt
                    handle.seek(0, os.SEEK_END)
                    if handle.tell() == 0:
                        handle.write(b"\0")
                        handle.flush()
                    handle.seek(0)
                    msvcrt.locking(handle.fileno(), msvcrt.LK_LOCK, 1)
                else:
                    import fcntl
                    fcntl.flock(handle.fileno(), fcntl.LOCK_EX)
                try:
                    yield path
                finally:
                    if os.name == "nt":
                        handle.seek(0)
                        msvcrt.locking(handle.fileno(), msvcrt.LK_UNLCK, 1)
                    else:
                        fcntl.flock(handle.fileno(), fcntl.LOCK_UN)

    def _read(self, path):
        try:
            with open(path, encoding="utf-8") as handle:
                data = json.load(handle)
        except FileNotFoundError:
            return None
        if not isinstance(data, dict):
            raise ValueError("account state is not a JSON object")
        return data

    def _sync_directory(self):
        if os.name != "nt":
            fd = os.open(self.directory, os.O_RDONLY)
            try:
                os.fsync(fd)
            finally:
                os.close(fd)

    def _write(self, path, data):
        fd, tmp = tempfile.mkstemp(prefix=os.path.basename(path) + ".", suffix=".tmp", dir=self.directory)
        try:
            with os.fdopen(fd, "w", encoding="utf-8") as handle:
                json.dump(data, handle, ensure_ascii=False, indent=2)
                handle.flush()
                os.fsync(handle.fileno())
            os.replace(tmp, path)
            self._sync_directory()
        finally:
            if os.path.exists(tmp):
                os.unlink(tmp)

    def _normalize(self, path, data):
        if data is not None and ("_generation" not in data or "_revision" not in data):
            data.setdefault("_generation", uuid.uuid4().hex)
            data.setdefault("_revision", 0)
            self._write(path, data)
        return data

    def load(self, account):
        with self.locked(account) as path:
            return copy.deepcopy(self._normalize(path, self._read(path)))

    def mutate(self, account, fn, generation=None, create=False):
        """Return committed fresh snapshot, or None for absent/wrong generation.

        fn modifies the latest dict; returning False aborts. Creation is explicit
        and must only be used by initial binding, never asynchronous completions.
        """
        with self.locked(account) as path:
            data = self._normalize(path, self._read(path))
            if data is None:
                if not create or generation is not None:
                    return None
                data = {"account": str(account), "_generation": uuid.uuid4().hex, "_revision": 0}
            if generation is not None and data.get("_generation") != generation:
                return None
            if fn(data) is False:
                return None
            prune_terminal(data)
            data["account"] = str(account)
            data["_revision"] += 1
            data["updated_at"] = time.time()
            self._write(path, data)
            return copy.deepcopy(data)

    def save(self, account, snapshot):
        """Compatibility CAS. Never implicitly creates/resurrects an account."""
        def replace(data):
            if (data.get("_generation"), data.get("_revision")) != (
                    snapshot.get("_generation"), snapshot.get("_revision")):
                return False
            data.clear()
            data.update(copy.deepcopy(snapshot))
        return self.mutate(account, replace, generation=snapshot.get("_generation"))

    def delete(self, account, generation=None):
        with self.locked(account) as path:
            data = self._normalize(path, self._read(path))
            if data is None or (generation is not None and data.get("_generation") != generation):
                return False
            os.unlink(path)
            self._sync_directory()
            return True


# ==================== 签到 URL 解析 / 签到接口（原 ginkgostu_signin.py 独有部分） ====================
# 来源：hk 主机 /home/openapi/source/dufe/ginkgostu_sign_in/ginkgo_sign_in.py
#       （2026-09-14 拉取，md5 92a6753fe39cd025a954bc8e6553f29a）
# AES / SSO 登录 / 会话部分与上方轮询登录共用一份实现，不再重复。


class NoSignInCodeError(Exception):
    """The numeric-code activity exists but has no code configured."""


class UnsupportedSigninType(Exception):
    """The activity requires a flow this module cannot safely automate."""


SIGNIN_TYPE_NAMES = {
    0: "二维码签到",
    1: "数字码签到",
    2: "普通签到",
    3: "直播签到",
}


def signin_type_label(value):
    try:
        signin_type = int(value)
    except (TypeError, ValueError):
        return "未知签到类型"
    return SIGNIN_TYPE_NAMES.get(signin_type, "未知签到类型(%s)" % signin_type)


def parse_course_url(url):
    parts = urlsplit(str(url or ""))
    m = re.fullmatch(
        r"/course/detail/(?P<courseId>[0-9A-Za-z]+)/(?P<batchNo>[^/]+)/"
        r"(?P<weekTimeId>\d+)/signin/(?P<signInId>\d+)/(?P<type>\d+)/?",
        parts.path)
    try:
        port = parts.port
    except ValueError:
        port = -1
    if (parts.scheme != "https" or parts.username is not None or parts.password is not None
            or port not in (None, 443)
            or (parts.hostname or "").lower() != "ginkgostu.dufe.edu.cn" or not m):
        raise ValueError(
            "URL is not a sign-in page (missing /signin/<signInId>/<type>)")
    return m.groupdict()


def _request(sess, method, *args, **kwargs):
    """Also guard fake sessions and HTTPError.response before any JSON parsing."""
    try:
        response = getattr(sess, method)(*args, **kwargs)
    except requests.RequestException as exc:
        raise UpstreamError(safe_error(exc)) from None
    return check_response(response, "signin")


def api_headers(token, fingerprint=None):
    return {
        "User-Agent": UA,
        "Origin": GINKGO,
        "Referer": GINKGO + "/",
        "Content-Type": "application/x-www-form-urlencoded",
        "token": token,
        "fingerprint": fingerprint or uuid.uuid4().hex,
    }


def call_api(sess, path, data, token):
    return _request(sess, "post", GINKGO + path, data=data, timeout=20,
                    headers=api_headers(token))


def commit_sign_in(sess, token, sign_in_id, batch_no, referer, code):
    """
    POST /student/signIn/InTheSign to actually check in with the given code.

    Mirrors the browser request:
        headers: batchno, fingerprint, token, Referer (course page)
        body:    signInId=<id>&code=<code>

    Returns (ok: bool, message: str, raw: dict|None).
    """
    headers = api_headers(token)
    if batch_no:
        # NOTE: the browser sends a "batchno" header that does not always equal
        # the batchNo segment of the course URL.  We pass the URL segment here;
        # the server appears not to enforce it.
        headers["batchno"] = str(batch_no)
    headers["Referer"] = referer

    try:
        r = _request(sess, "post",
            GINKGO + "/student/signIn/InTheSign",
            data={"signInId": str(sign_in_id), "code": str(code)},
            timeout=20,
            headers=headers,
        )
    except (requests.RequestException, UpstreamError) as e:
        return False, "request to InTheSign failed: %s" % safe_error(e), None

    if r.status_code != 200:
        return False, "InTheSign returned HTTP %d" % r.status_code, None

    try:
        j = r.json()
    except ValueError:
        return False, "InTheSign returned non-JSON", None

    if not isinstance(j, dict):
        return False, "Malformed InTheSign response", None
    api_code = str(j.get("code", ""))
    if api_code == "2000":
        return True, "accepted", j
    # Do not echo untrusted server messages/codes: they may contain credentials.
    return False, "InTheSign refused the sign-in request", j


def fetch_sign_in_info(username, password, course_url, do_commit=True, sess=None, token=None,
                       avoid_duplicate=False):
    route = parse_course_url(course_url)
    sign_in_id = route["signInId"]
    batch_no = route.get("batchNo")
    owns_session = sess is None or not token

    try:
        if owns_session:
            # 无外部登录态时自登录（保持向后兼容）；本函数负责关闭自建会话。
            sess, token, _student_name = login(username, password)
        # 否则：sess/token 由调用方（轮询缓存会话）提供，跳过 SSO 登录

        r = call_api(sess, "/student/signIn/getSignInInfo",
                     {"signInId": sign_in_id}, token)
        check_response(r, "getSignInInfo")
        try:
            j = r.json()
        except ValueError:
            raise UpstreamError("getSignInInfo returned non-JSON") from None

        if not isinstance(j, dict):
            raise UpstreamError("Malformed getSignInInfo response")
        data = j.get("data") or {}
        if not isinstance(data, dict):
            raise UpstreamError("Malformed getSignInInfo data")
        model = data.get("signInModel") or {}
        if not isinstance(model, dict):
            raise UpstreamError("Malformed getSignInInfo model")
        code = model.get("code")
        raw_type = model.get("type")
        if raw_type is None or raw_type == "":
            raw_type = route.get("type")
        try:
            signin_type = int(raw_type)
        except (TypeError, ValueError):
            raise UpstreamError("Malformed sign-in type") from None
        type_name = signin_type_label(signin_type)

        if signin_type == 1 and (code is None or code == ""):
            raise NoSignInCodeError("数字码签到未配置签到码")
        if signin_type == 2:
            code = ""

        result = {
            "signInId":   sign_in_id,
            "signInCode": code,
            "type":       signin_type,
            "typeName":   type_name,
            "title":      model.get("title"),
            "status":     data.get("signStatus"),
        }

        if do_commit:
            if avoid_duplicate and str(result["status"]) == "1":
                # A prior attempt may have committed before its response was lost.
                # Confirm the current state instead of blindly posting InTheSign again.
                result["commit"] = {"success": True, "message": "已确认此前签到成功"}
            elif signin_type not in (1, 2):
                result["commit"] = {
                    "success": False,
                    "message": "%s暂不支持自动签到，需特定流程" % type_name,
                }
            else:
                ok, reason, _raw = commit_sign_in(
                    sess, token, sign_in_id, batch_no, course_url, code
                )
                result["commit"] = {"success": ok, "message": reason}

        return result
    finally:
        if owns_session and sess is not None:
            try:
                sess.close()
            except Exception:
                pass



# ==================== 常量 ====================
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
STATE_DIR = os.path.join(BASE_DIR, "ginkgostu")

GINKGO = "https://ginkgostu.dufe.edu.cn"
SSO = "https://sso.dufe.edu.cn"
NOTICE_PAGE = GINKGO + "/notice"
SYSTEM_PAGE = GINKGO + "/notice/system"

# SSO 登录页固定 CSRF 常量（同时作为密码加密的 key/iv 来源）
TOKEN_CONST = "616246bd4a7266f46764686240a106f0"

UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
      "(KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36")

HTTP_TIMEOUT = 25
PAGE_SIZE = 20            # 单类单次抓取条数
LIST_MAX = 20             # .ginkgo notice <int> 上限
PUSH_MAX = 5              # 单轮最多逐条推送的新消息数
# 去重账本 seen_keys 有意保持"精确且不截断"：它是唯一能阻止旧消息被重新当成
# 新消息推送的凭据（通知接口只回最新 20 条，实际增长很慢）。
ACCOUNT_CACHE_SECONDS = 5
ACCOUNT_CONCURRENCY = 5        # 并发预备：同时轮询的账号数上限
RATE_LIMIT_COOLDOWN = 1800     # 触发限流后的建议冷却时间（秒）
# 限流判定统一走 is_rate_limited()（自带错误码与文案集合），此处不再重复定义。

TZ8 = timezone(timedelta(hours=8))

_requests_log_silenced = False


# ==================== 纯 Python AES-128（取自 openapi ginkgostu_sign_in 服务，验证过） ====================
_SBOX = [
    0x63, 0x7c, 0x77, 0x7b, 0xf2, 0x6b, 0x6f, 0xc5, 0x30, 0x01, 0x67, 0x2b, 0xfe, 0xd7, 0xab, 0x76,
    0xca, 0x82, 0xc9, 0x7d, 0xfa, 0x59, 0x47, 0xf0, 0xad, 0xd4, 0xa2, 0xaf, 0x9c, 0xa4, 0x72, 0xc0,
    0xb7, 0xfd, 0x93, 0x26, 0x36, 0x3f, 0xf7, 0xcc, 0x34, 0xa5, 0xe5, 0xf1, 0x71, 0xd8, 0x31, 0x15,
    0x04, 0xc7, 0x23, 0xc3, 0x18, 0x96, 0x05, 0x9a, 0x07, 0x12, 0x80, 0xe2, 0xeb, 0x27, 0xb2, 0x75,
    0x09, 0x83, 0x2c, 0x1a, 0x1b, 0x6e, 0x5a, 0xa0, 0x52, 0x3b, 0xd6, 0xb3, 0x29, 0xe3, 0x2f, 0x84,
    0x53, 0xd1, 0x00, 0xed, 0x20, 0xfc, 0xb1, 0x5b, 0x6a, 0xcb, 0xbe, 0x39, 0x4a, 0x4c, 0x58, 0xcf,
    0xd0, 0xef, 0xaa, 0xfb, 0x43, 0x4d, 0x33, 0x85, 0x45, 0xf9, 0x02, 0x7f, 0x50, 0x3c, 0x9f, 0xa8,
    0x51, 0xa3, 0x40, 0x8f, 0x92, 0x9d, 0x38, 0xf5, 0xbc, 0xb6, 0xda, 0x21, 0x10, 0xff, 0xf3, 0xd2,
    0xcd, 0x0c, 0x13, 0xec, 0x5f, 0x97, 0x44, 0x17, 0xc4, 0xa7, 0x7e, 0x3d, 0x64, 0x5d, 0x19, 0x73,
    0x60, 0x81, 0x4f, 0xdc, 0x22, 0x2a, 0x90, 0x88, 0x46, 0xee, 0xb8, 0x14, 0xde, 0x5e, 0x0b, 0xdb,
    0xe0, 0x32, 0x3a, 0x0a, 0x49, 0x06, 0x24, 0x5c, 0xc2, 0xd3, 0xac, 0x62, 0x91, 0x95, 0xe4, 0x79,
    0xe7, 0xc8, 0x37, 0x6d, 0x8d, 0xd5, 0x4e, 0xa9, 0x6c, 0x56, 0xf4, 0xea, 0x65, 0x7a, 0xae, 0x08,
    0xba, 0x78, 0x25, 0x2e, 0x1c, 0xa6, 0xb4, 0xc6, 0xe8, 0xdd, 0x74, 0x1f, 0x4b, 0xbd, 0x8b, 0x8a,
    0x70, 0x3e, 0xb5, 0x66, 0x48, 0x03, 0xf6, 0x0e, 0x61, 0x35, 0x57, 0xb9, 0x86, 0xc1, 0x1d, 0x9e,
    0xe1, 0xf8, 0x98, 0x11, 0x69, 0xd9, 0x8e, 0x94, 0x9b, 0x1e, 0x87, 0xe9, 0xce, 0x55, 0x28, 0xdf,
    0x8c, 0xa1, 0x89, 0x0d, 0xbf, 0xe6, 0x42, 0x68, 0x41, 0x99, 0x2d, 0x0f, 0xb0, 0x54, 0xbb, 0x16,
]
_INV_SBOX = [0] * 256
for _i, _v in enumerate(_SBOX):
    _INV_SBOX[_v] = _i
_RCON = [0x01, 0x02, 0x04, 0x08, 0x10, 0x20, 0x40, 0x80, 0x1B, 0x36]


def _gmul(a, b):
    p = 0
    for _ in range(8):
        if b & 1:
            p ^= a
        hi = a & 0x80
        a = (a << 1) & 0xFF
        if hi:
            a ^= 0x1B
        b >>= 1
    return p


def _expand_key(key):
    w = [list(key[4 * i:4 * i + 4]) for i in range(4)]
    for i in range(4, 44):
        t = list(w[i - 1])
        if i % 4 == 0:
            t = t[1:] + t[:1]
            t = [_SBOX[b] for b in t]
            t[0] ^= _RCON[i // 4 - 1]
        w.append([w[i - 4][j] ^ t[j] for j in range(4)])
    return w


def _add_round_key(s, w, rnd):
    for c in range(4):
        for r in range(4):
            s[r][c] ^= w[rnd * 4 + c][r]


def _sub_bytes(s, box):
    for r in range(4):
        for c in range(4):
            s[r][c] = box[s[r][c]]


def _shift_rows(s):
    for r in range(1, 4):
        s[r] = s[r][r:] + s[r][:r]


def _inv_shift_rows(s):
    for r in range(1, 4):
        s[r] = s[r][-r:] + s[r][:-r]


def _mix_columns(s):
    for c in range(4):
        a = [s[r][c] for r in range(4)]
        s[0][c] = _gmul(a[0], 2) ^ _gmul(a[1], 3) ^ a[2] ^ a[3]
        s[1][c] = a[0] ^ _gmul(a[1], 2) ^ _gmul(a[2], 3) ^ a[3]
        s[2][c] = a[0] ^ a[1] ^ _gmul(a[2], 2) ^ _gmul(a[3], 3)
        s[3][c] = _gmul(a[0], 3) ^ a[1] ^ a[2] ^ _gmul(a[3], 2)


def _inv_mix_columns(s):
    for c in range(4):
        a = [s[r][c] for r in range(4)]
        s[0][c] = _gmul(a[0], 14) ^ _gmul(a[1], 11) ^ _gmul(a[2], 13) ^ _gmul(a[3], 9)
        s[1][c] = _gmul(a[0], 9) ^ _gmul(a[1], 14) ^ _gmul(a[2], 11) ^ _gmul(a[3], 13)
        s[2][c] = _gmul(a[0], 13) ^ _gmul(a[1], 9) ^ _gmul(a[2], 14) ^ _gmul(a[3], 11)
        s[3][c] = _gmul(a[0], 11) ^ _gmul(a[1], 13) ^ _gmul(a[2], 9) ^ _gmul(a[3], 14)


class _AES128:
    def __init__(self, key):
        if len(key) != 16:
            raise ValueError("AES-128 requires a 16-byte key, got %d" % len(key))
        self.w = _expand_key(key)

    def encrypt_block(self, block):
        s = [[block[r + 4 * c] for c in range(4)] for r in range(4)]
        _add_round_key(s, self.w, 0)
        for rnd in range(1, 10):
            _sub_bytes(s, _SBOX)
            _shift_rows(s)
            _mix_columns(s)
            _add_round_key(s, self.w, rnd)
        _sub_bytes(s, _SBOX)
        _shift_rows(s)
        _add_round_key(s, self.w, 10)
        return bytes(s[r][c] for c in range(4) for r in range(4))

    def decrypt_block(self, block):
        s = [[block[r + 4 * c] for c in range(4)] for r in range(4)]
        _add_round_key(s, self.w, 10)
        for rnd in range(9, 0, -1):
            _inv_shift_rows(s)
            _sub_bytes(s, _INV_SBOX)
            _add_round_key(s, self.w, rnd)
            _inv_mix_columns(s)
        _inv_shift_rows(s)
        _sub_bytes(s, _INV_SBOX)
        _add_round_key(s, self.w, 0)
        return bytes(s[r][c] for c in range(4) for r in range(4))


def _pkcs7_unpad(data):
    if not data:
        return data
    n = data[-1]
    if 1 <= n <= 16 and data[-n:] == bytes([n]) * n:
        return data[:-n]
    return data


def _zero_pad(data):
    if len(data) % 16:
        data += b"\x00" * (16 - len(data) % 16)
    return data


def aes_ecb_decrypt_b64(cipher_b64, key_text):
    """滑块答案：密文 base64 + 以 token 前 16 字节为密钥的 AES-128-ECB/PKCS7"""
    aes = _AES128(key_text[:16].encode("utf-8"))
    raw = base64.b64decode(cipher_b64)
    out = b"".join(aes.decrypt_block(raw[i:i + 16]) for i in range(0, len(raw), 16))
    return _pkcs7_unpad(out)


def aes_cbc_encrypt_b64_zero(plain, key_text):
    """登录表单密码：AES-128-CBC + ZeroPadding，key=iv=__token__ 前 16 字节"""
    key = key_text[:16].encode("utf-8")
    aes = _AES128(key)
    data = _zero_pad(plain.encode("utf-8"))
    out, prev = b"", key
    for i in range(0, len(data), 16):
        blk = bytes(x ^ y for x, y in zip(data[i:i + 16], prev))
        prev = aes.encrypt_block(blk)
        out += prev
    return base64.b64encode(out).decode("ascii")


# ==================== 文本 / 时间工具 ====================
_TAG_RE = re.compile(r"<[^>]+>")


def _html_to_text(raw):
    if not raw:
        return ""
    text = re.sub(r"(?i)<br\s*/?>", "\n", raw)
    text = re.sub(r"(?i)</(p|div|tr|li)>", "\n", text)
    text = _TAG_RE.sub("", text)
    text = _html.unescape(text)
    text = re.sub(r"[ \t\u3000]+", " ", text)
    text = re.sub(r"\n{2,}", "\n", text)
    return text.strip()


def _clip(text, limit):
    text = (text or "").strip()
    if len(text) <= limit:
        return text
    return text[:limit - 1] + "…"


def _now_ts():
    return time.time()


def _fmt_ts(ts):
    if not ts:
        return ""
    return datetime.fromtimestamp(ts, TZ8).strftime("%Y-%m-%d %H:%M:%S")


def _parse_ts(value):
    text = (value or "").strip()
    if not text:
        return 0.0
    if text.endswith("Z"):
        text = text[:-1] + "+00:00"
    m = re.search(r"([+-]\d{2})(\d{2})$", text)
    if m:
        text = text[:m.start()] + m.group(1) + ":" + m.group(2)
    try:
        dt = datetime.fromisoformat(text)
    except ValueError:
        try:
            dt = datetime.strptime(text[:19], "%Y-%m-%d %H:%M:%S")
        except ValueError:
            return 0.0
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=TZ8)
    return dt.timestamp()


def _stamp():
    return datetime.now(TZ8).strftime("%Y-%m-%d %H:%M:%S")


# ==================== 异常 ====================
class AuthError(Exception):
    """账号密码错误 / 登录被拒"""


class UpstreamError(Exception):
    """远端接口异常或返回结构变化"""


# ==================== 状态文件 ====================
def _ensure_dir():
    os.makedirs(STATE_DIR, exist_ok=True)


def _account_file(account):
    return os.path.join(STATE_DIR, "%s.json" % account)


def _store():
    # Construct on demand so deployments/tests may relocate STATE_DIR safely.
    return AccountStore(STATE_DIR)


def _load_account(account):
    return _store().load(account)


def _mutate_account(account, fn, generation=None, create=False):
    def update(data):
        result = fn(data)
        if result is not False:
            data["updated"] = _stamp()
        return result
    result = _store().mutate(account, update, generation=generation, create=create)
    _state_cache["ts"] = 0.0
    return result


def _save_account(account, data):
    """Legacy snapshot API: stale snapshots fail closed; creation is bind-only."""
    result = _store().save(account, data)
    _state_cache["ts"] = 0.0
    return result


def _delete_account(account, generation=None):
    result = _store().delete(account, generation)
    _state_cache["ts"] = 0.0
    return result


_state_cache = {"ts": 0.0, "data": []}
_state_lock = threading.Lock()


def list_accounts(force=False):
    """列出全部已绑定账号（5 秒缓存，轮询循环每秒调用）"""
    now = time.time()
    with _state_lock:
        if not force and now - _state_cache["ts"] < ACCOUNT_CACHE_SECONDS:
            return _state_cache["data"]
    result = []
    try:
        names = [n for n in os.listdir(STATE_DIR) if n.endswith(".json")]
    except OSError:
        names = []
    for name in names:
        if not name[:-5].isdigit():
            continue
        try:
            data = _load_account(name[:-5])
        except (OSError, ValueError):
            # Corrupt/unreadable files are never treated as fresh accounts or
            # overwritten; isolate one bad account from the rest of the monitor.
            print("[ginkgo] 跳过无法读取的账号状态 %s" % name)
            continue
        if data:
            result.append(data)
    with _state_lock:
        _state_cache["ts"] = now
        _state_cache["data"] = result
    return result


def find_by_uid(uid):
    """统一身份：一个 uid 只允许绑定一个白果云账号"""
    for data in list_accounts(force=True):
        if str(data.get("user_id")) == str(uid):
            return data
    return None


def find_by_account(account):
    try:
        return _load_account(account)
    except (OSError, ValueError):
        # 与 list_accounts 同一约定：损坏/无法读取的状态文件按「未绑定」处理并打日志，
        # 绝不让 JSONDecodeError 逃出指令层（后续写入路径仍有自己的防护）。
        print("[ginkgo] 无法读取账号状态 %s，按未绑定处理" % account)
        return None


# ==================== 白果云协议层（同步，供 asyncio.to_thread 调用） ====================
def new_session():
    s = requests.Session()
    s.trust_env = False
    s.headers.update({"User-Agent": UA, "Accept-Language": "zh-CN,zh;q=0.9"})
    s.hooks["response"] = [lambda response, *args, **kwargs: check_response(response)]
    return s


def get_unify_config(sess):
    r = sess.get(GINKGO + "/student/getLoginUnifySet", timeout=HTTP_TIMEOUT)
    _raise_if_rate_limited(r, "白果云 getLoginUnifySet")
    r.raise_for_status()
    try:
        cfg = r.json()["data"]
    except (ValueError, KeyError, TypeError) as e:
        raise UpstreamError("getLoginUnifySet 返回异常: %s" % e)
    if not isinstance(cfg, dict):
        raise UpstreamError("getLoginUnifySet data 结构异常")
    if not cfg.get("loginUrl") or not cfg.get("clientId"):
        raise UpstreamError("getLoginUnifySet 缺少 loginUrl/clientId")
    return cfg


def open_authorize(sess, cfg):
    params = {
        "response_type": cfg.get("responseType") or "code",
        "client_id": cfg["clientId"],
        "redirect_uri": cfg.get("rollBackUrl") or (GINKGO + "/login"),
    }
    r = sess.get(cfg["loginUrl"], params=params, timeout=HTTP_TIMEOUT, allow_redirects=True)
    _raise_if_rate_limited(r, "SSO 授权页")
    r.raise_for_status()
    m = re.search(r'name="__token__"\s+value="([^"]+)"', r.text)
    if not m:
        raise UpstreamError("SSO 登录页未找到 __token__")
    return (m.group(1) or TOKEN_CONST), r.url


def exploit_slider(sess, referer):
    """滑块验证码：服务端直接下发答案（data.d 用 data.token 前 16 字节 EC B 解密即 x）"""
    url = (SSO + "/auth/widget?layer=image_verify&widget=Slider"
                  "&action=get_verify&type=login_image_verify")
    r = sess.post(url, data={"width": 280, "height": 155, "block_size": 80},
                  headers={"X-Requested-With": "XMLHttpRequest", "Referer": referer},
                  timeout=HTTP_TIMEOUT)
    _raise_if_rate_limited(r, "SSO 滑块验证码")
    r.raise_for_status()
    try:
        d = r.json()["data"]
        token, cipher = d["token"], d["d"]
    except (ValueError, KeyError, TypeError) as e:
        raise UpstreamError("滑块挑战返回异常: %s" % e)
    x = aes_ecb_decrypt_b64(cipher, token).decode("utf-8", "replace")
    return {"token": token, "x": x}


_ERR_HINTS = ("密码", "账号", "用户", "锁定", "失效", "验证", "错误", "不存在",
              "频繁", "限流", "限速", "稍后")


def _extract_login_error(page_text):
    """从 SSO 失败页里提取提示文案（登录失败时返回 200 + 错误信息）"""
    for pat in (r'id="error_container"[^>]*>(.*?)</div>',
                r'class="[^"]*(?:error|tip)[^"]*"[^>]*>(.*?)</div>'):
        m = re.search(pat, page_text or "", re.S)
        if m:
            txt = _html_to_text(m.group(1))
            if txt and any(h in txt for h in _ERR_HINTS):
                return _clip(txt, 120)
    return ""


def submit_login(sess, post_url, csrf, username, password, slider, referer):
    form = {
        "fingerprint": _uuid.uuid4().hex,
        "username": username,
        "password": aes_cbc_encrypt_b64_zero(password, csrf),
        "verify_token": slider["token"],
        "verify_code": slider["x"],
        "__token__": csrf,
    }
    return sess.post(post_url, data=form, timeout=HTTP_TIMEOUT, allow_redirects=False,
                     headers={"Referer": referer, "Origin": SSO,
                              "Content-Type": "application/x-www-form-urlencoded"})


def exchange_token(sess, code):
    r = sess.post(GINKGO + "/student/authLogin", data={"code": code}, timeout=HTTP_TIMEOUT,
                  headers={"Content-Type": "application/x-www-form-urlencoded"})
    _raise_if_rate_limited(r, "authLogin")
    try:
        j = r.json()
    except ValueError as e:
        raise UpstreamError("authLogin 返回非 JSON: %s" % e)
    if not isinstance(j, dict):
        raise UpstreamError("authLogin 返回结构异常")
    if str(j.get("code")) != "2000":
        if _looks_rate_limited(None, j):
            raise RateLimitedError("authLogin 被限流: %s" % json.dumps(j, ensure_ascii=False)[:200])
        raise AuthError("换取登录态被拒: %s" % json.dumps(j, ensure_ascii=False)[:200])
    data = j.get("data")
    if not isinstance(data, dict):
        raise UpstreamError("authLogin data 结构异常")
    return data


def login(account, password):
    """完整走一遍 SSO 流程，返回 (session, token, student_name)。

    成功时会话所有权转交调用方；任何失败都必须关闭已创建的 SSO 会话。
    """
    sess = new_session()
    try:
        cfg = get_unify_config(sess)
        csrf, page_url = open_authorize(sess, cfg)
        slider = exploit_slider(sess, page_url)
        resp = submit_login(sess, page_url, csrf, account, password, slider, page_url)
        check_response(resp, "SSO 登录")
        loc = resp.headers.get("Location", "") or ""
        m = re.search(r"[?&]code=([A-Za-z0-9_\-]+)", loc)
        if not m:
            if resp.status_code == 200:
                hint = _extract_login_error(resp.text)
            else:
                hint = ""
            if _looks_rate_limited(None, None, hint):
                raise RateLimitedError("SSO 登录被限流：%s" % (hint or "访问过于频繁"))
            raise AuthError(hint or "账号或密码错误（登录未返回授权码）")
        data = exchange_token(sess, m.group(1))
        token = data.get("token")
        if not token:
            raise UpstreamError("authLogin 未返回 token")
        return sess, token, (data.get("studentName") or "").strip()
    except BaseException:
        try:
            sess.close()
        except Exception:
            pass
        raise


class GinkgoSession:
    """单账号登录态（含失效自动重登）"""

    def __init__(self, account, password):
        self.account = account
        self.password = password
        self.req = None
        self.token = None
        self.student_name = ""
        self.lock = threading.Lock()

    def _login_locked(self):
        old_req = self.req
        req, token, student_name = login(self.account, self.password)
        self.req, self.token, self.student_name = req, token, student_name
        if old_req is not None and old_req is not req:
            try:
                old_req.close()
            except Exception:
                pass

    def ensure(self):
        with self.lock:
            if self.req is None or not self.token:
                self._login_locked()
            return self.req, self.token

    def refresh(self):
        with self.lock:
            self._login_locked()
            return self.req, self.token


_sessions = {}
_sessions_lock = threading.Lock()


def get_session(account, password):
    with _sessions_lock:
        sess = _sessions.get(account)
        if sess is None or sess.password != password:
            if sess is not None and sess.req is not None:
                try:
                    sess.req.close()
                except Exception:
                    pass
            sess = GinkgoSession(account, password)
            _sessions[account] = sess
    return sess


def _api_headers(token):
    return {
        "User-Agent": UA,
        "Origin": GINKGO,
        "Referer": GINKGO + "/",
        "Content-Type": "application/x-www-form-urlencoded",
        "token": token,
        "fingerprint": _uuid.uuid4().hex,
    }


def _looks_rate_limited(status=None, payload=None, text=""):
    return is_rate_limited(status, payload, text)


def _raise_if_rate_limited(r, what):
    return check_response(r, what)


def _safe_json(r, what):
    check_response(r, what)
    try:
        payload = r.json()
    except ValueError:
        raise UpstreamError("%s 返回非 JSON（HTTP %s）" % (what, r.status_code))
    if not isinstance(payload, dict):
        raise UpstreamError("%s JSON 必须为对象" % what)
    return payload


def _need_relogin(r, payload):
    if r.status_code in (401, 403):
        return True
    if isinstance(payload, dict) and str(payload.get("code")) in ("4000", "4001", "4010"):
        return True
    return False


def _api_request(sess, method, path, params, body):
    req, token = sess.ensure()
    headers = _api_headers(token)
    if method == "GET":
        return req.get(GINKGO + path, params=params, headers=headers, timeout=HTTP_TIMEOUT)
    headers["Content-Type"] = "application/json"
    return req.post(GINKGO + path, data=json.dumps(body if body is not None else {}),
                    headers=headers, timeout=HTTP_TIMEOUT)


def _check_limited(r, what):
    return check_response(r, what)


def _call(sess, method, path, params=None, body=None, what="接口"):
    r = _api_request(sess, method, path, params, body)
    _check_limited(r, what)
    payload = _safe_json(r, what)
    if _looks_rate_limited(None, payload):
        raise RateLimitedError("%s 触发频率限制：%s"
                               % (what, payload.get("message") or payload.get("code")))
    if _need_relogin(r, payload):
        sess.refresh()
        r = _api_request(sess, method, path, params, body)
        _check_limited(r, what)
        payload = _safe_json(r, what)
        if _looks_rate_limited(None, payload):
            raise RateLimitedError("%s 触发频率限制：%s"
                                   % (what, payload.get("message") or payload.get("code")))
    if str(payload.get("code")) != "2000":
        raise UpstreamError("%s 返回 code=%s message=%s"
                            % (what, payload.get("code"), payload.get("message")))
    return payload


def _normalize_notice(raw):
    title = (raw.get("title") or "").strip() or "(无标题)"
    ts = _parse_ts(raw.get("createTime"))
    return {
        "kind": "notice",
        "kinder": "通知",
        "key": "N%s" % raw.get("id"),
        "id": str(raw.get("id") or ""),
        "title": title,
        "text": _html_to_text(raw.get("content") or ""),
        "ts": ts,
        "time": _fmt_ts(ts),
        "link": NOTICE_PAGE,
        "sender": (raw.get("createUser") or "").strip(),
        "jump": "",
    }


def _normalize_busi(raw):
    text = _html_to_text(raw.get("content") or "")
    jump = (raw.get("jumpUrl") or "").strip()
    if jump.startswith("http"):
        link = jump
    elif jump.startswith("/"):
        link = GINKGO + jump
    else:
        link = SYSTEM_PAGE
    ts = _parse_ts(raw.get("createTime"))
    return {
        "kind": "busi",
        "kinder": "互动消息",
        "key": "B%s" % (raw.get("uuid") or ""),
        "uuid": str(raw.get("uuid") or ""),
        "id": "",
        "title": _clip(text, 120) or "(无内容)",
        "text": text,
        "ts": ts,
        "time": _fmt_ts(ts),
        "link": link,
        "sender": "",
        "jump": jump,
    }


def fetch_all(sess, page_size=PAGE_SIZE):
    """取两类消息（均为全部，含已读；去重由调用方按 key 处理）"""
    j1 = _call(sess, "GET", "/student/message/getMessageAll",
               params={"pageSize": page_size, "pageNo": 1, "readFlag": 0}, what="通知列表")
    notices = [_normalize_notice(x) for x in ((j1.get("data") or {}).get("list") or [])]
    j2 = _call(sess, "POST", "/student/busiMessage/list", body={}, what="互动消息列表")
    busi = [_normalize_busi(x) for x in (j2.get("data") or [])]
    return notices, busi


def _parse_signin_route(link):
    return parse_course_url(link)


def _signin_type_label(value):
    return signin_type_label(value)


def _validated_signin_item(item):
    """Return an annotated copy only for a real interactive sign-in activity."""
    if (not isinstance(item, dict) or item.get("kind") != "busi"
            or "签到" not in str(item.get("text") or "")):
        return None
    link = str(item.get("link") or "").strip()
    try:
        route = _parse_signin_route(link)
    except (ValueError, TypeError, AttributeError):
        return None
    row = dict(item)
    # Persist/display only the canonical path: query parameters may carry tokens
    # and must never reach receipts or push logs.
    row["link"] = GINKGO + urlsplit(link).path
    # The message UUID remains the notice watermark; signInId is the activity
    # identity, preventing different messages for one activity from overlapping.
    row["signin_activity_key"] = "S%s" % route["signInId"]
    try:
        row["signin_type"] = int(route["type"])
    except (TypeError, ValueError, KeyError):
        return None
    row["signin_type_name"] = _signin_type_label(row["signin_type"])
    return row


def _latest_signin_activity(items):
    valid = [row for row in (_validated_signin_item(item) for item in (items or [])) if row]
    return max(valid, key=lambda row: row.get("ts") or 0) if valid else None


def _annotate_signin_items(items):
    """Annotate valid jobs for automatic discovery; leave all notices intact."""
    result = []
    for item in items:
        result.append(_validated_signin_item(item) or item)
    return result


def _normalize_signin_payload(data):
    if isinstance(data, dict) and data.get("type") is not None:
        data = dict(data)
        try:
            data["type"] = int(data["type"])
        except (TypeError, ValueError):
            pass
        data["typeName"] = _signin_type_label(data.get("type"))
    return data


def _signin_result_message(data, ok):
    if ok:
        return "签到成功"
    if isinstance(data, dict):
        try:
            signin_type = int(data.get("type"))
        except (TypeError, ValueError):
            signin_type = None
        if signin_type not in (None, 1, 2):
            return "%s暂不支持自动签到，需特定流程" % _signin_type_label(signin_type)
    return "签到失败（上游未确认提交成功）"


def _call_sign_in_local(account, password, url, check_status=False):
    """进程内本地签到：复用轮询缓存登录态（不再每次完整 SSO 登录）。
    轮询 token 失效时 refresh 会话重试一次；仍失败按原语义返回失败。
    重试提交时 check_status=True，先以 getSignInInfo 确认未签到，避免重复提交。"""
    try:
        gs = get_session(account, password)
        req, token = gs.ensure()
        try:
            data = fetch_sign_in_info(
                account, password, url, do_commit=True, sess=req, token=token,
                avoid_duplicate=check_status)
        except (UpstreamError, AuthError):
            # 缓存 token 失效：强制 refresh 轮询会话并重试一次
            req, token = gs.refresh()
            data = fetch_sign_in_info(
                account, password, url, do_commit=True, sess=req, token=token,
                avoid_duplicate=check_status)
    except RateLimitedError as e:
        return False, str(e), None, True
    except NoSignInCodeError:
        return False, "数字码签到未配置签到码", None, False
    except UnsupportedSigninType as e:
        # This text is locally authored from the normalized type, not upstream data.
        return False, str(e), None, False
    except Exception as e:
        try:
            reason = safe_error(e)
        except RateLimitedError as limited:
            return False, str(limited), None, True
        return False, reason, None, _looks_rate_limited(None, None, reason)
    if not isinstance(data, dict):
        return False, "签到模块返回非对象 JSON", None, False
    data = _normalize_signin_payload(data)
    commit = data.get("commit")
    if not isinstance(commit, dict):
        return False, "签到失败（缺少明确提交结果）", data, False
    limited = _looks_rate_limited(None, commit)
    if limited:
        return False, "签到请求触发上游频率限制", data, True
    ok = commit.get("success") is True
    return ok, _signin_result_message(data, ok), data, False


def call_sign_in_api(account, password, url, check_status=False):
    """自动签到统一入口（进程内本地 SSO + 签到）。
    返回 (ok, message, data, rate_limited)
    """
    return _call_sign_in_local(account, password, url, check_status=check_status)


# ==================== 轮询规则 ====================
RULE_ROWS = (
    ("每日 00:00-05:00", 3600, "1 小时"),
    ("每日 05:00-07:00", 300, "5 分钟"),
    ("每日 07:00-11:30", 10, "10 秒"),
    ("每日 11:30-12:30", 300, "5 分钟"),
    ("每日 12:30-15:30", 10, "10 秒"),
    ("每日 15:30-18:00", 300, "5 分钟"),
    ("每日 18:00-21:00", 10, "10 秒"),
    ("每日 21:00-24:00", 300, "5 分钟"),
)


def current_interval(now=None):
    """返回 (间隔秒数, 档位说明)；时间为 UTC+8；全周统一（周末不再单独降频）"""
    now = now or datetime.now(TZ8)
    t = now.time()
    if t < dtime(5, 0):
        return 3600, "00:00-05:00 → 1 小时一次"
    if t < dtime(7, 0):
        return 300, "05:00-07:00 → 5 分钟一次"
    if t < dtime(11, 30):
        return 10, "07:00-11:30 → 10 秒一次"
    if t < dtime(12, 30):
        return 300, "11:30-12:30 → 5 分钟一次"
    if t < dtime(15, 30):
        return 10, "12:30-15:30 → 10 秒一次"
    if t < dtime(18, 0):
        return 300, "15:30-18:00 → 5 分钟一次"
    if t < dtime(21, 0):
        return 10, "18:00-21:00 → 10 秒一次"
    return 300, "21:00-24:00 → 5 分钟一次"


# 保活：只对「通知推送和自动签到都关」的账号生效——这类账号完全不轮询，
# 登录态长期不用会失效；每天 11:55 与 23:55 各探活一次，走
# fetch_all → _call 的 _need_relogin → sess.refresh() 路径刷新登录态。
# 每个时段带 KEEPALIVE_GRACE 秒宽限：进程稍晚启动（如 11:55:30 重启）也能补触发，
# 已触发过的时段不重复。处于限流冷却期（rate_limited_until 未到）的账号跳过。
KEEPALIVE_TIMES = ((11, 55), (23, 55))
KEEPALIVE_GRACE = 600


def _next_keepalive(now_ts):
    """下一次保活时段的时间戳（严格未来），供状态展示。"""
    now = datetime.fromtimestamp(now_ts, TZ8)
    candidates = []
    for day_offset in (0, 1):
        day = now + timedelta(days=day_offset)
        for hour, minute in KEEPALIVE_TIMES:
            target = day.replace(hour=hour, minute=minute, second=0, microsecond=0).timestamp()
            if target > now_ts:
                candidates.append(target)
    return min(candidates)


def _due_keepalive(now_ts, last_fired=0):
    """返回此刻应当触发的保活时段戳（无则 None）。

    时段到点即触发；时段过后 KEEPALIVE_GRACE 秒内仍可补触发（跨重启兜底）；
    已经触发过的时段（last_fired）不再重复触发。
    """
    now = datetime.fromtimestamp(now_ts, TZ8)
    due = []
    for day_offset in (0, -1):
        day = now + timedelta(days=day_offset)
        for hour, minute in KEEPALIVE_TIMES:
            target = day.replace(hour=hour, minute=minute, second=0, microsecond=0).timestamp()
            if target <= now_ts < target + KEEPALIVE_GRACE and target > (last_fired or 0):
                due.append(target)
    return min(due) if due else None


def rule_text():
    _, label = current_interval()
    lines = ["白果云轮询规则（UTC+8）", "━━━━━━━━━━━━"]
    for name, _sec, human in RULE_ROWS:
        lines.append("• %s  %s" % (name, human))
    lines.append("━━━━━━━━━━━━")
    lines.append("当前档位：%s" % label)
    lines.append("轮询在「通知推送」或「自动签到」开启时运行；两者都关则停止轮询")
    lines.append("仅开自动签到（通知推送关闭）时，只在 10 秒档时段轮询，其余时段停止")
    lines.append("通知与签到都关的账号：每天 11:55 / 23:55 各保活探活一次（仅刷新登录态）")
    return "\n".join(lines)


HELP_TEXT = (
    "白果云（东北财经大学）服务\n"
    "━━━━━━━━━━━━\n"
    "• .ginkgo help              显示本帮助\n"
    "• .ginkgo rule              查看轮询规则\n"
    "• .ginkgo bind <账号> <密码> 绑定账号（返回学生姓名）\n"
    "• .ginkgo unbind [账号]      解绑账号\n"
    "• .ginkgo status            查看绑定与开关状态\n"
    "• .ginkgo sign              强制执行最新有效签到活动\n"
    "• .ginkgo notice <N>        查看最新 N 条通知（通知+互动消息）\n"
    "• .ginkgo notice <on|off>   开关通知推送\n"
    "• .ginkgo auto-sign-in <on|off>  开关自动签到\n"
    "━━━━━━━━━━━━\n"
    "触发频率限制时会自动关闭推送与签到并通知你\n"
    "每人仅可绑定一个账号"
)


# ==================== 命令层（异步） ====================
_async_states = weakref.WeakKeyDictionary()


def _async_state():
    loop = asyncio.get_running_loop()
    if loop not in _async_states:
        _async_states[loop] = {"accounts": {}, "users": {},
                               "sem": asyncio.Semaphore(ACCOUNT_CONCURRENCY)}
    return _async_states[loop]


def _account_lock(account):
    return _async_state()["accounts"].setdefault(str(account), asyncio.Lock())


def _serialized_command(fn):
    """Commands and background work share account locks AND concurrency budget."""
    @wraps(fn)
    async def wrapped(channel, msg_type, target_id, user_id, *args):
        state = _async_state()
        async with state["users"].setdefault(str(user_id), asyncio.Lock()):
            if fn.__name__ == "_cmd_bind":
                match = re.match(r"\.ginkgo\s+(?:bind|绑定)\s+(\S+)", str(args[0]), re.I)
                account = match.group(1) if match else ""
            else:
                mine = await asyncio.to_thread(find_by_uid, user_id)
                account = mine.get("account") if mine else ""
            async with _account_lock(account):
                async with state["sem"]:
                    return await fn(channel, msg_type, target_id, user_id, *args)
    return wrapped


async def _network_thread(fn, *args):
    """Keep the account lock/budget until a cancelled thread call really stops."""
    task = asyncio.create_task(asyncio.to_thread(fn, *args))
    try:
        return await asyncio.shield(task)
    except asyncio.CancelledError:
        while not task.done():
            try:
                await asyncio.shield(task)
            except asyncio.CancelledError:
                continue
            except Exception:
                break
        if not task.cancelled():
            task.exception()  # retrieve any worker error before propagating cancellation
        raise


async def _reply(channel, msg_type, target_id, content):
    await send_message(channel, msg_type, target_id, content)


def _switch_text(on):
    return "开" if on else "关"


def _push_label(uid):
    """推送目标展示（统一身份：TG 私聊 + 每个已绑定 QQ 私聊镜像）"""
    labels = []
    for name, _target_id in identity.push_targets(uid):
        label = "TG 私聊" if name == "tg" else "%s 私聊" % name.upper()
        if label not in labels:
            labels.append(label)
    return " + ".join(labels) if labels else "无可用通道"


def _push_hint(data):
    if data.get("notice_push") or data.get("auto_sign_in"):
        return "轮询已运行"
    return "通知推送与自动签到都关闭，轮询未运行"


def _format_item(index, item):
    return "%d. [%s] %s\n   %s\n   %s" % (
        index, item.get("kinder", ""), _clip(item.get("title") or "", 60),
        item.get("time") or "", item.get("link") or NOTICE_PAGE)


def _signin_text(item, ok, message, payload, attempts=None):
    return receipt_text(item, ok, message, payload, attempts)


@_serialized_command
async def _cmd_bind(channel, msg_type, target_id, user_id, raw_text):
    m = re.match(r"\.ginkgo\s+(?:bind|绑定)\s+(\S+)\s+(\S.*)$", raw_text.strip(), re.I | re.S)
    if not m:
        await _reply(channel, msg_type, target_id,
                     "用法: .ginkgo bind <账号> <密码>\n例如 .ginkgo bind 2025211228 密码")
        return
    account = m.group(1).strip()
    password = m.group(2).strip()
    if not account.isdigit():
        await _reply(channel, msg_type, target_id, "账号应为学号（纯数字）")
        return
    if not password:
        await _reply(channel, msg_type, target_id, "密码不能为空")
        return

    old_other = await asyncio.to_thread(find_by_account, account)
    # 同一身份重复绑定同一账号允许（用于更新密码/重建水位）；只有他人占用才拒绝。
    # 注意：账号数据里不再保存 channel（统一身份改造后推送目标由 identity 决定），
    # 因此这里只能按 user_id 判定归属。
    if old_other and str(old_other.get("user_id")) != str(user_id):
        await _reply(channel, msg_type, target_id,
                     "该账号已被其他用户绑定，如确需绑定请联系管理员处理")
        return

    mine = await asyncio.to_thread(find_by_uid, user_id)
    if mine and mine.get("account") != account:
        await _reply(channel, msg_type, target_id,
                     "你已绑定账号 %s，请先 .ginkgo unbind %s 再绑定新账号"
                     % (mine.get("account"), mine.get("account")))
        return

    rebind = old_other is not None
    await _reply(channel, msg_type, target_id, "正在登录白果云校验账号，请稍候…")
    try:
        sess = get_session(account, password)
        await _network_thread(sess.ensure)
        # Only the first successful bind establishes a baseline. A rebind must
        # never fetch-and-union arrivals into the old discovery watermark.
        seen = []
        if not rebind:
            notices, busi = await _network_thread(fetch_all, sess)
            seen = list(dict.fromkeys(item["key"] for item in notices + busi))
    except Exception as exc:
        reason, limited = _exception_reason(exc)
        if limited and old_other:
            await _handle_rate_limited(old_other, reason)
        await _reply(channel, msg_type, target_id, "绑定失败：%s" % reason)
        return

    name = sess.student_name or ""
    baseline_note = ("（沿用原水位、待投递通知和签到任务，不吞掉期间新通知）" if rebind else
                     "（已建立基线：此前 %d 条历史通知不补推）" % len(seen))
    def bind_update(data):
        if "user_id" in data and str(data["user_id"]) != str(user_id):
            return False
        # A first-bind contender must not overwrite a concurrent completed bind.
        if not rebind and "password" in data:
            return False
        data.update(password=password, student_name=name, user_id=user_id)
        data.setdefault("notice_push", False)
        data.setdefault("auto_sign_in", False)
        data.setdefault("created", _stamp())
        if not rebind:
            data.update(seen_keys=seen, last_poll=0)
        audit(data, "rebound" if rebind else "bound", account)
    try:
        data = await asyncio.to_thread(_mutate_account, account, bind_update,
                                       old_other.get("_generation") if rebind else None, not rebind)
    except OSError:
        await _reply(channel, msg_type, target_id, "绑定失败：状态写入异常")
        return
    if data is None:
        await _reply(channel, msg_type, target_id, "绑定状态已变化，请重新执行绑定")
        return

    await _reply(channel, msg_type, target_id,
                 "绑定成功：%s\n通知推送：%s（.ginkgo notice on 开启）\n自动签到：%s（.ginkgo auto-sign-in on 开启）\n%s"
                 % (name or account, _switch_text(data["notice_push"]),
                    _switch_text(data["auto_sign_in"]), baseline_note))


async def _cmd_unbind(channel, msg_type, target_id, user_id, raw_text):
    m = re.match(r"\.ginkgo\s+(?:unbind|解绑)(?:\s+(\S+))?\s*$", raw_text.strip(), re.I)
    if not m:
        # 正则失配（如 ".ginkgo unbind 123 456"）时 arg 会退化为 ""；
        # 绝不能按无参处理直接进入删除流程，否则误输入会静默解绑
        await _reply(channel, msg_type, target_id,
                     "用法: .ginkgo unbind [账号]\n账号参数只能是单个学号，请勿包含空格")
        return
    arg = (m.group(1) or "").strip()
    mine = await asyncio.to_thread(find_by_uid, user_id)
    if not mine:
        await _reply(channel, msg_type, target_id, "你尚未绑定任何白果云账号")
        return
    if arg and arg != mine.get("account"):
        await _reply(channel, msg_type, target_id,
                     "你绑定的是 %s；若要解绑请发送 .ginkgo unbind %s"
                     % (mine.get("account"), mine.get("account")))
        return
    account = mine.get("account")
    await asyncio.to_thread(_delete_account, account, mine.get("_generation"))
    with _sessions_lock:
        _sessions.pop(account, None)
    _next_check.pop(account, None)
    _keepalive_last.pop(account, None)
    _poll_fail.pop(account, None)
    _recovery_notice_at.pop(account, None)
    await _reply(channel, msg_type, target_id, "已解绑账号 %s" % account)


async def _cmd_status(channel, msg_type, target_id, user_id):
    mine = await asyncio.to_thread(find_by_uid, user_id)
    if not mine:
        await _reply(channel, msg_type, target_id,
                     "尚未绑定账号，使用 .ginkgo bind <账号> <密码> 绑定")
        return
    account = mine.get("account")
    interval, label = current_interval()
    now_ts = _now_ts()
    notice_push = bool(mine.get("notice_push"))
    auto_sign_in = bool(mine.get("auto_sign_in"))
    if notice_push or auto_sign_in:
        if not notice_push and interval != 10:
            # 仅开自动签到：当前档位暂停轮询，等下一个 10 秒档
            next_text = "当前档位暂停（仅在 10 秒档时段轮询）"
        else:
            left = max(0, int(_next_check.get(account, 0) - now_ts))
            next_text = "%d 秒后" % left if left else "即将执行"
    elif int(mine.get("rate_limited_until") or 0) > now_ts:
        next_text = "未运行（限流冷却中，暂停保活探活）"
    elif not mine.get("password"):
        next_text = "未运行（缺少密码，保活探活不可用）"
    else:
        ka = _keepalive_last.get(account, 0)
        ka = ka if ka and ka > now_ts else _next_keepalive(now_ts)
        next_text = "未常规轮询；下次保活探活 %s（%d 分钟后）" % (
            _fmt_ts(ka), max(0, int((ka - now_ts) // 60)))
    manual_jobs = [job for job in mine.get("signin_jobs", {}).values()
                   if signin_source(job) == "manual"]
    manual_pending = sum(job.get("status") in ("pending", "running") for job in manual_jobs)
    recent_manual = max(manual_jobs,
                        key=lambda job: job.get("finished_at") or job.get("created_at") or 0,
                        default=None)
    if manual_pending and not (notice_push or auto_sign_in):
        next_text = "手动签到任务调度中/等待重试（%d 条）" % manual_pending
    lines = [
        "白果云账号状态",
        "━━━━━━━━━━━━",
        "账号：%s（%s）" % (account, mine.get("student_name") or "姓名未知"),
        "推送通道：%s" % _push_label(mine.get("user_id")),
        "通知推送：%s" % _switch_text(mine.get("notice_push")),
        "自动签到：%s" % _switch_text(mine.get("auto_sign_in")),
        "手动签到待执行：%d" % manual_pending,
        "最近手动任务：%s" % (recent_manual.get("status") if recent_manual else "无"),
        "轮询档位：%s" % label,
        "下次轮询：%s" % next_text,
        "上次轮询：%s" % (_fmt_ts(mine.get("last_poll")) or "尚未轮询"),
    ]
    until = int(mine.get("rate_limited_until") or 0)
    if until > _now_ts():
        lines.append("限流冷却：剩余 %d 分钟（%s 触发：%s）"
                     % (max(1, int((until - _now_ts()) // 60)),
                        mine.get("rate_limited_at") or "未知",
                        _clip(mine.get("rate_limited_reason") or "", 60)))
    await _reply(channel, msg_type, target_id, "\n".join(lines))


@_serialized_command
async def _cmd_notice_list(channel, msg_type, target_id, user_id, count):
    mine = await asyncio.to_thread(find_by_uid, user_id)
    if not mine:
        await _reply(channel, msg_type, target_id,
                     "尚未绑定账号，使用 .ginkgo bind <账号> <密码> 绑定")
        return
    count = max(1, min(count, LIST_MAX))
    account = mine.get("account")
    password = mine.get("password") or ""
    try:
        sess = get_session(account, password)
        notices, busi = await _network_thread(fetch_all, sess)
    except Exception as exc:
        reason, limited = _exception_reason(exc)
        if limited:
            await _handle_rate_limited(mine, reason)
        await _reply(channel, msg_type, target_id, "获取通知失败：%s" % reason)
        return

    items = notices + busi
    items.sort(key=lambda x: x.get("ts") or 0, reverse=True)
    items = items[:count]
    if not items:
        await _reply(channel, msg_type, target_id, "暂无通知")
        return
    lines = ["白果云最新通知（%d 条）" % len(items), "━━━━━━━━━━━━"]
    for i, item in enumerate(items, 1):
        lines.append(_format_item(i, item))
    await _reply(channel, msg_type, target_id, "\n".join(lines))


@_serialized_command
async def _cmd_sign(channel, msg_type, target_id, user_id):
    """Fetch current messages and durably enqueue the latest valid activity."""
    mine = await asyncio.to_thread(find_by_uid, user_id)
    if not mine:
        await _reply(channel, msg_type, target_id,
                     "尚未绑定账号，使用 .ginkgo bind <账号> <密码> 绑定")
        return
    account = mine.get("account")
    try:
        sess = get_session(account, mine.get("password") or "")
        _notices, busi = await _network_thread(fetch_all, sess)
    except Exception as exc:
        if isinstance(exc, AuthError):
            reason, limited = "账号认证失败，请检查绑定信息或重新绑定", False
        else:
            reason, limited = _exception_reason(exc)
        if limited:
            await _handle_rate_limited(mine, reason)
            await _reply(channel, msg_type, target_id,
                         "获取签到活动失败：上游请求过于频繁，已关闭通知推送和自动签到")
        else:
            await _reply(channel, msg_type, target_id, "获取签到活动失败：%s" % reason)
        return
    item = _latest_signin_activity(busi)
    if not item:
        await _reply(channel, msg_type, target_id, "没有找到可签到活动")
        return

    outcome = {}
    def create(latest):
        action, key, job = enqueue_signin(latest, item, "manual", _now_ts())
        outcome.update(action=action, key=key, run_id=(job or {}).get("run_id"))
    updated = await asyncio.to_thread(_mutate_account, account, create, mine.get("_generation"))
    if updated is None:
        await _reply(channel, msg_type, target_id, "绑定状态已变化，请重试")
        return

    action, key = outcome.get("action"), outcome.get("key")
    job = updated.get("signin_jobs", {}).get(key, {})
    if action == "active":
        result_text = "同一活动已有%s签到任务待执行，不重复创建" % (
            "手动" if signin_source(job) == "manual" else "自动")
    elif action == "succeeded":
        receipt_key = "signin:" + key + ((":" + str(job.get("run_id"))) if job.get("run_id") else "")
        receipt_status = (updated.get("outbox", {}).get(receipt_key) or {}).get("status")
        suffix = ("，结果通知仍在投递" if receipt_status in ("pending", "sending") else
                  "，结果通知投递失败" if receipt_status == "failed" else "")
        result_text = "该活动已成功处理过%s，不再次签到" % suffix
    elif action == "reset":
        result_text = "旧失败/取消任务已保留审计摘要，已重置为新的手动签到任务并立即调度"
    elif action == "created":
        result_text = "已创建持久化手动签到任务并立即调度"
    else:
        result_text = "任务状态异常，未创建手动签到任务，请稍后重试"
    old_hint = ""
    if item.get("ts") and _now_ts() - item["ts"] > 86400:
        old_hint = "\n提示：该活动距今较久，可能已结束"
        if action in ("created", "reset"):
            old_hint += "，仍已按手动签到入队"
    await _reply(channel, msg_type, target_id,
                 "白果云签到活动\n标题：%s\n签到类型：%s\n时间：%s\n链接：%s\n%s%s" % (
                     _clip(item.get("title") or "(无标题)", 120),
                     item.get("signin_type_name") or _signin_type_label(item.get("signin_type")),
                     item.get("time") or "未知", item.get("link") or "", result_text, old_hint))


async def _cmd_notice_switch(channel, msg_type, target_id, user_id, value):
    mine = await asyncio.to_thread(find_by_uid, user_id)
    if not mine:
        await _reply(channel, msg_type, target_id,
                     "尚未绑定账号，使用 .ginkgo bind <账号> <密码> 绑定")
        return
    account = mine["account"]
    pause_epoch = mine.get("rate_pause_epoch")
    def change(data):
        if value == "on" and data.get("rate_pause_epoch") != pause_epoch:
            return False
        data["notice_push"] = (value == "on")
        audit(data, "notice_switch", account, enabled=(value == "on"))
    mine = await asyncio.to_thread(_mutate_account, account, change, mine.get("_generation"))
    if mine is None:
        await _reply(channel, msg_type, target_id, "绑定状态已变化，请重试")
        return
    if mine["notice_push"]:
        _next_check[account] = 0  # 立即触发一次轮询
        await _reply(channel, msg_type, target_id,
                     "已开启通知推送，轮询已运行" + _cooldown_warning(mine))
    else:
        manual_pending = any(signin_source(job) == "manual"
                             and job.get("status") in ("pending", "running")
                             for job in mine.get("signin_jobs", {}).values())
        if manual_pending:
            suffix = "，手动签到任务仍继续执行"
        elif mine.get("auto_sign_in"):
            suffix = "，因自动签到仍开启，将在 10 秒档时段轮询"
        else:
            suffix = "，轮询已停止（每天 11:55 / 23:55 保活探活）"
        await _reply(channel, msg_type, target_id, "已关闭通知推送" + suffix)


async def _cmd_auto_sign_in(channel, msg_type, target_id, user_id, value):
    mine = await asyncio.to_thread(find_by_uid, user_id)
    if not mine:
        await _reply(channel, msg_type, target_id,
                     "尚未绑定账号，使用 .ginkgo bind <账号> <密码> 绑定")
        return
    account = mine["account"]
    pause_epoch = mine.get("rate_pause_epoch")
    def change(data):
        if value == "on" and data.get("rate_pause_epoch") != pause_epoch:
            return False
        data["auto_sign_in"] = (value == "on")
        if value != "on":
            cancel_signins(data, "用户关闭自动签到，已取消剩余重试", source="auto")
        audit(data, "signin_switch", account, enabled=(value == "on"))
    mine = await asyncio.to_thread(_mutate_account, account, change, mine.get("_generation"))
    if mine is None:
        await _reply(channel, msg_type, target_id, "绑定状态已变化，请重试")
        return
    if mine["auto_sign_in"]:
        _next_check[account] = 0
        await _reply(channel, msg_type, target_id,
                     "已开启自动签到：发现「你收到来自 xxxxx 的签到」类互动消息时自动签到，结果私聊告知"
                     + _cooldown_warning(mine))
    else:
        manual_pending = any(signin_source(job) == "manual"
                             and job.get("status") in ("pending", "running")
                             for job in mine.get("signin_jobs", {}).values())
        if manual_pending:
            suffix = "，手动签到任务仍继续执行"
        elif mine.get("notice_push"):
            suffix = "，因通知推送仍开启，轮询继续"
        else:
            suffix = "，轮询已停止（每天 11:55 / 23:55 保活探活）"
        await _reply(channel, msg_type, target_id, "已关闭自动签到" + suffix)


async def handle_ginkgo(channel, msg_type, target_id, user_id, raw_text):
    """dispatch 调用入口。群聊与私聊均可使用；推送一律走私聊。

    dispatch 用 create_task 发起且不 retrieve 异常：任何逃出子命令的异常
    （损坏的账号 JSON、存储 I/O 错误、脏数据结构等）都必须在此兜底，
    打日志并尽量回复用户，避免任务静默死亡、用户零反馈。
    """
    try:
        await _handle_ginkgo(channel, msg_type, target_id, user_id, raw_text)
    except Exception as exc:
        import traceback
        print("[ginkgo] 指令处理异常 (%s)" % type(exc).__name__)
        traceback.print_exc()
        try:
            await _reply(channel, msg_type, target_id,
                         "白果云指令执行内部错误（%s），请稍后重试或联系管理员"
                         % type(exc).__name__)
        except Exception:
            pass            # 回复失败不再二次抛出


async def _handle_ginkgo(channel, msg_type, target_id, user_id, raw_text):
    text = (raw_text or "").strip()
    body = text[len(".ginkgo"):].strip()
    parts = body.split(maxsplit=1)
    sub = parts[0].lower() if parts else ""
    arg = parts[1].strip() if len(parts) > 1 else ""

    if sub in ("", "help", "-h", "帮助"):
        await _reply(channel, msg_type, target_id, HELP_TEXT)
        return
    if sub in ("rule", "规则"):
        await _reply(channel, msg_type, target_id, rule_text())
        return
    if sub in ("bind", "绑定"):
        await _cmd_bind(channel, msg_type, target_id, user_id, text)
        return
    if sub in ("unbind", "解绑"):
        await _cmd_unbind(channel, msg_type, target_id, user_id, text)
        return
    if sub in ("status", "状态"):
        await _cmd_status(channel, msg_type, target_id, user_id)
        return
    if sub in ("sign", "签到"):
        if arg:
            await _reply(channel, msg_type, target_id, "用法: .ginkgo sign")
        else:
            await _cmd_sign(channel, msg_type, target_id, user_id)
        return
    if sub == "notice":
        low = arg.lower()
        if low in ("on", "off"):
            await _cmd_notice_switch(channel, msg_type, target_id, user_id, low)
            return
        if low in ("help", "-h", "帮助", ""):
            await _reply(channel, msg_type, target_id,
                         "用法: .ginkgo notice <N|on|off>\n• <N> 查看最新 N 条通知（1-%d）\n"
                         "• on/off 开关通知推送" % LIST_MAX)
            return
        if not arg.isdigit():
            await _reply(channel, msg_type, target_id,
                         "参数应为正整数或 on/off，例如 .ginkgo notice 5")
            return
        await _cmd_notice_list(channel, msg_type, target_id, user_id, int(arg))
        return
    if sub in ("auto-sign-in", "auto_sign_in", "autosignin", "自动签到"):
        low = arg.lower()
        if low in ("on", "off"):
            await _cmd_auto_sign_in(channel, msg_type, target_id, user_id, low)
            return
        await _reply(channel, msg_type, target_id, "用法: .ginkgo auto-sign-in <on|off>")
        return

    await _reply(channel, msg_type, target_id,
                 "未知子指令：%s\n发送 .ginkgo help 查看帮助" % sub)


# ==================== 轮询与推送 ====================
_next_check = {}
_keepalive_last = {}  # account -> 最近已触发的保活时段戳（通知与签到都关的账号用）
_running = {}                      # account -> asyncio.Task（防止同一账号重叠轮询）


PUSH_LOG = os.path.join(STATE_DIR, "push.log")
PUSH_LOG_MAX = 2 * 1024 * 1024


def _push_log(line):
    """推送审计日志：记录每次推送的时间/对象/结果，便于事后核查"""
    try:
        os.makedirs(STATE_DIR, exist_ok=True)
        if os.path.exists(PUSH_LOG) and os.path.getsize(PUSH_LOG) > PUSH_LOG_MAX:
            os.replace(PUSH_LOG, PUSH_LOG + ".1")
        with open(PUSH_LOG, "a", encoding="utf-8") as f:
            f.write("[%s] %s\n" % (_stamp(), line))
    except OSError as e:
        print("[ginkgo] 推送日志写入失败: %s" % e)


def _cooldown_warning(data):
    """限流冷却期内的提示（仅提示，不阻止用户手动恢复）"""
    until = int(data.get("rate_limited_until") or 0)
    left = until - _now_ts()
    if left <= 0:
        return ""
    return ("\n注意：%s 因触发限流被自动关闭过，建议再等 %d 分钟"
            % (data.get("rate_limited_at") or "上次", max(1, int(left // 60))))


def _poll_account(account, keepalive=False):
    """Fetch only if enabled, then commit discovery and durable work atomically.

    keepalive=True 时（账号两个开关都关）只做一次取数探活以刷新登录态，
    不写发现水位/seen_keys，避免用户重新开启推送后丢失期间的通知。
    """
    data = _load_account(account)
    if not data:
        return None
    enabled = data.get("notice_push") or data.get("auto_sign_in")
    if not enabled and not keepalive:
        return None
    password = data.get("password") or ""
    if not password:
        return None
    notices, busi = fetch_all(get_session(account, password))
    if not enabled:
        return True
    def commit(latest):
        # Switches may change while the upstream request is in flight. Never
        # restore them from this snapshot, and don't consume disabled arrivals.
        if not (latest.get("notice_push") or latest.get("auto_sign_in")):
            return False
        discover(latest, _annotate_signin_items(notices + busi),
                           _notice_push_text, _now_ts())
    return _mutate_account(account, commit, data.get("_generation"))


async def _notify(data, content, tag="通知", targets=None):
    """异步推送走统一身份扇出：TG 私聊 + 每个已绑定 QQ 私聊镜像。

    targets 可选：限定本次扇出的目标（持久化 outbox 重试时只补投未成功目标）。
    返回 {(channel_name, target_id): bool}；逐目标写 push.log 审计（成功/失败）。
    """
    uid = data.get("user_id")
    all_targets = identity.push_targets(uid) if targets is None else list(targets)
    if not all_targets:
        print("[ginkgo] 身份 %s 无可用的推送目标，跳过推送" % uid)
        _push_log("%s uid=%s 失败：无可用的推送目标" % (tag, uid))
        return {}
    results = {}
    # 白果云已有持久 outbox 和逐目标去重；这里只替换 render/send 阶段，
    # 不再叠加第二层 fanout，避免 TG/QQ 镜像重复。
    from agent.modes import ModeStore
    from channel import get_channel, send_message
    import notification_render
    import notifications
    for name, target_id in all_targets:
        mode = ModeStore().get_mode(name, uid)
        rendered = content
        if mode == "chat":
            event = notifications.Event(
                kind="ginkgo",
                dedupe_key=f"ginkgo:{uid}:{tag}:{_uuid.uuid4().hex}",
                legacy_text=content,
                uid=uid,
                payload={"text": content},
            )
            rendered = await notification_render.render_event(event, mode, target_id)
        ch = get_channel(name)
        ok = False
        if ch is not None:
            ok = await send_message(ch, "private", target_id, rendered, mirror=False)
        results[(name, target_id)] = bool(ok)
    brief = _clip(str(content).replace("\n", " / "), 160)
    for (name, target_id), ok in results.items():
        _push_log("%s uid=%s -> %s/%s %s | %s"
                  % (tag, uid, name, target_id, "成功" if ok else "失败", brief))
    return results


def _notice_push_text(item):
    return "【白果云·新%s】\n%s\n%s\n%s" % (
        item.get("kinder", "消息"), _clip(item.get("title") or "", 120),
        item.get("time") or "", item.get("link") or "")


def _exception_reason(exc):
    """Credential-free exception classification, including requests HTTP errors."""
    try:
        return safe_error(exc), False
    except RateLimitedError:
        return "上游请求过于频繁，请稍后重试", True


# ==================== 轮询失败追踪 / 恢复通知 ====================
# 失败细节写本机 nssm.log（带时间戳）；失败恢复后的首次成功轮询按阈值推送一条"已恢复"。
POLL_FAIL_NOTICE_COUNT = 3        # 连续失败达到此次数，恢复时也告知（即便失败未单独推送过）
POLL_FAIL_NOTICE_SECONDS = 120    # 或失败持续达到此秒数
RECOVERY_NOTICE_COOLDOWN = 1800   # 同一账号两次"已恢复"通知的最小间隔（秒）

_poll_fail = {}            # account -> {count, first_at, last_at, reason, auth, notified}
_recovery_notice_at = {}   # account -> 上次"已恢复"通知时间戳


def _dur_text(seconds):
    seconds = max(0, int(seconds))
    if seconds < 60:
        return "%d 秒" % seconds
    if seconds < 3600:
        return "%d 分钟" % (seconds // 60)
    return "%d 小时 %d 分钟" % (seconds // 3600, (seconds % 3600) // 60)


def _failure_detail(exc, reason):
    """本地日志用的失败细节（绝不进入推送文案）。

    优先用自有异常文案（AuthError/UpstreamError 等，可能带上游片段）；requests 异常
    补上"方法 + 路径"，**查询串一律剥掉**，避免任何参数/令牌写入日志。
    """
    if isinstance(exc, (AuthError, UpstreamError, NoSignInCodeError, UnsupportedSigninType)):
        detail = re.sub(r"\s+", " ", str(exc)).strip()
        if detail:
            return detail[:200]
    request = getattr(exc, "request", None)
    if request is not None:
        path = urlsplit(str(getattr(request, "url", "") or "")).path
        method = str(getattr(request, "method", "") or "")
        if path:
            return ("%s %s" % (method, path)).strip()
    return reason


def _note_poll_failure(account, exc, reason, keepalive=False, notified=False):
    """记录一次轮询失败，并把原因写进本机日志（时间戳 + 类型 + 连败次数 + 细节）。"""
    now = _now_ts()
    state = _poll_fail.get(account)
    if state is None:
        state = _poll_fail[account] = {"count": 0, "first_at": now, "auth": False,
                                       "notified": False, "reason": reason}
    state.update(count=state["count"] + 1, last_at=now, reason=reason,
                 auth=state["auth"] or isinstance(exc, AuthError),
                 notified=state["notified"] or notified)
    kind = ("认证" if isinstance(exc, AuthError) else
            "限流" if isinstance(exc, RateLimitedError) else "瞬时")
    print("[ginkgo] %s [%s] %s失败 #%d (%s/%s): %s | %s" % (
        account, _stamp(), "保活探活" if keepalive else "轮询", state["count"],
        type(exc).__name__, kind, reason, _failure_detail(exc, reason)))
    return state


async def _note_poll_recovered(account, snapshot):
    """失败后首次成功：写日志；达到阈值且不在冷却期时推送"已恢复/重新登录成功"。"""
    state = _poll_fail.pop(account, None)
    if not state:
        return
    now = _now_ts()
    print("[ginkgo] %s [%s] 轮询恢复正常（此前连续失败 %d 次，历时 %s，最近原因：%s）"
          % (account, _stamp(), state["count"], _dur_text(now - state["first_at"]),
             state["reason"]))
    substantial = (state["count"] >= POLL_FAIL_NOTICE_COUNT
                   or (now - state["first_at"]) >= POLL_FAIL_NOTICE_SECONDS)
    if not (state["notified"] or substantial):
        return                                  # 一闪而过的失败不打扰用户
    last_notice = _recovery_notice_at.get(account)
    if last_notice is not None and now - last_notice < RECOVERY_NOTICE_COOLDOWN:
        return                                  # 抖动期间避免通知风暴
    text = ("【白果云·已恢复】\n%s\n失败 %d 次（历时 %s），最近原因：%s\n恢复时间：%s" % (
        "重新登录成功，轮询已恢复正常。" if state["auth"] else "轮询已恢复正常。",
        state["count"], _dur_text(now - state["first_at"]),
        _clip(state.get("reason") or "", 120), _fmt_ts(now)))
    result = await asyncio.to_thread(_mutate_account, account,
        lambda latest: enqueue(latest, "control:recovered:" + _uuid.uuid4().hex, "control", text),
        (snapshot or {}).get("_generation"))
    if result is not None:
        _recovery_notice_at[account] = now


def _pause_state(data, reason):
    # All switches, cancellations and control alarm share the same disk commit.
    data.update(notice_push=False, auto_sign_in=False, rate_limited_at=_stamp(),
                rate_pause_epoch=_uuid.uuid4().hex,
                rate_limited_until=int(_now_ts() + RATE_LIMIT_COOLDOWN),
                rate_limited_reason=str(reason)[:200])
    cancel_signins(data, "触发限流，已取消剩余自动重试：%s" % reason, source="auto")
    text = ("【白果云·已暂停】\n检测到白果云频率限制：%s\n"
            "已关闭通知推送和自动签到，自动签到重试已取消。\n"
            "手动签到任务会在冷却结束后继续；建议冷却 %d 分钟后手动恢复：\n"
            ".ginkgo notice on / .ginkgo auto-sign-in on"
            % (reason, RATE_LIMIT_COOLDOWN // 60))
    enqueue(data, "control:limited:" + _uuid.uuid4().hex, "control", text)
    audit(data, "rate_limited", data["account"], reason=str(reason))


async def _handle_rate_limited(data, exc):
    account = data.get("account")
    # Do not interpolate arbitrary upstream exception strings into persisted logs.
    reason = "上游请求过于频繁，请稍后重试"
    state = _poll_fail.get(account)
    if state is not None:
        state["notified"] = True      # 限流暂停也会推送，恢复时应当告知
    result = await asyncio.to_thread(_mutate_account, account,
        lambda latest: _pause_state(latest, reason), data.get("_generation"))
    _next_check.pop(account, None)
    return result


async def _deliver_outbox(account):
    snapshot = await asyncio.to_thread(_load_account, account)
    if not snapshot:
        return
    # Controls/receipts cannot be starved behind paused notices or a large backlog.
    rows = sorted(snapshot.get("outbox", {}).items(),
                  key=lambda pair: (pair[1].get("kind") == "notice", pair[1].get("created_at", 0)))
    count = 0
    for key, row in rows:
        if count >= PUSH_MAX:
            break
        if row.get("status") in OUTBOX_TERMINAL:
            continue          # sent/failed 都是终态，不再重试
        if (row.get("next_attempt_at") or 0) > _now_ts():
            continue
        claim = _uuid.uuid4().hex
        latest = await asyncio.to_thread(_mutate_account, account,
            lambda data: claim_outbox(data, key, claim, _now_ts()), snapshot["_generation"])
        if not latest:
            continue
        count += 1
        acknowledged, reason, succeeded = False, "", []
        try:
            # Recheck after reservation; off/unbind are deliberately not blocked
            # behind network semaphores. Already in-flight sends cannot be undone.
            current = await asyncio.to_thread(_load_account, account)
            if not current or current["_generation"] != latest["_generation"]:
                return
            claimed = (latest.get("outbox") or {}).get(key) or {}
            # 逐目标去重：sent_targets 里已有的目标绝不再发（避免重复轰炸 TG）
            done = set(claimed.get("sent_targets") or [])
            wanted = [t for t in identity.push_targets(current.get("user_id"))
                      if target_key(t[0], t[1]) not in done]
            if row.get("kind") == "notice" and not current.get("notice_push"):
                reason = "notice delivery disabled"
            elif not wanted:
                if done:
                    acknowledged = True      # 所有目标此前都已投递成功
                else:
                    # 无可用推送目标：与旧 _notify 行为一致，仍写审计行
                    reason = "无可用的推送目标"
                    print("[ginkgo] 身份 %s 无可用的推送目标，跳过推送"
                          % current.get("user_id"))
                    _push_log("%s uid=%s 失败：无可用的推送目标"
                              % (row.get("kind", "通知"), current.get("user_id")))
            else:
                results = await _notify(current, claimed.get("text", ""),
                                        tag=row.get("kind", "通知"), targets=wanted)
                succeeded = [t for t, ok in results.items() if ok]
                acknowledged = bool(results) and len(succeeded) == len(results)
                if not acknowledged:
                    missing = ["%s/%s" % t for t, ok in results.items() if not ok]
                    reason = "未成功目标: %s" % ",".join(missing)
        except Exception as exc:
            reason = "delivery error (%s)" % type(exc).__name__
        await asyncio.to_thread(_mutate_account, account,
            lambda data: ack_outbox(data, key, claim, acknowledged, reason,
                                              _now_ts(), succeeded),
            latest["_generation"])


async def _run_signin_jobs(account):
    snapshot = await asyncio.to_thread(_load_account, account)
    if not snapshot:
        return
    if not snapshot.get("auto_sign_in") and any(
            job.get("status") in ("pending", "running")
            and signin_source(job) == "auto"
            for job in snapshot.get("signin_jobs", {}).values()):
        snapshot = await asyncio.to_thread(_mutate_account, account,
            lambda data: (False if data.get("auto_sign_in") else
                          cancel_signins(data, "自动签到已关闭，取消剩余重试",
                                                   source="auto")), snapshot["_generation"])
        if not snapshot:
            return
    due = [(key, job) for key, job in snapshot.get("signin_jobs", {}).items()
           if (job.get("status") in ("pending", "running")
               and signin_enabled(snapshot, job)
               and (job.get("next_attempt_at") or 0) <= _now_ts())]
    for key, _job in due[:1]:
        claim = _uuid.uuid4().hex
        latest = await asyncio.to_thread(_mutate_account, account,
            lambda data: claim_signin(data, key, claim, _now_ts()), snapshot["_generation"])
        if not latest:
            continue
        job = latest.get("signin_jobs", {}).get(key, {})
        if job.get("claim_id") != claim:
            continue  # A crashed sixth attempt was finalized, never attempt seven.
        if job.get("status") != "running":
            await asyncio.to_thread(_mutate_account, account,
                lambda data: abandon_signin_claim(data, key, claim), latest["_generation"])
            continue
        current = await asyncio.to_thread(_load_account, account)
        if not current or current.get("_generation") != latest["_generation"]:
            return
        current_job = current.get("signin_jobs", {}).get(key, {})
        if (not signin_enabled(current, current_job)
                or current_job.get("status") != "running"
                or current_job.get("claim_id") != claim):
            if current_job.get("claim_id") == claim:
                await asyncio.to_thread(_mutate_account, account,
                    lambda data: abandon_signin_claim(data, key, claim), latest["_generation"])
            return
        try:
            result = await _network_thread(
                call_sign_in_api, account, current.get("password") or "",
                job["item"].get("link") or "", job.get("attempts", 0) > 1)
            if not isinstance(result, (tuple, list)) or len(result) != 4:
                raise ValueError("malformed signin transport result")
            _ok, message, payload, limited = result
            message = str(message or "未返回明确提交成功结果")
            # Explicit throttling always wins, even if commit.success says True.
            limited = bool(limited) or is_rate_limited(None, {"message": message})
        except Exception as exc:
            message, limited = _exception_reason(exc)
            payload = None
        if limited:
            await _handle_rate_limited(latest, message)
            return
        await asyncio.to_thread(_mutate_account, account,
            lambda data: complete_signin(data, key, claim, message, payload, _now_ts()),
            latest["_generation"])


async def _poll_and_push(data, sem=None, do_poll=True, keepalive=False):
    """Serialize account network work; all callers share the global budget."""
    account = data.get("account")
    try:
        async with _account_lock(account):
            async with _async_state()["sem"]:
                current = await asyncio.to_thread(_load_account, account)
                if not current or current["_generation"] != data.get("_generation"):
                    return
                await _deliver_outbox(account)
                if do_poll:
                    try:
                        outcome = await _network_thread(_poll_account, account, keepalive)
                        if outcome:
                            # 只有真的完成取数才算成功；被开关变化跳过（None）不算恢复
                            await _note_poll_recovered(account, current)
                    except Exception as exc:
                        reason, limited = _exception_reason(exc)
                        if limited:
                            _note_poll_failure(account, exc, reason, keepalive, notified=True)
                            await _handle_rate_limited(current, reason)
                        elif isinstance(exc, AuthError):
                            # 保活探活失败不打扰用户；正常轮询失败会推送"登录失败"
                            _note_poll_failure(account, exc, reason, keepalive,
                                               notified=not keepalive)
                            if not keepalive:
                                await asyncio.to_thread(_mutate_account, account,
                                    lambda latest: enqueue(latest,
                                        "control:auth:" + str(int(_now_ts() // 600)), "control",
                                        "【白果云·登录失败】\n%s\n如已修改密码请重新绑定" % reason), current["_generation"])
                                _next_check[account] = _now_ts() + 600
                        else:
                            _note_poll_failure(account, exc, reason, keepalive)
                    await _deliver_outbox(account)
                await _run_signin_jobs(account)
                await _deliver_outbox(account)
    except Exception as exc:
        # Persisted reservations are recovered after their lease, without stale writes.
        print("[ginkgo] %s 工作异常 (%s)" % (account, type(exc).__name__))


async def monitor_loop():
    """One-second durable retry scheduling, independent from discovery intervals."""
    print("[ginkgo] 白果云通知轮询已启动（并发上限 %d）" % ACCOUNT_CONCURRENCY)
    try:
        while True:
            try:
                interval, _label = current_interval()
                now = _now_ts()
                for account, task in list(_running.items()):
                    if task.done():
                        _running.pop(account, None)
                        if not task.cancelled():
                            task.result()
                for data in await asyncio.to_thread(list_accounts, True):
                    account = data.get("account")
                    if not account or account in _running:
                        continue
                    notice_push = bool(data.get("notice_push"))
                    auto_sign_in = bool(data.get("auto_sign_in"))
                    enabled = notice_push or auto_sign_in
                    keepalive = False
                    keepalive_slot = None
                    if not enabled:
                        # 通知与签到都关：不常规轮询，仅每天 11:55 / 23:55 各保活探活
                        # 一次（每次时段有 KEEPALIVE_GRACE 秒宽限，跨重启可补触发；
                        # 限流冷却期内跳过，避免破坏限流自停）
                        paused = int(data.get("rate_limited_until") or 0) > now
                        if not paused and data.get("password"):
                            keepalive_slot = _due_keepalive(now, _keepalive_last.get(account, 0))
                            if keepalive_slot:
                                keepalive = True
                    elif not notice_push and interval != 10:
                        # 仅开自动签到（通知推送关闭）：只在 10 秒档时段轮询，
                        # 其余时段停止轮询（签到活动只在上课时段出现）
                        enabled = False
                    do_poll = bool((enabled or keepalive) and now >= _next_check.get(account, 0))
                    if not enabled and not keepalive:
                        _next_check.pop(account, None)
                    if not do_poll and not has_due(data, now):
                        continue
                    if do_poll:
                        _next_check[account] = now + interval
                        if keepalive_slot:
                            # 确认本次真的会执行取数探活（通过 _next_check 检查）后才登记
                            # 槽位；先登记后跳过的旧写法会让该时段当天不再补触发
                            _keepalive_last[account] = keepalive_slot
                            print("[ginkgo] %s [%s] 触发保活探活（通知与签到都关）"
                                  % (account, _stamp()))
                    _running[account] = asyncio.create_task(
                        _poll_and_push(data, do_poll=do_poll, keepalive=keepalive))
            except Exception as exc:
                print("[ginkgo] 轮询调度异常 (%s)" % type(exc).__name__)
            await asyncio.sleep(1)
    finally:
        # Do not release account locks while to_thread network calls still run.
        # Graceful shutdown drains workers; abrupt death uses persisted leases.
        if _running:
            await asyncio.gather(*list(_running.values()), return_exceptions=True)
        _running.clear()