#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
线报酷 · 托管推送脚本（Python 版）
v20261019120000 —— 与服务端 client_config 下发的 rowpass_ver 对应；不一致时本脚本会提示升级。

适用：青龙面板（Python3）/ VPS / NAS / 旧手机 Termux 等任何有 Python 3.8+ 的设备。零第三方依赖（仅标准库）。

用法（青龙面板）：
  1. Ucenter → 信息推送设置 → 托管推送：生成密钥（32 位，只显示一次）
  2. 环境变量：
       XB_HOST      站点域名（默认 new.xianbao.fun）
       XB_CLIENTS   JSON 数组：[{"uid":123,"key":"32hex","name":"主号"}, ...]（托管多个线报酷账号；渠道/规则在线报酷用户中心设置、实时生效）
       XB_INTERVAL  轮询秒（默认 5，下限 5）
  3. 定时任务：task xianbao_push.py（脚本常驻轮询；任务崩溃由青龙自动重拉）

行为与官方 JS 版（xianbao_push.js）完全同构：
  client_config（拉规则/渠道/feed）→ push_delta 增量轮询 → rowPass 本地匹配 →
  client_build 认领（服务端渲染请求规格；dup=1 表示被扩展/另一端抢先）→ 发送 → client_report 回报。
  弹窗渠道（type=browser）仅浏览器扩展执行：服务端不下发、本脚本也不会拿到。
