#!/usr/bin/env python3 """ Oracle Jump Query - AI Skill (Improved Version) 通过中转服务查询远程 Oracle 存储过程元数据 改进点： 1. 增加重试机制（可配置重试次数） 2. 增加详细日志输出（方便诊断问题） 3. 超时时间可配置 4. 增加 Agent 状态检查 5. 增加请求超时分级（快速命令 vs 慢速命令） """ import json import os import sys import time import logging import io import re import html import requests import base64 import hashlib import platform import uuid import secrets from typing import Optional, Dict, Any, List, Tuple import argparse from datetime import datetime, timezone # 修复 Windows 控制台 UTF-8 输出 if sys.platform == 'win32': sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace') sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8', errors='replace') # ============================================================ # 配置（修改为你的中转服务地址和认证信息） # ============================================================ TRANSIT_URL = "https://ts.henlo.net" AUTH_TOKEN = "" DEFAULT_SERVER_ID = "server-001" DEFAULT_TIMEOUT = 60 # 默认超时（秒） MAX_RETRIES = 3 # 最大重试次数 RETRY_DELAY = 2 # 重试延迟（秒） BOS_API_URL = "http://bos.henlo.net/servlet/doserverdata" # BOS 统一接口地址 MAX_AWR_DOWNLOAD_BYTES = 50 * 1024 * 1024 AWR_DATE_PATTERN = re.compile(r"^\d{8}$") # ============================================================ # 版本号 # ============================================================ SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) SKILL_DIR = os.path.dirname(SCRIPT_DIR) VERSION_FILE = os.path.join(SKILL_DIR, "VERSION") NO_PROXY_HOSTS = "bos.henlo.net,124.220.155.147,127.0.0.1,localhost" for _proxy_key in ("NO_PROXY", "no_proxy"): _existing_no_proxy = os.environ.get(_proxy_key, "") _items = [item.strip() for item in _existing_no_proxy.split(",") if item.strip()] for _host in NO_PROXY_HOSTS.split(","): if _host not in _items: _items.append(_host) os.environ[_proxy_key] = ",".join(_items) def get_version(): """读取 VERSION 文件中的版本号""" try: with open(VERSION_FILE, "r", encoding="utf-8") as f: return f.read().strip() except Exception: return "0.0.0" VERSION = get_version() # 表空间查询 SQL（需要 SELECT_CATALOG_ROLE 或 DBA 权限，建议 DBA 授权） TABLESPACE_SQL_DBA = r""" SELECT df.tablespace_name AS "表空间名称", ROUND(df.total_bytes / 1048576, 1) AS "总大小MB", ROUND((df.total_bytes - NVL(fs.free_bytes, 0)) / 1048576, 1) AS "已用MB", ROUND(NVL(fs.free_bytes, 0) / 1048576, 1) AS "剩余MB", ROUND((df.total_bytes - NVL(fs.free_bytes, 0)) / df.total_bytes * 100, 1) AS "使用率%", df.autoextensible AS "自动拓展", ROUND(df.max_bytes / 1048576, 1) AS "最大可拓展MB", ROUND((df.total_bytes - NVL(fs.free_bytes, 0)) / NULLIF(df.max_bytes, 0) * 100, 1) AS "最大使用率%" FROM ( SELECT tablespace_name, SUM(bytes) AS total_bytes, SUM(CASE WHEN autoextensible = 'YES' THEN maxbytes ELSE bytes END) AS max_bytes, MAX(CASE WHEN autoextensible = 'YES' THEN 'YES' ELSE 'NO' END) AS autoextensible FROM dba_data_files GROUP BY tablespace_name ) df LEFT JOIN ( SELECT tablespace_name, SUM(bytes) AS free_bytes FROM dba_free_space GROUP BY tablespace_name ) fs ON df.tablespace_name = fs.tablespace_name ORDER BY 6 DESC """ # 表空间查询回退 SQL（仅 USER_FREE_SPACE，始终可用） TABLESPACE_SQL_FREE = """ SELECT tablespace_name AS "表空间名称", ROUND(SUM(bytes) / 1048576, 2) AS "剩余MB" FROM user_free_space GROUP BY tablespace_name ORDER BY tablespace_name """ # DBA 授权 VIEW 名称：DBA 执行一次授权 SQL 后即可使用 DBA_TS_VIEW = "HENLO_TS_USAGE" DBA_TS_VIEW_FULL = f"sys.{DBA_TS_VIEW}" TABLESPACE_SQL_DBA_VIEW = f"""SELECT * FROM {DBA_TS_VIEW}""" TABLESPACE_SQL_DBA_VIEW_FULL = f"""SELECT * FROM {DBA_TS_VIEW_FULL}""" # 尝试使用 DBA 创建的专用函数（返回 SYS_REFCURSOR） TABLESPACE_SQL_DBA_FUNC = """SELECT sys.get_tablespace_usage FROM DUAL""" _TS_FUNC_COLUMNS = ["表空间名称", "总大小MB", "已用MB", "剩余MB", "使用率%"] _TS_FUNC_RE = re.compile(r'\[(\w+)\s+([\d.]+)\s+([\d.]+)\s+([\d.]+)\s+([\d.]+)\]') # 表空间使用率字段索引（完整 8 列表） _TS_COL_NAMES = 0 _TS_COL_TOTAL = 1 _TS_COL_USED = 2 _TS_COL_FREE = 3 _TS_COL_PCT = 4 _TS_COL_AUTO = 5 _TS_COL_MAX = 6 _TS_COL_MAXPCT = 7 def _check_ts_warnings(data_str: str) -> bool: """ 智能检查表空间告警： - 优先按表解析（ | ... | 格式的 markdown 行） - 自动扩展（YES）→ 看最大使用率%列 - 非自动扩展（NO）→ 看当前使用率%列 - 若解析失败则回退到字符串模糊匹配 - 回退路径忽略最大使用率%列，仅匹配当前使用率%列 - 函数查询路径（5 列）一律匹配第 5 列（使用率%） """ has_warning = False records = [] # (名称, 有效使用率) for line in data_str.split("\n"): line = line.strip() if not line.startswith("|") or line.startswith("|---"): continue cells = [c.strip() for c in line.split("|")] cells = [c for c in cells if c] # 去掉首尾空单元格 if len(cells) >= 5: name = cells[_TS_COL_NAMES] if len(cells) >= 8: # 完整 8 列表（VIEW / DBA 路径） if cells[_TS_COL_AUTO] == "YES": try: rate = float(cells[_TS_COL_MAXPCT]) records.append((name, rate)) if rate >= 95: has_warning = True except ValueError: pass else: try: rate = float(cells[_TS_COL_PCT]) records.append((name, rate)) if rate >= 95: has_warning = True except ValueError: pass elif len(cells) == 5: # 函数查询 5 列表 try: rate = float(cells[_TS_COL_PCT]) records.append((name, rate)) if rate >= 95: has_warning = True except ValueError: pass if has_warning: print("⚠️ 警告：以下表空间使用率过高，请尽快扩容！") # 对已触发告警的记录排序输出 warned = [(n, r) for n, r in records if r >= 95] warned.sort(key=lambda x: -x[1]) for n, r in warned: print(f" 🔴 {n}: {r}%") return has_warning # 给 DBA 的授权 SQL（仅在无权限时输出） DBA_GRANT_SQL = f'''-- ════════════════════════════════════════════════════════ -- 以下 SQL 请交给 DBA 在 Oracle 中执行一次 -- 作用：创建一个授权视图，让普通用户 bosnds3 也能查询表空间完整信息 -- 此视图使用定义者权限（默认），以 DBA 身份运行，不会暴露其他数据 -- ════════════════════════════════════════════════════════ CREATE OR REPLACE VIEW {DBA_TS_VIEW} AS SELECT df.tablespace_name AS "表空间名称", ROUND(df.total_bytes / 1048576, 1) AS "总大小MB", ROUND((df.total_bytes - NVL(fs.free_bytes, 0)) / 1048576, 1) AS "已用MB", ROUND(NVL(fs.free_bytes, 0) / 1048576, 1) AS "剩余MB", ROUND((df.total_bytes - NVL(fs.free_bytes, 0)) / df.total_bytes * 100, 1) AS "使用率%", df.autoextensible AS "自动拓展", ROUND(df.max_bytes / 1048576, 1) AS "最大可拓展MB", ROUND((df.total_bytes - NVL(fs.free_bytes, 0)) / NULLIF(df.max_bytes, 0) * 100, 1) AS "最大使用率%" FROM ( SELECT tablespace_name, SUM(bytes) AS total_bytes, SUM(CASE WHEN autoextensible = 'YES' THEN maxbytes ELSE bytes END) AS max_bytes, MAX(CASE WHEN autoextensible = 'YES' THEN 'YES' ELSE 'NO' END) AS autoextensible FROM dba_data_files GROUP BY tablespace_name ) df LEFT JOIN ( SELECT tablespace_name, SUM(bytes) AS free_bytes FROM dba_free_space GROUP BY tablespace_name ) fs ON df.tablespace_name = fs.tablespace_name ORDER BY 6 DESC; GRANT SELECT ON {DBA_TS_VIEW_FULL} TO bosnds3; ''' def _parse_func_result(data_str): """从 SYS_REFCURSOR 返回的 Go 结构体中提取表格数据""" matches = _TS_FUNC_RE.findall(data_str) if not matches: return None # 构建 Markdown 表格 header = "| " + " | ".join(_TS_FUNC_COLUMNS) + " |" sep = "|---" * len(_TS_FUNC_COLUMNS) + "|" rows = [] for m in matches: row = "| " + " | ".join(m) + " |" rows.append(row) return header + "\n" + sep + "\n" + "\n".join(rows) def query_tablespace(server_id): """ 查询表空间使用情况，按优先级尝试四种方式： 1. DBA_DATA_FILES 直接查询（需 SELECT_CATALOG_ROLE） 2. DBA 创建的专用函数 get_tablespace_usage（返回 SYS_REFCURSOR） 3. DBA 授权的专用 VIEW HENLO_TS_USAGE 4. USER_FREE_SPACE 简单查询（始终可用，仅剩余空间） """ schema = "BOSNDS3" # 第 1 步：尝试 DBA 查询 result = query(server_id, "execute_query", schema, "", timeout=DEFAULT_TIMEOUT, max_retries=1, sql=TABLESPACE_SQL_DBA) if result.get("success"): return result err = (result.get("error") or "").lower() is_perm_error = "ora-00942" in err or "insufficient privileges" in err if not is_perm_error: return result # 第 2 步：尝试 DBA 创建的专用函数（返回 SYS_REFCURSOR） logger.info("DBA 视图不可用，尝试 DBA 专用函数...") result_func = query(server_id, "execute_query", schema, "", timeout=DEFAULT_TIMEOUT, max_retries=1, sql=TABLESPACE_SQL_DBA_FUNC) if result_func.get("success"): raw_data = result_func.get("data", "") parsed = _parse_func_result(raw_data) if parsed: result_func["_via_func"] = True result_func["data"] = parsed return result_func else: # 解析失败但数据可用，保留原始数据 logger.info("函数返回了数据但格式无法解析，保留原始输出") result_func["_via_func"] = True result_func["_parse_failed"] = True return result_func # 第 3 步：尝试 DBA 授权 VIEW（无 schema 前缀） logger.info("DBA 函数不可用，尝试 DBA 授权 VIEW...") result2 = query(server_id, "execute_query", schema, "", timeout=DEFAULT_TIMEOUT, max_retries=1, sql=TABLESPACE_SQL_DBA_VIEW) if result2.get("success"): result2["_via_view"] = True return result2 # 第 3b 步：尝试 DBA 授权 VIEW（带 sys. 前缀，某些库可能创建在 SYS 下） logger.info("VIEW 无前缀不可用，尝试 sys. 前缀...") result2b = query(server_id, "execute_query", schema, "", timeout=DEFAULT_TIMEOUT, max_retries=1, sql=TABLESPACE_SQL_DBA_VIEW_FULL) if result2b.get("success"): result2b["_via_view"] = True return result2b # 第 4 步：全部失败，输出授权 SQL + 回退到 USER_FREE_SPACE logger.info("所有 DBA 方案不可用，回退到 USER_FREE_SPACE") _print_dba_grant_sql() result3 = query(server_id, "execute_query", schema, "", timeout=DEFAULT_TIMEOUT, max_retries=MAX_RETRIES, sql=TABLESPACE_SQL_FREE) if result3.get("success"): result3["_fallback"] = True return result3 def _print_dba_grant_sql(): """打印给 DBA 执行的授权 SQL 和说明""" print("\n" + "═" * 70) print("⚠️ 当前 Oracle 用户无 DBA 系统视图权限，无法获取完整表空间信息。") print("\n请将以下 SQL 交给 DBA 执行一次：") print("─" * 70) print(DBA_GRANT_SQL) print("─" * 70) print("📋 DBA 执行授权后，再次查询即可获得完整信息。") print("=" * 70) # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) def get_script_dir(): """获取脚本自身所在目录""" return os.path.dirname(os.path.abspath(__file__)) def get_config(): """Load local runtime config, creating it from the checked-in template when absent.""" cfg_path = os.path.join(get_script_dir(), "config.json") template_path = os.path.join(get_script_dir(), "config.template.json") for path in (cfg_path, "config.json"): try: with open(path, "r", encoding="utf-8-sig") as f: return json.load(f) except FileNotFoundError: continue except json.JSONDecodeError as exc: logger.warning("Invalid local config %s: %s", path, exc) break cfg = {"transit_url": TRANSIT_URL, "server_id": "", "access_token": "", "expires_at": "", "user_name": "", "client_code": "", "client_title": "", "client_list": []} try: with open(template_path, "r", encoding="utf-8-sig") as f: cfg.update(json.load(f)) except (FileNotFoundError, json.JSONDecodeError): pass save_config(cfg) return cfg def save_config(cfg: dict): """保存配置到脚本同目录的 config.json""" cfg_path = os.path.join(get_script_dir(), "config.json") with open(cfg_path, "w", encoding="utf-8") as f: json.dump(cfg, f, ensure_ascii=False, indent=2) logger.info(f"配置已保存到 {cfg_path}") def get_client_code(client: Dict[str, Any]) -> str: """Return normalized client code from server response.""" return str(client.get("code") or client.get("clientCode") or "").strip() def get_client_title(client: Dict[str, Any]) -> str: """Return normalized client title/name from server response.""" return str(client.get("title") or client.get("name") or "").strip() def normalize_client(client: Dict[str, Any]) -> Dict[str, Any]: """Normalize client response while preserving status fields.""" normalized = dict(client) normalized["code"] = get_client_code(client) normalized["title"] = get_client_title(client) return normalized def normalize_client_list(clients: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """Normalize client list and drop entries without code.""" result = [] seen = set() for client in clients or []: normalized = normalize_client(client) code = normalized.get("code", "") if not code or code in seen: continue seen.add(code) result.append(normalized) return result def merge_client_lists(base: List[Dict[str, Any]], updates: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """Merge client metadata by code, preserving title from either source.""" merged: Dict[str, Dict[str, Any]] = {} order: List[str] = [] for client in normalize_client_list(base) + normalize_client_list(updates): code = client.get("code", "") if not code: continue if code not in merged: merged[code] = {} order.append(code) merged[code].update(client) if not merged[code].get("title"): merged[code]["title"] = get_client_title(client) return [merged[code] for code in order] def sync_client_cache(cfg: Dict[str, Any], clients: List[Dict[str, Any]]) -> bool: """Write refreshed clients into config and refresh current client title.""" normalized = normalize_client_list(clients) cfg["client_list"] = normalized current_code = cfg.get("client_code") or cfg.get("server_id") if current_code: for client in normalized: if client.get("code") == current_code: cfg["client_code"] = client.get("code", current_code) cfg["client_title"] = client.get("title", cfg.get("client_title", "")) cfg["server_id"] = cfg["client_code"] break return True def refresh_client_cache(cfg: Optional[Dict[str, Any]] = None, quiet: bool = False) -> List[Dict[str, Any]]: """Refresh the authorized client list from transit server and persist it locally.""" cfg = cfg or get_config() or {} transit_url = cfg.get("transit_url", TRANSIT_URL) if not ensure_logged_in(cfg): return [] clients: List[Dict[str, Any]] = normalize_client_list(cfg.get("client_list", [])) try: resp = requests.get(f"{transit_url}/api/me", headers=make_headers(), timeout=10) if resp.status_code == 401: if not quiet: print("❌ 中转机登录态已失效，请重新登录") return clients resp.raise_for_status() me = resp.json() if me.get("expires_at"): cfg["expires_at"] = me.get("expires_at") if me.get("user", {}).get("name"): cfg["user_name"] = me.get("user", {}).get("name") except requests.exceptions.RequestException as e: if not quiet: print(f"⚠️ 从 /api/me 刷新 client 失败: {e}") try: resp = requests.get(f"{transit_url}/api/clients", headers=make_headers(), timeout=10) if resp.status_code == 401: if not quiet: print("❌ 中转机登录态已失效，请重新登录") return clients resp.raise_for_status() data = resp.json() # /api/clients is authoritative for the current authorization snapshot. # Replace instead of merging so revoked clients disappear from local cache. clients = normalize_client_list(data.get("clients", [])) except requests.exceptions.RequestException as e: if not quiet: print(f"⚠️ 从 /api/clients 刷新 client 失败: {e}") return clients if sync_client_cache(cfg, clients): save_config(cfg) return clients def find_client_matches(clients: List[Dict[str, Any]], selector: str) -> Tuple[List[Dict[str, Any]], str]: """Find client by code/title/name. Exact matches win over fuzzy matches.""" selector = (selector or "").strip() if not selector: return [], "empty" normalized = normalize_client_list(clients) needle = selector.lower() exact_code = [c for c in normalized if c.get("code", "").lower() == needle] if exact_code: return exact_code, "code" exact_title = [c for c in normalized if c.get("title", "").lower() == needle] if exact_title: return exact_title, "title" fuzzy = [] seen = set() for client in normalized: code = client.get("code", "") title = client.get("title", "") if needle in code.lower() or needle in title.lower(): if code not in seen: seen.add(code) fuzzy.append(client) return fuzzy, "fuzzy" def print_client_candidates(clients: List[Dict[str, Any]]): """Print selectable client candidates.""" for client in normalize_client_list(clients): code = client.get("code", "") title = client.get("title", "") online = client.get("online") authorized = client.get("authorized") status = "" if online is not None or authorized is not None: online_text = "在线" if online else "离线" auth_text = "已授权" if authorized else "未授权" status = f" ({online_text}, {auth_text})" print(f" - {code} ({title}){status}") def ts_api_call(method: str, params: dict, timeout: int = 15) -> Dict[str, Any]: """ 调用 BOS 统一接口 Args: method: 接口方法名（如 ts_login, ts_client_check） params: 请求参数 timeout: 超时时间（秒） Returns: dict: {"code": 0/1, "success": True/False, "message": ..., "data": ...} """ payload = {"method": method, "params": params} try: logger.info(f"调用 BOS 接口: {method}") resp = requests.post(BOS_API_URL, json=payload, timeout=timeout) resp.raise_for_status() result = resp.json() logger.info(f"BOS 接口响应: code={result.get('code')}, success={result.get('success')}") return result except requests.exceptions.ConnectionError as e: return {"code": 1, "success": False, "message": f"无法连接 BOS 接口: {e}", "data": None} except requests.exceptions.Timeout: return {"code": 1, "success": False, "message": f"BOS 接口超时（{timeout}秒）", "data": None} except Exception as e: return {"code": 1, "success": False, "message": f"BOS 接口错误: {e}", "data": None} def cmd_clients(): """获取当前用户可访问 client 列表及在线状态（通过中转服务的 /api/clients 接口）""" config = get_config() or {} if not ensure_logged_in(config): return clients = refresh_client_cache(config) if not clients: print("没有可用的 client") return print(f"共 {len(clients)} 个 client：") print("-" * 60) for cl in clients: code = cl.get("code", "?") title = cl.get("title", "?") online = cl.get("online", False) authorized = cl.get("authorized", False) status_icon = "[OK]" if online and authorized else ("[?]" if online else "[X]") auth_text = "已授权" if authorized else "未授权" online_text = "在线" if online else "离线" print(f" {status_icon} {code} - {title} ({online_text}, {auth_text})") print("-" * 60) def get_server_status(client_code: str = "") -> Dict[str, Any]: """Get server health status from transit server.""" config = get_config() or {} transit_url = config.get("transit_url", TRANSIT_URL) if not ensure_logged_in(config): return {"success": False, "error": "login required"} if not client_code: client_code = config.get("client_code", "") if not client_code: return {"success": False, "error": "clientCode is required"} try: resp = requests.get( f"{transit_url}/api/server_status", params={"client_code": client_code}, headers=make_headers(), timeout=DEFAULT_TIMEOUT + 5, ) if resp.status_code == 401: return {"success": False, "error": "login required or token expired"} if resp.status_code == 403: try: return {"success": False, "error": resp.json().get("error", resp.text)} except Exception: return {"success": False, "error": resp.text} resp.raise_for_status() return resp.json() except requests.exceptions.RequestException as e: return {"success": False, "error": str(e)} def cmd_checkup(client_code: str = ""): """Print server health report as a Markdown table.""" result = get_server_status(client_code) if not result.get("success"): print(f"Error: {result.get('error', 'unknown error')}") return data = result.get("data", {}) or {} fields = [ "clientCode", "info_cpu_rate", "info_memory_rate", "info_memory", "bak_path", "bak_lastdate", "computer", ] print("| " + " | ".join(fields) + " |") print("| " + " | ".join(["---"] * len(fields)) + " |") print("| " + " | ".join(str(data.get(f, "") or "") for f in fields) + " |") if data.get("log_saved") is False: print(f"\nNote: server log was not saved to BOS: {data.get('log_error', '')}") def _load_sys_function_bundle(schema: str = "") -> str: schema = (schema or "").strip() if schema != "" and not re.fullmatch(r"[A-Za-z][A-Za-z0-9_$#]*", schema): raise ValueError("schema must be a valid Oracle identifier") root = os.path.join(SKILL_DIR, "references", "sys-functions") sections = [] for name in sorted(f for f in os.listdir(root) if f.lower().endswith(".sql")): with open(os.path.join(root, name), "r", encoding="utf-8-sig") as f: sections.append(f"-- ===== {name} =====\n" + f.read().replace("", schema).rstrip()) return "\n\n".join(sections) + "\n" def _load_sys_function_sql(filename: str, schema: str) -> str: schema = (schema or "").strip().upper() if not re.fullmatch(r"[A-Za-z][A-Za-z0-9_$#]*", schema): raise ValueError("schema must be a valid Oracle identifier") if filename != os.path.basename(filename) or not filename.lower().endswith(".sql"): raise ValueError("invalid SYS function filename") path = os.path.join(SKILL_DIR, "references", "sys-functions", filename) with open(path, "r", encoding="utf-8-sig") as f: return f.read().replace("", schema) def _resolve_agent_schema(client_code: str) -> str: version_result = query(client_code, "version", timeout=15) version_data = version_result.get("data") or {} if isinstance(version_data, str): try: version_data = json.loads(version_data) except json.JSONDecodeError: version_data = {} schema = (version_data.get("schema") or "").strip().upper() if not re.fullmatch(r"[A-Za-z][A-Za-z0-9_$#]*", schema): return "" return schema def check_inspection_sys_functions(client_code: str) -> Dict[str, Any]: result = query(client_code, "sys_function_check", timeout=30) if not result.get("success"): return {"success": False, "error": result.get("error", "SYS function check failed")} data = result.get("data") or {} if isinstance(data, str): try: data = json.loads(data) except json.JSONDecodeError: data = {} if not data.get("success"): schema = (data.get("schema") or "").strip() if not schema: schema = _resolve_agent_schema(client_code) return { "success": False, "error": "SYS.HENLO_ORA_MONITOR is missing, incompatible, or not executable", "agent_schema": schema, "sys_function_check": data, } return {"success": True, "data": data} def _inspection_grant_result(result: Dict[str, Any]) -> Dict[str, Any]: schema = (result.get("agent_schema") or "").strip() if not schema or not re.fullmatch(r"[A-Za-z][A-Za-z0-9_$#]*", schema): result["grant_sql_required"] = True result["grant_sql"] = "" result["error"] = "SYS function preflight failed and the Agent Oracle schema could not be determined" result["next_step"] = "请检查 Agent oracle.schema 配置；Skill 不会输出包含 占位符的授权 SQL" return result result["grant_sql_required"] = True result["agent_schema"] = schema.upper() result["grant_sql"] = _load_sys_function_bundle(schema.upper()) result["next_step"] = f"请由客户 DBA 使用 SYS 执行 grant_sql（授权账号 {schema.upper()}），完成后重新生成巡检报告" return result def inspection_report(client_code: str = "", timeout: int = 90, report_type: str = "manual") -> Dict[str, Any]: """Generate and archive a server inspection report through transit server.""" config = get_config() or {} transit_url = config.get("transit_url", TRANSIT_URL) if not ensure_logged_in(config): return {"success": False, "error": "login required"} if not client_code: client_code = config.get("client_code", "") if not client_code: return {"success": False, "error": "clientCode is required"} sys_check = check_inspection_sys_functions(client_code) if not sys_check.get("success"): return _inspection_grant_result(sys_check) payload = { "client_code": client_code, "report_type": report_type, "timeout": timeout, } try: resp = requests.post( f"{transit_url}/api/inspection_report", json=payload, headers=make_headers(), timeout=max(30, timeout + 15), ) if resp.status_code in (400, 401, 403, 404, 502, 503, 504): try: return {"success": False, "error": resp.json().get("error", resp.text), "status_code": resp.status_code} except Exception: return {"success": False, "error": resp.text, "status_code": resp.status_code} resp.raise_for_status() return resp.json() except requests.exceptions.RequestException as e: return {"success": False, "error": str(e)} def inspection_report_latest(client_code: str = "", refresh: bool = False) -> Dict[str, Any]: """Read the latest archived inspection report, optionally refreshing it first.""" config = get_config() or {} transit_url = config.get("transit_url", TRANSIT_URL) if not ensure_logged_in(config): return {"success": False, "error": "login required"} if not client_code: client_code = config.get("client_code", "") if not client_code: return {"success": False, "error": "clientCode is required"} if refresh: sys_check = check_inspection_sys_functions(client_code) if not sys_check.get("success"): return _inspection_grant_result(sys_check) params = {"client_code": client_code} if refresh: params["refresh"] = "true" try: resp = requests.get( f"{transit_url}/api/inspection_report/latest", params=params, headers=make_headers(), timeout=120 if refresh else DEFAULT_TIMEOUT + 5, ) if resp.status_code in (400, 401, 403, 404, 502, 503, 504): try: return {"success": False, "error": resp.json().get("error", resp.text), "status_code": resp.status_code} except Exception: return {"success": False, "error": resp.text, "status_code": resp.status_code} resp.raise_for_status() return resp.json() except requests.exceptions.RequestException as e: return {"success": False, "error": str(e)} def inspection_report_get(report_id: str) -> Dict[str, Any]: """Read an archived inspection report by ID.""" config = get_config() or {} transit_url = config.get("transit_url", TRANSIT_URL) if not ensure_logged_in(config): return {"success": False, "error": "login required"} report_id = str(report_id or "").strip() if not report_id: return {"success": False, "error": "report id is required"} try: resp = requests.get( f"{transit_url}/api/inspection_report/{report_id}", headers=make_headers(), timeout=DEFAULT_TIMEOUT + 5, ) if resp.status_code in (400, 401, 403, 404, 502, 503, 504): try: return {"success": False, "error": resp.json().get("error", resp.text), "status_code": resp.status_code} except Exception: return {"success": False, "error": resp.text, "status_code": resp.status_code} resp.raise_for_status() return resp.json() except requests.exceptions.RequestException as e: return {"success": False, "error": str(e)} def _awr_client_code(client_code: str = "") -> str: cfg = get_config() or {} return (client_code or cfg.get("client_code") or cfg.get("server_id") or "").strip() def awr_status(client_code: str = "") -> Dict[str, Any]: """Read AWR readiness without triggering report generation.""" cfg = get_config() or {} if not ensure_logged_in(cfg): return {"success": False, "error": "login required"} client_code = _awr_client_code(client_code) if not client_code: return {"success": False, "error": "clientCode is required"} try: resp = requests.get( f"{str(cfg.get('transit_url') or TRANSIT_URL).rstrip('/')}/api/awr/status", params={"client_code": client_code}, headers=make_headers(), timeout=DEFAULT_TIMEOUT, ) if not resp.ok: try: return {"success": False, "error": resp.json().get("error", resp.text), "status_code": resp.status_code} except Exception: return {"success": False, "error": resp.text, "status_code": resp.status_code} result = resp.json() data = result.get("data") or {} if isinstance(data, dict) and data.get("status") == "permission_required": result["dba_sql_required"] = True schema = _resolve_agent_schema(client_code) if schema: result["agent_schema"] = schema result["dba_sql"] = _load_sys_function_sql("henlo_awr_export.sql", schema) result["next_step"] = f"请由客户 DBA 在确认 Diagnostics Pack 授权后执行 dba_sql（授权账号 {schema}），Skill 不会自动执行" else: result["agent_schema"] = "" result["dba_sql"] = "" result["next_step"] = "请检查 Agent oracle.schema 配置；Skill 不会输出包含 占位符的 SQL" return result except requests.exceptions.RequestException as e: return {"success": False, "error": str(e)} def awr_list(client_code: str = "") -> Dict[str, Any]: """List Agent-generated AWR reports without downloading their HTML.""" cfg = get_config() or {} if not ensure_logged_in(cfg): return {"success": False, "error": "login required"} client_code = _awr_client_code(client_code) if not client_code: return {"success": False, "error": "clientCode is required"} try: resp = requests.get( f"{str(cfg.get('transit_url') or TRANSIT_URL).rstrip('/')}/api/awr/list", params={"client_code": client_code}, headers=make_headers(), timeout=DEFAULT_TIMEOUT, ) if not resp.ok: try: return {"success": False, "error": resp.json().get("error", resp.text), "status_code": resp.status_code} except Exception: return {"success": False, "error": resp.text, "status_code": resp.status_code} return resp.json() except requests.exceptions.RequestException as e: return {"success": False, "error": str(e)} def _safe_awr_client_part(client_code: str) -> str: value = re.sub(r"[^A-Za-z0-9_-]+", "_", client_code).strip("_") return value or "CLIENT" def download_awr_report(client_code: str, date: str, output_path: str = "") -> Dict[str, Any]: """Download the exact AWR HTML supplied by Agent through transit-server.""" cfg = get_config() or {} if not ensure_logged_in(cfg): return {"success": False, "error": "login required"} client_code = _awr_client_code(client_code) date = (date or "").strip() if not client_code: return {"success": False, "error": "clientCode is required"} if not AWR_DATE_PATTERN.fullmatch(date): return {"success": False, "error": "date must use yyyyMMdd format"} if not output_path: out_dir = os.path.join(SKILL_DIR, "outputs") os.makedirs(out_dir, exist_ok=True) output_path = os.path.join(out_dir, f"AWR-{_safe_awr_client_part(client_code)}-{date}.html") output_path = os.path.abspath(output_path) parent = os.path.dirname(output_path) if parent: os.makedirs(parent, exist_ok=True) temp_path = output_path + ".part" try: if os.path.exists(temp_path): os.remove(temp_path) except OSError as e: return {"success": False, "error": str(e)} total = 0 try: with requests.get( f"{str(cfg.get('transit_url') or TRANSIT_URL).rstrip('/')}/api/awr/download", params={"client_code": client_code, "date": date}, headers=make_headers(), timeout=120, stream=True, ) as resp: if not resp.ok: try: error = resp.json().get("error", resp.text) except Exception: error = resp.text return {"success": False, "error": error, "status_code": resp.status_code} content_length = resp.headers.get("Content-Length", "") if content_length.isdigit() and int(content_length) > MAX_AWR_DOWNLOAD_BYTES: return {"success": False, "error": f"AWR report exceeds {MAX_AWR_DOWNLOAD_BYTES} bytes"} with open(temp_path, "wb") as f: for chunk in resp.iter_content(chunk_size=64 * 1024): if not chunk: continue total += len(chunk) if total > MAX_AWR_DOWNLOAD_BYTES: raise ValueError(f"AWR report exceeds {MAX_AWR_DOWNLOAD_BYTES} bytes") f.write(chunk) os.replace(temp_path, output_path) return {"success": True, "client_code": client_code, "date": date, "path": output_path, "size": total} except (requests.exceptions.RequestException, OSError, ValueError) as e: try: if os.path.exists(temp_path): os.remove(temp_path) except OSError: pass return {"success": False, "error": str(e)} def cmd_awr_status(client_code: str = "", as_json: bool = False): result = awr_status(client_code) if as_json or not result.get("success"): print(json.dumps(result, ensure_ascii=False, indent=2)) return data = result.get("data") or {} print(f"AWR status: {data.get('status', '-')}") print(f"Client: {result.get('client_code') or _awr_client_code(client_code)}") print(f"Available: {data.get('available', False)}") print(f"Enabled: {data.get('enabled', False)}") print(f"Diagnostics Pack confirmed: {data.get('license_confirmed', False)}") if data.get("latest_date"): print(f"Latest date: {data.get('latest_date')}") if data.get("reason"): print(f"Reason: {data.get('reason')}") if result.get("dba_sql_required"): if result.get("dba_sql"): print(f"Agent schema: {result.get('agent_schema')}") print("DBA SQL (manual execution only; Diagnostics Pack authorization must be confirmed first):") print(result.get("dba_sql")) print(f"Next step: {result.get('next_step', '')}") def cmd_awr_list(client_code: str = "", as_json: bool = False): result = awr_list(client_code) if as_json or not result.get("success"): print(json.dumps(result, ensure_ascii=False, indent=2)) return rows = result.get("data") or [] if not rows: print("No AWR reports are available.") return print("DATE STATUS BEGIN_SNAP END_SNAP GENERATED_AT REASON") for row in rows: print(f"{row.get('date', '-'):<9} {row.get('status', '-'):<8} {str(row.get('begin_snap', '-')):<10} {str(row.get('end_snap', '-')):<8} {row.get('generated_at', '-')} {row.get('reason', '')}") def cmd_awr_download(client_code: str, date: str, output_path: str = "", as_json: bool = False): result = download_awr_report(client_code, date, output_path) if as_json or not result.get("success"): print(json.dumps(result, ensure_ascii=False, indent=2)) return print(f"AWR report: {result['path']}") print(f"Size: {result['size']} bytes") def _inspection_report_payload(result: Dict[str, Any]) -> Dict[str, Any]: data = result.get("data") or {} if isinstance(data, dict) and data.get("report_json"): report_json = data.get("report_json") if isinstance(report_json, str): try: return json.loads(report_json) except Exception: return {} if isinstance(report_json, dict): return report_json if isinstance(data, dict) and ("memory" in data or "disks" in data or "alerts" in data): return data return {} def _fmt_report_number(value): if value is None or value == "": return "" if isinstance(value, float) and value.is_integer(): return str(int(value)) return str(value) def _html_escape(value) -> str: if value is None: return "" return html.escape(str(value), quote=True) def _pct(value) -> str: if value is None or value == "": return "" try: return f"{float(value):.2f}%" except Exception: return _html_escape(value) def _gb(value) -> str: if value is None or value == "": return "" try: return f"{float(value):.2f}G" except Exception: return _html_escape(value) def _tag(level: str, text: str = "") -> str: level = str(level or "").lower() if level in ("critical", "bad", "error", "failed"): cls, label = "bad", text or "重大异常" elif level in ("warning", "warn", "partial"): cls, label = "warn", text or "普通异常" elif level in ("pending", "unknown", "not_checked"): cls, label = "pending", text or "待接入" else: cls, label = "ok", text or "正常" return f'{_html_escape(label)}' def _status_level_from_percent(value, warn=80, bad=90): try: n = float(value) except Exception: return "pending" if n >= bad: return "critical" if n >= warn: return "warning" return "ok" def _cache_hit_level(value): try: n = float(value) except Exception: return "pending" if n < 80: return "critical" if n < 90: return "warning" return "ok" def _render_process_summary(processes): if not processes: return "未返回关键进程状态", "pending" rows = [] missing = False for process in processes: if isinstance(process, str): rows.append(_html_escape(process)) continue name = process.get("name") or "unknown" running = bool(process.get("running")) count = process.get("count", 0) missing = missing or not running label = f"{name}：{'运行中' if running else '未运行'}" if running: label += f"（{count} 个）" rows.append(_html_escape(label)) return "
".join(rows), "warning" if missing else "ok" def _find_ops(payload: Dict[str, Any], item_name: str) -> Dict[str, Any]: for item in payload.get("oracle_ops") or []: if isinstance(item, dict) and item.get("item") == item_name: return item return {} def _ops_count(item: Dict[str, Any]) -> int: rows = item.get("rows") if isinstance(item, dict) else [] return len(rows or []) def _render_ops_rows(item: Dict[str, Any], empty_text: str, max_rows: int = 6) -> str: rows = item.get("rows") if isinstance(item, dict) else [] if not rows: return f'{_html_escape(empty_text)}' columns = item.get("columns") or [] html_rows = [] for row in rows[:max_rows]: data = {str(columns[i]).upper(): row[i] for i in range(min(len(columns), len(row)))} session = f"{data.get('USERNAME', '')} / {data.get('SID', '')},{data.get('SERIAL_NO', '')}".strip(" /,") detail = data.get("SQL_TEXT", "") source = " / ".join(value for value in (data.get("MACHINE", ""), data.get("PROGRAM", "")) if value) html_rows.append( "" f"{_html_escape(data.get('ELAPSED_SEC', ''))} 秒" f"{_html_escape(session)}" f"{_html_escape(data.get('SQL_ID', ''))}" f"{_html_escape(data.get('EVENT', ''))}" f"{_tag('warning' if item.get('status') == 'warning' else 'ok')}" f"{_html_escape(source)}
{_html_escape(detail)}
" "" ) return "\n".join(html_rows) def _render_tablespace_cards(payload: Dict[str, Any]) -> str: cards = [] tablespace_item = _find_ops(payload, "tablespace") datafile_item = _find_ops(payload, "datafiles") datafile_columns = [str(value).upper() for value in (datafile_item.get("columns") or [])] datafiles = {} for row in datafile_item.get("rows") or []: data = {datafile_columns[i]: row[i] for i in range(min(len(datafile_columns), len(row)))} name = data.get("TABLESPACE_NAME", "") if name and (name not in datafiles or float(data.get("MAX_GB") or 0) > float(datafiles[name].get("MAX_GB") or 0)): datafiles[name] = data columns = [str(value).upper() for value in (tablespace_item.get("columns") or [])] for row in tablespace_item.get("rows") or []: data = {columns[i]: row[i] for i in range(min(len(columns), len(row)))} name = data.get("TABLESPACE_NAME") or (row[0] if row else "") datafile = datafiles.get(name, {}) used = data.get("USED_PCT", "") max_gb = datafile.get("MAX_GB", "") current_gb = data.get("TOTAL_GB", "") used_gb = data.get("USED_GB", "") free_gb = data.get("FREE_GB", "") max_used = datafile.get("FILE_USED_PCT", "") try: extend_free = round(float(max_gb) - float(used_gb), 2) if max_gb != "" else "" except Exception: extend_free = "" level_value = max_used if max_used != "" else used level = _status_level_from_percent(level_value, 85, 95) metrics = [("当前大小", f"{current_gb}G"), ("已用", f"{used_gb}G"), ("当前剩余", f"{free_gb}G"), ("自动扩展上限", f"{max_gb}G" if max_gb != "" else "未返回"), ("可扩展剩余", f"{extend_free}G" if extend_free != "" else "未返回"), ("最大空间使用率", f"{level_value}%" if level_value != "" else "未返回")] metric_html = "".join(f'
{_html_escape(k)}
{_html_escape(v)}
' for k, v in metrics) advice = "当前未达到重大阈值，建议持续观察。" if level == "ok" else "空间使用率较高，建议评估扩容、清理历史数据或确认自动扩展上限。" cards.append('
' f'
{_html_escape(name)}
{_tag(level)}
' f'
{metric_html}
' f'
{_html_escape(advice)}
') if not cards: return '
未返回表空间明细。
' return "\n".join(cards) def _render_alert_rows(payload: Dict[str, Any]) -> str: alerts = payload.get("alerts") or [] failed = payload.get("failed_items") or [] rows = [] for alert in alerts: level = alert.get("level", "warning") if isinstance(alert, dict) else "warning" title = alert.get("title", "异常") if isinstance(alert, dict) else str(alert) desc = alert.get("desc", "") if isinstance(alert, dict) else "" rows.append(f"{_tag(level)}{_html_escape(title)}{_html_escape(desc)}按运维规范排查并持续观察。") for item in failed: name = item.get("item", "巡检项") if isinstance(item, dict) else "巡检项" err = item.get("error", "") if isinstance(item, dict) else str(item) rows.append(f"{_tag('warning')}{_html_escape(name)}{_html_escape(err)}检查授权、网络或 Agent 日志后重试。") if not rows: rows.append(f"{_tag('ok')}未发现异常本次巡检未产生告警。保持观察。") return "\n".join(rows) def render_inspection_report_html(result: Dict[str, Any]) -> str: payload = _inspection_report_payload(result) record = result.get("data") if isinstance(result.get("data"), dict) else {} client = result.get("client_code") or payload.get("client_code") or record.get("client_code") or get_server_id() report_no = result.get("report_no") or record.get("report_no") or f"CHECK-{client}" collected_at = payload.get("collected_at") or record.get("collected_at") or "" server = payload.get("server") or {} memory = payload.get("memory") or {} disks = payload.get("disks") or [] oracle = payload.get("oracle") or {} connections = server.get("connections") or {} process_summary, process_level = _render_process_summary(server.get("processes")) cache_hit = oracle.get("cache_hit") or {} buffer_cache = cache_hit.get("buffer_cache_percent") library_cache = cache_hit.get("library_cache_percent") alerts = payload.get("alerts") or [] overall = "warning" if alerts or payload.get("failed_items") else "ok" cpu_level = _status_level_from_percent(server.get("cpu_percent"), 80, 90) mem_level = _status_level_from_percent(memory.get("used_percent"), 80, 90) oracle_level = "ok" if oracle.get("online") else "critical" blocking = _find_ops(payload, "blocking_locks") long_tx = _find_ops(payload, "long_transactions") slow = _find_ops(payload, "active_slow_sql") ip_addresses = ", ".join(server.get("ip_addresses") or []) or "待采集" os_text = " ".join(str(value) for value in (server.get("os"), server.get("os_version")) if value) or "待采集" connection_text = "采集失败" connection_level = "pending" if connections.get("available"): connection_text = "总计 {total}；已建立 {established}；监听 {listening}；TIME_WAIT {time_wait}；CLOSE_WAIT {close_wait}".format( total=connections.get("total", 0), established=connections.get("established", 0), listening=connections.get("listening", 0), time_wait=connections.get("time_wait", 0), close_wait=connections.get("close_wait", 0)) connection_level = "ok" disk_rows = [] disk_burst = "未爆满" for disk in disks: level = disk.get("status") or _status_level_from_percent(disk.get("used_percent"), 85, 95) if level in ("warning", "critical", "bad"): disk_burst = "存在风险" disk_rows.append( "" f"{_html_escape(disk.get('drive'))}" f"{_pct(disk.get('used_percent'))}" f"{_gb(disk.get('used_gb'))} / {_gb(disk.get('free_gb'))}" f"读 {_html_escape(disk.get('read_mb_s', 0))} MB/s
写 {_html_escape(disk.get('write_mb_s', 0))} MB/s" f"{_tag(level)}" f"24h 使用率峰值 {_pct(disk.get('used_peak_24h_percent'))}。" "" ) if not disk_rows: disk_rows.append(f"未返回磁盘明细。") disk_rows.append(f"磁盘爆满检测{_html_escape(disk_burst)}普通：≥ 85%
重大：≥ 95%-{_tag('warning' if disk_burst != '未爆满' else 'ok')}按各分区当前使用率和 24h 峰值判断。") conclusion = "本次巡检未发现明显异常。" if alerts: conclusion = "本次巡检发现：" + "；".join(_html_escape(a.get("title", "")) for a in alerts if isinstance(a, dict)) + "。建议按异常明细处理。" return f""" 服务器巡检报告

