shop-java/tools/verify_yml.py

# -*- coding: utf-8 -*-
"""读取 application.yml 验证 MySQL/Redis 连通性（不打印任何密码）"""
import re
import socket
import pymysql
import urllib.parse
from _root import ROOT

YML = str(ROOT / "yoshop-java/src/main/resources/application.yml")
text = open(YML, encoding="utf-8").read()

def get(pattern, default=""):
    m = re.search(pattern, text, re.M)
    return m.group(1).strip().strip('"\'') if m else default

url = get(r"url:\s*(\S+)")
user = get(r"username:\s*(\S+)")
# redis 段：从 "redis:" 行开始截取，避免误读 server.port
redis_section = text[text.index("redis:"):]
rhost = re.search(r"host:\s*(\S+)", redis_section).group(1).strip("\"'")
rport = int(re.search(r"port:\s*(\d+)", redis_section).group(1))
# datasource 段的 password 是第一个，redis 段的是最后一个
pwds = re.findall(r"password:\s*(.+)", text)
pwd = pwds[0].strip().strip('"\'') if pwds else ""
rpwd = pwds[-1].strip().strip('"\'') if pwds else ""

m = re.match(r"jdbc:mysql://([^:/]+):(\d+)/([^?]+)", url)
db_host, db_port, db_name = m.group(1), int(m.group(2)), m.group(3)

print(f"MySQL 目标: {db_host}:{db_port}/{db_name}  用户: {user}  密码: {'<已读取,共%d位>' % len(pwd)}")
try:
    conn = pymysql.connect(host=db_host, port=db_port, user=user, password=pwd,
                           database=db_name, connect_timeout=5)
    with conn.cursor() as c:
        c.execute("SELECT VERSION()")
        ver = c.fetchone()[0]
        c.execute("SELECT COUNT(*) FROM information_schema.tables WHERE table_schema=%s", (db_name,))
        tables = c.fetchone()[0]
    conn.close()
    print(f"[MYSQL OK] 版本 {ver}，{db_name} 库共 {tables} 张表")
except Exception as e:
    print(f"[MYSQL FAIL] {e}")

print(f"Redis 目标: {rhost}:{rport}  密码: {'<无>' if not rpwd else '<已读取,共%d位>' % len(rpwd)}")
try:
    s = socket.create_connection((rhost, rport), timeout=5)
    def cmd(*a):
        l = f"*{len(a)}\r\n".encode()
        for x in a:
            b = x.encode()
            l += f"${len(b)}\r\n".encode() + b + b"\r\n"
        s.sendall(l)
        return s.recv(200)
    if rpwd:
        r = cmd("AUTH", rpwd)
        if not r.startswith(b"+OK"):
            print(f"[REDIS FAIL] AUTH 失败")
            raise SystemExit
    r = cmd("PING")
    print("[REDIS OK] PING ->", r.decode().strip())
    s.close()
except Exception as e:
    print(f"[REDIS FAIL] {e}")