shop-java/tools/redis_probe.py

# -*- coding: utf-8 -*-
"""最小 RESP 客户端，用于验证 Redis 连通性与键值（不依赖 redis-py）。"""
import socket
import sys


def cmd(sock, *args):
    data = b"*%d\r\n" % len(args)
    for a in args:
        b = a.encode() if isinstance(a, str) else a
        data += b"$%d\r\n%s\r\n" % (len(b), b)
    sock.sendall(data)
    return read_reply(sock)


def read_reply(sock):
    line = readline(sock)
    if not line:
        return None
    t, rest = line[:1], line[1:]
    if t == b"+":
        return rest.decode()
    if t == b"-":
        return Exception(rest.decode())
    if t == b":":
        return int(rest)
    if t == b"$":
        n = int(rest)
        if n == -1:
            return None
        buf = b""
        while len(buf) < n + 2:
            buf += sock.recv(n + 2 - len(buf))
        return buf[:n].decode()
    if t == b"*":
        n = int(rest)
        if n == -1:
            return None
        return [read_reply(sock) for _ in range(n)]
    return line


def readline(sock):
    buf = b""
    while not buf.endswith(b"\r\n"):
        c = sock.recv(1)
        if not c:
            return buf
        buf += c
    return buf[:-2]


def main():
    host, port, pwd = "192.168.1.177", 6379, "@@Qinys12346.."
    s = socket.create_connection((host, port), timeout=5)
    print("AUTH:", cmd(s, "AUTH", pwd))
    print("PING:", cmd(s, "PING"))
    print("DBSIZE:", cmd(s, "DBSIZE"))
    s.close()


if __name__ == "__main__":
    main()