服务器巡检报告

本报告由 Agent 自动采集服务器与 Oracle 数据库巡检结果后生成，用于运维归档、工单报备和客户沟通。
报告编号：{_html_escape(report_no)}
巡检时间：{_html_escape(collected_at)}
巡检方式：手动巡检
总体状态：{_tag(overall)}

一、基础信息

客户名称{_html_escape(client)}服务器编号{_html_escape(client)}
服务器 IP{_html_escape(ip_addresses)}计算机名{_html_escape(server.get('computer') or '待采集')}
操作系统{_html_escape(os_text)}数据库状态{'在线' if oracle.get('online') else '异常'}
启动时间{_html_escape(server.get('boot_time') or '待采集')}运行时长{_html_escape(server.get('uptime') or '待采集')}

二、服务器资源巡检

2.1 基础运行状态
巡检项结果值预警规则状态说明
服务器在线状态{'在线' if server.get('online') else '离线'}离线即重大异常{_tag('ok' if server.get('online') else 'critical')}Agent WebSocket 在线，最近心跳正常。
CPU 使用率{_pct(server.get('cpu_percent'))}普通：≥ 80%
重大：≥ 90%		{_tag(cpu_level)}当前 CPU 使用率按 Agent 实时采集值判断。
运行时长{_html_escape(server.get('uptime') or '待采集')}采集失败时标记待接入{_tag('ok' if server.get('uptime') else 'pending')}启动时间：{_html_escape(server.get('boot_time') or '待采集')}。
关键进程{process_summary}任一配置进程未运行即异常{_tag(process_level)}检查项由 Agent 配置 inspection_processes 定义。
TCP 连接数{_html_escape(connection_text)}CLOSE_WAIT 持续增长需关注{_tag(connection_level)}{_html_escape(connections.get('error') or '来自 netstat -ano -p tcp 实时统计。')}
2.2 内存状态
巡检项结果值预警规则状态说明
实时内存使用率{_pct(memory.get('used_percent'))}普通：≥ 80%
重大：≥ 90%		{_tag(mem_level)}当前内存使用率按 Agent 实时采集值判断。
已用 / 剩余内存{_gb(memory.get('used_gb'))} / {_gb(memory.get('free_gb'))}剩余低于 10% 为重大异常{_tag(mem_level)}总内存 {_gb(memory.get('total_gb'))}。
内存峰值1h {_pct(memory.get('peak_1h_percent'))}
24h {_pct(memory.get('peak_24h_percent'))}		普通：≥ 85%
重大：≥ 95%		{_tag(_status_level_from_percent(memory.get('peak_24h_percent'),85,95))}来自 Agent 本地滚动指标缓存。
内存溢出检测{len(memory.get('oom_events') or [])} 条事件发现 OOM / 相关错误即异常{_tag('warning' if memory.get('oom_events') else 'ok')}按采集周期内内存异常事件汇总。
2.3 磁盘状态
{''.join(disk_rows)}
分区使用率已用 / 剩余空间读写速率状态说明

