Python实现每2小时覆盖写入TXT文件（Shodan监控场景）

问题分析 #

你的代码存在几个核心问题导致定时功能失效：

1. api.stream.alert()是无限阻塞的数据流循环，后面的while循环永远无法执行，程序会一直卡在这个循环里处理实时数据。
2. 定时清空的逻辑完全脱离了数据流处理流程，没有和实时数据写入结合。
3. port变量的处理有风险：如果banner中没有port字段，会触发未定义变量的错误。

解决方案 #

核心思路是在处理实时Shodan数据流的过程中，定期检查时间阈值，每到2小时就清空文件，之后继续追加新数据。具体步骤：

• 设定2小时（7200秒）为一个周期
• 记录每次清空文件的时间点
• 每次处理新的banner数据前，判断是否已到周期节点，是则清空文件并更新时间点
• 安全处理port字段，避免异常
• 实时将有效数据追加写入文件

修正后的代码 #

import time
from shodan import Shodan
from shodan.helpers import get_ip
from shodan.cli.helpers import get_api_key

file_path = 'ips.txt'
# 设定2小时为一个周期（单位：秒）
CYCLE_DURATION = 7200

# 初始化API连接
api = Shodan(get_api_key())
# 记录上次清空文件的时间
last_reset_time = time.time()

# 先清空一次初始文件
with open(file_path, 'w') as f:
    f.write('')

# 处理Shodan实时告警数据流
for banner in api.stream.alert():
    # 检查是否到达周期，需要清空文件
    if time.time() - last_reset_time >= CYCLE_DURATION:
        with open(file_path, 'w') as f:
            f.write('')
        last_reset_time = time.time()
        print("已清空文件，开始新周期的数据记录")
    
    # 安全获取port字段，避免KeyError
    port = banner.get('port')
    if not port:
        continue  # 没有port字段的banner直接跳过
    
    ip = get_ip(banner)
    data = f"{ip}:{port}"
    
    # 追加写入数据
    with open(file_path, 'a') as f:
        f.write(f"{data}\n")

代码说明 #

• 周期控制：通过last_reset_time记录每次清空文件的时间，每次处理数据前判断是否超过2小时，是则清空文件并更新时间。
• 异常防护：用banner.get('port')替代直接索引，避免没有port字段时的报错，没有port的banner直接跳过。
• 初始清空：程序启动时先清空一次文件，确保第一个周期的数据是全新的。
• 实时写入：每获取到有效banner数据就立即追加写入文件，保证数据实时性。

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