Compare commits

..

4 Commits

Author SHA1 Message Date
liangtianyu
25a016977c 不处理退出信号 2025-12-14 20:20:30 +08:00
mark.tian
8541dc77ec Merge branch 'main' of https://git.shiliulizi.cn/mark.tian/ai-talk-callback 2025-12-13 13:43:04 +08:00
mark.tian
666a073475 增加全局未处理异常处理 2025-12-13 13:42:29 +08:00
mark.tian
3337c751a5 增加全局未处理异常的处理 2025-12-13 13:41:59 +08:00

154
main.py
View File

@@ -1,19 +1,20 @@
from datetime import datetime
import os
import traceback
import signal
import tempfile
import time
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from contextlib import asynccontextmanager
from sqlalchemy import text
import redis.asyncio as redis
import subprocess
import sys
import os
from app.config import settings
from app.database import engine
from app.routes import router
from app.logger import LoggerManager, get_main_logger
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
import redis.asyncio as redis
from sqlalchemy import text
# 初始化日志系统
@@ -21,11 +22,36 @@ LoggerManager.setup_logging()
logger = get_main_logger()
def global_exception_handler(exc_type, exc_value, exc_traceback):
if issubclass(exc_type, KeyboardInterrupt):
sys.__excepthook__(exc_type, exc_value, exc_traceback)
return
# 获取格式化的异常信息
error_msg = "捕获到未处理的异常:\n"
error_msg += f"异常类型: {exc_type.__name__}\n"
error_msg += f"异常信息: {exc_value}\n"
error_msg += "堆栈跟踪:\n"
error_msg += "".join(traceback.format_tb(exc_traceback))
error_msg += f"异常时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n"
error_msg += "-" * 50 + "\n"
# 输出到控制台
print(error_msg)
# 输出到文件
logger.error(error_msg)
# 注册全局异常处理器
sys.excepthook = global_exception_handler
@asynccontextmanager
async def lifespan(app: FastAPI):
# 启动时初始化
logger.info("🚀 应用启动中...")
# Redis 连接对象
redis_client = None
@@ -52,10 +78,10 @@ async def lifespan(app: FastAPI):
# 执行 ping 命令验证连接
await redis_client.ping()
logger.info("✅ Redis 连接验证成功")
# 存储到应用状态中供其他组件使用
app.state.redis_client = redis_client
except Exception as redis_error:
logger.error(f"❌ Redis 连接验证失败: {redis_error}")
raise
@@ -72,7 +98,7 @@ async def lifespan(app: FastAPI):
finally:
# 关闭时清理
logger.info("🛑 应用关闭中...")
# 关闭 Redis 连接
if redis_client:
try:
@@ -80,7 +106,7 @@ async def lifespan(app: FastAPI):
logger.info("🔴 Redis 连接已关闭")
except Exception as e:
logger.warning(f"⚠️ 关闭 Redis 连接时出现警告: {e}")
logger.info("👋 应用已关闭")
@@ -137,17 +163,18 @@ async def health_check():
raise HTTPException(status_code=404, detail="Not Found")
logger.debug("💓 健康检查接口被访问")
return {"status": "healthy"}
# 全局变量存储进程
processes = []
def signal_handler(signum, frame):
"""信号处理器,用于优雅关闭所有服务"""
print(f"\n🛑 接收到信号 {signum},正在关闭所有服务...")
# 逆序关闭进程(最后启动的最先关闭)
for i, process in enumerate(reversed(processes)):
if process and process.poll() is None: # 进程仍在运行
@@ -161,86 +188,104 @@ def signal_handler(signum, frame):
process.kill() # 强制杀死进程
except Exception as e:
print(f"❌ 关闭进程时出错: {e}")
print("👋 所有服务已关闭")
sys.exit(0)
def start_all_services():
"""启动所有服务"""
global processes
print("\n🚀 AI Talk Callback API 一键启动所有服务")
print("=" * 60)
# 注册信号处理器
signal.signal(signal.SIGINT, signal_handler) # Ctrl+C
signal.signal(signal.SIGTERM, signal_handler) # 终止信号
# signal.signal(signal.SIGINT, signal_handler) # Ctrl+C
# signal.signal(signal.SIGTERM, signal_handler) # 终止信号
try:
# 1. 启动 Celery Worker
print("🌿 启动 Celery Worker...")
worker_process = subprocess.Popen([
sys.executable, "-m", "celery",
"-A", "app.celery_app",
"worker",
'--loglevel=info',
'--pool=solo',
'--concurrency=1',
'--time-limit=300', # 5分钟任务超时
'--soft-time-limit=240' # 4分钟软超时
])
worker_process = subprocess.Popen(
[
sys.executable,
"-m",
"celery",
"-A",
"app.celery_app",
"worker",
"--loglevel=info",
"--pool=solo",
"--concurrency=1",
"--time-limit=300", # 5分钟任务超时
"--soft-time-limit=240", # 4分钟软超时
]
)
processes.append(worker_process)
time.sleep(2) # 等待 Worker 启动
# 2. 启动 Celery Beat
print("\n📅 启动 Celery Beat...")
beat_process = subprocess.Popen([
sys.executable, "-m", "celery",
"-A", "app.celery_app",
"beat",
'--loglevel=info',
f'--schedule={os.path.join(tempfile.gettempdir(), "celerybeat-schedule")}'
])
beat_process = subprocess.Popen(
[
sys.executable,
"-m",
"celery",
"-A",
"app.celery_app",
"beat",
"--loglevel=info",
f'--schedule={os.path.join(tempfile.gettempdir(), "celerybeat-schedule")}',
]
)
processes.append(beat_process)
time.sleep(2) # 等待 Beat 启动
# 3. 启动 Flower 监控(如果启用)
if settings.flower_enabled:
print("\n📊 启动 Flower 监控服务...")
flower_cmd = [
sys.executable, "-m", "celery",
"-A", "app.celery_app",
sys.executable,
"-m",
"celery",
"-A",
"app.celery_app",
f"--broker={settings.celery_broker_url}",
"flower",
f"--port={settings.flower_port}"
f"--port={settings.flower_port}",
]
if settings.flower_basic_auth:
flower_cmd.append(f"--basic_auth={settings.flower_basic_auth}")
if settings.flower_url_prefix:
flower_cmd.append(f"--url_prefix={settings.flower_url_prefix}")
flower_process = subprocess.Popen(flower_cmd)
processes.append(flower_process)
time.sleep(2) # 等待 Flower 启动
# 4. 启动 FastAPI 应用
print("\n🚀 启动 FastAPI 应用...")
api_cmd = [
sys.executable, "-m", "uvicorn",
sys.executable,
"-m",
"uvicorn",
"main:app",
"--host", "0.0.0.0",
"--port", "8000"
"--host",
"0.0.0.0",
"--port",
"8000",
]
# 添加调试模式(如果配置了)
if settings.debug:
api_cmd.append("--reload")
api_process = subprocess.Popen(api_cmd)
processes.append(api_process)
time.sleep(2) # 等待 API 启动
print("\n" + "=" * 60)
print("✅ 所有服务启动完成!")
print("\n🌐 服务地址:")
@@ -251,7 +296,7 @@ def start_all_services():
print(f" 📊 监控界面: {settings.flower_url}")
print("\n💡 使用 Ctrl+C 可以优雅关闭所有服务")
print("=" * 60)
# 等待所有进程
while True:
# 检查是否有进程异常退出
@@ -260,15 +305,16 @@ def start_all_services():
print(f"❌ 进程 {i+1} 异常退出,退出码: {process.returncode}")
signal_handler(signal.SIGINT, None)
return
time.sleep(1) # 每秒检查一次
except KeyboardInterrupt:
signal_handler(signal.SIGINT, None)
except Exception as e:
print(f"❌ 启动服务时出错: {e}")
signal_handler(signal.SIGINT, None)
if __name__ == "__main__":
# 直接启动所有服务
start_all_services()