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