运行Splunk查询Python脚本无法获取正确单值结果求助

我用Python脚本执行Splunk查询，脚本能正常运行，但无法得到预期的单值结果（比如492）。尝试过Splunk官方库也没成功，求帮忙解决。代码如下：

import urllib
import httplib2 #import library
import json
import pprint
import time
import re
from xml.dom import minidom

searchquery = 'search index="movable_in" sourcetype="movable:in:assets" | stats avg(exposure_score)'

myhttp = httplib2.Http()
baseurl = 'https://xxxx.splunkxxx.com:8089'
usernamesp = 'xxxx'
passwordsp = 'xxxx'


def get_splunk_result(searchquery):
    # Step 1: Get a session key
    servercontent = myhttp.request(f'{baseurl}/services/auth/login', 'POST', headers={},
                                   body=urllib.parse.urlencode({'username': usernamesp, 'password': passwordsp}))[1]
    sessionkey = minidom.parseString(servercontent).getElementsByTagName('sessionKey')[0].childNodes[0].nodeValue
    # print ("====>sessionkey:  %s  <====" % sessionkey)
    sid = ''
    # ------------------
    if not searchquery.startswith('search'):
        searchquery = f'search {searchquery}'

    # Step 2: Get a sid with the search query
    i = 0
    while True:
        time.sleep(1)
        try:
            searchjob = myhttp.request(f'{baseurl}/services/search/jobs', 'POST',
                                       headers={F'Authorization': F'Splunk %s' % sessionkey},
                                       body=urllib.parse.urlencode({'search': searchquery}))[1]
            sid = minidom.parseString(searchjob).getElementsByTagName('sid')[0].childNodes[0].nodeValue
            break
        except:
            i = i + 1
            # print(i)
            if (i > 30): break
    # print("====>SID:  %s  <====" % sid)
    # Step 3: Get search status

    myhttp.add_credentials(usernamesp, passwordsp)
    servicessearchstatusstr = '/services/search/jobs/%s/' % sid

    isnotdone = True
    while isnotdone:
        searchstatus = myhttp.request(f'{baseurl}{servicessearchstatusstr}', 'GET')[1]
        isdonestatus = re.compile('isDone">(0|1)')
        strstatus = str(searchstatus)
        isdonestatus = isdonestatus.search(strstatus).groups()[0]
        if (isdonestatus == '1'):
            isnotdone = False
# Step 4: Get the search result

    services_search_results_str = '/services/search/jobs/%s/results?output_mode=json_rows&count=0' % sid
    searchresults = myhttp.request(f'{baseurl}{services_search_results_str}', 'GET')[1]

    searchresults = json.loads(searchresults)
    # searchresults = splunk_result(searchresults)
    return searchresults


output = get_splunk_result(searchquery)
print(output)

问题原因与修复方案 #

1. 核心问题：未提取单值结果 #

当前脚本返回的是完整的json_rows格式结构，包含字段定义和结果行，但没有从中提取具体的数值。需要从返回的rows数组中取出唯一的统计值。

2. 关键修复点 #

修改结果解析部分，提取单值：

# 替换原返回结果的处理代码
searchresults = json.loads(searchresults)
# 提取单值：取第一行第一列的数值
if searchresults.get('rows') and len(searchresults['rows']) > 0:
    single_value = searchresults['rows'][0][0]
    return single_value
else:
    return None  # 无结果时返回None

3. 其他优化建议 #

• 状态检查更可靠：用JSON格式获取任务状态，避免正则匹配字符串的不稳定：

  # 替换原状态检查循环
  while isnotdone:
      searchstatus = myhttp.request(f'{baseurl}{servicessearchstatusstr}?output_mode=json', 'GET',
                                    headers={'Authorization': f'Splunk {sessionkey}'})[1]
      status_data = json.loads(searchstatus)
      if status_data['entry'][0]['content']['isDone']:
          isnotdone = False
      time.sleep(1)
  

• 移除重复认证：已经用session key做认证，不需要再调用myhttp.add_credentials，避免冲突。
• 精准异常捕获：不要用except:捕获所有异常，改为捕获具体类型（如HTTPError、ExpatError），方便排查问题。

完整修复后的代码 #

import urllib
import httplib2
import json
import time
from xml.dom import minidom

searchquery = 'search index="movable_in" sourcetype="movable:in:assets" | stats avg(exposure_score)'

myhttp = httplib2.Http()
baseurl = 'https://xxxx.splunkxxx.com:8089'
usernamesp = 'xxxx'
passwordsp = 'xxxx'


def get_splunk_result(searchquery):
    # Step 1: 获取会话密钥
    servercontent = myhttp.request(f'{baseurl}/services/auth/login', 'POST', headers={},
                                   body=urllib.parse.urlencode({'username': usernamesp, 'password': passwordsp}))[1]
    sessionkey = minidom.parseString(servercontent).getElementsByTagName('sessionKey')[0].childNodes[0].nodeValue

    sid = ''
    # 确保查询以search开头
    if not searchquery.startswith('search'):
        searchquery = f'search {searchquery}'

    # Step 2: 创建搜索任务获取SID
    i = 0
    while True:
        time.sleep(1)
        try:
            searchjob = myhttp.request(f'{baseurl}/services/search/jobs', 'POST',
                                       headers={'Authorization': f'Splunk {sessionkey}'},
                                       body=urllib.parse.urlencode({'search': searchquery}))[1]
            sid = minidom.parseString(searchjob).getElementsByTagName('sid')[0].childNodes[0].nodeValue
            break
        except Exception as e:
            i += 1
            if i > 30:
                raise TimeoutError("创建搜索任务超时") from e

    # Step 3: 等待搜索完成
    servicessearchstatusstr = f'/services/search/jobs/{sid}/'
    isnotdone = True
    while isnotdone:
        searchstatus = myhttp.request(f'{baseurl}{servicessearchstatusstr}?output_mode=json', 'GET',
                                      headers={'Authorization': f'Splunk {sessionkey}'})[1]
        status_data = json.loads(searchstatus)
        if status_data['entry'][0]['content']['isDone']:
            isnotdone = False
        time.sleep(1)

    # Step 4: 获取并解析结果
    services_search_results_str = f'/services/search/jobs/{sid}/results?output_mode=json_rows&count=0'
    searchresults = myhttp.request(f'{baseurl}{services_search_results_str}', 'GET',
                                   headers={'Authorization': f'Splunk {sessionkey}'})[1]

    searchresults = json.loads(searchresults)
    # 提取单值结果
    if searchresults.get('rows') and len(searchresults['rows']) > 0:
        return searchresults['rows'][0][0]
    return None


output = get_splunk_result(searchquery)
print(output)

内容的提问来源于stack exchange，提问作者user3754136