QingLong/AI/AiService/Test/TestStream.py

from fastapi import FastAPI
from fastapi.responses import StreamingResponse, PlainTextResponse
import time
import socket

app = FastAPI()

# 获取本机所有 IPv4 地址
def get_local_ips():
    ips = []
    hostname = socket.gethostname()
    try:
        # 获取所有 IP 地址
        addrs = socket.getaddrinfo(hostname, None, family=socket.AF_INET)  # 只获取 IPv4 地址
        for addr in addrs:
            ip = addr[4][0]
            if ip not in ips:
                ips.append(ip)
    except Exception as e:
        print(f"获取 IP 地址失败: {e}")
    return ips

# 模拟逐字生成数据的函数（SSE 格式）
def generate_data(text):
    for char in text:
        time.sleep(0.1)  # 模拟延迟
        yield f" {char}".encode("utf-8")  # 按照 SSE 格式返回数据

# 根路由，返回提示信息
@app.get("/")
def root():
    return PlainTextResponse("Hello ApiStream")

# 流式返回数据（SSE）
@app.get("/stream")
def stream_data():
    text = "你好，这是一个流式返回数据的示例。这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例这是一个流式返回数据的示例"
    return StreamingResponse(
        generate_data(text),
        media_type="text/event-stream",  # 使用 text/event-stream
        headers={
            "Cache-Control": "no-cache",  # 禁用缓存
            "Connection": "keep-alive",   # 保持连接
            "Access-Control-Allow-Origin": "*",  # 允许跨域
            "X-Accel-Buffering": "no"     # 禁用 Nginx 缓冲（如果有代理）
        }
    )

# 运行应用
if __name__ == "__main__":
    
    import uvicorn

    # 获取本机所有 IPv4 地址
    ips = get_local_ips()
    if not ips:
        print("无法获取本机 IP 地址，使用默认地址 127.0.0.1")
        ips = ["127.0.0.1"]

    # 打印所有 IP 地址
    print("服务将在以下 IP 地址上运行：")
    for ip in ips:
        print(f"http://{ip}:8000")

    # 启动 FastAPI 应用，绑定到所有 IP 地址
    uvicorn.run(app, host="0.0.0.0", port=8000)