"""只读采集测试环境结构，结果只能写入工作区外临时目录。 依赖 PyYAML、JDK 17+、本地 MySQL Connector/J；不启动任何业务服务。 凭据从本地 Nacos 文件在内存中读取，不打印，不作为进程参数传递。 """ import argparse import base64 import datetime import json import pathlib import re import subprocess import urllib.error import urllib.parse import urllib.request import yaml SERVICES = ("system", "infra", "report", "device", "launcher", "task", "blacklist", "bpm", "rule") def read_yaml(path): """Spring 配置允许多 YAML document，按顺序递归合并。""" result = {} def merge(target, source): for key, value in source.items(): if isinstance(value, dict) and isinstance(target.get(key), dict): merge(target[key], value) else: target[key] = value for doc in yaml.safe_load_all(path.read_text(encoding="utf-8")): if isinstance(doc, dict): merge(result, doc) return result def stamp(): return datetime.datetime.now(datetime.timezone.utc).isoformat() def save(path, value): # main 已拒绝工作区内输出；这里仅保存可复核临时证据，不生成交付文档。 path.write_text(json.dumps(value, ensure_ascii=False, indent=2, default=str), encoding="utf-8") def request_json(url, username, password, sql=None): authorization = base64.b64encode(f"{username}:{password}".encode()).decode() req = urllib.request.Request(url, data=None if sql is None else sql.encode(), headers={ "Authorization": f"Basic {authorization}", "Content-Type": "text/plain; charset=utf-8" }, method="GET" if sql is None else "POST") # 内网数据源不经系统 HTTP 代理，避免将凭据转交代理。 opener = urllib.request.build_opener(urllib.request.ProxyHandler({})) with opener.open(req, timeout=25) as response: return json.load(response) def identifier(value): return "`" + value.replace("`", "``") + "`" def td_collect(config, database): parsed = urllib.parse.urlsplit(re.sub(r"^jdbc:TAOS-(?:RS|WS):", "http:", config["url"], flags=re.I)) endpoint = f"http://{parsed.netloc}/rest/sql" queries = [] def sql(statement): queries.append(statement) payload = request_json(endpoint, config["username"], config["password"], statement) if payload.get("code", 0) != 0: # 服务端描述可能含环境信息；不保存原始错误文本。 raise RuntimeError(f"TDengine_code_{payload.get('code')}") columns = [col[0] for col in payload.get("column_meta", [])] return [dict(zip(columns, row)) for row in payload.get("data", [])] result = {"database": database, "captured_at": stamp(), "queries": queries, "objects": {}} try: result["server_version"] = sql("SELECT SERVER_VERSION()") stables = sql(f"SHOW {identifier(database)}.STABLES") result["stables"] = stables # 仅统计系统目录中的子表数量，不读取设备标签值或业务数据。 safe_db = database.replace("'", "''") result["child_counts"] = sql("SELECT stable_name, COUNT(*) AS child_count FROM information_schema.ins_tables " f"WHERE db_name='{safe_db}' GROUP BY stable_name") normal = sql("SELECT table_name FROM information_schema.ins_tables " f"WHERE db_name='{safe_db}' AND (stable_name IS NULL OR stable_name='')") result["normal_tables"] = normal names = [(next(iter(row.values())), "stable") for row in stables] names += [(row["table_name"], "normal") for row in normal] for name, kind in names: try: result["objects"][name] = {"kind": kind, "columns": sql(f"DESCRIBE {identifier(database)}.{identifier(name)}")} except Exception as exc: result["objects"][name] = {"kind": kind, "status": "error", "error": safe_error(exc)} result["status"] = "partial" if any(v.get("status") == "error" for v in result["objects"].values()) else "ok" except Exception as exc: result.update(status="error", error=safe_error(exc)) return result def safe_error(exc): if isinstance(exc, urllib.error.HTTPError): return f"HTTP_{exc.code}" if isinstance(exc, RuntimeError) and str(exc).startswith("TDengine_code_"): return str(exc) return type(exc).__name__ def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--workspace", type=pathlib.Path, required=True) parser.add_argument("--output", type=pathlib.Path, required=True) parser.add_argument("--java", required=True) parser.add_argument("--mysql-jar", required=True) parser.add_argument("--extra-td", action="append", default=[], help="代码中确认、但公共配置缺失的同实例数据库") args = parser.parse_args() workspace, output = args.workspace.resolve(), args.output.resolve() if output == workspace or workspace in output.parents: parser.error("输出必须位于工作区之外的临时目录") output.mkdir(parents=True, exist_ok=True) config_dir = workspace / "doc/数据库设计管理/nacos" datasource = read_yaml(config_dir / "datasource.yaml")["datasource"] manifest = {"captured_at": stamp(), "sources": {}, "service_bindings": {}, "unresolved": [], "scope": "metadata_only"} selected = set() for service in SERVICES: config = read_yaml(config_dir / f"{service}-server.yaml") bindings = config.get("spring", {}).get("datasource", {}).get("dynamic", {}).get("datasource", {}) manifest["service_bindings"][service] = {} for alias, connection in bindings.items(): match = re.fullmatch(r"\$\{datasource\.([^.]+)\.url\}", str(connection.get("url", ""))) if not match: manifest["unresolved"].append({"service": service, "alias": alias, "reason": "non_reference_url"}) continue key = match.group(1) manifest["service_bindings"][service][alias] = key if key not in datasource: manifest["unresolved"].append({"service": service, "alias": alias, "source": key, "reason": "missing_datasource"}) else: selected.add(key) identities = {} for key in sorted(selected): config = datasource[key] url = config["url"] kind = "mysql" if url.startswith("jdbc:mysql:") else "tdengine" normalized = re.sub(r"^jdbc:(?:mysql|TAOS-RS|TAOS-WS):", "http:", url, flags=re.I) parsed = urllib.parse.urlsplit(normalized) database = parsed.path.strip("/") identity = (kind, parsed.netloc, database) if identity in identities: manifest["sources"][key] = dict(manifest["sources"][identities[identity]], shared_with=identities[identity]) continue identities[identity] = key filename = f"{kind}-{database}.json" target = output / filename if target.exists(): result = json.loads(target.read_text(encoding="utf-8")) elif kind == "mysql": # 不继承 URL 中的写入/重试选项，连接和读取均有界。 jdbc = f"jdbc:mysql://{parsed.netloc}/{database}?useSSL=false&allowPublicKeyRetrieval=true&connectTimeout=5000&socketTimeout=25000&characterEncoding=UTF-8" stdin = "\n".join(base64.b64encode(str(x).encode()).decode() for x in (jdbc, config["username"], config["password"], database)) + "\n" try: proc = subprocess.run([args.java, "-cp", args.mysql_jar, str(pathlib.Path(__file__).with_name("MysqlMetadata.java"))], input=stdin.encode(), stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=120) result = json.loads(proc.stdout.decode("utf-8")) if proc.returncode == 0 else {"status": "error", "error": "Java_process_failed", "exit_code": proc.returncode} except Exception as exc: result = {"status": "error", "error": safe_error(exc)} result.update(database=database, captured_at=stamp()) save(target, result) else: result = td_collect(config, database) save(target, result) manifest["sources"][key] = {"kind": kind, "database": database, "file": filename, "status": result["status"]} print(f"{kind} {database}: {result['status']}", flush=True) td_config = next((datasource[k] for k in sorted(selected) if "TAOS-" in datasource[k]["url"]), None) for database in args.extra_td: if not re.fullmatch(r"[A-Za-z0-9_]+", database) or td_config is None: parser.error("extra-td 需要合法标识符及已有 TDengine 连接") target = output / f"tdengine-{database}.json" result = json.loads(target.read_text(encoding="utf-8")) if target.exists() else td_collect(td_config, database) save(target, result) manifest["sources"][f"code:{database}"] = {"kind": "tdengine", "database": database, "file": target.name, "status": result["status"], "origin": "code_qualified_database_on_configured_instance"} print(f"tdengine {database}: {result['status']}", flush=True) es = read_yaml(config_dir / "application-common.yaml").get("easy-es", {}) target = output / "elasticsearch.json" if target.exists(): result = json.loads(target.read_text(encoding="utf-8")) else: try: base = f"{es.get('schema', 'http')}://{str(es['address']).split(',')[0]}" auth = (es.get("username", ""), es.get("password", "")) version = request_json(base, *auth).get("version", {}).get("number") mappings = request_json(base + "/_mapping?expand_wildcards=open", *auth) aliases = request_json(base + "/_alias?expand_wildcards=open", *auth) result = {"status": "ok", "server_version": version, "mappings": mappings, "aliases": aliases, "captured_at": stamp()} except Exception as exc: result = {"status": "error", "error": safe_error(exc), "captured_at": stamp()} save(target, result) manifest["elasticsearch"] = {"file": target.name, "status": result["status"]} save(output / "manifest.json", manifest) print(f"elasticsearch: {result['status']}; unresolved bindings: {len(manifest['unresolved'])}", flush=True) if __name__ == "__main__": main()