Requests 流式请求

import requests

# 设置stream=True启用流式请求
response = requests.get('https://httpbin.org/stream/20', stream=True)

# 重要：在使用流式响应时，需要手动检查状态码
if response.status_code == 200:
    # 使用iter_content()或iter_lines()方法迭代处理数据
    for chunk in response.iter_content(chunk_size=1024):
        # 处理每个数据块
        if chunk:  # 过滤掉keep-alive空块
            print(f"收到 {len(chunk)} 字节数据")
            # 在这里处理chunk数据

    print("流式请求完成！")
else:
    print(f"请求失败，状态码: {response.status_code}")

# 重要：在使用流式请求后，应该手动关闭响应
response.close()

import requests

# 使用with语句自动管理连接
with requests.get('https://httpbin.org/stream/5', stream=True) as response:
    # 检查响应状态
    if response.status_code == 200:
        for line in response.iter_lines():
            if line:  # 过滤空行
                print(f"收到行数据: {line.decode('utf-8')}")
    else:
        print(f"请求失败: {response.status_code}")

# with块结束后，响应会自动关闭，无需手动调用response.close()

import requests
import json

def process_json_stream(url):
    """
    处理JSON流式API响应
    示例API返回每行一个JSON对象
    """
    try:
        response = requests.get(url, stream=True)
        response.raise_for_status()

        for line in response.iter_lines():
            if line:  # 过滤空行
                try:
                    # 解析JSON数据
                    data = json.loads(line.decode('utf-8'))

                    # 处理数据
                    print(f"收到数据: {data}")

                    # 这里可以添加业务逻辑
                    # 例如：存储到数据库、实时分析等

                except json.JSONDecodeError as e:
                    print(f"JSON解析错误: {e}")
                    continue

    except requests.exceptions.RequestException as e:
        print(f"请求失败: {e}")
    finally:
        if 'response' in locals():
            response.close()

# 使用示例
if __name__ == "__main__":
    # 使用一个返回JSON流的测试API
    test_url = "https://httpbin.org/stream/10"
    process_json_stream(test_url)

import requests
import re

def listen_to_sse(url):
    """
    监听服务器推送事件(SSE)
    SSE格式: data: {json}\n\n
    """
    try:
        # 设置长时间超时
        response = requests.get(url, stream=True, timeout=30)
        response.raise_for_status()

        print("开始监听SSE流...")

        buffer = ""
        for chunk in response.iter_content(chunk_size=1024, decode_unicode=True):
            if chunk:
                buffer += chunk

                # 查找完整的事件（以\n\n分隔）
                while "\n\n" in buffer:
                    event, buffer = buffer.split("\n\n", 1)

                    # 提取data字段
                    data_match = re.search(r'data:\s*(.+)$', event, re.MULTILINE)
                    if data_match:
                        data_str = data_match.group(1)
                        print(f"收到事件: {data_str}")

                        # 这里可以处理事件数据
                        # 例如：更新UI、触发回调等

    except requests.exceptions.Timeout:
        print("连接超时，尝试重新连接...")
    except requests.exceptions.RequestException as e:
        print(f"连接错误: {e}")
    finally:
        if 'response' in locals():
            response.close()

# 使用示例（需要实际的SSE端点）
# if __name__ == "__main__":
#     sse_url = "https://example.com/api/events"
#     listen_to_sse(sse_url)

import requests
import json
import time

class StreamListener:
    """流式API监听器"""

    def __init__(self, api_url, auth_token):
        self.api_url = api_url
        self.headers = {
            'Authorization': f'Bearer {auth_token}',
            'Content-Type': 'application/json'
        }
        self.running = False

    def on_data(self, data):
        """处理收到的数据"""
        print(f"收到数据: {data.get('text', 'No text')[:50]}...")
        # 这里可以添加自定义处理逻辑
        return True

    def on_error(self, error):
        """处理错误"""
        print(f"流错误: {error}")
        return False

    def start(self):
        """开始监听流"""
        self.running = True

        while self.running:
            try:
                # 设置流式请求
                response = requests.post(
                    self.api_url,
                    headers=self.headers,
                    json={'track': 'python'},  # 跟踪关键词
                    stream=True,
                    timeout=90
                )

                print("连接成功，开始接收数据流...")

                # 处理数据流
                for line in response.iter_lines():
                    if not self.running:
                        break

                    if line:
                        try:
                            data = json.loads(line.decode('utf-8'))

                            # 调用数据处理方法
                            if not self.on_data(data):
                                break

                        except json.JSONDecodeError:
                            print(f"无法解析JSON: {line[:100]}")
                            continue

            except requests.exceptions.Timeout:
                print("连接超时，5秒后重试...")
                time.sleep(5)
                continue
            except requests.exceptions.RequestException as e:
                print(f"请求错误: {e}")
                if not self.on_error(str(e)):
                    break
                time.sleep(10)  # 等待后重试

    def stop(self):
        """停止监听"""
        self.running = False
        print("停止监听流...")

# 使用示例（需要实际的API凭证）
# if __name__ == "__main__":
#     listener = StreamListener(
#         api_url="https://api.twitter.com/2/tweets/search/stream",
#         auth_token="YOUR_BEARER_TOKEN"
#     )
#
#     try:
#         listener.start()
#     except KeyboardInterrupt:
#         listener.stop()