Python多线程配合requests爬取IPv6设备数据运行报错排查求助

问题排查及修复方案 #

1. 基础语法错误 #

• 导入模块拼写错误：代码第一行导入的jso是笔误，正确应为json，该问题会直接导致程序启动报错
  修复：将导入语句改为

  import requests, queue, threading, urllib3, json, pandas as pd, os, time, datetime, inspect
  

2. IP地址读取错误 #

• 从ip_list.txt读取IP时，每一行末尾自带换行符\n，拼接进IPv6请求URL后会导致地址格式非法，无法建立连接
  修复：读取IP时去掉首尾空白字符，修改读取代码为：

  with open (root+ "/ip_list.txt") as ips:
      device_ip = [ip.strip() for ip in ips if ip.strip()]
  

3. 请求逻辑缺陷 #

• 所有requests请求未设置超时时间，部分无法连通的IP会导致线程永久阻塞；同时未校验登录请求的返回结果，直接复用未验证的Cookie请求Dashboard，大概率返回401无权限
  修复：给所有请求添加timeout参数，同时增加登录结果校验（判断规则可根据设备实际返回调整）：

  # 登录请求示例
  r = requests.post(
      f"https://[{device_ip}]/?q=index.login&mimosa_ajax=1",
      {"username":"configure", "password":self.pas},
      verify=False,
      timeout=10
  )
  # 新增登录校验，可根据设备实际返回调整判断条件
  if r.status_code != 200 or "login_success" not in r.text:
      st2 = "Login Failed"
      self.out_q.put([device_ip,st2,"",""])
      return
  

4. 异常处理逻辑漏洞 #

• finally块的触发规则是无论try块是否抛出异常、是否return都会执行，目前代码中如果登录请求抛出ConnectionError，已经将st设为Offline并return，虽然不会触发后续Dashboard请求，但逻辑存在隐患，建议将Dashboard请求的逻辑移到try块内部，不要放在finally里。

5. Pandas写入逻辑错误 #

• 写DataFrame时self.df[device_details[0]] = device_details是新增列的逻辑，不是新增行，会导致输出的表格格式完全错误
  修复：修改为新增行的写法：

  self.df.loc[len(self.df)] = device_details
  

6. 队列等待逻辑缺陷 #

• 主函数只等待了任务队列q的结束，没有等待输出队列out_q的任务全部执行完，就直接输出df并保存，会导致部分设备数据丢失
  修复：在q.join()后新增out_q.join()

7. 网络环境校验 #

• 确认运行脚本的主机已配置可用的IPv6地址，且到目标设备的443端口无防火墙、安全组拦截，可手动用命令测试单个设备的登录接口是否正常：
  curl -k -X POST -d "username=configure&password=你的密码" "https://[设备IPv6地址]/?q=index.login&mimosa_ajax=1"

修复后完整代码 #

import requests, queue, threading, urllib3, json, pandas as pd, os, time, datetime, inspect
num_threads = 2
root = os.path.dirname(os.path.abspath(inspect.getfile(inspect.currentframe())))
with open (root+ "/ip_list.txt") as ips:
    device_ip = [ip.strip() for ip in ips if ip.strip()]
class Writer_Worker(threading.Thread):
    def __init__(self, queue, df, *args, **kwargs):
        if not queue:
            print("Device Queue not specified")
            exit(1)
        self.out_q = queue
        self.df = df
        super().__init__(*args, **kwargs)
    def run(self):
        while True:
            try:
                device_details = self.out_q.get(timeout=3)
            except queue.Empty:
                return
            self.df.loc[len(self.df)] = device_details
            self.out_q.task_done()
class Worker(threading.Thread):
    def __init__(self, queue, out_queue, device_password, *args, **kwargs):
        if not queue:
            print("Device Queue not specified")
            exit(1)
        self.queue = queue
        self.pas = device_password
        self.out_q = out_queue
        super().__init__(*args, **kwargs)
    def run(self):
        while True:
            try:
                device_ip = self.queue.get(timeout=3)
            except queue.Empty:
                return
            self.connect_to_device_and_process(device_ip)
            self.queue.task_done()
    def connect_to_device_and_process(self, device_ip):
        st = "Online"
        try:
            r = requests.post(
                f"https://[{device_ip}]/?q=index.login&mimosa_ajax=1",
                {"username":"configure", "password":self.pas},
                verify=False,
                timeout=10
            )
            # 校验登录是否成功
            if r.status_code != 200:
                st2 = "Login Failed"
                self.out_q.put([device_ip,st2,"",""])
                return
            # 请求Dashboard
            r = requests.get(
                f"https://[{device_ip}]/cgi/dashboard.php",
                cookies=r.cookies,
                verify=False,
                timeout=10
            )
            if r.status_code == 401:
                st2 = "Password Error"
                self.out_q.put([device_ip,st2,"",""])
            else:
                data = json.loads(r.content.decode())
                output5 = data ['config'] ['Spectrum_Power']
                auto_power = output5['Auto_Power'].replace('2', 'Max Power').replace('1', 'Min Power').replace('0', 'off')
                self.out_q.put([device_ip,st,auto_power,output5['AutoConfig']])
        except requests.exceptions.ConnectionError:
                st = "Offline"
                self.out_q.put([device_ip,st,"",""])
                return
def main():
    start = time.time()
    urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
    pas = input("Enter Device Password:")
    df =pd.DataFrame(columns = ["IP","Status","Auto_Power","AutoChannel"])
    q = queue.Queue(len(device_ip))
    for ip in device_ip:
        q.put_nowait(ip)
    out_q = queue.Queue(len(device_ip))
    Writer_Worker(out_q, df).start()
    for _ in range(num_threads):
        Worker(q, out_q, pas).start()
    q.join()
    out_q.join()
    print(df)
    df.to_excel('iBridge_C5x_Audit_Report.xlsx', sheet_name='Detail', index = False)
    
if __name__ == "__main__":
    main()

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