"""

import hashlib
import json
import os
import re
import sys
import time
import threading
import urllib.parse
import urllib.request
import urllib.error
import ssl

VER = '20261022020000'
STALE_SEC = 7200  # 时效保护：与扩展/服务端 2h 窗口同口径

# 客户端 UA（20261020）：与 JS 版共用 XianbaoPush/ 前缀，便于服务端日志归并。
# 此前 JS 版完全不发 UA → WAF 记为 btwaf_null；配合 chunked 编码被判扫描器封 IP。
# 注意 Python 的 urllib 会自动补 Content-Length（所以本版没有 chunked 问题），
# 只有 JS 版需要手工补 —— 详见 xianbao_push.js 的 postHeaders() 注释。
UA = 'XianbaoPush/' + VER + ' (Python ' + sys.version.split()[0] + '; ' + sys.platform + ')'

# ---------- 桌面/启动器形态（20261020b 补齐） ----------
# 背景：三个启动器（.bat / .command / .sh）在没装 Node 时会回落到 Python，但本脚本此前
# 完全没有 app-mode —— 不认 XB_LAUNCHER / XB_CODE，也不引导粘贴接入码，于是「有 Python
# 没 Node」的用户双击后必然看到「未配置 XB_CLIENTS」然后退出（启动器还会误报成
# 「接入码写入失败」）。这里补齐与 JS 版同构的最小集。
IS_APP = os.environ.get('XB_LAUNCHER') == '1'

def _app_dir():
    """数据目录：优先脚本所在目录（开机自启时 cwd 可能是 / 或 System32）。"""
    try:
        d = os.path.dirname(os.path.abspath(__file__))
        if os.path.isdir(d) and os.access(d, os.W_OK):
            return d
    except Exception:
        pass
    return os.path.expanduser('~')

DATA_DIR = _app_dir() if IS_APP else (os.getcwd() if os.path.isdir(os.getcwd()) else '.')
# 配置刷新间隔（20261020）：原 600 秒太长——站长反馈改完设置要等 10 分钟才生效。降到 60 秒。
# 想「立刻」生效：在脚本目录新建空文件 xb_reload，下一轮 tick（≤5 秒）就会重拉配置并自动删除它。
CFG_TTL = 60
RELOAD_FLAG = 'xb_reload'

# 重启即重建基线（20261020b，站长定稿）：每次启动后每个 feed 的第一轮只建水位、不推送。
# 否则重开程序会把停机期间积压的条目一次性全推（瞬间 20+ 条 × build+report），
# 直接撞服务端 60 次/分钟限速（日志成片「请求过于频繁」），对接收方也是突发流量。
RATE_MAX = 50
RATE_WIN = 60.0
_rate_hits = {}
_rate_lock = threading.Lock()

# 配置来源：脚本同目录 xb_config.json 优先（KB 教程方式 A），缺省回落环境变量
FILECFG = None
try:
    _fp = os.path.join(DATA_DIR, 'xb_config.json')
    if os.path.isfile(_fp):
        with open(_fp, 'r', encoding='utf-8') as _f:
            _j = json.load(_f)
            if isinstance(_j, dict):
                FILECFG = _j
except Exception:
    FILECFG = None

# 首次接入（app/启动器形态）：没有配置文件 → 优先环境变量 XB_CODE，其次交互粘贴。
def _parse_code(raw):
    """接入码 XB1.<base64url(zlib(json))> 或配置文件全文或裸 clients 数组。"""
    s = str(raw or '').strip()
    if not s:
        return None
    s = s.replace('\u201c', '"').replace('\u201d', '"')
    obj = None
    if s[:4].upper() == 'XB1.':
        try:
            b = s[4:].replace('-', '+').replace('_', '/')
            b += '=' * (-len(b) % 4)
            import base64
            import zlib
            buf = base64.b64decode(b)
            if len(buf) > 1 and buf[0] == 0x1f:
                buf = zlib.decompress(buf[1:], -15)
            obj = json.loads(buf.decode('utf-8'))
        except Exception:
            return None
    elif s[0] == '{':
        try:
            obj = json.loads(s)
        except Exception:
            return None
    elif s[0] == '[':
        try:
            obj = {'clients': json.loads(s)}
        except Exception:
            return None
    if not isinstance(obj, dict) and not isinstance(obj, list):
        return None
    if isinstance(obj, list):
        obj = {'clients': obj}
    clients = obj.get('clients')
    if not isinstance(clients, list):
        clients = [obj] if obj.get('uid') else []
    ok = []
    for c in clients:
        if not isinstance(c, dict):
            continue
        try:
            uid = int(c.get('uid') or 0)
        except Exception:
            continue
        key = str(c.get('key') or '').strip().lower()
        if uid > 0 and re.fullmatch(r'[0-9a-f]{32}', key or ''):
            ok.append({'uid': uid, 'key': key, 'name': str(c.get('name') or ('uid%d' % uid))})
    if not ok:
        return None
    return {
        'host': re.sub(r'^https?://|/+$', '', str(obj.get('host') or obj.get('site_host') or '')),
        'interval': max(5, int(obj.get('interval') or 5)),
        'clients': ok,
    }

def _app_first_run():
    """没有配置时的一次性接入。返回 (clients, host)；无法完成则退出。"""
    code = str(os.environ.get('XB_CODE') or '').strip()
    if code:
        cfg = _parse_code(code)
        if not cfg:
            print('  ✗ 环境变量 XB_CODE 里的接入码无法识别（可能复制不全）。', flush=True)
            sys.exit(1)
        try:
            with open(os.path.join(DATA_DIR, 'xb_config.json'), 'w', encoding='utf-8') as f:
                json.dump({'host': cfg['host'] or 'new.xianbao.fun', 'interval': cfg['interval'],
                           'clients': cfg['clients']}, f, ensure_ascii=False, indent=2)
        except Exception:
            print('  ✗ 无法写入配置到 ' + DATA_DIR, flush=True)
            sys.exit(1)
        print('  ✓ 已用 XB_CODE 写入配置：' + '、'.join(c['name'] for c in cfg['clients']), flush=True)
        return cfg['clients'], (cfg['host'] or 'new.xianbao.fun')

    if not sys.stdin or not sys.stdin.isatty():
        print('  ✗ 还没有接入配置，而且当前不是交互终端。', flush=True)
        print('    请任选一种方式：', flush=True)
        print('      ① 在真实终端里直接运行本程序（会提示粘贴接入码）', flush=True)
        print('      ② 设环境变量 XB_CODE=XB1.xxxxx 后再运行', flush=True)
        print('      ③ 把用户中心生成的 xb_config.json 放到本目录', flush=True)
        sys.exit(1)

    print('', flush=True)
    print('  ┌──────────────────────────────────────────────┐', flush=True)
    print('  │           线报酷 · 设备托管推送              │', flush=True)
    print('  └──────────────────────────────────────────────┘', flush=True)
    print('', flush=True)
    print('  第一次使用，需要粘贴「接入码」（XB1. 开头）。', flush=True)
    print('  获取方式：线报酷 → 用户中心 → 信息推送设置 → 设备托管', flush=True)
    print('            → 一键生成配置 → 复制「⓪ 接入码」那一串', flush=True)
    print('  这台机器可以托管多个账号（帮家人朋友一起挂）：', flush=True)
    print('  每个账号的主人在他自己的用户中心生成码发给你，逐枚粘贴即可。', flush=True)
    print('', flush=True)
    clients, host = [], ''
    while True:
        try:
            raw = input('  粘贴接入码后按回车（直接回车 = 粘完了）：')
        except (EOFError, KeyboardInterrupt):
            print('\n  已取消。', flush=True)
            _app_hold()
            sys.exit(1)
        if not raw.strip():
            break
        cfg = _parse_code(raw)
        if not cfg:
            print('  ✗ 接入码无法识别（可能复制不全，或不是最新生成的），请整串复制再试。', flush=True)
            continue
        ch = re.sub(r'^https?://|/+$', '', str(cfg.get('host') or ''), flags=re.I)
        if host and ch and ch != host:
            print('  ✗ 这枚接入码来自不同站点（%s）：本机已托管 %s。所有账号的码必须从同一个站生成，这一枚先跳过。' % (ch, host), flush=True)
            continue
        if ch:
            host = ch
        before = len(clients)
        for c in cfg['clients']:
            dup = next((x for x in clients if int(x['uid']) == int(c['uid'])), None)
            if dup:
                dup['key'] = c['key']
                dup['name'] = c.get('name') or dup.get('name') or ('uid%d' % c['uid'])
                print('  · 账号 %s（uid %d）的接入码已更新' % (dup['name'], c['uid']), flush=True)
            else:
                clients.append(dict(c))
                print('  + 已添加账号：%s（uid %d）' % (clients[-1]['name'], c['uid']), flush=True)
        if len(clients) == before:
            print('  · 没有变化（可能粘了同一枚）', flush=True)
        print('  当前共 %d 个账号。继续粘下一枚，或直接回车完成。' % len(clients), flush=True)
    if not clients:
        print('  ✗ 一枚接入码都没有，无法启动。重新运行本程序再粘一次。', flush=True)
        _app_hold()
        sys.exit(1)
    try:
        with open(os.path.join(DATA_DIR, 'xb_config.json'), 'w', encoding='utf-8') as f:
            json.dump({'host': host or 'new.xianbao.fun', 'interval': 5, 'clients': clients},
                      f, ensure_ascii=False, indent=2)
    except Exception:
        print('  ✗ 无法写入配置到 ' + DATA_DIR + '\n    请把本程序放到你有写权限的文件夹后再试。', flush=True)
        _app_hold()
        sys.exit(1)
    print('', flush=True)
    print('  ✓ 接入成功！账号 %d 个：%s' % (len(clients), '、'.join(c['name'] for c in clients)), flush=True)
    print('    配置已保存到：' + os.path.join(DATA_DIR, 'xb_config.json'), flush=True)
    print('    以后随时可以再加账号：在脚本文件夹里新建一个文本文档，命名为 xb_accounts（保留 .txt 后缀也算）→ 下一轮自动打印账号面板。', flush=True)
    print('', flush=True)
    return clients, (host or 'new.xianbao.fun')

def _app_hold():
    """出错时别让窗口一闪而过（桌面形态）。"""
    if not IS_APP:
        return
    try:
        if sys.stdin and sys.stdin.isatty():
            input('  按回车键关闭窗口…')
    except Exception:
        pass

if IS_APP and not FILECFG:
    _cl, _h = _app_first_run()
    FILECFG = {'host': _h, 'clients': _cl}

HOST = re.sub(r'^https?://|/+$', '', (FILECFG or {}).get('host') or os.environ.get('XB_HOST', 'new.xianbao.fun'), flags=re.I)
INTERVAL = max(5, int((FILECFG or {}).get('interval') or os.environ.get('XB_INTERVAL', '5') or 5))
CTX = ssl.create_default_context()

# ---------- 多用户配置 ----------
def _parse_clients():
    raw = (json.dumps(FILECFG['clients'], ensure_ascii=False) if FILECFG and isinstance(FILECFG.get('clients'), list) else '') or os.environ.get('XB_CLIENTS') or os.environ.get('XB_CLIENT') or ''
    out = []
    try:
        arr = json.loads(raw)
        items = arr if isinstance(arr, list) else [arr]
        for c in items:
            if not isinstance(c, dict):
                continue
            uid = int(c.get('uid') or 0)
            key = str(c.get('key') or '').strip().lower()
            if uid > 0 and re.fullmatch(r'[0-9a-f]{32}', key or ''):
                out.append({'uid': uid, 'key': key, 'name': str(c.get('name') or ('uid%d' % uid))})
    except Exception:
        pass
    return out

CLIENTS = _parse_clients()
if not CLIENTS:
    print('[xianbao_push] 未配置 XB_CLIENTS（示例：[{"uid":123,"key":"32位密钥","name":"主号"}]），退出。')
    sys.exit(0)

# ---------- HTTP ----------
def http_req(method, url_path, headers=None, body=None, timeout=15):
    """返回 (status, body_str)；任何网络错误抛异常"""
    url = 'https://' + HOST + url_path
    req = urllib.request.Request(url, data=body.encode('utf-8') if body is not None else None, method=method)
    for k, v in (headers or {}).items():
        req.add_header(k, v)
    req.add_header('User-Agent', UA)
    with urllib.request.urlopen(req, timeout=timeout, context=CTX) as resp:
        return resp.getcode(), resp.read().decode('utf-8', 'replace')

def api_post(client, filename, params):
    body = urllib.parse.urlencode(params)
    return http_req('POST', '/zb_users/plugin/xbpush/api/' + filename, {
        'Content-Type': 'application/x-www-form-urlencoded',
        'X-Xbpush-Key': client['key'],
    }, body)

def _rate_wait(uid):
    """每 uid 60 秒滑动窗，最多 RATE_MAX 次（服务端 key 通道上限 60/分钟，留余量给配置刷新与心跳）。"""
    with _rate_lock:
        now = time.time()
        arr = [t for t in _rate_hits.get(uid, []) if now - t < RATE_WIN]
        _rate_hits[uid] = arr
        if len(arr) < RATE_MAX:
            return 0.0
        return max(0.25, RATE_WIN - (now - arr[0]) + 0.1)

def _rate_mark(uid):
    with _rate_lock:
        _rate_hits.setdefault(uid, []).append(time.time())

def api_post_push(client, filename, params):
    """build/report 走限速版（每轮推送最密集的两次请求）。"""
    w = _rate_wait(client['uid'])
    if w > 0:
        time.sleep(w)
    _rate_mark(client['uid'])
    return api_post(client, filename, params)

# ---------- 每用户状态 ----------
class ClientState:
    def __init__(self, client):
        self.client = client
        self.cfg = None
        self.cfg_at = 0.0
        self.rowpass_warned = False
        self.feed_names = []
        self.backoff = 0
        self.next_at = 0.0
        self.last_beat = 0.0
        self.wm = {'seeded': {}}
        self.boot_baseline = {}
        self.agg = {}  # 聚合缓冲（20261020b）：key "渠道id:规则ID" → {ch, rule, items, flushAt}
        self.last_err = ''
        self.key_invalid = False  # 20261021：账号面板用
        self.wm_file = os.path.join(DATA_DIR, 'xianbao_push_wm_%d.json' % client['uid'])
        self._load_wm()
        self.lock = threading.Lock()

    def _load_wm(self):
        try:
            with open(self.wm_file, 'r', encoding='utf-8') as f:
                d = json.load(f)
                if isinstance(d, dict):
                    self.wm = d
                    if not isinstance(self.wm.get('seeded'), dict):
                        self.wm['seeded'] = {}
        except Exception:
            pass

    def save_wm(self):
        try:
            tmp = self.wm_file + '.tmp'
            with open(tmp, 'w', encoding='utf-8') as f:
                json.dump(self.wm, f, ensure_ascii=False)
            os.replace(tmp, self.wm_file)
        except Exception:
            pass

def log(st, tag, msg):
    t = time.localtime()
    print('[%02d:%02d:%02d][%s][%s] %s' % (t.tm_hour, t.tm_min, t.tm_sec, st.client['name'], tag, msg), flush=True)

# ---------- 匹配引擎（与 JS 版 words/anyHit/scopedHit/rowPass/scopePass/inTuitime 同构） ----------
def words(s):
    return [w.strip() for w in re.split(r'[\r\n|#]+', str(s or '')) if w.strip()]

def scope_mode(row):
    return 'split' if str((row or {}).get('title_scope') or '').strip() == 'split' else 'all'

def scope_words(field):
    out = {'title': [], 'content': [], 'both': []}
    for w in words(field):
        m = re.match(r'^(标题|内容)[:：]\s*([\s\S]+)$', w)
        if m:
            out['title' if m.group(1) == '标题' else 'content'].append(m.group(2).strip())
        else:
            out['both'].append(w)
    return out

def any_hit(ws, hay):
    hay = str(hay).lower()
    for w in ws:
        w = str(w or '').strip()
        if not w:
            continue
        if '+' not in w:
            if w.lower() in hay:
                return True
        else:
            parts = [p.strip() for p in w.split('+') if p.strip()]
            if parts and all(p.lower() in hay for p in parts):
                return True
    return False

def any_hit_exact(ws, hay):
    """作者（楼主）维度：整串相等（20261021）。屏蔽 ab 不再连带 abc / xab。

    与 any_hit 的唯一区别是不做子串包含，且不支持 + 组合词（组合是"同时含"语义，
    与"名字等于某值"矛盾）。与 JS 版 anyHitExact、服务端 xbpush_any_hit_exact、
    xb_kw_matches_exact 同口径。区分大小写——与 ES author.raw 的 keyword term 一致。
    """
    hay = str(hay or '').strip()
    if not hay:
        return False
    for w in ws:
        w = str(w or '').strip()
        if w and w == hay:
            return True
    return False

def scoped_hit(field, mode, it):
    field = str(field or '')
    if mode != 'split':
        return any_hit(words(field), it['title'] + '\n' + it['content'])
    sc = scope_words(field)
    if sc['title'] and any_hit(sc['title'], it['title']):
        return True
    if sc['content'] and any_hit(sc['content'], it['content']):
        return True
    if sc['both'] and any_hit(sc['both'], it['title'] + '\n' + it['content']):
        return True
    return False

def price_from_title(title):
    last = None
    for m in re.finditer(r'(?:¥|￥)\s*([0-9]+(?:\.[0-9]+)?)|([0-9]+(?:\.[0-9]+)?)\s*元', title):
        last = m.group(1) or m.group(2)
    return float(last) if last is not None else None

def norm_item(raw, is_zdm, linkhost):
    try:
        rid = int(raw.get('id'))
    except (TypeError, ValueError):
        return None
    title = str(raw.get('title') or '').strip()
    if not rid or not title:
        return None
    catename = str(raw.get('catename') or raw.get('category_name') or '')
    cateid = int(raw.get('cateid') or 0) or (19 if is_zdm else 0)
    price = float(raw['price']) if raw.get('price') not in (None, '') else price_from_title(title)
    content = re.sub(r'\s+', ' ', str(raw.get('content') or ''))
    u = str(raw.get('url') or '')
    lh = linkhost or ('https://' + HOST + '/')
    url_abs = u if u.startswith('http') else (lh + u.lstrip('/') if u else lh + '?id=%d' % rid)
    is_z = is_zdm or cateid == 19
    mall = str(raw.get('mall_name') if raw.get('mall_name') is not None else '')
    if not mall:
        dash = catename.rfind('-')
        if dash != -1 and dash < len(catename) - 1:
            mall = catename[dash + 1:].strip()
    try:
        ts = int(raw.get('shijianchuo') or raw.get('posttime') or 0)
    except (TypeError, ValueError):
        ts = 0
    return {
        'id': rid, 'title': title, 'content': content, 'cateid': cateid, 'catename': catename,
        'louzhu': str(raw.get('louzhu') or ''), 'price': price, 'is_zdm': is_z, 'mall': mall,
        'tags': ['全站', '首页', '分类页:' + catename] + (['值得买'] if is_z else []),
        'shijianchuo': ts,
        '_raw': raw,
    }

def row_pass(row, it):
    need = False
    gjc, pbc = words(row.get('title_gjc')), words(row.get('title_pbc'))
    mode = scope_mode(row)
    if gjc:
        need = True
        if not scoped_hit(row.get('title_gjc'), mode, it):
            return False
    if pbc:
        need = True
        if scoped_hit(row.get('title_pbc'), mode, it):
            return False
    cg, cp = words(row.get('category_gjc')), words(row.get('category_pbc'))
    if cg:
        need = True
        if not any_hit(cg, it['catename']):
            return False
    if cp:
        need = True
        if any_hit(cp, it['catename']):
            return False
    mg, mp = words(row.get('mall_gjc')), words(row.get('mall_pbc'))
    mall = str(it.get('mall') if it.get('mall') is not None else '')
    if mg:
        need = True
        if mall == '' or not any_hit(mg, mall):
            return False
    if mp:
        need = True
        if mall != '' and any_hit(mp, mall):
            return False
    # 作者维度（20261021）：作者名按完全匹配（any_hit_exact），屏蔽 ab 不连带 abc/xab
    lg, lp = words(row.get('louzhu_gjc')), words(row.get('louzhu_pbc'))
    if lg:
        need = True
        if it['louzhu'] == '' or not any_hit_exact(lg, it['louzhu']):
            return False
    if lp:
        need = True
        if any_hit_exact(lp, it['louzhu']):
            return False
    mi, mx = str(row.get('Miprice') or '').strip(), str(row.get('Mxprice') or '').strip()
    if mi or mx:
        need = True
        if it['price'] is not None:
            if mi and it['price'] < float(mi):
                return False
            if mx and it['price'] > float(mx):
                return False
    fw = str(row.get('fanwei') or '').strip()
    if fw:
        need = True
        hit = False
        for tok in words(fw):
            if tok in ('本站', '首页', '主列表', '全站'):
                if not it['is_zdm']:
                    hit = True
                    break
                continue
            if tok == '值得买':
                if '值得买' in it['tags']:
                    hit = True
                    break
                continue
            if tok == '分类页':
                hit = True
                break
            if tok.startswith('分类页:'):
                sub = tok[4:].strip()
                if sub and sub in it['catename']:
                    hit = True
                    break
            elif tok in it['catename']:
                hit = True
                break
        if not hit:
            return False
    return need

def scope_pass(scope, it):
    if not scope:
        return True
    for s0 in scope:
        s = str(s0).strip()
        if s in ('', 'all'):
            return True
        if s == 'zdm' and it['is_zdm']:
            return True
        if re.fullmatch(r'\d+', s) and int(s) == it['cateid']:
            return True
    return False

def in_tuitime(tt):
    if not tt or not tt.get('switch'):
        return True
    ranges_str = str(tt.get('ranges') or '').strip()
    if not ranges_str:
        return True
    d = time.localtime()
    now = '%02d:%02d' % (d.tm_hour, d.tm_min)
    is_weekend = d.tm_wday in (5, 6)
    for line0 in re.split(r'[\r\n]+', ranges_str):
        line = line0.strip()
        if line.startswith('工作日'):
            if is_weekend:
                continue
            line = line[3:].strip()
        elif line.startswith('周末'):
            if not is_weekend:
                continue
            line = line[2:].strip()
        m = re.match(r'^(\d{1,2}:\d{2})\s*-\s*(\d{1,2}:\d{2})$', line)
        if not m:
            continue
        a, b = m.group(1), m.group(2)
        if b <= a:
            if now >= a or now <= b:
                return True
            continue
        if a <= now <= b:
            return True
    return False

# ---------- 拉配置 ----------
def _cfg_sig(cfg):
    """配置指纹（20261020）：条数 + 启用数 + 内容短哈希。
    只用条数会漏报「规则启用/停用」「关键词改了」这类最常见变更（实测漏报过）。"""
    if not isinstance(cfg, dict):
        return ''
    rules = cfg.get('rules') or []
    chans = [c for c in (cfg.get('local_channels') or []) if isinstance(c, dict) and c.get('status') == 1]
    on_rules = [r for r in rules if isinstance(r, dict) and int(r.get('Status', 1) or 0) == 1]
    feeds = ','.join(cfg.get('feeds') or [])
    h = hashlib.md5()
    for r in on_rules:
        h.update(('~'.join(str(x) for x in [
            r.get('ID'), r.get('Status'), r.get('title_gjc'), r.get('title_pbc'),
            r.get('category_gjc'), r.get('category_pbc'), r.get('louzhu_gjc'),
            r.get('Miprice'), r.get('Mxprice'), ','.join(map(str, r.get('channels') or [])),
            r.get('agg_mode'), r.get('agg_num'), r.get('agg_sec'), r.get('tt_switch')])).encode('utf-8'))
    for c in chans:
        h.update(('~'.join(str(x) for x in [c.get('id'), c.get('name'), c.get('type'), c.get('status')])).encode('utf-8'))
    return '%d|%d|%d|%s|%s' % (len(rules), len(on_rules), len(chans), feeds, h.hexdigest()[:8])

def refresh_config(st):
    code, body = api_post(st.client, 'client_config.php', {'uid': st.client['uid']})
    j = json.loads(body or '{}')
    if j.get('code') != 1:
        msg = str(j.get('msg') or '')
        # 密钥失效识别（20261021，与 JS 版同构）：该账号标红、不抛异常（好账号继续跑）
        if j.get('ratelimit'):
            raise RuntimeError('config: %s (HTTP %s)' % (msg, code))
        if re.search(r'nologin|密钥|失效|登录', msg, re.I):
            if not st.key_invalid:
                st.key_invalid = True
                st.last_err = '密钥已失效'
                log(st, 'err', '该账号密钥已失效（主人可能重置了密钥）——请在「设备托管」页重新生成接入码后重试；其他账号不受影响')
            return
        raise RuntimeError('config: %s (HTTP %s)' % (msg or code, code))
    if st.key_invalid:
        st.key_invalid = False
        st.last_err = ''
        log(st, 'sys', '密钥已恢复，该账号重新上线')
    st.last_err = ''
    # 变更留痕（20261020）：刷新变频繁后 config ok 每 60 秒一条，只在真变化时额外打一行
    prev_sig = _cfg_sig(st.cfg)
    st.cfg = j['data']
    st.cfg_at = time.time()
    # 聚合桶随规则走（20261020b）：配置真变了就把残留桶强制放出，避免用旧规则/旧渠道对象继续发。
    # 只在本脚本已有桶时动作；prev_sig 为空（首轮）不刷。
    if prev_sig and prev_sig != _cfg_sig(j['data']) and st.agg:
        try:
            _flush_agg(st, force=True)
        except Exception:
            pass  # 冲刷失败由服务端 30 分钟死认领兜底
        st.agg = {}
    if str(j['data'].get('rowpass_ver') or '') != VER and not st.rowpass_warned:
        st.rowpass_warned = True
        log(st, 'sys', '提示：服务端规则引擎版本 %s 与本脚本 %s 不一致，建议更新脚本（地址见 Ucenter 托管推送页）' % (j['data'].get('rowpass_ver'), VER))
    st.feed_names = []
    for f in (j['data'].get('feeds') or ['/plus/json/push.json']):
        m = re.search(r'/plus/json/(push(?:_\d+)?)\.json', str(f))
        if m:
            st.feed_names.append(m.group(1))
    q = j['data'].get('quota') or {}
    lmax = ('/%d' % q['local_max']) if q.get('local_max', 0) and q['local_max'] > 0 else '（不限）'
    cur_sig = _cfg_sig(j['data'])
    if prev_sig and prev_sig != cur_sig:
        a = prev_sig.split('|')
        b = cur_sig.split('|')
        diff = []
        if a[0] != b[0]: diff.append('规则条数 %s → %s 条' % (a[0], b[0]))
        if a[1] != b[1]: diff.append('启用规则 %s → %s 条' % (a[1], b[1]))
        if a[2] != b[2]: diff.append('启用托管渠道 %s → %s 个' % (a[2], b[2]))
        if a[3] != b[3]: diff.append('feed 清单有变化')
        if not diff: diff.append('规则/渠道内容有变化（关键词或开关）')
        log(st, 'sys', '⚡ 配置已更新：' + ' / '.join(diff))
    log(st, 'sys', 'config ok：规则 %d 条 / 挂机渠道 %d 个 / feed %s / 今日挂机 %s%s' % (
        len(j['data'].get('rules') or []), len(j['data'].get('local_channels') or []),
        ','.join(st.feed_names), q.get('local_sent', 0), lmax))
    # 面板统计（20261021）：从服务端 quota 取「今日已推」，供账号面板显示
    st.sent_today = int(q.get('local_sent', 0) or 0)

def beat(st):
    try:
        http_req('POST', '/zb_users/plugin/xbpush/api/client_heartbeat.php', {
            'Content-Type': 'application/x-www-form-urlencoded',
            'X-Xbpush-Key': st.client['key'],
        }, 'uid=%d' % st.client['uid'], timeout=8)
    except Exception:
        pass

# ---------- 发送一条 ----------
def send_request(rq):
    """服务端渲染好的 url/method/headers/body；返回 (status, body)"""
    u = urllib.parse.urlparse(rq['url'])
    body = rq.get('body')
    data = body.encode('utf-8') if body else None
    req = urllib.request.Request(rq['url'], data=data, method=(rq.get('method') or 'GET').upper())
    for k, v in (rq.get('headers') or {}).items():
        req.add_header(k, v)
    req.add_header('User-Agent', UA)
    with urllib.request.urlopen(req, timeout=15, context=CTX) as resp:
        return resp.getcode(), resp.read().decode('utf-8', 'replace')[:256 * 1024]

# ---------- 聚合缓冲（20261020b：托管脚本此前完全没有聚合） ----------
# 背景：规则里的「聚合N条 / 聚合N秒」以前只对浏览器扩展生效（扩展有 S.agg 桶），
# 本脚本每条线报都立刻单发，agg_mode 只在 _cfg_sig 里当指纹用、从不参与发送。
# 于是同一条规则「用扩展跑」与「用脚本跑」收到的消息形态不同，用户设了「聚合10秒」在脚本上完全没效果。
#
# 现在与扩展 background.js / xianbao_push.js 同构：
#   - 桶键 = 渠道id + ':' + 规则ID（同一渠道不同规则各攒各的）
#   - num：攒够 N 条立即冲刷；不足则等，最早一条临近时效（STALE_SEC-120）兜底发
#   - sec：入桶满 N 秒后整组发出（flushAt 时间戳，不依赖精确 timer）
#   - one / 未设置：维持原行为，立即单发（逐字节不变）
#
# 与扩展的有意差异：脚本是长驻进程、tick 每 5 秒必然调 flush_agg，不需要扩展那样
# 在 alarm 里单独补一次 scheduleAggTimers；这里也不另起线程（避免与 tick 并发双发同一批）。

def _agg_key(ch, rule):
    rid = rule.get('ID') if isinstance(rule, dict) else None
    if rid is None or rid == '':
        rid = 0
    return '%s:%s' % (ch['id'], rid)

def _agg_mode_of(rule):
    m = rule.get('agg_mode') if isinstance(rule, dict) else None
    return m if m in ('num', 'sec') else 'one'

def _agg_offer(st, ch, it, rule):
    """送进聚合桶；到条件就地冲刷。返回 True=已入桶（由 flush 统一发）；False=调用方应立即单发。"""
    mode = _agg_mode_of(rule)
    if mode == 'one':
        return False
    k = _agg_key(ch, rule)
    b = st.agg.get(k)
    if not b:
        b = {'ch': ch, 'rule': rule, 'items': [], 'flushAt': 0.0}
        st.agg[k] = b
    b['ch'] = ch
    b['rule'] = rule
    # 入桶去重：同一条只留一份（与扩展 aggBucketHas 同口径）
    for x in b['items']:
        if x and x.get('id') == it.get('id'):
            return True
    b['items'].append(it)
    if mode == 'num':
        n = max(2, min(50, int(rule.get('agg_num') or 10)))
        if len(b['items']) >= n:
            _agg_flush(st, k)
    else:
        if not b['flushAt']:
            sec = max(10, min(7200, int(rule.get('agg_sec') or 60)))
            b['flushAt'] = time.time() + sec
    return True

def _flush_agg(st, force=False):
    """tick 尾部调用：把到点的桶冲刷掉（sec 到时间 / num 临近时效兜底 / 规则被改成 one 的残留桶）。"""
    if not st.agg:
        return
    now = time.time()
    for k in list(st.agg.keys()):
        b = st.agg.get(k)
        if not b or not b.get('items'):
            st.agg.pop(k, None)
            continue
        mode = _agg_mode_of(b.get('rule'))
        due = False
        if mode == 'sec':
            due = bool(b.get('flushAt')) and b['flushAt'] <= now
        elif mode == 'num':
            first = b['items'][0]
            oldest = float(first.get('shijianchuo') or 0)
            due = oldest > 0 and (now - oldest) >= (STALE_SEC - 120)
        else:
            due = True  # 规则中途改成 one/停用：残留桶立即放出，别憋着
        if due or force:
            _agg_flush(st, k)

def _agg_flush(st, k):
    """冲刷一个桶：整批一次认领 + 一次构建 + 一次发送，按 aid 逐行上报。"""
    b = st.agg.get(k) if st.agg else None
    if not b or not b.get('ch') or not b.get('items'):
        if st.agg:
            st.agg.pop(k, None)
        return
    items = list(b['items'])
    ch, rule = b['ch'], b['rule']
    st.agg.pop(k, None)
    if not items:
        return
    if len(items) == 1:
        dispatch(st, ch, items[0], rule)
        return
    dispatch_batch(st, ch, items, rule)

def dispatch_batch(st, ch, items, rule):
    """聚合批量发送（items>1）：一次请求承载整批，按服务端回的 aids 逐行上报。"""
    if not items:
        return
    try:
        code, body = api_post_push(st.client, 'client_build.php', {
            'uid': st.client['uid'], 'cid': ch['id'],
            'items': json.dumps([x['_raw'] for x in items], ensure_ascii=False),
            'rule': str((rule.get('ID') if isinstance(rule, dict) else '') or ''),
            'zdm': 1 if items[0].get('is_zdm') else 0,
        })
    except Exception as e:
        log(st, 'err', 'build 网络异常 cid=%s aid=%s×%d：%s（服务端 30 分钟死认领回收后可由其他端补推）'
            % (ch['id'], items[0].get('id'), len(items), e))
        return
    try:
        j = json.loads(body or '{}')
    except Exception:
        j = {'code': 0, 'msg': 'bad json: ' + (body or '')[:80]}
    if j.get('code') != 1:
        log(st, 'err', 'build cid=%s 批%d条：%s' % (ch['id'], len(items), j.get('msg') or ('HTTP %s' % code)))
        return
    if j.get('dup'):
        return  # 整批都已被别端认领
    reqs = j.get('requests') or []
    if not reqs:
        return
    by_id = {x['id']: x for x in items}
    label = items[0]['title'][:30] + ' 等%d条' % len(items)
    for rq in reqs:
        if rq.get('sleep_before', 0) > 0:
            time.sleep(min(rq['sleep_before'], 30))
        http, resp, err = 0, '', rq.get('error') or ''
        if not err and rq.get('url'):
            try:
                http, resp = send_request(rq)
                resp = resp[:2000]
                # 与 dispatch() 同口径：HTTP 4xx/5xx 立即算失败
                if http >= 400:
                    err = 'HTTP %d%s' % (http, _brief_err(resp))
            except Exception as e:
                err = str(e)
        elif not rq.get('url') and not err:
            err = '空请求地址'
        # 该请求承载哪些文章：服务端回 aids（聚合=整批；GET 降级逐条=各自一条）
        aids = [int(a) for a in (rq.get('aids') or [])] or [x['id'] for x in items]
        rows = []
        for aid in aids:
            x = by_id.get(aid)
            rows.append({'cid': ch['id'], 'aid': aid,
                         'title': (x['title'] if x else items[0]['title']),
                         'http': http, 'resp': resp, 'err': err})
        verdicts = report_batch(st, rows)
        line = err
        # 服务端判定优先（它知道 Bark 的 code / 企微的 errcode）
        if verdicts:
            bad = [v for v in verdicts if not v['ok']]
            if bad:
                line = bad[0]['err'] or line or '服务端判定失败'
            elif not err:
                line = ''
        log(st, 'push', ('✗ ' if line else '✓ ') + (ch.get('name') or ch.get('type')) + '：' + label
            + ((' — ' + line) if line else (' — 已送达' + (('（HTTP %d）' % http) if http else ''))))

def report_batch(st, rows):
    """批量上报（一次 POST 交多行）。返回 [{aid,ok,err}]；None=上报本身失败。"""
    if not rows:
        return None
    try:
        code, body = api_post_push(st.client, 'client_report.php', {
            'uid': st.client['uid'],
            'results': json.dumps(rows, ensure_ascii=False),
        })
        j = json.loads(body or '{}')
        items = ((j.get('data') or {}).get('items') or [])
        if not items:
            return None
        return [{'aid': int(x.get('aid') or 0), 'ok': int(x.get('ok') or 0) == 1,
                 'err': str(x.get('err') or '')} for x in items]
    except Exception:
        return None  # 回报失败不重试（服务端 30 分钟死认领回收兜底）

def dispatch(st, ch, it, rule):
    try:
        code, body = api_post_push(st.client, 'client_build.php', {
            'uid': st.client['uid'], 'cid': ch['id'],
            'item': json.dumps(it['_raw'], ensure_ascii=False),
            'rule': str(rule.get('ID') or ''), 'zdm': 1 if it['is_zdm'] else 0,
        })
    except Exception as e:
        log(st, 'err', 'build 网络异常 cid=%s aid=%s：%s（服务端 30 分钟死认领回收后可由其他端补推）' % (ch['id'], it['id'], e))
        return
    try:
        j = json.loads(body or '{}')
    except Exception:
        j = {'code': 0, 'msg': 'bad json: ' + (body or '')[:80]}
    if j.get('code') != 1:
        extra = '（弹窗渠道归扩展）' if j.get('reason') == 'browser_channel_ext_only' else ''
        log(st, 'err', 'build cid=%s aid=%s：%s%s' % (ch['id'], it['id'], j.get('msg') or ('HTTP %s' % code), extra))
        return
    if j.get('dup'):
        return  # 已被扩展/工作台/另一托管端认领——去重兜底
    for rq in (j.get('requests') or []):
        if rq.get('sleep_before', 0) > 0:
            time.sleep(min(rq['sleep_before'], 30))
        http, resp, err = 0, '', rq.get('error') or ''
        if not err and rq.get('url'):
            try:
                http, resp = send_request(rq)
                resp = resp[:2000]
                # 20261020b：HTTP 层拒绝立即算失败。此前只看请求是否发出，
                # Bark 回 HTTP 400「failed to get device token」也打勾 ✓（用户看到满屏成功、手机没收到）。
                # 口径与服务端 client_report.php 的 `if ($ok && $http >= 400)` 一致。
                if http >= 400:
                    err = 'HTTP %d%s' % (http, _brief_err(resp))
            except Exception as e:
                err = str(e)
        elif not rq.get('url') and not err:
            err = '空请求地址'
        # 服务端按渠道预设 check() 做业务层判活（只有它知道 Bark 的 code、企微的 errcode），
        # 把结论拿回来显示：本地日志从此说「对方到底收了没」。
        verdict = report(st, [{'cid': ch['id'], 'aid': it['id'], 'title': it['title'], 'http': http, 'resp': resp, 'err': err}])
        if verdict is not None and not verdict['ok']:
            err = verdict['err'] or err or '服务端判定失败'
        log(st, 'push', ('✗ ' if err else '✓ ') + (ch.get('name') or ch.get('type')) + '：' + it['title'][:40]
            + ((' — ' + err) if err else (' — 已送达' + (('（HTTP %d）' % http) if http else ''))))

def _brief_err(resp):
    """从返回体里抠一句人话错误（Bark/企微/TG 都把原因放在 message/errmsg/description）。"""
    s = str(resp or '').strip()
    if not s:
        return ''
    try:
        j = json.loads(s)
        m = j.get('message') or j.get('msg') or j.get('errmsg') or j.get('description') or j.get('error')
        if not m and isinstance(j.get('data'), dict):
            m = j['data'].get('message')
        if m:
            return '：' + str(m)[:120]
    except Exception:
        pass
    return '：' + re.sub(r'\s+', ' ', s)[:120]

def report(st, results):
    """上报结果；返回服务端判定 {ok, err}（None=上报本身失败，不影响已发生的推送）。"""
    try:
        code, body = api_post_push(st.client, 'client_report.php', {
            'uid': st.client['uid'],
            'results': json.dumps(results, ensure_ascii=False),
        })
        j = json.loads(body or '{}')
        items = ((j.get('data') or {}).get('items') or [])
        if not items:
            return None
        it = items[0]
        verdict = {'ok': int(it.get('ok') or 0) == 1, 'err': str(it.get('err') or '')}
        # 面板统计（20261021）：最近一次失败原因供「账号面板」显示
        if not verdict['ok'] and verdict['err']:
            st.last_err = verdict['err']
        elif verdict['ok']:
            st.last_err = ''
        return verdict
    except Exception:
        return None  # 回报失败不重试（服务端 30 分钟死认领回收兜底）

# ---------- xb_add 加账号（20261022，文件触发式；py 版不进交互循环的补账号入口） ----------
def _do_add():
    """目录出现 xb_add 空文件后触发：粘贴一枚接入码，合并进 xb_config.json 的 clients，重启生效。"""
    print('', flush=True)
    print('  检测到 xb_add：给这台机器再加一个账号。', flush=True)
    print('  请把新账号主人在「线报酷 → 用户中心 → 信息推送设置 → 设备托管」', flush=True)
    print('  生成的「⓪ 接入码」（XB1. 开头）整串粘贴进来。', flush=True)
    interactive = bool(sys.stdin and getattr(sys.stdin, 'isatty', lambda: False)())
    if not interactive:
        print('  ✗ 当前不是交互终端（如青龙定时任务），无法在这里输入接入码。', flush=True)
        print('    请直接编辑 xb_config.json，在 clients 数组里照格式加一行，再重启脚本。', flush=True)
        return
    try:
        raw = input('  接入码：').strip()
    except (EOFError, KeyboardInterrupt):
        print('\n  已取消。', flush=True)
        return
    cfg = _parse_code(raw)
    if not cfg:
        print('  ✗ 接入码无法识别（可能复制不全，或不是最新生成的），请整串复制再试。', flush=True)
        print('    本次未改动 xb_config.json。', flush=True)
        return
    ch = re.sub(r'^https?://|/+$', '', str(cfg.get('host') or ''), flags=re.I)
    if ch and HOST and ch != HOST:
        print('  ✗ 这枚接入码来自不同站点（%s）：本机已托管 %s，拒绝加入。' % (ch, HOST), flush=True)
        return
    try:
        with open(os.path.join(DATA_DIR, 'xb_config.json'), 'r', encoding='utf-8') as f:
            obj = json.load(f)
        if not isinstance(obj.get('clients'), list):
            raise ValueError
    except Exception:
        print('  ✗ 读取 xb_config.json 失败，本次未改动。请手动检查该文件。', flush=True)
        return
    for c in cfg['clients']:
        dup = next((x for x in obj['clients'] if int(x.get('uid') or 0) == int(c['uid'])), None)
        if dup:
            dup['key'] = c['key']
            dup['name'] = c.get('name') or dup.get('name') or ('uid%d' % c['uid'])
            print('  · 账号 %s（uid %d）的接入码已更新（重启后生效）' % (dup['name'], c['uid']), flush=True)
        else:
            obj['clients'].append(dict(c))
            print('  + 已添加账号：%s（uid %d）' % (c['name'], c['uid']), flush=True)
    try:
        with open(os.path.join(DATA_DIR, 'xb_config.json'), 'w', encoding='utf-8') as f:
            json.dump(obj, f, ensure_ascii=False, indent=2)
    except Exception:
        print('  ✗ 写回 xb_config.json 失败，请检查目录写权限。', flush=True)
        return
    print('  当前共 %d 个账号。重启脚本（关掉窗口重新运行）后新账号开始托管。' % len(obj['clients']), flush=True)
    print('', flush=True)

# ---------- 账号面板（20261021，与 JS 版同构；py 版为快照式） ----------
def print_panel(states_map):
    """打印一次账号快照。py 版不进交互循环（青龙/无人值守友好），改账号用 xb_accounts 文件 + 重启。"""
    rows = list(states_map.values())
    print('', flush=True)
    print('  线报酷 · 账号面板（%d 个账号）' % len(rows), flush=True)
    print('  ' + '-' * 74, flush=True)
    print('  %-12s %-8s %-6s %-6s %-6s %s' % ('账号', 'uid', '规则', '渠道', '今日', '状态 / 最近错误'), flush=True)
    print('  ' + '-' * 74, flush=True)
    for st in rows:
        rules = '…'
        chans = '…'
        if st.cfg:
            rs = st.cfg.get('rules') or []
            rules = sum(1 for r in rs if (r.get('Status') if r.get('Status') is not None else 1) == 1)
            cs = st.cfg.get('local_channels') or []
            chans = sum(1 for c in cs if c.get('status') == 1)
        if st.key_invalid:
            status = '✖ 密钥已失效（请账号主人重新生成接入码）'
        elif st.last_err:
            status = '✖ ' + str(st.last_err)[:44]
        elif st.cfg is not None and not (st.cfg.get('rules') or []):
            status = '▲ 没有启用的推送规则'
        elif st.cfg is not None and not any((c.get('status') == 1) for c in (st.cfg.get('local_channels') or [])):
            status = '▲ 没有可用的托管渠道'
        else:
            status = '● 正常'
        print('  %-12s %-8s %-6s %-6s %-6s %s' % (
            str(st.client['name'])[:12], st.client['uid'], rules, chans, st.sent_today, status), flush=True)
    print('  ' + '-' * 74, flush=True)
    print('  （加账号：本目录新建空文件 xb_add 后重启；导出：直接看 xb_config.json）', flush=True)
    print('', flush=True)

# ---------- 主轮询 ----------
def tick(st):
    now = time.time()
    if st.backoff > 0 and now < st.next_at:
        return
    try:
        # 手动立即刷新（20261020）：脚本目录存在 xb_reload 空文件 → 本轮立刻重拉配置并删掉它
        try:
            # 20261022：双名兜底，接受 xb_reload.txt（小白记事本建文件友好）
            for _name in (RELOAD_FLAG, RELOAD_FLAG + '.txt'):
                _flag = os.path.join(DATA_DIR, _name)
                if os.path.isfile(_flag):
                    os.remove(_flag)
                    st.cfg_at = 0
                    log(st, 'sys', '检测到 xb_reload，立即重新拉取配置')
                    break
        except Exception:
            pass  # 无写权限等异常不打断主循环
        if not st.cfg or now - st.cfg_at > CFG_TTL:
            refresh_config(st)
        if not st.last_beat or now - st.last_beat > 600:
            st.last_beat = now
            threading.Thread(target=beat, args=(st,), daemon=True).start()

        fresh_total = 0
        for name in st.feed_names:
            is_zdm = name == 'push_19'
            since = int(st.wm.get(name) or 0)
            code, body = http_req('GET', '/plus/json/push_delta.php?name=%s&since=%d' % (name, since), timeout=12)
            if code != 200:
                raise RuntimeError('delta %s HTTP %s' % (name, code))
            j = json.loads(body or '{}')
            items = j.get('a') or []
            # 水位回卷自愈（20261021）：since > 服务端当前最大 id（m）只可能是水位记在旧 ID 空间
            # （实例：zbp_xianbao_zdm 10-20 重建 26M→6万）⇒ 增量恒空、值得买永久停推。
            # 处置：水位重置为 m（跳过积压、从最新续推，勿补推历史），下一轮起恢复正常增量。
            try:
                srv_max = int(j.get('m') or 0)
            except (TypeError, ValueError):
                srv_max = 0
            if since > 0 and srv_max > 0 and since > srv_max:
                st.wm[name] = srv_max
                st.save_wm()
                log(st, 'sys', '%s 水位回卷自愈：%d → %d（站点数据源 ID 空间变更，跳过积压从最新续推）'
                    % (name, since, srv_max))
                continue
            if not items:
                continue

            if not st.wm['seeded'].get(name):
                max_id = since
                for raw in items:
                    try:
                        max_id = max(max_id, int(raw.get('id')))
                    except (TypeError, ValueError):
                        pass
                st.wm[name] = max_id
                st.wm['seeded'][name] = 1
                st.save_wm()
                log(st, 'sys', '%s 基线同步 %d 条（不推送历史）' % (name, len(items)))
                continue

            # 每次重启后的第一轮：同样只建水位、不推送（20261020b，站长定稿）。
            # 水位跨进程持久化，所以「重启」必须显式处理，否则停机期间积压的条目会被一次性补推。
            if not st.boot_baseline.get(name):
                st.boot_baseline[name] = 1
                max_id = since
                for raw in items:
                    try:
                        max_id = max(max_id, int(raw.get('id')))
                    except (TypeError, ValueError):
                        pass
                skipped = 0
                for raw in items:
                    try:
                        if int(raw.get('id')) > since:
                            skipped += 1
                    except (TypeError, ValueError):
                        pass
                st.wm[name] = max_id
                st.save_wm()
                log(st, 'sys', '启动基线：%s 跳到最新 id=%d%s' % (
                    name, max_id,
                    ('（跳过停机期间的 %d 条，不补推）' % skipped) if skipped else '（无积压）'))
                continue

            linkhost = st.cfg.get('linkhost') or st.cfg.get('host')
            fresh = []
            cur_max = int(st.wm.get(name) or 0)
            for raw in items:
                it = norm_item(raw, is_zdm, linkhost)
                if not it:
                    continue
                if it['id'] > cur_max:
                    if it['shijianchuo'] > 0 and it['shijianchuo'] < now - STALE_SEC:
                        continue  # 时效保护
                    fresh.append(it)
            fresh.sort(key=lambda x: x['id'])
            if fresh:
                fresh_total += len(fresh)
            # 匹配（per-item 异常隔离：一条异常不丢同批其他条目）
            for it in fresh:
                try:
                    for rule in (st.cfg.get('rules') or []):
                        if int(rule.get('Status', 1) or 0) != 1:
                            continue
                        if not row_pass(rule, it):
                            continue
                        bind = rule.get('channels') if isinstance(rule.get('channels'), list) else ['all']
                        want_all = 'all' in bind
                        want = {str(int(b)): 1 for b in bind if str(b) != 'all'}
                        for ch in (st.cfg.get('local_channels') or []):
                            if ch.get('status') != 1:
                                continue
                            if not want_all and not want.get(str(ch['id'])):
                                continue
                            if not scope_pass(ch.get('scope'), it):
                                continue
                            if rule.get('tt_switch') and not in_tuitime({'switch': rule.get('tt_switch'), 'ranges': rule.get('tt_ranges')}):
                                continue
                            if not in_tuitime(ch.get('tuitime')):
                                continue
                            # 聚合路由（20261020b）：num/sec 进桶由 _flush_agg 统一发；one（默认）立即单发
                            if not _agg_offer(st, ch, it, rule):
                                dispatch(st, ch, it, rule)
                except Exception as e:
                    log(st, 'err', '条目处理异常 aid=%s：%s' % (it['id'], e))
            # 水位在 fresh 全部处理完后推进（中途异常本轮不推，下轮重拉补推）
            max_id = int(st.wm.get(name) or 0)
            for raw in items:
                try:
                    max_id = max(max_id, int(raw.get('id')))
                except (TypeError, ValueError):
                    pass
            st.wm[name] = max_id
            st.save_wm()
        # 聚合冲刷（20261020b）：sec 窗口到点 / num 临近时效兜底 / 规则被改成 one 的残留桶。
        # 必须在 feed 循环之后、且不受「本轮无新条目」影响——sec 桶可能只是没新线报，
        # 那也要照常到点发出，否则一直不推（扩展 v0.17.2 踩过同一个坑）。
        _flush_agg(st)
        if st.backoff > 0:
            log(st, 'sys', '恢复正常轮询')
            st.backoff = 0
    except Exception as e:
        st.backoff = min(st.backoff + 1, 20)
        wait = min(30 * (2 ** st.backoff), 600)
        st.next_at = time.time() + wait
        log(st, 'err', '本轮失败（退避 %ds）：%s' % (wait, e))

def run_client(client):
    st = ClientState(client)
    _PY_STATES[client['uid']] = st
    while True:
        try:
            tick(st)
        except Exception as e:
            log(st, 'err', 'tick 外层异常：%s' % e)
        time.sleep(INTERVAL)

# 账号面板用的全量状态（20261021）：run_client 注册进去；面板读快照
_PY_STATES = {}

if __name__ == '__main__':
    print('[xianbao_push] py v%s 启动：host=%s interval=%ds，托管用户 %s' % (
        VER, HOST, INTERVAL, '、'.join(c['name'] for c in CLIENTS)), flush=True)
    if IS_APP:
        # 桌面/启动器形态：窗口一关就停推，必须把这三件事说清楚
        print('  配置目录：' + DATA_DIR, flush=True)
        print('  日志位置：' + ('就是本窗口（可最小化，别关）' if (sys.stdout and sys.stdout.isatty())
                              else os.path.join(DATA_DIR, 'xianbao_push.log')), flush=True)
        print('  关闭本窗口 = 停止推送。', flush=True)
        print('', flush=True)
    for c in CLIENTS:
        threading.Thread(target=run_client, args=(c,), daemon=True).start()
        st = ClientState(c)
        def beat_loop(st=st):
            while True:
                beat(st)
                time.sleep(600)
        threading.Thread(target=beat_loop, daemon=True).start()
    # xb_accounts 触发器（20261021）：目录里出现空文件 → 打印一次账号快照并删掉。
    # 每 10 秒查一次，比挂在 tick 里更稳（tick 异常退避时也能响应）。
    if CLIENTS:
        def accounts_watch():
            # 20261022：双名兜底 xb_accounts / xb_accounts.txt（Windows 记事本默认带 .txt，
            # 「隐藏扩展名」开着时改名极易变成 xb_accounts.txt.txt，小白翻车点）
            fps = [os.path.join(DATA_DIR, n) for n in ('xb_accounts', 'xb_accounts.txt')]
            while True:
                try:
                    for fp in fps:
                        if os.path.isfile(fp):
                            os.remove(fp)
                            time.sleep(1)  # 等 run_client 把状态注册进来
                            print_panel(_PY_STATES)
                            break
                except Exception:
                    pass
                time.sleep(10)
        threading.Thread(target=accounts_watch, daemon=True).start()
    # xb_add 触发器（20261022）：目录里出现空文件 → 交互粘贴一枚接入码，合并写回 xb_config.json。
    # py 版不进交互循环，这是唯一免手改 JSON 的加账号入口；追加后需重启脚本生效。
    if CLIENTS:
        def add_watch():
            fps = [os.path.join(DATA_DIR, n) for n in ('xb_add', 'xb_add.txt')]
            while True:
                try:
                    for fp in fps:
                        if os.path.isfile(fp):
                            os.remove(fp)
                            _do_add()
                            break
                except Exception:
                    pass
                time.sleep(10)
        threading.Thread(target=add_watch, daemon=True).start()
    try:
        while True:
            time.sleep(3600)
    except (KeyboardInterrupt, SystemExit):
        print('')
        print('[xianbao_push] 退出。', flush=True)
    except Exception as e:
        # 桌面形态：任何异常都不要让窗口无声消失
        print('[xianbao_push] 异常退出：%s' % e, flush=True)
        _app_hold()