如何优化调用Tenable API拉取漏洞报告的Python脚本执行效率？

核心瓶颈分析 #

你的代码99.9%的耗时都集中在1000次串行调用Tenable API的环节，按2700秒总耗时算单次API请求平均耗时2.7秒，后续的遍历逻辑总耗时不会超过10毫秒，优化重点完全放在API调用侧即可。

优化方案 #

1. 优先用批量API查询（性能提升1000倍左右） #

Tenable的sc.analysis.vulns接口支持多条件组合查询，你不需要逐个IP发请求，把所有待查询的IP拼接为or连接的过滤条件，仅发1次请求就能拿到所有IP的扫描结果，总耗时直接降到单次API请求的水平（2~3秒）。

2. 不支持批量查询时用并发请求（性能提升20~50倍） #

如果受限于Tenable账号权限/接口限制没法批量查询，就用多线程并发调用API，按常规API限流阈值开20_{50个线程，总耗时可以降到50}150秒区间。

3. 冗余逻辑&细节优化 #

• 提前读取CSV的IP并去重，避免重复查询同一个IP浪费请求
• 原代码里的while循环遍历得到的scan_data、scan_count完全没有被使用，属于无效代码，要么直接删掉，要么和结果统计逻辑合并
• 读取CSV时用上下文管理器with open避免文件句柄泄漏

4. 缓存优化（多轮运行时性能提升近100%） #

如果脚本需要多次运行，把已经查询过的IP结果存在本地JSON/redis缓存，下次查询优先读缓存，不需要重复调用API，后续运行耗时直接降到毫秒级。

优化后代码示例 #

批量查询版本（优先推荐） #

import time
import csv

def get_ten(sc):
    now = time.time()
    # 读取所有IP并去重
    with open('full.csv', 'r', encoding='utf-8') as f:
        ip_set = {x[15] for x in csv.reader(f) if x[15] != 'PrivateIpAddress'}
    
    # 构造批量过滤条件
    filters = []
    ip_list = list(ip_set)
    for idx, ip in enumerate(ip_list):
        filters.append(('ip', '=', ip))
        if idx != len(ip_list) - 1:
            filters.append(('or',))
    
    # 单次请求拉取所有IP的漏洞数据
    all_vuln_data = sc.analysis.vulns(*filters, tool='sumseverity', sortDirection='desc')
    
    # 按IP分组统计结果（可根据实际业务需求调整）
    ip_stat = {}
    for item in all_vuln_data:
        ip = item['ip']
        if ip not in ip_stat:
            ip_stat[ip] = {'scan_data': [], 'scan_count': 0}
        ip_stat[ip]['scan_data'].append(item['count'])
        ip_stat[ip]['scan_count'] += int(item['count'])
    
    print(f"执行耗时：{time.time() - now}")
    return ip_stat

并发查询版本（批量不可用时用） #

import time
import csv
from concurrent.futures import ThreadPoolExecutor, as_completed

def query_ip_vuln(sc, ip):
    """单IP查询逻辑，作为并发任务"""
    return ip, sc.analysis.vulns(('ip', '=', ip), tool='sumseverity', sortDirection='desc')

def get_ten(sc):
    now = time.time()
    # 读取所有IP并去重
    with open('full.csv', 'r', encoding='utf-8') as f:
        ip_set = {x[15] for x in csv.reader(f) if x[15] != 'PrivateIpAddress'}
    
    ip_stat = {}
    # 并发数根据Tenable API限流规则调整，一般20~50是安全区间
    with ThreadPoolExecutor(max_workers=20) as pool:
        # 提交所有并发任务
        tasks = [pool.submit(query_ip_vuln, sc, ip) for ip in ip_set]
        # 处理任务返回结果
        for task in as_completed(tasks):
            ip, vulns = task.result()
            scan_data = []
            scan_count = 0
            for scan in vulns:
                count = scan['count']
                scan_data.append(count)
                scan_count += int(count)
            ip_stat[ip] = {'scan_data': scan_data, 'scan_count': scan_count}
    
    print(f"执行耗时：{time.time() - now}")
    return ip_stat

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