51 lines
1.4 KiB
Python
51 lines
1.4 KiB
Python
from fastapi import FastAPI, Query
|
|
import httpx
|
|
import asyncio
|
|
import uvicorn
|
|
from main import send_sip_message
|
|
|
|
app = FastAPI()
|
|
|
|
async def main():
|
|
"""主函数内容"""
|
|
print("执行main函数...")
|
|
# 执行发送SIP消息
|
|
send_sip_message()
|
|
return {"message": "main function executed - SIP message sent"}
|
|
|
|
async def test():
|
|
"""测试函数内容"""
|
|
print("执行test函数...")
|
|
# 在这里添加你的test函数逻辑
|
|
return {"message": "test function executed"}
|
|
|
|
@app.api_route("/api/router", methods=["GET", "POST", "PUT", "DELETE", "PATCH", "HEAD", "OPTIONS"])
|
|
async def router(path: str = Query(...)):
|
|
"""路由处理器"""
|
|
try:
|
|
# 根据path参数执行不同的函数
|
|
if path == "floor":
|
|
result = await main()
|
|
elif path == "test":
|
|
result = await test()
|
|
else:
|
|
return {"error": f"Unknown path: {path}"}
|
|
|
|
return {"success": True, "data": result}
|
|
|
|
except Exception as e:
|
|
return {"success": False, "error": str(e)}
|
|
|
|
@app.api_route("/", methods=["GET", "POST", "PUT", "DELETE", "PATCH", "HEAD", "OPTIONS"])
|
|
async def root():
|
|
"""健康检查端点"""
|
|
return {"message": "FastAPI server is running"}
|
|
|
|
if __name__ == "__main__":
|
|
# 监听所有IPv6地址的3001端口
|
|
uvicorn.run(
|
|
"app:app",
|
|
host="::", # 监听所有IPv6地址
|
|
port=3001,
|
|
reload=True
|
|
) |