Update 测绘站采集.py

py/测绘站采集.py

@@ -36,10 +36,9 @@ from fake_useragent import UserAgent  # 需要先安装：pip install fake-usera
 os.makedirs('playlist', exist_ok=True)
 
 # 配置参数
-DELAY_RANGE = (3, 6)  # 随机延迟时间范围（秒）
-MAX_RETRIES = 3  # 最大重试次数
-REQUEST_TIMEOUT = 10  # 请求超时时间（秒）
-
+DELAY_RANGE = (3, 6)     # 随机延迟时间范围（秒）
+MAX_RETRIES = 3          # 最大重试次数
+REQUEST_TIMEOUT = 10     # 请求超时时间（秒）
 
 def get_random_header():
     """生成随机请求头"""
@@ -49,7 +48,6 @@ def get_random_header():
         'Referer': 'https://fofa.info/'
     }
 
-
 def safe_request(url):
     """带重试机制的请求函数"""
     for attempt in range(MAX_RETRIES):
@@ -74,25 +72,32 @@ def safe_request(url):
             return response.text
 
         except Exception as e:
-            print(f"请求失败（第{attempt + 1}次重试）: {str(e)}")
+            print(f"请求失败（第{attempt+1}次重试）: {str(e)}")
             if attempt == MAX_RETRIES - 1:
                 raise
 
-
 def validate_video(url, mcast):
-    """验证视频流链接连通性"""
+    """验证视频流有效性"""
     video_url = f"{url}/rtp/{mcast}"
     print(f"正在验证: {video_url}")
 
     try:
-        response = requests.get(video_url, headers=get_random_header(), timeout=REQUEST_TIMEOUT)
+        # 发送请求，尝试下载 1 千字节的数据
+        response = requests.get(video_url, headers=get_random_header(), timeout=REQUEST_TIMEOUT, stream=True)
         response.raise_for_status()
-        return True
+
+        content_length = 0
+        for chunk in response.iter_content(chunk_size=1024):
+            if chunk:
+                content_length += len(chunk)
+                if content_length >= 64:
+                    break
+        return content_length >= 16
+
     except Exception as e:
         print(f"视频验证异常: {str(e)}")
         return False
 
-
 def main():
     # 获取需要处理的文件列表
     files = [f.split('.')[0] for f in os.listdir('rtp') if f.endswith('.txt')]
@@ -146,10 +151,10 @@ def main():
                     dst.write(modified + '\n')
             print(f"已生成播放列表: {output_file}")
 
-
 if __name__ == '__main__':
     main()
-    
+
+
 print('对playlist文件夹里面的所有txt文件进行去重处理')
 def remove_duplicates_keep_order(folder_path):
     for filename in os.listdir(folder_path):
@@ -198,6 +203,7 @@ def get_ip_key(url):
 
 # 设置固定的文件夹路径
 folder_path = 'playlist'
+
 # 确保文件夹路径存在
 if not os.path.isdir(folder_path):
     print("指定的文件夹不存在。")