三、数据库巡检结果

3.1 基础状态
巡检项结果值预警阈值状态说明
数据库服务在线状态{'在线' if oracle.get('online') else '异常'}不可连接即重大异常{_tag(oracle_level)}{_html_escape(oracle.get('error') or 'Agent 可正常连接 Oracle，基础查询成功。')}
数据库会话数{_html_escape(oracle.get('active_sessions')) if oracle.get('active_sessions') is not None else '采集失败'} 个 ACTIVE
{_html_escape(oracle.get('total_sessions')) if oracle.get('total_sessions') is not None else '-'} 个用户会话		ACTIVE 普通：≥ 50
重大：≥ 100		{_tag(_status_level_from_percent(oracle.get('active_sessions'),50,100))}来源：{_html_escape(oracle.get('session_count_source') or '未采集')}。
Buffer Cache 命中率{_pct(buffer_cache) if buffer_cache is not None else '采集失败'}普通：< 90%
重大：< 80%		{_tag(_cache_hit_level(buffer_cache))}来自 v$sysstat。
Library Cache 命中率{_pct(library_cache) if library_cache is not None else '采集失败'}普通：< 90%
重大：< 80%		{_tag(_cache_hit_level(library_cache))}来自 v$librarycache。
事务运行状态长事务 {_ops_count(long_tx)} 条普通：≥ 1 条
重大：≥ 3 条		{_tag('warning' if _ops_count(long_tx) else 'ok')}按 Oracle 运维监控函数返回结果统计。
阻塞会话{_ops_count(blocking)} 条普通：≥ 1 条
重大：≥ 3 条		{_tag('warning' if _ops_count(blocking) else 'ok')}未返回记录时视为未发现阻塞。
3.2 慢 SQL 检测明细
慢 SQL 不只展示数量，还应保留 SQL_ID、会话、耗时、等待事件和 SQL 文本，便于后续定位来源对象或业务过程。
{_render_ops_rows(slow, '未发现当前慢 SQL。')}
巡检项用户/会话SQL_ID等待事件状态SQL 文本 / 溯源
3.3 表空间检测明细
表空间报告应列出当前空间和自动扩展上限。若数据文件允许自动扩展，异常判断优先参考“最大空间使用率”和“剩余可扩展空间”，避免只按当前大小误报。
{_render_tablespace_cards(payload)}
3.4 存储过程 / 自定义 SQL 巡检明细
如果巡检项来自自定义 SQL 或存储过程，需要展示执行对象、SQL/过程内容、返回字段和异常判定依据。
巡检名称执行对象返回结果状态SQL / 存储过程内容
Oracle 运维监控函数SYS.HENLO_ORA_MONITOR{'可执行' if payload.get('oracle_ops') else '未返回'}{_tag('ok' if payload.get('oracle_ops') else 'warning')}Agent 优先调用 SYS 下的巡检函数，失败时回退到当前 schema。
SELECT SYS.HENLO_ORA_MONITOR(:item, :minutes, :top_n) FROM DUAL

