在线人数显示

murat 5天前

这是在代码块中执行一次显示一次吗

这是在代码块中执行一次显示一次吗

黑白404 2026-09-22 18:36:40

只支持明道云私有化，我自己测试过可以用，你可以试试 """ 统计明道云（HAP）在线人数（基于 Redis 键统计） 原理：明道云会在 Redis 中写入三类带 TTL 的键，分别对应不同时间窗口的在线状态： - s:wskr:d1:<连接ID>:<userId> WebSocket 连接键，断开即失效（当天在线） - s:pc:a:b2:<accountId> 活跃账号键，TTL 约 3 天（3 天内活跃） - h:a:s:i:<md_pss_id> 登录会话键，TTL 上限 7 天（7 天内登录） """ import redis import json # Redis 连接配置（改成你自己的）如果是多台填写随机一台即可 REDIS_HOST = '127.0.0.1' REDIS_PORT = 6379 REDIS_PASSWORD = 'your-password' REDIS_DB = 0 def scan_keys(r, pattern): cursor, keys = 0, [] while True: cursor, batch = r.scan(cursor=cursor, match=pattern, count=2000) keys.extend(batch) if cursor == 0: return keys def main(): r = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, db=REDIS_DB, password=REDIS_PASSWORD, protocol=2, socket_timeout=15) r.ping() print('Redis 连接成功\n') # 1) 当天在线：统计 WebSocket 连接键，按 userId 去重 ws_users = set() for k in scan_keys(r, 's:wskr:d1:*'): parts = k.decode('utf-8').split(':') if len(parts) >= 5 and parts[4] != 'milliseconds': ws_users.add(parts[4]) # 2) 3 天内活跃：活跃账号键，一键一账号 pc_accounts = {k.decode('utf-8').split(':')[-1] for k in scan_keys(r, 's:pc:a:b2:*')} # 3) 7 天内登录：解析会话键的 account 字段，按 AccountId 去重 accounts = set() for k in scan_keys(r, 'h:a:s:i:*'): value = r.hget(k, 'account') if value: try: accounts.add(json.loads(value).get('AccountId')) except json.JSONDecodeError: pass print(f'当天在线（WebSocket 连接）: {len(ws_users)} 人') print(f'3 天内活跃账号: {len(pc_accounts)} 人') print(f'7 天内登录用户: {len(accounts)} 人') r.close() if __name__ == '__main__': try: main() except redis.AuthenticationError: print('Redis 认证失败: 密码错误') except redis.ConnectionError as e: print(f'Redis 连接失败: {e}')

只支持明道云私有化，我自己测试过可以用，你可以试试

"""
统计明道云（HAP）在线人数（基于 Redis 键统计）
原理：明道云会在 Redis 中写入三类带 TTL 的键，分别对应不同时间窗口的在线状态：
- s:wskr:d1:<连接ID>:<userId>   WebSocket 连接键，断开即失效（当天在线）
- s:pc:a:b2:<accountId>         活跃账号键，TTL 约 3 天（3 天内活跃）
- h:a:s:i:<md_pss_id>           登录会话键，TTL 上限 7 天（7 天内登录）
"""

import redis
import json

# Redis 连接配置（改成你自己的）如果是多台填写随机一台即可
REDIS_HOST = '127.0.0.1'
REDIS_PORT = 6379
REDIS_PASSWORD = 'your-password'
REDIS_DB = 0

def scan_keys(r, pattern):
    cursor, keys = 0, []
    while True:
        cursor, batch = r.scan(cursor=cursor, match=pattern, count=2000)
        keys.extend(batch)
        if cursor == 0:
            return keys

def main():
    r = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, db=REDIS_DB,
                    password=REDIS_PASSWORD, protocol=2, socket_timeout=15)
    r.ping()
    print('Redis 连接成功\n')

    # 1) 当天在线：统计 WebSocket 连接键，按 userId 去重
    ws_users = set()
    for k in scan_keys(r, 's:wskr:d1:*'):
        parts = k.decode('utf-8').split(':')
        if len(parts) >= 5 and parts[4] != 'milliseconds':
            ws_users.add(parts[4])

    # 2) 3 天内活跃：活跃账号键，一键一账号
    pc_accounts = {k.decode('utf-8').split(':')[-1]
                   for k in scan_keys(r, 's:pc:a:b2:*')}

    # 3) 7 天内登录：解析会话键的 account 字段，按 AccountId 去重
    accounts = set()
    for k in scan_keys(r, 'h:a:s:i:*'):
        value = r.hget(k, 'account')
        if value:
            try:
                accounts.add(json.loads(value).get('AccountId'))
            except json.JSONDecodeError:
                pass

    print(f'当天在线（WebSocket 连接）: {len(ws_users)} 人')
    print(f'3 天内活跃账号: {len(pc_accounts)} 人')
    print(f'7 天内登录用户: {len(accounts)} 人')

    r.close()

if __name__ == '__main__':
    try:
        main()
    except redis.AuthenticationError:
        print('Redis 认证失败: 密码错误')
    except redis.ConnectionError as e:
        print(f'Redis 连接失败: {e}')

任向晖 2026-09-11 18:00:03

目的是什么？

目的是什么？