290 lines
8.7 KiB
Python
290 lines
8.7 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Oracle Jump Query - AI Skill
|
|
通过中转服务查询远程 Oracle 存储过程元数据
|
|
|
|
使用方式:
|
|
python oracle_skill.py analyze BOS M_RETAIL_SUBMIT
|
|
python oracle_skill.py list BOS
|
|
python oracle_skill.py source BOS M_RETAIL_SUBMIT
|
|
python oracle_skill.py deps BOS M_RETAIL_SUBMIT
|
|
python oracle_skill.py tables BOS M_RETAIL_SUBMIT
|
|
python oracle_skill.py servers
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
import requests
|
|
|
|
# ============================================================
|
|
# 配置（修改为你的中转服务地址和认证信息）
|
|
# ============================================================
|
|
TRANSIT_URL = "http://bos.henlo.net:6357"
|
|
AUTH_TOKEN = ""
|
|
DEFAULT_SERVER_ID = "server-001"
|
|
|
|
|
|
def get_script_dir():
|
|
"""获取脚本自身所在目录"""
|
|
return os.path.dirname(os.path.abspath(__file__))
|
|
|
|
|
|
def get_config():
|
|
"""从 config.json 加载配置（优先脚本目录，其次当前目录）"""
|
|
# 优先从脚本同目录加载
|
|
cfg_path = os.path.join(get_script_dir(), "config.json")
|
|
try:
|
|
with open(cfg_path, "r", encoding="utf-8") as f:
|
|
cfg = json.load(f)
|
|
return cfg
|
|
except FileNotFoundError:
|
|
pass
|
|
# 其次从当前目录加载
|
|
try:
|
|
with open("config.json", "r", encoding="utf-8") as f:
|
|
cfg = json.load(f)
|
|
return cfg
|
|
except FileNotFoundError:
|
|
return None
|
|
|
|
|
|
def make_headers():
|
|
"""构建 HTTP 请求头"""
|
|
headers = {"Content-Type": "application/json"}
|
|
token = AUTH_TOKEN
|
|
cfg = get_config()
|
|
if cfg:
|
|
token = cfg.get("auth_token", token)
|
|
if token:
|
|
headers["Authorization"] = f"Bearer {token}"
|
|
return headers
|
|
|
|
|
|
def get_transit_url():
|
|
"""获取中转服务 URL"""
|
|
cfg = get_config()
|
|
if cfg and "transit_url" in cfg:
|
|
return cfg["transit_url"]
|
|
return TRANSIT_URL
|
|
|
|
|
|
def get_server_id():
|
|
"""获取默认服务器 ID"""
|
|
cfg = get_config()
|
|
if cfg and "server_id" in cfg:
|
|
return cfg["server_id"]
|
|
return DEFAULT_SERVER_ID
|
|
|
|
|
|
def query(server_id, action, schema="", name="", timeout=60):
|
|
"""发送查询请求到中转服务"""
|
|
url = f"{get_transit_url()}/api/query"
|
|
payload = {
|
|
"server_id": server_id,
|
|
"action": action,
|
|
"schema": schema,
|
|
"name": name,
|
|
"timeout": timeout,
|
|
}
|
|
|
|
try:
|
|
resp = requests.post(url, json=payload, headers=make_headers(), timeout=timeout + 5)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
except requests.exceptions.ConnectionError:
|
|
return {"success": False, "error": "无法连接中转服务，请检查地址和端口"}
|
|
except requests.exceptions.Timeout:
|
|
return {"success": False, "error": "请求超时，Agent 可能未响应"}
|
|
except requests.exceptions.HTTPError as e:
|
|
return {"success": False, "error": f"HTTP 错误: {e}"}
|
|
|
|
|
|
def list_servers():
|
|
"""列出所有在线的 Agent 服务器"""
|
|
url = f"{get_transit_url()}/api/servers"
|
|
try:
|
|
resp = requests.get(url, headers=make_headers(), timeout=10)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
except Exception as e:
|
|
return {"error": str(e)}
|
|
|
|
|
|
def health_check():
|
|
"""检查中转服务健康状态"""
|
|
url = f"{get_transit_url()}/api/health"
|
|
try:
|
|
resp = requests.get(url, timeout=5)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
except Exception as e:
|
|
return {"error": str(e)}
|
|
|
|
|
|
def print_result(result):
|
|
"""格式化输出结果"""
|
|
if result.get("success"):
|
|
data = result.get("data", "")
|
|
print(data)
|
|
else:
|
|
error = result.get("error", "未知错误")
|
|
print(f"❌ 错误: {error}")
|
|
|
|
|
|
def interactive_mode():
|
|
"""交互式模式"""
|
|
print("=" * 50)
|
|
print(" Oracle Jump Query - AI Skill")
|
|
print(" 输入 help 查看命令，exit 退出")
|
|
print("=" * 50)
|
|
|
|
# Check transit server
|
|
health = health_check()
|
|
if "error" in health:
|
|
print(f"⚠️ 中转服务不可用: {health['error']}")
|
|
print("请检查 config.json 中的 transit_url 配置")
|
|
else:
|
|
print(f"✅ 中转服务正常 (在线 Agent: {health.get('agents', 0)})")
|
|
|
|
print()
|
|
|
|
while True:
|
|
try:
|
|
line = input("> ").strip()
|
|
except (EOFError, KeyboardInterrupt):
|
|
print("\n再见！")
|
|
break
|
|
|
|
if not line:
|
|
continue
|
|
|
|
parts = line.split()
|
|
cmd = parts[0].lower()
|
|
|
|
if cmd == "exit" or cmd == "quit":
|
|
print("再见！")
|
|
break
|
|
|
|
elif cmd == "help":
|
|
print("""
|
|
可用命令:
|
|
analyze <schema> <procedure> - 分析存储过程（源码+依赖+表+触发器）
|
|
list <schema> - 列出 schema 下的所有存储过程
|
|
source <schema> <procedure> - 获取存储过程源码
|
|
deps <schema> <procedure> - 获取存储过程依赖
|
|
tables <schema> <procedure> - 获取相关表结构
|
|
servers - 列出在线 Agent
|
|
health - 检查中转服务状态
|
|
server <server_id> - 切换目标服务器
|
|
exit - 退出
|
|
""")
|
|
|
|
elif cmd == "servers":
|
|
result = list_servers()
|
|
if "error" in result:
|
|
print(f"❌ {result['error']}")
|
|
else:
|
|
servers = result.get("servers", [])
|
|
if not servers:
|
|
print("没有在线的 Agent")
|
|
else:
|
|
for s in servers:
|
|
print(f" 🟢 {s['server_id']}")
|
|
|
|
elif cmd == "health":
|
|
result = health_check()
|
|
if "error" in result:
|
|
print(f"❌ {result['error']}")
|
|
else:
|
|
print(f"✅ 中转服务正常 (在线 Agent: {result.get('agents', 0)})")
|
|
|
|
elif cmd == "server":
|
|
if len(parts) < 2:
|
|
print(f"当前服务器: {get_server_id()}")
|
|
else:
|
|
global DEFAULT_SERVER_ID
|
|
DEFAULT_SERVER_ID = parts[1]
|
|
print(f"切换到服务器: {parts[1]}")
|
|
|
|
elif cmd == "analyze":
|
|
if len(parts) < 3:
|
|
print("用法: analyze <schema> <procedure>")
|
|
continue
|
|
result = query(get_server_id(), "analyze_procedure", parts[1], parts[2], timeout=120)
|
|
print_result(result)
|
|
|
|
elif cmd == "list":
|
|
if len(parts) < 2:
|
|
print("用法: list <schema>")
|
|
continue
|
|
result = query(get_server_id(), "list_procedures", parts[1], timeout=30)
|
|
print_result(result)
|
|
|
|
elif cmd == "source":
|
|
if len(parts) < 3:
|
|
print("用法: source <schema> <procedure>")
|
|
continue
|
|
result = query(get_server_id(), "get_source", parts[1], parts[2], timeout=60)
|
|
print_result(result)
|
|
|
|
elif cmd == "deps":
|
|
if len(parts) < 3:
|
|
print("用法: deps <schema> <procedure>")
|
|
continue
|
|
result = query(get_server_id(), "get_dependencies", parts[1], parts[2], timeout=60)
|
|
print_result(result)
|
|
|
|
elif cmd == "tables":
|
|
if len(parts) < 3:
|
|
print("用法: tables <schema> <procedure>")
|
|
continue
|
|
result = query(get_server_id(), "get_tables", parts[1], parts[2], timeout=60)
|
|
print_result(result)
|
|
|
|
else:
|
|
print(f"未知命令: {cmd}，输入 help 查看帮助")
|
|
|
|
|
|
def main():
|
|
if len(sys.argv) == 1:
|
|
interactive_mode()
|
|
return
|
|
|
|
cmd = sys.argv[1].lower()
|
|
|
|
if cmd == "servers":
|
|
result = list_servers()
|
|
print(json.dumps(result, indent=2, ensure_ascii=False))
|
|
|
|
elif cmd == "health":
|
|
result = health_check()
|
|
print(json.dumps(result, indent=2, ensure_ascii=False))
|
|
|
|
elif cmd == "analyze" and len(sys.argv) >= 4:
|
|
result = query(get_server_id(), "analyze_procedure", sys.argv[2], sys.argv[3], timeout=120)
|
|
print_result(result)
|
|
|
|
elif cmd == "list" and len(sys.argv) >= 3:
|
|
result = query(get_server_id(), "list_procedures", sys.argv[2], timeout=30)
|
|
print_result(result)
|
|
|
|
elif cmd == "source" and len(sys.argv) >= 4:
|
|
result = query(get_server_id(), "get_source", sys.argv[2], sys.argv[3], timeout=60)
|
|
print_result(result)
|
|
|
|
elif cmd == "deps" and len(sys.argv) >= 4:
|
|
result = query(get_server_id(), "get_dependencies", sys.argv[2], sys.argv[3], timeout=60)
|
|
print_result(result)
|
|
|
|
elif cmd == "tables" and len(sys.argv) >= 4:
|
|
result = query(get_server_id(), "get_tables", sys.argv[2], sys.argv[3], timeout=60)
|
|
print_result(result)
|
|
|
|
else:
|
|
print(__doc__)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|