四、异常明细

{_render_alert_rows(payload)}
异常等级异常类型异常说明处理建议

五、巡检结论

{'普通异常' if overall == 'warning' else '正常'}
{conclusion}
生成来源：Oracle Jump Query Agent / 固定格式 HTML 巡检报告 Demo
""" def write_inspection_report_html(result: Dict[str, Any], path: str = "") -> str: payload = _inspection_report_payload(result) client = result.get("client_code") or payload.get("client_code") or get_server_id() report_no = result.get("report_no") or f"inspection-report-{client}" if not path: out_dir = os.path.join(SKILL_DIR, "outputs") os.makedirs(out_dir, exist_ok=True) safe_name = re.sub(r"[^A-Za-z0-9_.-]+", "_", f"{report_no}-{client}") path = os.path.join(out_dir, f"{safe_name}.html") else: parent = os.path.dirname(os.path.abspath(path)) if parent: os.makedirs(parent, exist_ok=True) html_text = render_inspection_report_html(result) with open(path, "w", encoding="utf-8") as f: f.write(html_text) return os.path.abspath(path) def _markdown_escape(value) -> str: if value is None: return "" return str(value).replace("\\", "\\\\").replace("|", "\\|").replace("\r\n", "
").replace("\n", "
") def render_inspection_report_markdown(result: Dict[str, Any]) -> str: payload = _inspection_report_payload(result) record = result.get("data") if isinstance(result.get("data"), dict) else {} client = result.get("client_code") or payload.get("client_code") or record.get("client_code") or get_server_id() report_no = result.get("report_no") or record.get("report_no") or f"CHECK-{client}" collected_at = payload.get("collected_at") or record.get("collected_at") or "" server = payload.get("server") or {} memory = payload.get("memory") or {} disks = payload.get("disks") or [] oracle = payload.get("oracle") or {} alerts = payload.get("alerts") or [] failed = payload.get("failed_items") or [] missing = payload.get("missing_permissions") or payload.get("missing_grants") or [] connections = server.get("connections") or {} cache_hit = oracle.get("cache_hit") or {} process_items = [] for process in server.get("processes") or []: if isinstance(process, str): process_items.append(process) else: process_items.append(f"{process.get('name', 'unknown')}={'运行中' if process.get('running') else '未运行'}({process.get('count', 0)})") process_text = "；".join(process_items) or "未返回关键进程状态" stale = result.get("stale") overall = "普通异常" if alerts or failed else "正常" lines = [ f"# 服务器巡检报告 - {_markdown_escape(client)}", "", f"- 报告编号：{_markdown_escape(report_no)}", f"- 巡检时间：{_markdown_escape(collected_at)}", f"- 报告类型：{'历史快照' if stale else '实时报告'}", f"- 总体状态：{overall}", "", "## 一、基础信息", "", "| 项目 | 内容 |", "| --- | --- |", f"| 客户/服务器编号 | {_markdown_escape(client)} |", f"| 计算机名 | {_markdown_escape(server.get('computer') or '待确认')} |", f"| 服务器 IP | {_markdown_escape(', '.join(server.get('ip_addresses') or []) or '待采集')} |", f"| 操作系统 | {_markdown_escape(' '.join(str(value) for value in (server.get('os'), server.get('os_version')) if value) or '待采集')} |", f"| 启动时间 | {_markdown_escape(server.get('boot_time') or '待采集')} |", f"| 服务器状态 | {'在线' if server.get('online') else '离线'} |", f"| Oracle 状态 | {'在线' if oracle.get('online') else '异常'} |", "", "## 二、服务器资源巡检", "", "### 2.1 基础运行状态", "", "| 巡检项 | 结果值 | 状态 |", "| --- | --- | --- |", f"| 服务器在线状态 | {'在线' if server.get('online') else '离线'} | {'正常' if server.get('online') else '重大异常'} |", f"| CPU 使用率 | {_markdown_escape(_pct(server.get('cpu_percent')))} | {_markdown_escape(_status_level_from_percent(server.get('cpu_percent'), 80, 90))} |", f"| 运行时长 | {_markdown_escape(server.get('uptime') or '待采集')} | {'正常' if server.get('uptime') else '待接入'} |", f"| 关键进程 | {_markdown_escape(process_text)} | {'正常' if process_items and all((p if isinstance(p, str) else p.get('running')) for p in (server.get('processes') or [])) else '异常或待采集'} |", f"| TCP 连接数 | 总计 {_markdown_escape(connections.get('total', '-'))}；已建立 {_markdown_escape(connections.get('established', '-'))}；监听 {_markdown_escape(connections.get('listening', '-'))}；TIME_WAIT {_markdown_escape(connections.get('time_wait', '-'))}；CLOSE_WAIT {_markdown_escape(connections.get('close_wait', '-'))} | {'正常' if connections.get('available') else '采集失败'} |", "", "### 2.2 内存状态", "", "| 总量 | 已用 | 剩余 | 使用率 | 1h 峰值 | 24h 峰值 | OOM 事件 |", "| --- | --- | --- | --- | --- | --- | --- |", "| {total} | {used} | {free} | {used_pct} | {peak1} | {peak24} | {oom} |".format( total=_markdown_escape(_gb(memory.get("total_gb"))), used=_markdown_escape(_gb(memory.get("used_gb"))), free=_markdown_escape(_gb(memory.get("free_gb"))), used_pct=_markdown_escape(_pct(memory.get("used_percent"))), peak1=_markdown_escape(_pct(memory.get("peak_1h_percent"))), peak24=_markdown_escape(_pct(memory.get("peak_24h_percent"))), oom=len(memory.get("oom_events") or []), ), "", "### 2.3 磁盘状态", "", "| 分区 | 总量 | 已用 | 剩余 | 使用率 | 24h 峰值 | 读速率 MB/s | 写速率 MB/s | 状态 |", "| --- | --- | --- | --- | --- | --- | --- | --- | --- |", ] if disks: for disk in disks: lines.append( "| {drive} | {total} | {used} | {free} | {used_pct} | {peak24} | {read} | {write} | {status} |".format( drive=_markdown_escape(disk.get("drive")), total=_markdown_escape(_gb(disk.get("total_gb"))), used=_markdown_escape(_gb(disk.get("used_gb"))), free=_markdown_escape(_gb(disk.get("free_gb"))), used_pct=_markdown_escape(_pct(disk.get("used_percent"))), peak24=_markdown_escape(_pct(disk.get("used_peak_24h_percent"))), read=_markdown_escape(disk.get("read_mb_s", 0)), write=_markdown_escape(disk.get("write_mb_s", 0)), status=_markdown_escape(disk.get("status") or "unknown"), ) ) else: lines.append("| - | - | - | - | - | - | - | - | 未返回磁盘明细 |") lines.extend([ "", "## 三、数据库巡检结果", "", "| 巡检项 | 结果值 | 状态 |", "| --- | --- | --- |", f"| 数据库服务 | {'在线' if oracle.get('online') else '异常'} | {'正常' if oracle.get('online') else '重大异常'} |", f"| 活跃会话数 | {_markdown_escape(oracle.get('active_sessions') if oracle.get('active_sessions') is not None else '采集失败')} | {_markdown_escape(_status_level_from_percent(oracle.get('active_sessions'), 50, 100))} |", f"| 用户会话总数 | {_markdown_escape(oracle.get('total_sessions') if oracle.get('total_sessions') is not None else '采集失败')} | {_markdown_escape(oracle.get('session_count_source') or '未采集')} |", f"| Buffer Cache 命中率 | {_markdown_escape(_pct(cache_hit.get('buffer_cache_percent')) if cache_hit.get('buffer_cache_percent') is not None else '采集失败')} | {_markdown_escape(_cache_hit_level(cache_hit.get('buffer_cache_percent')))} |", f"| Library Cache 命中率 | {_markdown_escape(_pct(cache_hit.get('library_cache_percent')) if cache_hit.get('library_cache_percent') is not None else '采集失败')} | {_markdown_escape(_cache_hit_level(cache_hit.get('library_cache_percent')))} |", f"| 长事务 | {_ops_count(_find_ops(payload, 'long_transactions'))} 条 | {'普通异常' if _ops_count(_find_ops(payload, 'long_transactions')) else '正常'} |", f"| 阻塞会话 | {_ops_count(_find_ops(payload, 'blocking_locks'))} 条 | {'普通异常' if _ops_count(_find_ops(payload, 'blocking_locks')) else '正常'} |", f"| 当前慢 SQL | {_ops_count(_find_ops(payload, 'active_slow_sql'))} 条 | {'普通异常' if _ops_count(_find_ops(payload, 'active_slow_sql')) else '正常'} |", "", "## 四、异常明细", "", "| 等级 | 类型 | 说明 |", "| --- | --- | --- |", ]) if alerts or failed: for alert in alerts: if isinstance(alert, dict): lines.append(f"| {_markdown_escape(alert.get('level', 'warning'))} | {_markdown_escape(alert.get('title') or alert.get('type') or '异常')} | {_markdown_escape(alert.get('desc') or alert.get('value') or '')} |") else: lines.append(f"| warning | 异常 | {_markdown_escape(alert)} |") for item in failed: if isinstance(item, dict): lines.append(f"| warning | {_markdown_escape(item.get('item') or '巡检项')} | {_markdown_escape(item.get('error') or item.get('reason') or '')} |") else: lines.append(f"| warning | 巡检项 | {_markdown_escape(item)} |") else: lines.append("| normal | 未发现异常 | 本次巡检未产生告警。 |") lines.extend(["", "## 五、巡检结论", "", f"本次巡检总体状态：**{overall}**。"]) if missing: lines.extend(["", "### 缺失授权", ""]) lines.extend(f"- {_markdown_escape(item)}" for item in missing) lines.extend(["", "---", "", "生成来源：Oracle Jump Query Agent / 固定格式 Markdown 巡检报告", ""]) return "\n".join(lines) def write_inspection_report_markdown(result: Dict[str, Any], path: str = "") -> str: payload = _inspection_report_payload(result) client = result.get("client_code") or payload.get("client_code") or get_server_id() report_no = result.get("report_no") or f"inspection-report-{client}" if not path: out_dir = os.path.join(SKILL_DIR, "outputs") os.makedirs(out_dir, exist_ok=True) safe_name = re.sub(r"[^A-Za-z0-9_.-]+", "_", f"{report_no}-{client}") path = os.path.join(out_dir, f"{safe_name}.md") else: parent = os.path.dirname(os.path.abspath(path)) if parent: os.makedirs(parent, exist_ok=True) with open(path, "w", encoding="utf-8") as f: f.write(render_inspection_report_markdown(result)) return os.path.abspath(path) def _parse_report_cli_args(args): as_json = "--json" in args html_path = "" markdown_path = None clean = [] i = 0 while i < len(args): arg = args[i] if arg == "--json": i += 1 continue if arg == "--html": html_path = "" if i + 1 < len(args) and not args[i + 1].startswith("--"): html_path = args[i + 1] i += 2 continue i += 1 continue if arg in ("--markdown", "--md"): markdown_path = "" if i + 1 < len(args) and not args[i + 1].startswith("--"): markdown_path = args[i + 1] i += 2 continue i += 1 continue clean.append(arg) i += 1 return clean, as_json, html_path, markdown_path def _print_inspection_report(result: Dict[str, Any]): if not result.get("success"): print(f"Error: {result.get('error', 'unknown error')}") if result.get("status_code"): print(f"HTTP status: {result.get('status_code')}") return payload = _inspection_report_payload(result) record = result.get("data") if isinstance(result.get("data"), dict) else {} client = result.get("client_code") or payload.get("client_code") or record.get("client_code") or get_server_id() stale = result.get("stale") title_suffix = "历史快照" if stale else "实时报告" print(f"# 服务器巡检报告 - {client} ({title_suffix})") print() if result.get("report_no"): print(f"- 报告号: {result.get('report_no')}") if record.get("id"): print(f"- 归档 ID: {_fmt_report_number(record.get('id'))}") if payload.get("collected_at"): print(f"- 采集时间: {payload.get('collected_at')}") elif record.get("collected_at"): print(f"- 采集时间: {record.get('collected_at')}") if payload.get("duration_ms") is not None: print(f"- 耗时: {payload.get('duration_ms')} ms") if result.get("archive_saved") is not None: print(f"- 已归档: {result.get('archive_saved')}") if stale is not None: print(f"- 历史快照: {stale}") memory = payload.get("memory") or {} if memory: print() print("## 内存") print("| 总量GB | 已用GB | 剩余GB | 使用率 | 1h峰值 | 24h峰值 |") print("| --- | --- | --- | --- | --- | --- |") print("| {total} | {used} | {free} | {used_pct}% | {peak1}% | {peak24}% |".format( total=memory.get("total_gb", ""), used=memory.get("used_gb", ""), free=memory.get("free_gb", ""), used_pct=memory.get("used_percent", ""), peak1=memory.get("peak_1h_percent", ""), peak24=memory.get("peak_24h_percent", ""), )) disks = payload.get("disks") or [] if disks: print() print("## 磁盘") print("| 分区 | 总量GB | 已用GB | 剩余GB | 使用率 | 24h峰值 | 状态 |") print("| --- | --- | --- | --- | --- | --- | --- |") for disk in disks: print("| {drive} | {total} | {used} | {free} | {used_pct}% | {peak24}% | {status} |".format( drive=disk.get("drive", ""), total=disk.get("total_gb", ""), used=disk.get("used_gb", ""), free=disk.get("free_gb", ""), used_pct=disk.get("used_percent", ""), peak24=disk.get("used_peak_24h_percent", ""), status=disk.get("status", ""), )) alerts = payload.get("alerts") or [] if alerts: print() print("## 告警") for alert in alerts: title = alert.get("title") or alert.get("type") or "告警" desc = alert.get("desc") or alert.get("value") or "" level = alert.get("level") or "" print(f"- [{level}] {title}: {desc}") failed = payload.get("failed_items") or [] if failed: print() print("## 未完成项目") for item in failed: if isinstance(item, dict): print(f"- {item.get('item', '')}: {item.get('error') or item.get('reason') or ''}") else: print(f"- {item}") missing = payload.get("missing_permissions") or payload.get("missing_grants") or [] if missing: print() print("## 缺失授权") for item in missing: print(f"- {item}") oracle = payload.get("oracle") or {} oracle_text = json.dumps(oracle, ensure_ascii=False).lower() needs_sys = any(token in oracle_text for token in ("unknown item", "ora-00942", "henlo_ora_monitor", "instance status")) if needs_sys: _print_inspection_sys_guidance() def _print_inspection_sys_guidance(): """Show the DBA-applied SYS function artifacts when inspection grants are missing.""" skill_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) sys_dir = os.path.join(skill_root, "references", "sys-functions") files = ["henlo_ora_monitor.sql", "henlo_awr_export.sql"] print() print("## 需要 SYS 授权") print("当前巡检检测到 Oracle SYS 函数或动态性能视图授权尚未完成。请由 DBA/SYS 手工执行以下留档 SQL，先将 替换为 Agent 实际登录账号。Skill 不会自动执行授权。") print(f"SQL 留档目录: {sys_dir}") for name in files: path = os.path.join(sys_dir, name) print(f"- {path}") print("执行顺序：henlo_ora_monitor.sql；确认 Diagnostics Pack 授权后再执行 henlo_awr_export.sql。授权完成后重新生成巡检报告。") def download_inspection_report_html(result: Dict[str, Any], path: str = "") -> str: """Download the fixed HTML rendered by transit-server; Skill only stores it locally.""" cfg = get_config() or {} transit_url = str(cfg.get("transit_url") or get_transit_url() or TRANSIT_URL).rstrip("/") html_url = str(result.get("html_url") or "") if not html_url: report_no = str(result.get("report_no") or result.get("data", {}).get("report_no") or "") if report_no: html_url = f"{transit_url}/api/inspection_report/{report_no}/html" if html_url.startswith("/"): html_url = transit_url + html_url if not html_url: raise RuntimeError("transit-server did not return html_url") if result.get("archive_saved") is False: raise RuntimeError("巡检报告尚未完成归档，暂时无法下载 HTML；请稍后重试") response = None last_error = "" for attempt in range(5): try: response = requests.get(html_url, headers=make_headers(), timeout=30) if response.status_code != 404: response.raise_for_status() break last_error = "HTML archive is still pending" except requests.RequestException as exc: last_error = str(exc) if attempt < 4: time.sleep(1) if response is None or response.status_code == 404: raise RuntimeError(f"巡检报告 HTML 下载失败：{last_error}，请确认 transit-server 已完成归档") if not path: out_dir = os.path.join(SKILL_DIR, "outputs") os.makedirs(out_dir, exist_ok=True) report_no = str(result.get("report_no") or "inspection-report") path = os.path.join(out_dir, report_no + ".html") else: os.makedirs(os.path.dirname(os.path.abspath(path)), exist_ok=True) with open(path, "wb") as f: f.write(response.content) return path def cmd_agent_versions(client_codes=None, all_clients=False, as_json=False): cfg = get_config() or {} clients = refresh_client_cache(cfg, quiet=True) if not clients: result = {"success": False, "error": "无法获取客户列表：登录已失效或本地没有客户缓存，请先执行 login"} print(json.dumps(result, ensure_ascii=False, indent=2) if as_json else result["error"]) return wanted = {str(c).upper() for c in (client_codes or [])} rows = [] for client in clients: code = get_client_code(client) if not all_clients and code.upper() not in wanted: continue row = {"client_code": code, "title": get_client_title(client), "version": client.get("version", ""), "status": "offline"} try: result = query(code, "version", "", "", timeout=15, max_retries=0) if result.get("success"): data = result.get("data") if isinstance(data, str): data = json.loads(data) if isinstance(data, dict) and data.get("version"): row["version"] = data["version"] row["status"] = "online" except Exception as exc: row["error"] = str(exc) rows.append(row) if as_json: print(json.dumps({"success": True, "servers": rows}, ensure_ascii=False, indent=2)) else: for row in rows: print(f"{row['client_code']}: {row.get('version') or '-'} ({row['status']})") def _output_inspection_report(result: Dict[str, Any], as_json: bool, html_path="", markdown_path=None): if not result.get("success"): if as_json: print(json.dumps(result, ensure_ascii=False, indent=2)) else: _print_inspection_report(result) return artifacts = {"html": download_inspection_report_html(result, html_path or "")} if markdown_path is not None: artifacts["markdown"] = write_inspection_report_markdown(result, markdown_path) if as_json: output = dict(result) output["artifacts"] = artifacts print(json.dumps(output, ensure_ascii=False, indent=2)) return print(f"HTML report: {artifacts['html']}") if artifacts.get("markdown"): print(f"Markdown report: {artifacts['markdown']}") print() _print_inspection_report(result) def cmd_inspection_report(client_code: str = "", as_json: bool = False, timeout: int = 90, html_path="", markdown_path=None): result = inspection_report(client_code, timeout=timeout) _output_inspection_report(result, as_json, html_path, markdown_path) def cmd_inspection_latest(client_code: str = "", as_json: bool = False, refresh: bool = False, html_path="", markdown_path=None): result = inspection_report_latest(client_code, refresh=refresh) _output_inspection_report(result, as_json, html_path, markdown_path) def cmd_inspection_get(report_id: str, as_json: bool = False, html_path="", markdown_path=None): result = inspection_report_get(report_id) _output_inspection_report(result, as_json, html_path, markdown_path) def get_upgrade_admin_token(cfg: Optional[Dict[str, Any]] = None) -> str: """Get transit-server agent update admin token.""" token = os.environ.get("ORACLE_JUMP_UPGRADE_ADMIN_TOKEN", "").strip() if token: return token cfg = cfg or get_config() or {} token = str(cfg.get("upgrade_admin_token", "") or "").strip() if token: return token return "" def agent_update(client_code: str = "", timeout: int = 300) -> Dict[str, Any]: """Trigger a remote agent self-update through transit-server.""" config = get_config() or {} transit_url = config.get("transit_url", TRANSIT_URL) if not ensure_logged_in(config): return {"success": False, "error": "login required"} if not client_code: client_code = config.get("client_code") or config.get("server_id", "") if not client_code: return {"success": False, "error": "clientCode is required"} if timeout <= 0 or timeout > 600: timeout = 300 upgrade_token = get_upgrade_admin_token(config) if not upgrade_token: return { "success": False, "error": "upgrade admin token is not configured; set ORACLE_JUMP_UPGRADE_ADMIN_TOKEN or upgrade_admin_token", } payload = {"server_id": client_code, "timeout": timeout} headers = make_headers() headers["X-Upgrade-Admin-Token"] = upgrade_token try: resp = requests.post( f"{transit_url}/api/admin/agent_update/trigger", json=payload, headers=headers, timeout=timeout + 15, ) if resp.status_code in (400, 401, 403, 404, 502, 503, 504): try: return {"success": False, "error": resp.json().get("error", resp.text), "status_code": resp.status_code} except Exception: return {"success": False, "error": resp.text, "status_code": resp.status_code} resp.raise_for_status() return resp.json() except requests.exceptions.RequestException as e: return {"success": False, "error": str(e)} def cmd_agent_update(client_code: str = "", timeout: int = 300): """Print remote agent update trigger result.""" result = agent_update(client_code, timeout) if not result.get("success"): print(f"Error: {result.get('error', 'unknown error')}") if result.get("status_code"): print(f"HTTP status: {result.get('status_code')}") return print("Agent update triggered.") print(json.dumps(result, ensure_ascii=False, indent=2)) DEFAULT_OPS_REPORT_ITEMS = [ "active_slow_sql", "blocking_locks", "long_transactions", "tablespace", "datafiles", "undo", "memory", "io_waits", "ora_errors", "invalid_objects", ] def oracle_ops(item: str, client_code: str = "", minutes: int = 60, top_n: int = 20, timeout: int = 30) -> Dict[str, Any]: """Call transit server oracle ops endpoint.""" config = get_config() or {} transit_url = config.get("transit_url", TRANSIT_URL) if not ensure_logged_in(config): return {"success": False, "error": "login required"} if not client_code: client_code = config.get("client_code", "") payload = { "client_code": client_code, "item": item, "minutes": minutes, "top_n": top_n, "timeout": timeout, } try: resp = requests.post(f"{transit_url}/api/oracle_ops", json=payload, headers=make_headers(), timeout=timeout + 5) if resp.status_code in (400, 401, 403, 503): try: return {"success": False, "error": resp.json().get("error", resp.text)} except Exception: return {"success": False, "error": resp.text} resp.raise_for_status() return resp.json() except requests.exceptions.RequestException as e: return {"success": False, "error": str(e)} def oracle_ops_report(client_code: str = "", items=None, minutes: int = 60, top_n: int = 20, per_item_timeout: int = 30) -> Dict[str, Any]: """Call transit server oracle ops report endpoint.""" config = get_config() or {} transit_url = config.get("transit_url", TRANSIT_URL) if not ensure_logged_in(config): return {"success": False, "error": "login required"} if not client_code: client_code = config.get("client_code", "") payload = { "client_code": client_code, "items": items or DEFAULT_OPS_REPORT_ITEMS, "minutes": minutes, "top_n": top_n, "per_item_timeout": per_item_timeout, } try: resp = requests.post( f"{transit_url}/api/oracle_ops_report", json=payload, headers=make_headers(), timeout=max(30, per_item_timeout * len(payload["items"]) + 10), ) if resp.status_code in (400, 401, 403, 503): try: return {"success": False, "error": resp.json().get("error", resp.text)} except Exception: return {"success": False, "error": resp.text} resp.raise_for_status() return resp.json() except requests.exceptions.RequestException as e: return {"success": False, "error": str(e)} def _markdown_table(columns, rows): if not columns: return "未发现异常" lines = [] lines.append("| " + " | ".join(str(c) for c in columns) + " |") lines.append("| " + " | ".join(["---"] * len(columns)) + " |") for row in rows or []: values = [str(v) if v is not None else "" for v in row] while len(values) < len(columns): values.append("") lines.append("| " + " | ".join(values[:len(columns)]) + " |") if len(lines) == 2: lines.append("| " + " | ".join([""] * len(columns)) + " |") return "\n".join(lines) def _print_ops_item(data: Dict[str, Any]): title = data.get("title") or data.get("item") or "Oracle 运维监控" status = data.get("status", "") print(f"## {title}") if status: print(f"状态: {status}") if data.get("warnings"): print("\n告警:") for warning in data.get("warnings", []): print(f"- {warning}") if data.get("error"): print(f"\n错误: {data.get('error')}") print() print(_markdown_table(data.get("columns", []), data.get("rows", []))) if data.get("grant_sql_required") and data.get("grant_sql"): print("\n### SYS 授权函数安装 SQL") print("```sql") print(data.get("grant_sql")) print("```") def cmd_ops(item: str, client_code: str = ""): result = oracle_ops(item, client_code, timeout=60 if item == "history_top_sql" else 30) if not result.get("success"): print(f"Error: {result.get('error', 'unknown error')}") return _print_ops_item(result.get("data", {}) or {}) def cmd_ops_report(client_code: str = ""): result = oracle_ops_report(client_code) if not result.get("success"): print(f"Error: {result.get('error', 'unknown error')}") return client = result.get("client_code", client_code or get_server_id()) summary = result.get("summary", {}) or {} print(f"# Oracle 运维日报 - {client}") print() print("## 总览") print(f"- 高风险: {summary.get('high', 0)}") print(f"- 警告: {summary.get('warning', 0)}") print(f"- 正常: {summary.get('normal', 0)}") print(f"- 未完成: {summary.get('failed', 0)}") for item in result.get("items", []) or []: print() _print_ops_item(item) failed = result.get("failed_items", []) or [] if failed: print("\n## 未完成项目") print("| item | reason |") print("| --- | --- |") for f in failed: print(f"| {f.get('item', '')} | {f.get('error', '')} |") if result.get("grant_sql_required") and result.get("grant_sql"): print("\n## SYS 授权函数") print("当前库缺少或无法执行 `SYS.HENLO_ORA_MONITOR` / 当前账号 `HENLO_ORA_MONITOR`。") print("如果有 SYS 账号，优先按 Option A 安装；如果当前 Agent 登录账号已有 v$/dba_/AWR 查询权限，可按 Option B 安装到当前账号。") print("```sql") print(result.get("grant_sql")) print("```") def cmd_sys_functions(schema: str = "", as_json: bool = False): """Print every SYS function SQL currently required by the skill.""" schema = (schema or "").strip() if schema != "" and not re.fullmatch(r"[A-Za-z][A-Za-z0-9_$#]*", schema): raise ValueError("schema must be a valid Oracle identifier") root = os.path.join(SKILL_DIR, "references", "sys-functions") files = sorted(f for f in os.listdir(root) if f.lower().endswith(".sql")) items = [] for name in files: with open(os.path.join(root, name), "r", encoding="utf-8-sig") as f: items.append({"file": name, "sql": f.read().replace("", schema)}) if as_json: print(json.dumps({"success": True, "schema": schema, "files": items}, ensure_ascii=False, indent=2)) return print("-- Oracle Jump Query SYS function deployment bundle") print(f"-- Grant target schema: {schema}") print("-- Execute as SYS. This command only generates SQL.\n") for item in items: print(f"-- ===== {item['file']} =====") print(item["sql"].rstrip() + "\n") def cmd_login(secret_key: str, client_code: str = ""): """ 登录中转机并选择 client 流程： 1. 调中转机 /api/login，由中转机调用 BOS ts_login 2. 中转机验证 clientCode（客户服务器编号，可选）并签发 access_token 3. 保存到 config.json """ cfg = get_config() or {} transit_url = cfg.get("transit_url", TRANSIT_URL) try: resp = requests.post( f"{transit_url}/api/login", json={"secret_key": secret_key, "client_code": client_code}, timeout=20, ) if resp.status_code in (401, 403): try: err = resp.json().get("error", resp.text) except Exception: err = resp.text print(f"❌ 登录失败: {err}") return resp.raise_for_status() result = resp.json() except requests.exceptions.RequestException as e: print(f"❌ 登录失败，无法连接中转机: {e}") return if not result.get("success"): print(f"❌ 登录失败: {result.get('error', '未知错误')}") return current_client = result.get("current_client", {}) client_list = result.get("client_list", []) access_token = result.get("access_token", "") if not access_token: print("❌ 登录失败: 中转机未返回 access_token") return # 3. 保存配置 cfg.pop("secret_key", None) cfg["access_token"] = access_token cfg["expires_at"] = result.get("expires_at", "") cfg["user_name"] = result.get("user", {}).get("name", "") cfg["client_code"] = current_client.get("code", client_code) cfg["client_title"] = current_client.get("title", "") cfg["client_list"] = client_list cfg["server_id"] = cfg["client_code"] save_config(cfg) print(f"✅ 登录成功") print(f" 用户: {cfg.get('user_name', '')}") print(f" 指向 client: {cfg['client_code']} ({cfg.get('client_title', '')})") print(f" 过期时间: {cfg.get('expires_at', '')}") if len(client_list) > 1: print(f" 可切换 client: {', '.join(c['code'] for c in client_list)}") def _device_key_path() -> str: return os.path.join(os.path.expanduser("~"), ".oracle-jump-query", "device-key.json") def _load_device_key() -> Dict[str, Any]: with open(_device_key_path(), "r", encoding="utf-8") as f: return json.load(f) def _save_device_key(item: Dict[str, Any]) -> None: path = _device_key_path() os.makedirs(os.path.dirname(path), mode=0o700, exist_ok=True) with open(path, "x", encoding="utf-8") as f: json.dump(item, f, ensure_ascii=False, indent=2) try: os.chmod(path, 0o600) except OSError: pass def _device_identity() -> Tuple[str, str]: mac = ":".join(f"{(uuid.getnode() >> shift) & 0xff:02X}" for shift in range(40, -1, -8)) return hashlib.sha256(mac.encode("ascii")).hexdigest(), f"{mac[:8]}****{mac[-5:]}" def cmd_device_register(device_name: str = ""): """Generate an Ed25519 key locally and register this terminal as PENDING.""" try: from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey except ImportError: print("❌ 需要安装 cryptography: python -m pip install cryptography") return cfg = get_config() or {} if not ensure_logged_in(cfg): print("❌ 当前登录态已失效，请先执行 login") return try: key = _load_device_key() private = Ed25519PrivateKey.from_private_bytes(base64.b64decode(key["private_key"])) except (OSError, KeyError, ValueError, TypeError): private = Ed25519PrivateKey.generate() public = private.public_key().public_bytes_raw() key = {"device_id": "device-" + secrets.token_hex(12), "private_key": base64.b64encode(private.private_bytes_raw()).decode("ascii"), "public_key": base64.b64encode(public).decode("ascii"), "key_fingerprint": hashlib.sha256(public).hexdigest()} try: _save_device_key(key) except FileExistsError: # Another Skill process created the stable key first; never overwrite it. key = _load_device_key() mac_hash, mac_masked = _device_identity() body = {"device_id": key["device_id"], "device_name": device_name or platform.node() or "Trusted device", "computer_name": platform.node(), "os_name": platform.platform(), "key_algorithm": "ED25519", "public_key": key["public_key"], "key_fingerprint": key["key_fingerprint"], "mac_hash": mac_hash, "mac_masked": mac_masked, "description": "Registered by oracle-jump-query skill"} try: resp = requests.post(f"{cfg.get('transit_url', TRANSIT_URL)}/api/device/register", json=body, headers=make_headers(), timeout=20) data = resp.json() except (requests.RequestException, ValueError) as exc: print(f"❌ 可信设备注册失败: {exc}") return if not data.get("success"): print(f"❌ 可信设备注册失败: {data.get('error', resp.text)}") return print("✅ 可信设备注册请求已提交，当前状态为 PENDING") print(f" device_id: {key['device_id']}") print(f" fingerprint: {key['key_fingerprint']}") print(" 请在后台表单审批为 APPROVED 后执行 device_login") def cmd_device_login(client_code: str = ""): """Use the locally stored Ed25519 key to perform trusted-device login.""" try: from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey key = _load_device_key() private = Ed25519PrivateKey.from_private_bytes(base64.b64decode(key["private_key"])) except (ImportError, OSError, KeyError, ValueError, TypeError) as exc: print(f"❌ 本机可信设备密钥不可用: {exc}") return transit_url = (get_config() or {}).get("transit_url", TRANSIT_URL) identity = {"device_id": key["device_id"], "key_fingerprint": key["key_fingerprint"]} try: challenge_resp = requests.post(f"{transit_url}/api/device/challenge", json=identity, timeout=20) challenge_data = challenge_resp.json() if not challenge_data.get("success"): print(f"❌ 获取可信设备挑战失败: {challenge_data.get('error', challenge_resp.text)}") return challenge = challenge_data["challenge"] signature = base64.b64encode(private.sign(challenge.encode("utf-8"))).decode("ascii") resp = requests.post(f"{transit_url}/api/device/login", json={**identity, "challenge": challenge, "signature": signature, "client_code": client_code}, timeout=20) data = resp.json() except (requests.RequestException, ValueError) as exc: print(f"❌ 可信设备登录失败: {exc}") return if not data.get("success"): print(f"❌ 可信设备登录失败: {data.get('error', resp.text)}") return cfg = get_config() or {} cfg.update({"access_token": data.get("access_token", ""), "expires_at": data.get("expires_at", ""), "user_name": data.get("user", {}).get("name", ""), "client_code": data.get("current_client", {}).get("code", client_code), "client_title": data.get("current_client", {}).get("title", "")}) cfg["server_id"] = cfg.get("client_code", "") cfg["client_list"] = data.get("client_list", []) save_config(cfg) print("✅ 可信设备登录成功") print(f" 用户: {cfg.get('user_name', '')}") print(f" 指向 client: {cfg.get('client_code', '')}") def cmd_logout(): """登出，清除登录信息""" cfg = get_config() or {} token = cfg.get("access_token", "") transit_url = cfg.get("transit_url", TRANSIT_URL) if token: try: requests.post(f"{transit_url}/api/logout", headers=make_headers(), timeout=10) except Exception: pass # 保留 transit_url 和 server_id，清除登录相关 keep_keys = {"transit_url", "server_id", "default_schema"} new_cfg = {k: v for k, v in cfg.items() if k in keep_keys} save_config(new_cfg) print("✅ 已登出") def cmd_status(): """显示当前登录状态和选中的 client""" config = get_config() or {} transit_url = config.get("transit_url", TRANSIT_URL) print("当前状态：") print("-" * 40) if not config.get("access_token"): print(" 未登录，请使用 login [clientCode] 登录") print("-" * 40) return if is_token_expired(config): print(" 登录已过期，请重新执行 login [clientCode]") print(f" 过期时间：{config.get('expires_at', '')}") print("-" * 40) return try: resp = requests.get(f"{transit_url}/api/me", headers=make_headers(), timeout=5) if resp.status_code == 401: print(" 中转机登录态已失效，请重新登录") print("-" * 40) return resp.raise_for_status() me = resp.json() user = me.get("user", {}) client_list = me.get("client_list", []) if sync_client_cache(config, client_list): if me.get("expires_at"): config["expires_at"] = me.get("expires_at") if user.get("name"): config["user_name"] = user.get("name") save_config(config) print(f" 用户：{user.get('name', config.get('user_name', '未知'))}") print(f" client：{config.get('client_code', '未选择')} ({config.get('client_title', '')})") print(f" 过期时间：{me.get('expires_at', config.get('expires_at', ''))}") if client_list: print(f" 可用 client：") for cl in client_list: marker = " <--" if cl["code"] == config.get("client_code") else "" print(f" - {cl['code']} ({cl.get('title', '')}){marker}") except Exception as e: print(f" 无法从中转机获取状态: {e}") # 显示 Agent 在线状态 try: resp = requests.get(f"{transit_url}/api/clients", headers=make_headers(), timeout=5) if resp.status_code == 200: data = resp.json() if sync_client_cache(config, data.get("clients", [])): save_config(config) current_code = config.get("client_code", "") for cl in data.get("clients", []): if cl.get("code") == current_code: online = "在线" if cl.get("online") else "离线" auth = "已授权" if cl.get("authorized") else "未授权" print(f" Agent 状态：{online} | {auth}") break except Exception: pass # 静默失败 print("-" * 40) def cmd_switch(client_selector: str): """切换当前 client，支持 code 或 title/name 匹配。""" cfg = get_config() if not cfg: print("❌ 未登录，请先使用 login 登录") return if not ensure_logged_in(cfg): return client_list = cfg.get("client_list", []) matches, match_type = find_client_matches(client_list, client_selector) if not matches: print(f"未在本地缓存中找到 client: {client_selector}，正在刷新 client 列表...") client_list = refresh_client_cache(cfg) matches, match_type = find_client_matches(client_list, client_selector) if not matches: print(f"❌ client '{client_selector}' 不在可用列表中") print("可用 client:") print_client_candidates(client_list) return if len(matches) > 1: print(f"⚠️ client '{client_selector}' 匹配到多个结果，请指定更准确的 client code：") print_client_candidates(matches) return matched_client = normalize_client(matches[0]) cfg["client_code"] = matched_client["code"] cfg["client_title"] = matched_client.get("title", "") cfg["server_id"] = matched_client["code"] sync_client_cache(cfg, client_list) save_config(cfg) print(f"✅ 已切换到 client: {matched_client['code']} ({matched_client.get('title', '')})") def make_headers(): """构建 HTTP 请求头""" headers = {"Content-Type": "application/json"} token = "" cfg = get_config() if cfg: token = cfg.get("access_token", "") if not token: token = AUTH_TOKEN if token: headers["Authorization"] = f"Bearer {token}" return headers def parse_expires_at(value: str) -> Optional[datetime]: if not value: return None try: return datetime.fromisoformat(value.replace("Z", "+00:00")) except Exception: return None def is_token_expired(cfg: dict) -> bool: expires_at = parse_expires_at(cfg.get("expires_at", "")) if not expires_at: return False now = datetime.now(expires_at.tzinfo or timezone.utc) return now >= expires_at def _auto_device_login(cfg: Dict[str, Any]) -> bool: """Refresh the transit session with the locally approved trusted device.""" try: from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey key = _load_device_key() private = Ed25519PrivateKey.from_private_bytes(base64.b64decode(key["private_key"])) identity = {"device_id": key["device_id"], "key_fingerprint": key["key_fingerprint"]} transit_url = cfg.get("transit_url", TRANSIT_URL) challenge_data = requests.post(f"{transit_url}/api/device/challenge", json=identity, timeout=20).json() if not challenge_data.get("success"): return False challenge = challenge_data["challenge"] signature = base64.b64encode(private.sign(challenge.encode("utf-8"))).decode("ascii") data = requests.post(f"{transit_url}/api/device/login", json={**identity, "challenge": challenge, "signature": signature, "client_code": cfg.get("client_code", "")}, timeout=20).json() if not data.get("success") or not data.get("access_token"): return False current = data.get("current_client", {}) cfg.update({"access_token": data["access_token"], "expires_at": data.get("expires_at", ""), "user_name": data.get("user", {}).get("name", cfg.get("user_name", "")), "client_code": current.get("code", cfg.get("client_code", "")), "client_title": current.get("title", cfg.get("client_title", "")), "client_list": data.get("client_list", cfg.get("client_list", []))}) cfg["server_id"] = cfg.get("client_code", cfg.get("server_id", "")) save_config(cfg) print("✅ 中转 token 已通过可信设备自动续取") return True except (ImportError, OSError, KeyError, ValueError, TypeError, requests.RequestException): return False def ensure_logged_in(cfg: Optional[dict] = None) -> bool: cfg = cfg or get_config() or {} if not cfg.get("access_token") or is_token_expired(cfg): if _auto_device_login(cfg): return True if not cfg.get("access_token"): print("❌ 未登录，请先执行: python oracle_skill.py login [clientCode]") return False if is_token_expired(cfg): print("❌ 登录已过期，请重新执行: python oracle_skill.py login [clientCode]") print(" 注意：同一个 secretKey 在其他设备重新登录后，本设备也需要重新登录。") return False return True 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: str, action: str, schema: str = "", name: str = "", timeout: int = DEFAULT_TIMEOUT, max_retries: int = MAX_RETRIES, sql: str = "") -> Dict[str, Any]: """ 发送查询请求到中转服务（带重试机制） Args: server_id: 服务器 ID action: 操作类型（analyze_procedure, list_procedures, 等等） schema: Schema 名称 name: 存储过程名称 timeout: 超时时间（秒） max_retries: 最大重试次数 sql: SQL 语句（execute_query 用） Returns: dict: {"success": True/False, "data": ..., "error": ...} """ if not ensure_logged_in(): return {"success": False, "error": "login required"} url = f"{get_transit_url()}/api/query" payload = { "server_id": server_id, "action": action, "schema": schema, "name": name, "timeout": timeout, } if sql: payload["sql"] = sql last_error = None for attempt in range(max_retries + 1): try: logger.info(f"尝试 {attempt + 1}/{max_retries + 1}: POST {url}") logger.debug(f"Payload: {payload}") resp = requests.post( url, json=payload, headers=make_headers(), timeout=timeout + 3 # 额外 3 秒用于网络延迟（中转在 timeout 秒后即返回 504） ) resp.raise_for_status() result = resp.json() logger.info(f"请求成功: {result.get('success', 'unknown')}") return result except requests.exceptions.ConnectionError as e: last_error = f"无法连接中转服务 {url}，请检查地址和端口: {e}" logger.error(last_error) except requests.exceptions.Timeout as e: last_error = f"请求超时（{timeout}秒），Agent 可能未响应或处理时间过长: {e}" logger.error(last_error) except requests.exceptions.HTTPError as e: last_error = f"HTTP 错误: {e}" logger.error(last_error) except Exception as e: last_error = f"未知错误: {e}" logger.error(last_error) # 如果不是最后一次尝试，则等待后重试 if attempt < max_retries: logger.info(f"等待 {RETRY_DELAY} 秒后重试...") time.sleep(RETRY_DELAY) # 所有重试都失败了 return {"success": False, "error": last_error} def list_servers(max_retries: int = MAX_RETRIES): """列出所有在线的 Agent 服务器（带重试）""" if not ensure_logged_in(): return {"error": "login required"} url = f"{get_transit_url()}/api/servers" for attempt in range(max_retries + 1): try: logger.info(f"获取服务器列表: {url}") resp = requests.get(url, headers=make_headers(), timeout=10) resp.raise_for_status() return resp.json() except Exception as e: if attempt < max_retries: logger.warning(f"获取服务器列表失败，重试中... ({e})") time.sleep(RETRY_DELAY) else: logger.error(f"获取服务器列表失败: {e}") return {"error": str(e)} def health_check(max_retries: int = MAX_RETRIES): """检查中转服务健康状态（带重试）""" url = f"{get_transit_url()}/api/health" for attempt in range(max_retries + 1): try: logger.info(f"健康检查: {url}") resp = requests.get(url, timeout=5) resp.raise_for_status() return resp.json() except Exception as e: if attempt < max_retries: logger.warning(f"健康检查失败，重试中... ({e})") time.sleep(RETRY_DELAY) else: logger.error(f"健康检查失败: {e}") return {"error": str(e)} def check_agent_status(server_id: str) -> Dict[str, Any]: """ 检查 Agent 状态 Returns: dict: {"online": True/False, "server_id": ..., "error": ...} """ logger.info(f"检查 Agent 状态: {server_id}") result = list_servers() if "error" in result: return {"online": False, "error": result["error"]} servers = result.get("servers", []) for s in servers: if s.get("server_id") == server_id: return {"online": True, "server_id": server_id, "info": s} return {"online": False, "error": f"Agent {server_id} 不在线"} def query_with_permission(user_id: str, table_id: str, sql: str, timeout: int = DEFAULT_TIMEOUT) -> Dict[str, Any]: """ 执行带权限过滤的查询 流程： 1. 调用 get_user_perm 获取用户权限 2. 解析权限SQL 3. 拼接到原SQL的WHERE条件 4. 执行最终SQL Args: user_id: 用户ID table_id: 表ID (AD_TABLE.ID) sql: 原始SQL查询 timeout: 超时时间 Returns: dict: 查询结果 """ logger.info(f"查询用户 {user_id} 对表 {table_id} 的权限...") # 1. 获取权限 param = f"{user_id},{table_id}" perm_result = query(get_server_id(), "get_user_perm", "BOSNDS3", param, timeout=30, max_retries=MAX_RETRIES) if not perm_result.get("success"): return perm_result # 2. 解析权限SQL try: perm_data = json.loads(perm_result.get("data", "{}")) except json.JSONDecodeError: return {"success": False, "error": "权限数据解析失败"} perm_sql = perm_data.get("perm_sql", "") has_restriction = perm_data.get("has_restriction", False) logger.info(f"权限结果: has_restriction={has_restriction}") # 3. 拼接SQL if has_restriction and perm_sql: # 改进的 WHERE 拼接：找到最外层 WHERE 的位置 # 1. 去除注释 import re sql_no_comments = re.sub(r'--.*?$', '', sql, flags=re.MULTILINE) sql_no_comments = re.sub(r'/\*.*?\*/', '', sql_no_comments, flags=re.DOTALL) # 2. 找到最外层 WHERE（忽略子查询中的 WHERE） # 简单方法：找到最后一个 FROM 后的 WHERE（如果存在） # 更健壮的方法：使用 SQL 解析器，这里简化处理 sql_upper = sql_no_comments.upper() # 检查是否有 WHERE if "WHERE" in sql_upper: # 找到 WHERE 位置 where_pos = sql_upper.rfind("WHERE") # 检查 WHERE 是否在括号内（子查询） # 统计 WHERE 前的 '(' 和 ')' 数量 before_where = sql_no_comments[:where_pos] open_paren = before_where.count('(') close_paren = before_where.count(')') if open_paren == close_paren: # 最外层 WHERE，追加 AND final_sql = f"{sql} AND {perm_sql}" else: # WHERE 在子查询中，在末尾添加 WHERE # 检查是否已有 GROUP BY / ORDER BY / LIMIT if any(kw in sql_upper for kw in ["GROUP BY", "ORDER BY", "LIMIT", "FETCH"]): # 在这些关键字前插入 WHERE for kw in ["GROUP BY", "ORDER BY", "LIMIT", "FETCH"]: if kw in sql_upper: kw_pos = sql_upper.find(kw) final_sql = f"{sql[:kw_pos]} WHERE {perm_sql} {sql[kw_pos:]}" break else: # 在末尾添加 WHERE final_sql = f"{sql} WHERE {perm_sql}" else: # 无 WHERE，添加 WHERE # 检查是否已有 GROUP BY / ORDER BY / LIMIT if any(kw in sql_upper for kw in ["GROUP BY", "ORDER BY", "LIMIT", "FETCH"]): for kw in ["GROUP BY", "ORDER BY", "LIMIT", "FETCH"]: if kw in sql_upper: kw_pos = sql_upper.find(kw) final_sql = f"{sql[:kw_pos]} WHERE {perm_sql} {sql[kw_pos:]}" break else: # 在末尾添加 WHERE final_sql = f"{sql} WHERE {perm_sql}" logger.info(f"拼接权限条件: {perm_sql[:100]}...") else: # 全权限或无限制 final_sql = sql logger.info("用户无权限限制，直接执行原SQL") # 4. 执行最终SQL logger.info(f"执行最终SQL: {final_sql[:100]}...") return query(get_server_id(), "execute_query", "BOSNDS3", "", timeout=timeout, max_retries=MAX_RETRIES, sql=final_sql) def print_result(result: Dict[str, Any]): """格式化输出结果""" if result.get("success"): data = result.get("data", "") print(data) else: error = result.get("error", "未知错误") print(f"❌ 错误: {error}", file=sys.stderr) logger.error(f"查询失败: {error}") def interactive_mode(): """交互式模式""" # Global variable declarations (must be at the beginning of the function) global DEFAULT_TIMEOUT, MAX_RETRIES, DEFAULT_SERVER_ID print("=" * 50) print(" Oracle Jump Query - AI Skill (Improved)") 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: agents = health.get('agents', 0) print(f"✅ 中转服务正常 (在线 Agent: {agents})") # 检查默认 Agent 是否在线 agent_status = check_agent_status(get_server_id()) if agent_status["online"]: print(f"✅ Agent {get_server_id()} 在线") else: print(f"⚠️ Agent {get_server_id()} 不在线: {agent_status.get('error', '')}") 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 - 分析存储过程（源码+依赖+表+触发器） list - 列出 schema 下的所有存储过程 source - 获取存储过程源码 deps - 获取存储过程依赖 tables - 获取相关表结构 describe - 查询表结构（列、索引、行数） discover [schema] [domain] - 发现核心业务域（NL2SQL前置） nl2sql [schema] [domain] - 生成 NL2SQL Schema 字典 query - 执行 SELECT 查询（生产表必须带 WHERE 条件和行数限制） perm [col] - 查询用户数据权限 qperm - 带权限过滤的查询 login [clientCode]- 登录中转机 + 选择客户服务器 logout - 登出 status - 查看登录状态 + 当前 client switch - 按 code 或名称切换 client servers - 列出在线 Agent health - 检查中转服务状态 tablespace / tablespaces - 查询表空间使用情况 inspection_report [client] - 生成并归档巡检报告，默认导出 HTML（可加 --markdown） inspection_latest [client] - 查看最近报告并导出 HTML（可加 --refresh / --markdown） inspection_get - 查看指定归档报告并导出 HTML（可加 --markdown） awr_status [client] - 查看 AWR 授权、权限和生成状态 awr_list [client] - 列出 Agent 已生成的 AWR 报告 awr_download - 下载指定日期的 AWR HTML（yyyyMMdd） agent - 检查指定 Agent 状态 server - 切换目标服务器 timeout - 设置默认超时时间 retry - 设置重试次数 debug - 切换调试模式 version - 查看 Skill 版本号 exit - 退出 """) elif cmd == "version": print(f"Oracle Jump Query Skill v{VERSION}") print(f"脚本路径: {__file__}") print(f"Skill 目录: {SKILL_DIR}") elif cmd == "login": if len(parts) < 2: print("用法: login [clientCode]") continue cmd_login(parts[1], parts[2] if len(parts) >= 3 else "") elif cmd == "logout": cmd_logout() elif cmd == "status": cmd_status() elif cmd == "ops": if len(parts) < 2: print("用法: ops [clientCode]") continue cmd_ops(parts[1], parts[2] if len(parts) >= 3 else "") elif cmd == "ops_report": cmd_ops_report(parts[1] if len(parts) >= 2 else "") elif cmd == "sys_functions": args = [p for p in parts[1:] if not p.startswith("--")] cmd_sys_functions(args[0] if args else "", "--json" in parts) elif cmd == "inspection_report": args, as_json, html_path, markdown_path = _parse_report_cli_args(parts[1:]) cmd_inspection_report(args[0] if args else "", as_json=as_json, html_path=html_path, markdown_path=markdown_path) elif cmd == "inspection_latest": refresh = "--refresh" in parts args, as_json, html_path, markdown_path = _parse_report_cli_args([p for p in parts[1:] if p != "--refresh"]) cmd_inspection_latest(args[0] if args else "", as_json=as_json, refresh=refresh, html_path=html_path, markdown_path=markdown_path) elif cmd == "inspection_get": if len(parts) < 2: print("用法: inspection_get [--json] [--html [path]] [--markdown [path]]") continue args, as_json, html_path, markdown_path = _parse_report_cli_args(parts[1:]) cmd_inspection_get(args[0], as_json=as_json, html_path=html_path, markdown_path=markdown_path) elif cmd == "awr_status": args = [p for p in parts[1:] if p != "--json"] cmd_awr_status(args[0] if args else "", as_json="--json" in parts) elif cmd == "awr_list": args = [p for p in parts[1:] if p != "--json"] cmd_awr_list(args[0] if args else "", as_json="--json" in parts) elif cmd == "awr_download": args = [p for p in parts[1:] if not p.startswith("--")] if len(args) < 2: print("用法: awr_download [--output ] [--json]") continue output_path = "" if "--output" in parts: output_index = parts.index("--output") if output_index + 1 < len(parts): output_path = parts[output_index + 1] cmd_awr_download(args[0], args[1], output_path=output_path, as_json="--json" in parts) elif cmd == "agent_update": timeout = int(parts[2]) if len(parts) >= 3 and parts[2].isdigit() else 300 cmd_agent_update(parts[1] if len(parts) >= 2 else "", timeout) elif cmd == "switch": if len(parts) < 2: print("用法: switch ") continue cmd_switch(" ".join(parts[1:])) 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 == "agent": if len(parts) < 2: status = check_agent_status(get_server_id()) else: status = check_agent_status(parts[1]) if status["online"]: print(f"✅ Agent {status['server_id']} 在线") print(f" 信息: {status.get('info', {})}") else: print(f"❌ Agent 不在线: {status.get('error', '')}") elif cmd == "server": if len(parts) < 2: print(f"当前服务器: {get_server_id()}") else: DEFAULT_SERVER_ID = parts[1] print(f"切换到服务器: {parts[1]}") elif cmd == "timeout": if len(parts) < 2: print(f"当前超时时间: {DEFAULT_TIMEOUT} 秒") else: try: DEFAULT_TIMEOUT = int(parts[1]) print(f"超时时间设置为: {DEFAULT_TIMEOUT} 秒") except ValueError: print("用法: timeout <秒数>") elif cmd == "retry": if len(parts) < 2: print(f"当前重试次数: {MAX_RETRIES}") else: try: MAX_RETRIES = int(parts[1]) print(f"重试次数设置为: {MAX_RETRIES}") except ValueError: print("用法: retry <次数>") elif cmd == "debug": # 切换调试模式 if logger.level == logging.DEBUG: logger.setLevel(logging.INFO) print("调试模式: 关闭") else: logger.setLevel(logging.DEBUG) print("调试模式: 开启") elif cmd == "analyze": if len(parts) < 3: print("用法: analyze ") continue print(f"⏳ 正在分析 {parts[1]}.{parts[2]} (超时 {DEFAULT_TIMEOUT * 2} 秒)...") result = query(get_server_id(), "analyze_procedure", parts[1], parts[2], timeout=DEFAULT_TIMEOUT * 2, max_retries=MAX_RETRIES) print_result(result) elif cmd == "list": if len(parts) < 2: print("用法: list ") continue print(f"⏳ 正在获取存储过程列表...") result = query(get_server_id(), "list_procedures", parts[1], timeout=30, max_retries=MAX_RETRIES) print_result(result) elif cmd == "source": if len(parts) < 3: print("用法: source ") continue print(f"⏳ 正在获取源码...") result = query(get_server_id(), "get_source", parts[1], parts[2], timeout=DEFAULT_TIMEOUT, max_retries=MAX_RETRIES) print_result(result) elif cmd == "deps": if len(parts) < 3: print("用法: deps ") continue print(f"⏳ 正在获取依赖...") result = query(get_server_id(), "get_dependencies", parts[1], parts[2], timeout=DEFAULT_TIMEOUT, max_retries=MAX_RETRIES) print_result(result) elif cmd == "tables": if len(parts) < 3: print("用法: tables ") continue print(f"⏳ 正在获取表结构...") result = query(get_server_id(), "get_tables", parts[1], parts[2], timeout=DEFAULT_TIMEOUT, max_retries=MAX_RETRIES) print_result(result) elif cmd == "describe": if len(parts) < 3: print("用法: describe
") continue print(f"⏳ 正在查询表 {parts[1]}.{parts[2]} ...") result = query(get_server_id(), "describe_table", parts[1], parts[2], timeout=DEFAULT_TIMEOUT, max_retries=MAX_RETRIES) print_result(result) elif cmd == "discover": # NL2SQL: 发现核心业务域 schema = parts[1] if len(parts) > 1 else get_config().get("default_schema", "BOSNDS3") domain = parts[2] if len(parts) > 2 else "" print(f"⏳ 正在发现业务域 (schema={schema}, domain={domain or 'ALL'})...") result = query(get_server_id(), "schema_discovery", schema, domain, timeout=DEFAULT_TIMEOUT * 3, max_retries=MAX_RETRIES) print_result(result) elif cmd == "nl2sql": # NL2SQL: 生成完整 Schema schema = parts[1] if len(parts) > 1 else get_config().get("default_schema", "BOSNDS3") domain = parts[2] if len(parts) > 2 else "RETAIL" print(f"⏳ 正在生成 NL2SQL Schema (schema={schema}, domain={domain})...") result = query(get_server_id(), "generate_nl2sql_schema", schema, domain, timeout=DEFAULT_TIMEOUT * 3, max_retries=MAX_RETRIES) print_result(result) elif cmd == "query": # execute_query: 执行任意 SELECT if len(parts) < 2: print("用法: query ") print(" 示例: query SELECT * FROM M_RETAIL WHERE BILLDATE = 20260501 AND ROWNUM <= 20") continue # 支持可选 server_id 前缀：query henlo "SELECT ..." known_servers = ["henlo", "renben", "HENLO", "RENBEN"] sql_parts = parts[1:] server_override = None if parts[1].lower() in known_servers: server_override = parts[1].upper() sql_parts = parts[2:] if not sql_parts: print("用法: query ") continue sql = " ".join(sql_parts) server_id = server_override if server_override else get_server_id() print(f"⏳ 执行查询 [{server_id}]: {sql[:80]}...") result = query(server_id, "execute_query", "BOSNDS3", "", timeout=DEFAULT_TIMEOUT, max_retries=MAX_RETRIES, sql=sql) print_result(result) elif cmd == "perm": # get_user_perm: 获取用户权限 if len(parts) < 3: print("用法: perm [columnName]") print(" 示例: perm 1015 12964") print(" 示例: perm 940 12983 C_STORE_ID") continue user_id = parts[1] table_id = parts[2] col_name = parts[3] if len(parts) > 3 else "" param = f"{user_id},{table_id}" if not col_name else f"{user_id},{table_id},{col_name}" print(f"⏳ 查询用户 {user_id} 对表 {table_id} 的权限...") result = query(get_server_id(), "get_user_perm", "BOSNDS3", param, timeout=30, max_retries=MAX_RETRIES) print_result(result) elif cmd == "qperm": # query_with_perm: 带权限过滤的查询 if len(parts) < 4: print("用法: qperm ") print(" 示例: qperm 1015 12964 SELECT * FROM M_RETAIL WHERE BILLDATE=20260501") continue user_id = parts[1] table_id = parts[2] sql = " ".join(parts[3:]) print(f"⏳ 执行带权限过滤的查询 (用户={user_id}, 表={table_id})...") result = query_with_permission(user_id, table_id, sql) print_result(result) elif cmd == "clients": cmd_clients() elif cmd in ("tablespace", "tablespaces"): print(f"⏳ 正在查询表空间使用情况...") result = query_tablespace(get_server_id()) if result.get("success"): via_func = result.get("_via_func", False) via_view = result.get("_via_view", False) is_fallback = result.get("_fallback", False) parse_failed = result.get("_parse_failed", False) if via_func and parse_failed: print("\n📊 表空间使用情况一览（通过 DBA 专用函数，原始格式）：") elif via_func: print("\n📊 表空间使用情况一览（通过 DBA 专用函数）：") elif via_view: print("\n📊 表空间使用情况一览（通过 DBA 授权 VIEW）：") elif is_fallback: print("\n📊 表空间使用情况（用户视图 — 无DBA权限，仅显示名称和剩余空间）：") print("─" * 65) else: print("\n📊 表空间使用情况一览：") print("─" * 110) print(result.get("data", "")) print("─" * 65 if is_fallback else "─" * 110) if not is_fallback: _check_ts_warnings(result.get("data", "")) else: print_result(result) else: print(f"未知命令: {cmd}，输入 help 查看帮助") # === capabilities command (version + command discovery) === _cap_parser = argparse.ArgumentParser(prog="oracle_skill", add_help=False) _cap_subparsers = _cap_parser.add_subparsers(dest="subcmd") p = _cap_subparsers.add_parser('capabilities') p.add_argument('--json', action='store_true', help='Output pure JSON') def cmd_capabilities(args): """ Return Skill version and all available command definitions (JSON format) Args: args: Command line arguments, including --json flag """ # Define all available commands commands = [ { "name": "analyze", "description": "Analyze stored procedure: source + dependency tables + nested procedures + triggers", "parameters": { "type": "object", "properties": { "schema": { "type": "string", "description": "Database schema name (e.g. bosnds3)" }, "procedure": { "type": "string", "description": "Stored procedure name (e.g. M_RETAIL_SUBMIT)" } }, "required": ["schema", "procedure"] } }, { "name": "list", "description": "List all stored procedures in specified schema", "parameters": { "type": "object", "properties": { "schema": { "type": "string", "description": "Database schema name" } }, "required": ["schema"] } }, { "name": "source", "description": "Get stored procedure source code", "parameters": { "type": "object", "properties": { "schema": { "type": "string", "description": "Database schema name" }, "procedure": { "type": "string", "description": "Stored procedure name" } }, "required": ["schema", "procedure"] } }, { "name": "describe", "description": "Query table structure (field names, types, comments)", "parameters": { "type": "object", "properties": { "schema": { "type": "string", "description": "Database schema name" }, "table": { "type": "string", "description": "Table name" } }, "required": ["schema", "table"] } }, { "name": "search", "description": "Search tables or stored procedures (supports fuzzy search)", "parameters": { "type": "object", "properties": { "schema": { "type": "string", "description": "Database schema name" }, "keyword": { "type": "string", "description": "Search keyword" } }, "required": ["schema", "keyword"] } }, { "name": "query", "description": "Execute SELECT query (only allows SELECT statements; production table queries must include WHERE filters and row limits)", "parameters": { "type": "object", "properties": { "schema": { "type": "string", "description": "Database schema name" }, "sql": { "type": "string", "description": "SQL SELECT query statement. For production tables, include explicit WHERE filters and ROWNUM/page limits; do not query whole tables." } }, "required": ["schema", "sql"] } }, { "name": "discover", "description": "Schema exploration: list all tables or search specific tables", "parameters": { "type": "object", "properties": { "schema": { "type": "string", "description": "Database schema name" }, "filter": { "type": "string", "description": "Optional, table name filter condition (e.g. M_%)" } }, "required": ["schema"] } }, { "name": "nl2sql", "description": "Natural language to SQL (requires DeepSeek API)", "parameters": { "type": "object", "properties": { "schema": { "type": "string", "description": "Database schema name" }, "question": { "type": "string", "description": "Natural language question (e.g. 'Query sales of Huadu Store 2 in May')" } }, "required": ["schema", "question"] } }, { "name": "perm", "description": "Get user data permission SQL (via stored procedure get_userspermsql)", "parameters": { "type": "object", "properties": { "schema": { "type": "string", "description": "Database schema name" }, "userid": { "type": "string", "description": "User ID" }, "table": { "type": "string", "description": "Table name (e.g. M_RETAIL)" } }, "required": ["schema", "userid", "table"] } }, { "name": "qperm", "description": "Quick permission SQL retrieval (simplified version)", "parameters": { "type": "object", "properties": { "schema": { "type": "string", "description": "Database schema name" }, "userid": { "type": "string", "description": "User ID" }, "table": { "type": "string", "description": "Table name" } }, "required": ["schema", "userid", "table"] } }, { "name": "login", "description": "Login to transit server and select a client", "parameters": { "type": "object", "properties": { "secretKey": { "type": "string", "description": "BOS secret key for authentication" }, "clientCode": { "type": "string", "description": "Optional client/server code to select (must be in user's clientList). If omitted, the transit server selects the first available client." } }, "required": ["secretKey"] } }, { "name": "logout", "description": "Logout and clear login credentials", "parameters": { "type": "object", "properties": {} } }, { "name": "status", "description": "Show current login status and selected client", "parameters": { "type": "object", "properties": {} } }, { "name": "clients", "description": "Refresh the current user's authorized client list and online status using the existing transit-server access token.", "parameters": { "type": "object", "properties": {} } }, { "name": "ops", "description": "Run one Oracle daily operations monitor item through transit server and agent.", "parameters": { "type": "object", "properties": { "item": { "type": "string", "description": "Monitor item, such as active_slow_sql, blocking_locks, tablespace, memory, io_waits, ora_errors" }, "clientCode": { "type": "string", "description": "Optional client/server code. If omitted, the current client is used." } }, "required": ["item"] } }, { "name": "ops_report", "description": "Generate an Oracle daily operations report for a client.", "parameters": { "type": "object", "properties": { "clientCode": { "type": "string", "description": "Optional client/server code. If omitted, the current client is used." } }, "required": [] } }, { "name": "inspection_report", "description": "Generate and archive a server inspection report for a client.", "parameters": { "type": "object", "properties": { "clientCode": { "type": "string", "description": "Optional client/server code. If omitted, the current client is used." }, "json": { "type": "boolean", "description": "Output raw JSON instead of Markdown summary." }, "html": { "type": "string", "description": "HTML output path. HTML is generated by default; use an empty value to write to the default outputs directory." }, "markdown": { "type": "string", "description": "Optional Markdown output path. Use an empty value to write to the default outputs directory." } }, "required": [] } }, { "name": "inspection_latest", "description": "Read the latest archived server inspection report for a client.", "parameters": { "type": "object", "properties": { "clientCode": { "type": "string", "description": "Optional client/server code. If omitted, the current client is used." }, "refresh": { "type": "boolean", "description": "Generate a fresh report before returning it." }, "json": { "type": "boolean", "description": "Output raw JSON instead of Markdown summary." }, "html": { "type": "string", "description": "HTML output path. HTML is generated by default; use an empty value to write to the default outputs directory." }, "markdown": { "type": "string", "description": "Optional Markdown output path. Use an empty value to write to the default outputs directory." } }, "required": [] } }, { "name": "inspection_get", "description": "Read an archived server inspection report by report ID.", "parameters": { "type": "object", "properties": { "id": { "type": "string", "description": "TS_INSPECTION_REPORT.ID" }, "json": { "type": "boolean", "description": "Output raw JSON instead of Markdown summary." }, "html": { "type": "string", "description": "HTML output path. HTML is generated by default; use an empty value to write to the default outputs directory." }, "markdown": { "type": "string", "description": "Optional Markdown output path. Use an empty value to write to the default outputs directory." } }, "required": ["id"] } }, { "name": "agent_update", "description": "Trigger a remote Agent self-update through transit-server. The Agent downloads the upgrade package from OSS.", "parameters": { "type": "object", "properties": { "clientCode": { "type": "string", "description": "Optional client/server code. If omitted, the current client is used." }, "timeout": { "type": "integer", "description": "Optional timeout seconds, default 300, max 600." } }, "required": [] } }, { "name": "device_register", "description": "Generate a local Ed25519 device key and submit a trusted-device registration request.", "parameters": {"type": "object", "properties": {"deviceName": {"type": "string", "description": "Optional device display name."}}}, "required": [] }, { "name": "device_login", "description": "Authenticate through an approved trusted device using a local Ed25519 private key.", "parameters": {"type": "object", "properties": {"clientCode": {"type": "string", "description": "Optional client/server code."}}}, "required": [] }, { "name": "awr_status", "description": "Read AWR authorization, readiness and latest generated report status without generating a report.", "parameters": { "type": "object", "properties": { "clientCode": {"type": "string", "description": "Optional client/server code."}, "json": {"type": "boolean", "description": "Output machine-readable JSON."} }, "required": [] } }, { "name": "awr_list", "description": "List AWR reports already generated and stored by the target Agent.", "parameters": { "type": "object", "properties": { "clientCode": {"type": "string", "description": "Optional client/server code."}, "json": {"type": "boolean", "description": "Output machine-readable JSON."} }, "required": [] } }, { "name": "awr_download", "description": "Download an Agent-generated AWR HTML report through transit-server without local rendering.", "parameters": { "type": "object", "properties": { "clientCode": {"type": "string", "description": "Client/server code."}, "date": {"type": "string", "description": "AWR report date in yyyyMMdd format."}, "output": {"type": "string", "description": "Optional local output path."} }, "required": ["clientCode", "date"] } }, { "name": "switch", "description": "Switch to a different client by code or title/name", "parameters": { "type": "object", "properties": { "clientCode": { "type": "string", "description": "Client code or title/name to switch to" } }, "required": ["clientCode"] } }, { "name": "capabilities", "description": "Show Skill version and all available command definitions", "parameters": { "type": "object", "properties": { "json": { "type": "boolean", "description": "Whether to output pure JSON format" } } } } ] # Build return result result = { "version": VERSION, "skill_name": "oracle-jump-query", "description": "Oracle Jump Query Skill - Query remote Oracle database via transit server", "transit_server": "https://ts.henlo.net", "default_schema": "bosnds3", "commands": commands, "metadata": { "total_commands": len(commands), "supported_actions": [ "analyze", "list", "source", "describe", "search", "query", "discover", "nl2sql", "perm", "qperm", "login", "logout", "status", "switch", "clients", "device_register", "device_login", "ops", "ops_report", "inspection_report", "inspection_latest", "inspection_get", "awr_status", "awr_list", "awr_download", "agent_update", "capabilities" ], "python_version": "3.6+", "dependencies": ["requests"] } } # Output format if hasattr(args, 'json') and args.json: # Pure JSON output (no log interference) print(json.dumps(result, ensure_ascii=False, indent=2)) else: # Friendly format output print("Oracle Jump Query Skill - Capabilities") print("=" * 60) print("Version: " + result['version']) print("Transit Server: " + result['transit_server']) print("Default Schema: " + result['default_schema']) print("\nAvailable Commands (" + str(result['metadata']['total_commands']) + "):") print("-" * 60) for i, cmd in enumerate(commands, 1): print(str(i) + ". " + cmd['name']) print(" Description: " + cmd['description']) if 'parameters' in cmd and 'properties' in cmd['parameters']: params = cmd['parameters']['properties'] if params: print(" Parameters: " + ', '.join(params.keys())) print() print("=" * 60) print("For machine-readable output, use: --json") def main(): # Global variable declarations for main function global DEFAULT_TIMEOUT, MAX_RETRIES, DEFAULT_SERVER_ID 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 == "version": print(f"Oracle Jump Query Skill v{VERSION}") if len(sys.argv) >= 3 and sys.argv[2] == "agent": args = [a for a in sys.argv[3:] if a not in ("--all", "--json")] cmd_agent_versions(args, all_clients="--all" in sys.argv[3:], as_json="--json" in sys.argv[3:]) return elif cmd == "agent": if len(sys.argv) >= 3: status = check_agent_status(sys.argv[2]) else: status = check_agent_status(get_server_id()) print(json.dumps(status, 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=DEFAULT_TIMEOUT * 2, max_retries=MAX_RETRIES) print_result(result) elif cmd == "list" and len(sys.argv) >= 3: result = query(get_server_id(), "list_procedures", sys.argv[2], timeout=30, max_retries=MAX_RETRIES) 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=DEFAULT_TIMEOUT, max_retries=MAX_RETRIES) 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=DEFAULT_TIMEOUT, max_retries=MAX_RETRIES) 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=DEFAULT_TIMEOUT, max_retries=MAX_RETRIES) print_result(result) elif cmd == "describe" and len(sys.argv) >= 4: result = query(get_server_id(), "describe_table", sys.argv[2], sys.argv[3], timeout=DEFAULT_TIMEOUT, max_retries=MAX_RETRIES) print_result(result) elif cmd == "discover" and len(sys.argv) >= 2: schema = sys.argv[2] if len(sys.argv) > 2 else "BOSNDS3" domain = sys.argv[3] if len(sys.argv) > 3 else "" result = query(get_server_id(), "schema_discovery", schema, domain, timeout=DEFAULT_TIMEOUT * 3, max_retries=MAX_RETRIES) print_result(result) elif cmd == "nl2sql" and len(sys.argv) >= 2: schema = sys.argv[2] if len(sys.argv) > 2 else "BOSNDS3" domain = sys.argv[3] if len(sys.argv) > 3 else "RETAIL" result = query(get_server_id(), "generate_nl2sql_schema", schema, domain, timeout=DEFAULT_TIMEOUT * 3, max_retries=MAX_RETRIES) print_result(result) elif cmd == "query" and len(sys.argv) >= 2: sql = " ".join(sys.argv[2:]) result = query(get_server_id(), "execute_query", "BOSNDS3", "", timeout=DEFAULT_TIMEOUT, max_retries=MAX_RETRIES, sql=sql) print_result(result) elif cmd == "perm" and len(sys.argv) >= 4: user_id = sys.argv[2] table_id = sys.argv[3] col_name = sys.argv[4] if len(sys.argv) > 4 else "" param = f"{user_id},{table_id}" if not col_name else f"{user_id},{table_id},{col_name}" result = query(get_server_id(), "get_user_perm", "BOSNDS3", param, timeout=30, max_retries=MAX_RETRIES) print_result(result) elif cmd == "qperm" and len(sys.argv) >= 5: user_id = sys.argv[2] table_id = sys.argv[3] sql = " ".join(sys.argv[4:]) result = query_with_permission(user_id, table_id, sql) print_result(result) elif cmd == "login" and len(sys.argv) >= 3: cmd_login(sys.argv[2], sys.argv[3] if len(sys.argv) >= 4 else "") elif cmd == "logout": cmd_logout() elif cmd == "device_register": cmd_device_register(" ".join(sys.argv[2:]) if len(sys.argv) >= 3 else "") elif cmd == "device_login": cmd_device_login(sys.argv[2] if len(sys.argv) >= 3 else "") elif cmd == "status": cmd_status() elif cmd == "ops" and len(sys.argv) >= 3: cmd_ops(sys.argv[2], sys.argv[3] if len(sys.argv) >= 4 else "") elif cmd == "ops_report": cmd_ops_report(sys.argv[2] if len(sys.argv) >= 3 else "") elif cmd == "sys_functions": args = [p for p in sys.argv[2:] if not p.startswith("--")] cmd_sys_functions(args[0] if args else "", "--json" in sys.argv[2:]) elif cmd == "inspection_report": args, as_json, html_path, markdown_path = _parse_report_cli_args(sys.argv[2:]) cmd_inspection_report(args[0] if args else "", as_json=as_json, html_path=html_path, markdown_path=markdown_path) elif cmd == "inspection_latest": refresh = "--refresh" in sys.argv[2:] args, as_json, html_path, markdown_path = _parse_report_cli_args([p for p in sys.argv[2:] if p != "--refresh"]) cmd_inspection_latest(args[0] if args else "", as_json=as_json, refresh=refresh, html_path=html_path, markdown_path=markdown_path) elif cmd == "inspection_get" and len(sys.argv) >= 3: args, as_json, html_path, markdown_path = _parse_report_cli_args(sys.argv[2:]) cmd_inspection_get(args[0], as_json=as_json, html_path=html_path, markdown_path=markdown_path) elif cmd == "awr_status": args = [p for p in sys.argv[2:] if p != "--json"] cmd_awr_status(args[0] if args else "", as_json="--json" in sys.argv[2:]) elif cmd == "awr_list": args = [p for p in sys.argv[2:] if p != "--json"] cmd_awr_list(args[0] if args else "", as_json="--json" in sys.argv[2:]) elif cmd == "awr_download" and len(sys.argv) >= 4: output_path = "" if "--output" in sys.argv[4:]: output_index = sys.argv.index("--output") if output_index + 1 < len(sys.argv): output_path = sys.argv[output_index + 1] cmd_awr_download(sys.argv[2], sys.argv[3], output_path=output_path, as_json="--json" in sys.argv[4:]) elif cmd == "agent_update": timeout = int(sys.argv[3]) if len(sys.argv) >= 4 and sys.argv[3].isdigit() else 300 cmd_agent_update(sys.argv[2] if len(sys.argv) >= 3 else "", timeout) elif cmd == "switch" and len(sys.argv) >= 3: cmd_switch(" ".join(sys.argv[2:])) elif cmd == "clients": cmd_clients() elif cmd in ("tablespace", "tablespaces"): print(f"⏳ 正在查询表空间使用情况...") result = query_tablespace(get_server_id()) if result.get("success"): via_func = result.get("_via_func", False) via_view = result.get("_via_view", False) is_fallback = result.get("_fallback", False) parse_failed = result.get("_parse_failed", False) if via_func and parse_failed: print("\n📊 表空间使用情况一览（通过 DBA 专用函数，原始格式）：") elif via_func: print("\n📊 表空间使用情况一览（通过 DBA 专用函数）：") elif via_view: print("\n📊 表空间使用情况一览（通过 DBA 授权 VIEW）：") elif is_fallback: print("\n📊 表空间使用情况（用户视图 — 无DBA权限，仅显示名称和剩余空间）：") print("─" * 65) else: print("\n📊 表空间使用情况一览：") print("─" * 110) print(result.get("data", "")) print("─" * 65 if is_fallback else "─" * 110) if not is_fallback: _check_ts_warnings(result.get("data", "")) else: print_result(result) elif cmd == "capabilities": args = _cap_parser.parse_args(["capabilities"] + sys.argv[2:]) cmd_capabilities(args) return else: print(__doc__) if __name__ == "__main__": main()