不处理退出信号

Merge branch 'main' of https://git.shiliulizi.cn/mark.tian/ai-talk-callback
增加全局未处理异常处理
2025-12-14 20:20:30 +08:00 · 2025-12-13 13:43:04 +08:00 · 2025-12-13 13:42:29 +08:00 · 2025-12-13 13:41:59 +08:00 · 2025-12-13 09:27:44 +08:00 · 2025-12-13 09:22:19 +08:00
23 changed files with 2696 additions and 817 deletions
--- a/.env.example
+++ b/.env.example
@@ -1,16 +1,26 @@
 # 数据库配置
 DATABASE_URL=postgresql+asyncpg://user:password@localhost:5432/ai_talk_callback_db
-# Redis配置
+# Celery配置
-REDIS_URL=redis://localhost:6379/0
+CELERY_BROKER_URL=redis://localhost:6379/0
-REDIS_LOCK_TIMEOUT=30
+CELERY_RESULT_BACKEND=redis://localhost:6379/0
-REDIS_LOCK_RETRY_DELAY=0.1
+
-REDIS_LOCK_MAX_RETRIES=100
+# 任务配置
 ENABLED_PUSH_DATA_TO_DTC_TASK=false
 ENABLED_CALL_API_TASK=false
 # Flower监控配置
 FLOWER_ENABLED=true
 FLOWER_PORT=5555
 FLOWER_BASIC_AUTH=admin:admin123
 FLOWER_URL_PREFIX=
 FLOWER_URL=http://localhost:5555
 # 业务配置
 COUNT_THRESHOLD=3
 EXTERNAL_API_ENABLED=false
 EXTERNAL_API_RETRY_MAX=3
-EXTERNAL_API_URL=https://jeep-api.d2c.stlassac.com/api/openapi/customerApi/aiTaskResultFail
+EXTERNAL_API_URL=your_external_api_url_here
 # 日志配置
 LOG_LEVEL=INFO
@@ -20,6 +30,10 @@ LOG_BACKUP_COUNT=5
 LOG_FORMAT=%(asctime)s - %(name)s - %(levelname)s - %(message)s
 LOG_DATE_FORMAT=%Y-%m-%d %H:%M:%S
 # API配置
 API_AUTHORIZATION_TOKEN=your_bearer_token_here
 # 应用配置
 APP_NAME=AI Talk Callback API
 ENVIRONMENT=production # 环境: production, test, development
 DEBUG=false
--- a/.gitignore
+++ b/.gitignore
@@ -0,0 +1,159 @@
 # Byte-compiled / optimized / DLL files
 __pycache__/
 *.py[cod]
 *$py.class
 # C extensions
 *.so
 # Distribution / packaging
 .Python
 build/
 develop-eggs/
 dist/
 downloads/
 eggs/
 .eggs/
 lib/
 lib64/
 parts/
 sdist/
 var/
 wheels/
 share/python-wheels/
 *.egg-info/
 .installed.cfg
 *.egg
 MANIFEST
 # PyInstaller
 #  Usually these files are written by a python script from a template
 #  before PyInstaller builds the exe, so as to inject date/other infos into it.
 *.manifest
 *.spec
 # Installer logs
 pip-log.txt
 pip-delete-this-directory.txt
 # Unit test / coverage reports
 htmlcov/
 .tox/
 .nox/
 .coverage
 .coverage.*
 .cache
 nosetests.xml
 coverage.xml
 *.cover
 *.py,cover
 .hypothesis/
 .pytest_cache/
 cover/
 # Translations
 *.mo
 *.pot
 # Django stuff:
 *.log
 local_settings.py
 db.sqlite3
 db.sqlite3-journal
 # Flask stuff:
 instance/
 .webassets-cache
 # Scrapy stuff:
 .scrapy
 # Sphinx documentation
 docs/_build/
 # PyBuilder
 .pybuilder/
 target/
 # Jupyter Notebook
 .ipynb_checkpoints
 # IPython
 profile_default/
 ipython_config.py
 # pyenv
 #   For a library or package, you might want to ignore these files since the code is
 #   intended to run in multiple environments; otherwise, check them in:
 # .python-version
 # pipenv
 #   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
 #   However, in case of collaboration, if having platform-specific dependencies or dependencies
 #   having no cross-platform support, pipenv may install dependencies that don't work, or not
 #   install all needed dependencies.
 #Pipfile.lock
 # PEP 582; used by e.g. github.com/David-OConnor/pyflow
 __pypackages__/
 # Celery stuff
 celerybeat-schedule
 celerybeat.pid
 # SageMath parsed files
 *.sage.py
 # Environments
 .env
 .venv
 env/
 venv/
 ENV/
 env.bak/
 venv.bak/
 # Spyder project settings
 .spyderproject
 .spyproject
 # Rope project settings
 .ropeproject
 # mkdocs documentation
 /site
 # mypy
 .mypy_cache/
 .dmypy.json
 dmypy.json
 # Pyre type checker
 .pyre/
 # pytype static type analyzer
 .pytype/
 # Cython debug symbols
 cython_debug/
 logs
 *.log
 npm-debug.log*
 yarn-debug.log*
 yarn-error.log*
 pnpm-debug.log*
 lerna-debug.log*
 node_modules
 .DS_Store
 dist
 dist-ssr
 coverage
 *.local
 app.db
 .vscode
 celerybeat-schedule.*
--- a/JSON_LOG_EXAMPLES.md
+++ b/JSON_LOG_EXAMPLES.md
@@ -1,350 +0,0 @@
 # JSON格式日志输出示例
 ## 优化后的 `log_callback_request` 方法 - JSON格式日志
 ### 1. site_id 日志输出
 ```json
 📝 site_id: "test-site-123"
 ```
 ### 2. 请求头日志输出（JSON格式）
 ```json
 📋 请求头: {
  "content-type": "application/json",
  "authorization": "***REDACTED***",
  "x-api-key": "***REDACTED***",
  "user-agent": "python-httpx/0.25.2",
  "accept": "application/json",
  "content-length": "1024",
  "host": "localhost:8000"
 }
 ```
 ### 3. 请求体日志输出（JSON格式）
 ```json
 📄 请求体: {
  "count": 2,
  "data": [
    {
      "bill": 100,
      "duration": 60,
      "callid": "test-call-123",
      "calldate": "2024-01-01 12:00:00",
      "number": "13800138000",
      "numberid": "number-123",
      "customer_id": "customer-123",
      "status": 0,
      "status_str": "失败",
      "user_id": "user-123",
      "type": 1,
      "number_data": {
        "number": "13800138000",
        "province": "广东",
        "city": "深圳",
        "operator": "移动"
      },
      "group": {
        "id": 1,
        "name": "测试组"
      },
      "task": {
        "id": "task-123",
        "name": "测试任务"
      },
      "user": {
        "id": "user-123",
        "name": "测试用户"
      },
      "customer_data": {
        "name": "测试客户",
        "email": "test@example.com",
        "company": "测试公司",
        "extra": "额外信息"
      }
    }
  ]
 }
 ```
 ### 4. server_ip 日志输出（JSON格式）
 ```json
 🏠 server_ip: {
  "ip": "0.0.0.0",
  "port": 8000
 }
 ```
 ### 5. client_ip 日志输出（JSON格式）
 ```json
 🖥️ client_ip: {
  "ip": "127.0.0.1",
  "port": 52345
 }
 ```
 ### 6. callback_data 日志输出（JSON格式）
 ```json
 📦 callback_data: {
  "count": 2,
  "data_count": 1,
  "data_sample": {
    "bill": 100,
    "duration": 60,
    "callid": "test-call-123",
    "calldate": "2024-01-01 12:00:00",
    "number": "13800138000",
    "numberid": "number-123",
    "customer_id": "customer-123",
    "status": 0,
    "status_str": "失败",
    "user_id": "user-123",
    "type": 1,
    "number_data": {
      "number": "13800138000",
      "province": "广东",
      "city": "深圳",
      "operator": "移动"
    },
    "group": {
      "id": 1,
      "name": "测试组"
    },
    "task": {
      "id": "task-123",
      "name": "测试任务"
    },
    "user": {
      "id": "user-123",
      "name": "测试用户"
    },
    "customer_data": {
      "name": "测试客户",
      "email": "test@example.com",
      "company": "测试公司",
      "extra": "额外信息"
    }
  }
 }
 ```
 ## 完整的请求处理日志流程
 ### 单次请求的完整日志输出
 ```
 2024-12-02 16:30:15 - routes - INFO - 🔥 收到AI Talk回调请求: siteId=test-site-123, count=2, data_count=1
 2024-12-02 16:30:15 - routes - INFO - 📝 site_id: "test-site-123"
 2024-12-02 16:30:15 - routes - INFO - 📋 请求头: {
  "content-type": "application/json",
  "authorization": "***REDACTED***",
  "user-agent": "python-httpx/0.25.2",
  "accept": "application/json",
  "content-length": "1024",
  "host": "localhost:8000"
 }
 2024-12-02 16:30:15 - routes - INFO - 📄 请求体: {
  "count": 2,
  "data": [
    {
      "bill": 100,
      "duration": 60,
      "callid": "test-call-123",
      "calldate": "2024-01-01 12:00:00",
      "number": "13800138000",
      "numberid": "number-123",
      "customer_id": "customer-123",
      "status": 0,
      "status_str": "失败",
      "user_id": "user-123",
      "type": 1,
      "number_data": {
        "number": "13800138000",
        "province": "广东",
        "city": "深圳",
        "operator": "移动"
      },
      "group": {
        "id": 1,
        "name": "测试组"
      },
      "task": {
        "id": "task-123",
        "name": "测试任务"
      },
      "user": {
        "id": "user-123",
        "name": "测试用户"
      },
      "customer_data": {
        "name": "测试客户",
        "email": "test@example.com",
        "company": "测试公司",
        "extra": "额外信息"
      }
    }
  ]
 }
 2024-12-02 16:30:15 - routes - INFO - 🏠 server_ip: {
  "ip": "0.0.0.0",
  "port": 8000
 }
 2024-12-02 16:30:15 - routes - INFO - 🖥️ client_ip: {
  "ip": "127.0.0.1",
  "port": 52345
 }
 2024-12-02 16:30:15 - routes - INFO - 📦 callback_data: {
  "count": 2,
  "data_count": 1,
  "data_sample": {
    "bill": 100,
    "duration": 60,
    "callid": "test-call-123",
    "calldate": "2024-01-01 12:00:00",
    "number": "13800138000",
    "numberid": "number-123",
    "customer_id": "customer-123",
    "status": 0,
    "status_str": "失败",
    "user_id": "user-123",
    "type": 1,
    "number_data": {
      "number": "13800138000",
      "province": "广东",
      "city": "深圳",
      "operator": "移动"
    },
    "group": {
      "id": 1,
      "name": "测试组"
    },
    "task": {
      "id": "task-123",
      "name": "测试任务"
    },
    "user": {
      "id": "user-123",
      "name": "测试用户"
    },
    "customer_data": {
      "name": "测试客户",
      "email": "test@example.com",
      "company": "测试公司",
      "extra": "额外信息"
    }
  }
 }
 2024-12-02 16:30:15 - routes - INFO - ✅ 回调请求记录成功保存到数据库，ID: 123
 2024-12-02 16:30:15 - routes - INFO - 📞 count=2 < 3，调用外部API
 2024-12-02 16:30:15 - routes - DEBUG - 🔒 获取Redis锁成功: external_api_call_test-site-123_2
 ```
 ## JSON格式日志的优势
 ### 1. 结构化数据
 - 每个字段都以标准JSON格式输出
 - 便于程序解析和处理
 - 支持复杂的嵌套数据结构
 ### 2. 可读性强
 - JSON格式具有良好的层次结构
 - 缩进格式便于人工阅读
 - 支持中文字符（`ensure_ascii=False`）
 ### 3. 便于分析
 - 可以直接使用JSON工具解析
 - 支持日志分析工具（如ELK、Fluentd等）
 - 便于数据提取和统计
 ### 4. 安全性
 - 敏感信息自动过滤为 `***REDACTED***`
 - 保持数据结构完整性
 - 避免敏感信息泄露
 ## 日志分析示例
 ### 使用jq工具分析JSON日志
 ```bash
 # 提取所有site_id
 grep "📝 site_id:" logs/app.log | jq -r '.📝 site_id'
 # 提取所有client_ip信息
 grep "🖥️ client_ip:" logs/app.log | jq -r '.🖥️ client_ip.ip'
 # 统计不同count值的请求
 grep "📦 callback_data:" logs/app.log | jq -r '.📦 callback_data.count' | sort | uniq -c
 # 提取包含特定号码的请求
 grep "📄 请求体:" logs/app.log | jq 'select(.📄 请求_body.data[].number == "13800138000")'
 ```
 ### 使用Python分析JSON日志
 ```python
 import json
 import re
 # 解析日志中的JSON数据
 def parse_json_logs(log_file):
    site_ids = []
    client_ips = []
    with open(log_file, 'r', encoding='utf-8') as f:
        for line in f:
            if '📝 site_id:' in line:
                # 提取JSON部分
                json_str = line.split('📝 site_id: ')[1].strip()
                site_id = json.loads(json_str)
                site_ids.append(site_id)
            elif '🖥️ client_ip:' in line:
                json_str = line.split('🖥️ client_ip: ')[1].strip()
                client_info = json.loads(json_str)
                client_ips.append(client_info['ip'])
    return site_ids, client_ips
 ```
 ## 性能考虑
 ### 1. JSON序列化开销
 - 使用标准库`json.dumps()`
 - `ensure_ascii=False` 支持中文但略慢
 - 对于高频调用，可考虑关闭详细日志
 ### 2. 日志文件大小
 - JSON格式比纯文本占用更多空间
 - 建议合理设置日志轮转大小
 - 生产环境可考虑使用压缩存储
 ### 3. 内存使用
 - 大型请求体会占用较多内存
 - `callback_data` 只记录摘要信息，避免完整数据
 ## 配置建议
 ### 开发环境
 ```bash
 LOG_LEVEL=INFO  # 显示所有JSON日志
 ```
 ### 生产环境
 ```bash
 LOG_LEVEL=WARNING  # 只显示重要信息，减少JSON日志量
 ```
 ### 调试特定问题
 ```bash
 # 临时开启详细日志
 LOG_LEVEL=DEBUG
 # 问题解决后恢复
 LOG_LEVEL=INFO
 ```
 现在 `log_callback_request` 方法以JSON格式输出所有关键信息，便于日志分析和系统监控！
--- a/REDIS_SETUP.md
+++ b/REDIS_SETUP.md
@@ -0,0 +1,281 @@
 # Redis 配置指南
 ## 概述
 本应用使用 Redis 作为 Celery 的消息代理和结果后端。在启动应用前，需要确保 Redis 服务正常运行。
 ## Redis 安装
 ### Windows
 1. **下载 Redis for Windows**
   ```bash
   # 访问 https://github.com/microsoftarchive/redis/releases
   # 下载最新的 .msi 文件并安装
   ```
 2. **使用 WSL (推荐)**
   ```bash
   # 在 WSL 中安装
   sudo apt update
   sudo apt install redis-server
   sudo systemctl start redis-server
   sudo systemctl enable redis-server
   ```
 ### Linux
 ```bash
 # Ubuntu/Debian
 sudo apt update
 sudo apt install redis-server
 sudo systemctl start redis-server
 sudo systemctl enable redis-server
 # CentOS/RHEL
 sudo yum install redis
 sudo systemctl start redis
 sudo systemctl enable redis
 ```
 ### macOS
 ```bash
 # 使用 Homebrew
 brew install redis
 brew services start redis
 ```
 ## Redis 配置
 ### 基本配置
 编辑 Redis 配置文件 `/etc/redis/redis.conf`：
 ```ini
 # 设置密码（可选）
 requirepass your_redis_password
 # 设置最大内存
 maxmemory 256mb
 maxmemory-policy allkeys-lru
 # 持久化配置
 save 900 1
 save 300 10
 save 60 10000
 ```
 ### 环境变量配置
 在 `.env` 文件中配置：
 ```bash
 # Redis 配置
 CELERY_BROKER_URL=redis://localhost:6379/0
 CELERY_RESULT_BACKEND=redis://localhost:6379/0
 # 如果设置了密码
 # CELERY_BROKER_URL=redis://:your_password@localhost:6379/0
 # CELERY_RESULT_BACKEND=redis://:your_password@localhost:6379/0
 ```
 ## Redis 服务管理
 ### 启动 Redis
 ```bash
 # Windows
 redis-server
 # Linux (Systemd)
 sudo systemctl start redis-server
 # macOS (Homebrew)
 brew services start redis
 ```
 ### 停止 Redis
 ```bash
 # Linux
 sudo systemctl stop redis-server
 # macOS
 brew services stop redis
 # 手动停止
 redis-cli shutdown
 ```
 ### 重启 Redis
 ```bash
 # Linux
 sudo systemctl restart redis-server
 # macOS
 brew services restart redis
 ```
 ## 连接测试
 ### 测试基本连接
 ```bash
 # 使用 redis-cli
 redis-cli ping
 # 应该返回 PONG
 # 使用项目测试脚本
 python test_redis.py
 ```
 ### 检查 Redis 状态
 ```bash
 # 查看连接信息
 redis-cli info clients
 # 查看内存使用
 redis-cli info memory
 # 查看数据库信息
 redis-cli info keyspace
 ```
 ## 应用中的 Redis 集成
 ### 启动时检查
 应用启动时会自动：
 1. 验证 Redis 连接
 2. 创建 Redis 客户端
 3. 存储到应用状态中供其他组件使用
 ### 健康检查
 访问 `/health` 端点可以查看 Redis 连接状态：
 ```json
 {
  "status": "healthy",
  "redis": "connected"
 }
 ```
 ### 故障排除
 如果 Redis 连接失败，应用将无法启动。检查：
 1. Redis 服务是否运行
 2. 连接配置是否正确
 3. 防火墙设置
 4. 密码配置
 ## 生产环境建议
 ### 安全配置
 1. **设置密码**
   ```ini
   requirepass your_strong_password
   ```
 2. **绑定特定 IP**
   ```ini
   bind 127.0.0.1 10.0.0.1
   ```
 3. **禁用危险命令**
   ```ini
   rename-command FLUSHDB ""
   rename-command FLUSHALL ""
   rename-command KEYS ""
   rename-command CONFIG ""
   ```
 ### 性能优化
 1. **设置最大内存**
   ```ini
   maxmemory 1gb
   maxmemory-policy allkeys-lru
   ```
 2. **持久化配置**
   ```ini
   save 900 1
   save 300 10
   save 60 10000
   ```
 3. **连接限制**
   ```ini
   maxclients 1000
   ```
 ### 监控
 1. **使用 INFO 命令**
   ```bash
   redis-cli info server
   redis-cli info memory
   redis-cli info stats
   ```
 2. **监控工具**
   - RedisInsight
   - Redis Commander
   - 自定义监控脚本
 ## 常见问题
 ### Q: 应用启动失败，提示 Redis 连接错误
 **A:** 检查以下几点：
 1. Redis 服务是否启动：`redis-cli ping`
 2. 配置文件中的 URL 是否正确
 3. 防火墙是否阻止连接
 4. Redis 密码配置是否匹配
 ### Q: Redis 内存占用过高
 **A:** 
 1. 检查键的数量：`redis-cli dbsize`
 2. 设置过期策略：`CONFIG SET maxmemory-policy allkeys-lru`
 3. 清理无用键：`redis-cli FLUSHDB`
 ### Q: Celery 任务不执行
 **A:** 检查：
 1. Redis 连接状态
 2. Celery Worker 是否启动
 3. 队列配置是否正确
 ## 日志位置
 - 应用日志：`logs/app.log`
 - Redis 日志：`/var/log/redis/redis.log` (Linux)
 - Celery 日志：控制台输出或配置的日志文件
 ## 相关命令
 ```bash
 # 查看所有键
 redis-cli keys "*"
 # 查看特定键
 redis-cli get key_name
 # 删除键
 redis-cli del key_name
 # 清空数据库
 redis-cli flushdb
 # 查看信息
 redis-cli info
 # 监控命令
 redis-cli monitor
 ```
--- a/REFACTOR_GUIDE.md
+++ b/REFACTOR_GUIDE.md
@@ -0,0 +1,71 @@
 # 代码重构指南
 ## Service 模块重构
 ### 重构内容
 将以下业务逻辑函数从 `app/celery_tasks.py` 移动到独立的 `app/callback_service.py` 模块中：
 #### 移动的函数
 1. `save_callback_data_items` - 保存回调数据到数据库
 2. `get_uncompleted_callback_log` - 获取未完成的回调日志
 3. `get_callback_log_data` - 获取回调日志数据
 4. `check_phone_number_threshold` - 检查手机号阈值
 5. `call_external_api_with_retry` - 带重试的外部API调用
 6. `mark_callback_log_completed` - 标记回调日志为完成
 7. `_log_dtc_push_call` - 记录API调用日志
 ### 重构优势
 #### 1. 代码组织优化
 - **关注点分离**：Celery 任务专注于任务调度，业务逻辑独立到 Service 层
 - **代码复用**：Service 函数可以被其他模块直接调用，不限于 Celery 任务
 - **维护性提升**：业务逻辑集中管理，便于测试和维护
 #### 2. 模块职责清晰
 - `celery_tasks.py`: 负责任务定义和调度逻辑
 - `callback_service.py`: 负责回调处理的核心业务逻辑
 #### 3. 可测试性增强
 - Service 函数可以独立进行单元测试
 - 不依赖 Celery 环境，测试更加便捷
 ### 使用方式
 #### 在 Celery 任务中使用
 ```python
 from app.callback_service import (
    save_callback_data_items,
    get_uncompleted_callback_log,
    call_external_api_with_retry
 )
 # 直接调用服务函数
 callback_log_data = get_uncompleted_callback_log(conn)
 if callback_log_data:
    save_callback_data_items(conn, data_list, callback_log_id)
 ```
 #### 在其他模块中使用
 ```python
 from app.callback_service import check_phone_number_threshold
 # 直接使用业务逻辑
 exceeds, is_valid = check_phone_number_threshold(conn, phone_number)
 ```
 ### 向后兼容性
 - ✅ 所有现有功能保持不变
 - ✅ Celery 任务正常工作
 - ✅ API 接口行为一致
 - ✅ 数据库操作不变
 ### 文件结构
 ```
 app/
 ├── callback_service.py    # 新增：回调处理服务
 ├── celery_tasks.py        # 重构：仅包含任务定义
 ├── celery_app.py          # 保持不变
 └── ...                     # 其他文件保持不变
 ```
 这次重构提高了代码的可维护性和可测试性，同时保持了完全的向后兼容性。
--- a/START_GUIDE.md
+++ b/START_GUIDE.md
@@ -0,0 +1,59 @@
 # 启动说明
 ## 启动模式
 现在所有启动都通过 `main.py` 完成，支持以下模式：
 ### 1. 仅启动 API 服务
 ```bash
 python main.py --mode api
 # 或者
 python main.py
 ```
 ### 2. 仅启动 Celery Worker
 ```bash
 python main.py --mode worker
 ```
 ### 3. 仅启动 Celery Beat 调度器
 ```bash
 python main.py --mode beat
 ```
 ### 4. 启动完整服务栈 (API + Worker + Beat)
 ```bash
 python main.py --mode all
 ```
 ## 开发环境启动
 开发环境下推荐使用 `all` 模式：
 ```bash
 python main.py --mode all
 ```
 这将在同一个进程中启动：
 - FastAPI 应用 (端口 8000)
 - Celery Worker (后台线程)
 - Celery Beat 调度器 (后台线程)
 ## 生产环境启动
 生产环境下建议分别启动各个组件：
 ```bash
 # 终端1: 启动 API
 python main.py --mode api
 # 终端2: 启动 Worker
 python main.py --mode worker
 # 终端3: 启动 Beat
 python main.py --mode beat
 ```
 ## 注意事项
 - 使用 `--mode all` 时，Celery Worker 和 Beat 运行在后台线程中，适合开发和测试
 - 生产环境建议使用进程管理工具 (如 systemd, supervisor) 分别管理各个进程
 - 日志会统一输出到配置的日志文件中
--- a/app/api_config.py
+++ b/app/api_config.py
@@ -0,0 +1,64 @@
 """API接口配置文件"""
 # API配置
 API_CONFIG = {
    # 关闭清洗任务
    'task_1': {
        'url': 'http://agent-api.ai.telrobot.top/agent-api/new/user/8f6bae44-e02a-4393-a6cf-adb861bf7d8c/task_start/a3bdd5b6-1932-4bc9-95bf-dbc4bccb171f',
        'method': 'PUT',
        'headers': {
            'Content-Type': 'application/json',
            'User-Agent': 'Celery-Task/1.0'
        },
        'timeout': 30,
        'body': {
        }
    },
    # Ice跟进
    'task_2': {
        'url': 'http://agent-api.ai.telrobot.top/agent-api/new/user/8f6bae44-e02a-4393-a6cf-adb861bf7d8c/task_start/db17ba5f-666a-4dc0-a635-1d4d0059338d',
        'method': 'PUT',
        'headers': {
            'Content-Type': 'application/json',
            'User-Agent': 'Celery-Task/1.0'
        },
        'timeout': 30,
        'body': {
        }
    },
    # 战败清洗跟进
    'task_3': {
        'url': 'http://agent-api.ai.telrobot.top/agent-api/new/user/8f6bae44-e02a-4393-a6cf-adb861bf7d8c/task_start/2121d91e-c7d8-47cf-8a67-fa2f22e087ae',
        'method': 'PUT',
        'headers': {
            'Content-Type': 'application/json',
            'User-Agent': 'Celery-Task/1.0'
        },
        'timeout': 30,
        'body': {
        }
    }
 }
 '''
 API_CONFIG = {
    'task_test': {
        'url': 'http://localhost:8000/health',
        'method': 'GET',
        'headers': {
            'Content-Type': 'application/json',
            'User-Agent': 'Celery-Task/1.0'
        },
        'timeout': 30,
        'body': {
        }
    },
 }
 '''
 # 重试配置
 RETRY_CONFIG = {
    'max_retries': 3,
    'retry_delay': 60,  # 重试间隔（秒）
    'retry_on_status': [500, 502, 503, 504, 429]
 }
--- a/app/callback_service.py
+++ b/app/callback_service.py
@@ -0,0 +1,525 @@
 """
 回调处理服务模块
 包含与回调处理相关的业务逻辑函数
 """
 import json
 import asyncio
 import httpx
 from datetime import datetime
 from typing import Optional, Tuple, Dict, Any
 from fastapi import Request
 from sqlalchemy.ext.asyncio import AsyncSession
 from app.config import settings
 from app.logger import get_callback_service_logger
 from app.database import CallbackFailureLog
 logger = get_callback_service_logger()
 async def log_callback_request(
    db: AsyncSession,
    request: Request,
    site_id: str,
    callback_data: dict
 ) -> bool:
    """记录回调请求到数据库，返回操作是否成功"""
    import json
    # 获取客户端IP地址
    client_ip = request.client.host if request.client else None
    client_port = request.client.port if request.client else None
    # 获取服务器IP地址
    server_ip = None
    server_port = None
    if hasattr(request, 'scope') and 'server' in request.scope:
        server_host, server_port_info = request.scope['server']
        server_ip = server_host
        server_port = server_port_info
    # 准备请求头信息（直接记录原始请求头）
    request_headers = dict(request.headers)
    # 准备请求体信息（直接记录原始请求体）
    request_body = callback_data
    # 记录请求URL（JSON格式）
    logger.info(f"🌐 请求URL: {request.url}")
    # 记录site_id（JSON格式）
    logger.info(f"📝 site_id: {site_id}")
        # 记录server_ip（JSON格式）
    server_info = {
        "ip": server_ip,
        "port": server_port
    }
    logger.info(f"🏠 server_ip: {server_info}")
    # 记录client_ip（JSON格式）
    client_info = {
        "ip": client_ip,
        "port": client_port
    }
    logger.info(f"🖥️ client_ip: {client_info}")
    # 记录请求头（JSON格式）
    logger.info(f"📋 请求头: {json.dumps(request_headers, ensure_ascii=False, indent=2)}")
    # 记录请求体（JSON格式）
    logger.info(f"📄 请求体: {json.dumps(callback_data, ensure_ascii=False, indent=2)}")
    try:
        # 保存到数据库
        callback_log = CallbackFailureLog(
            site_id=site_id,
            remote_address=f"{client_ip}:{client_port}" if client_ip and client_port else client_ip,
            server_ip=f"{server_ip}:{server_port}" if server_ip and server_port else server_ip,
            request_url=str(request.url),
            request_headers=request_headers,  # 保存原始请求头
            request_body=request_body  # 保存从request.body获取的原始请求体
        )
        db.add(callback_log)
        await db.commit()
        logger.info(f"✅ 回调请求记录成功保存到数据库，ID: {callback_log.id}")
        # 返回成功标识
        return True
    except Exception as e:
        logger.error(f"❌ 保存回调请求到数据库失败: {e}", exc_info=True)
        # 返回失败标识
        return False
 async def save_callback_data_items(
    db: AsyncSession,
    callback_data_items: list,
    callback_failure_log_id: int
 ) -> bool:
    """保存callback_data.data中的数据到数据库
    Returns:
        bool: 保存是否成功，True表示成功，False表示失败或跳过保存
    """
    from sqlalchemy import text
    try:
        logger.info(f"📊 共传入callback_log_id {callback_failure_log_id} 的 {len(callback_data_items)} 条数据")
        # 检查是否已经存在该callback_log_id的数据
        existing_data_result = await db.execute(
            text("SELECT * FROM callback_failure_data WHERE callback_failure_log_id = :log_id"),
            {"log_id": callback_failure_log_id}
        )
        existing_data_list = existing_data_result.fetchall()
        if existing_data_list:
            logger.warning(f"⚠️ callback_log_id {callback_failure_log_id} 已存在 {len(existing_data_list)} 条数据，跳过保存")
            return True
        for item in callback_data_items:
            # 获取手机号
            number_data = item.get('number_data', {})
            phone_number = number_data.get('number')
            if not phone_number:
                continue
            # 获取任务ID
            task = item.get('task', {})
            task_id = task.get('id', '')
            # 获取用户ID
            user_id = item.get('user_id', '')
            # 获取状态信息
            status = item.get('status', 0)
            status_description = item.get('status_str', '')
            # 获取通话日期，默认使用当前时间
            calldate = datetime.now()
            if 'calldate' in item:
                try:
                    # 如果calldate是字符串，尝试解析为datetime
                    if isinstance(item['calldate'], str):
                        calldate = datetime.fromisoformat(item['calldate'].replace('Z', '+00:00'))
                    elif isinstance(item['calldate'], (int, float)):
                        # 如果是时间戳，转换为datetime
                        calldate = datetime.fromtimestamp(item['calldate'])
                except (ValueError, TypeError) as e:
                    logger.warning(f"⚠️ 解析calldate失败: {item.get('calldate')}, 使用当前时间, 错误: {e}")
                    calldate = datetime.now()
            # 将整个item转换为JSON字符串保存
            raw_data_json = json.dumps(item, ensure_ascii=False)
            # 直接插入数据库
            await db.execute(
                text("""
                    INSERT INTO callback_failure_data 
                    (callback_failure_log_id, phone_number, task_id, user_id, status, status_description, raw_data, calldate)
                    VALUES (:log_id, :phone_number, :task_id, :user_id, :status, :status_description, :raw_data, :calldate)
                """),
                {
                    "log_id": callback_failure_log_id,
                    "phone_number": phone_number,
                    "task_id": task_id,
                    "user_id": user_id,
                    "status": status,
                    "status_description": status_description,
                    "raw_data": raw_data_json,
                    "calldate": calldate
                }
            )
        await db.commit()
        logger.info(f"✅ 成功保存 {len(callback_data_items)} 条callback_data记录到数据库")
        return True
    except Exception as e:
        logger.error(f"❌ 保存callback_data到数据库失败: {e}", exc_info=True)
        await db.rollback()
        # 不重新抛出异常，避免影响主业务流程
        return False
 async def get_uncompleted_callback_log(db: AsyncSession) -> Tuple[bool, Optional[Tuple[int, str, str, str]]]:
    """获取一条未完成的回调请求（按创建时间取最小值）
    Returns:
        Tuple[bool, Optional[Tuple[int, str, str, str]]]: 
            - 第一个值表示查询是否成功（True表示成功，False表示失败）
            - 第二个值为回调日志数据元组或None
    """
    from sqlalchemy import text
    try:
        # 查询一条未完成的回调日志（按创建时间升序排列，取第一条）
        result = await db.execute(
            text("""
                SELECT id, site_id, request_headers, request_body 
                FROM callback_failure_logs 
                WHERE is_completed = false 
                ORDER BY created_at ASC 
                LIMIT 1
            """)
        )
        row = result.fetchone()
        if not row:
            logger.info("📋 没有找到未完成的回调日志")
            return True, None
        callback_log_id, site_id, request_headers, request_body = row
        request_headers_json = json.dumps(request_headers, ensure_ascii=False)
        request_body_json = json.dumps(request_body, ensure_ascii=False)
        return True, (callback_log_id, site_id, request_headers_json, request_body_json)
    except Exception as e:
        logger.error(f"❌ 查询未完成回调日志失败: {e}", exc_info=True)
        return False, None
 async def get_callback_log_data(db: AsyncSession, callback_log_id: int) -> Tuple[bool, Optional[str], Optional[str]]:
    """获取回调日志数据"""
    from sqlalchemy import text
    try:
        # 查询回调日志
        result = await db.execute(
            text("SELECT request_headers, request_body FROM callback_failure_logs WHERE id = :log_id"),
            {"log_id": callback_log_id}
        )
        row = result.fetchone()
        if not row:
            logger.error(f"❌ 未找到回调日志: {callback_log_id}")
            return False, None, None
        request_headers_json = json.dumps(row[0], ensure_ascii=False)
        request_body_json = json.dumps(row[1], ensure_ascii=False)
        return True, request_headers_json, request_body_json
    except Exception as e:
        logger.error(f"❌ 获取回调日志数据失败: {e}", exc_info=True)
        return False, None, None
 async def call_external_api_with_retry(
    db: AsyncSession,
    request_body: dict,
    request_headers: Dict[str, str],
    max_retries: int,
    callback_failure_log_id: int
 ) -> Tuple[bool, int]:
    """带重试的调用外部API接口"""
    for attempt in range(1, max_retries + 1):
        try:
            logger.info(f"🌐 尝试调用外部API接口，第{attempt}次")
            async with httpx.AsyncClient(timeout=30.0) as client:
                response = await client.post(
                    settings.external_api_url,
                    json=request_body,
                    headers=request_headers
                )
                # 记录推送日志
                await _log_dtc_push_call(
                    db=db,
                    callback_failure_log_id=callback_failure_log_id,
                    request_url=settings.external_api_url,
                    request_headers=request_headers,
                    request_body=request_body,
                    response_status=response.status_code,
                    response_headers=dict(response.headers),
                    response_body=response.text,
                    retry_count=attempt - 1
                )
                if response.status_code == 200:
                    logger.info(f"✅ 调用外部API接口成功，状态码: {response.status_code}")
                    return True, attempt - 1
                else:
                    logger.warning(f"⚠️ 外部API返回非成功状态码: {response.status_code}")
                    # 如果是客户端错误（4xx），不重试
                    if 400 <= response.status_code < 500:
                        logger.error(f"❌ 客户端错误，不重试: {response.status_code}")
                        return False, attempt - 1
        except httpx.TimeoutException:
            logger.warning(f"⏰ 调用外部API接口超时，第{attempt}次尝试")
        except httpx.RequestError as e:
            logger.warning(f"🌐 调用外部API接口请求错误，第{attempt}次尝试: {e}")
        except Exception as e:
            logger.error(f"❌ 调用外部API接口异常，第{attempt}次尝试: {e}", exc_info=True)
        # 如果不是最后一次尝试，等待一段时间再重试
        if attempt < max_retries:
            await asyncio.sleep(2 ** attempt)  # 指数退避
    # 所有重试都失败了
    logger.error(f"❌ 调用外部API接口失败，已重试{max_retries}次")
    return False, max_retries
 async def mark_callback_log_completed(db: AsyncSession, callback_log_id: int) -> bool:
    """标记CallbackFailureLog记录为已完成"""
    from sqlalchemy import text
    try:
        # 先检查推送日志中是否有响应成功的记录
        success_result = await db.execute(
            text("""
                SELECT COUNT(*) as success_count
                FROM external_api_logs 
                WHERE callback_failure_log_id = :log_id 
                AND response_status = 200
            """),
            {"log_id": callback_log_id}
        )
        success_count = success_result.fetchone()[0]
        # 检查失败记录数量
        failure_result = await db.execute(
            text("""
                SELECT COUNT(*) as failure_count
                FROM external_api_logs 
                WHERE callback_failure_log_id = :log_id 
                AND response_status != 200
            """),
            {"log_id": callback_log_id}
        )
        failure_count = failure_result.fetchone()[0]
        # 如果有成功记录，或者失败记录达到最大重试次数，则标记为完成
        if success_count == 0 and failure_count < settings.external_api_retry_max:
            logger.warning(f"⚠️ 回调日志 {callback_log_id} 没有成功的推送记录，且失败记录({failure_count}条)未达到最大重试次数({settings.external_api_retry_max}条)，不标记为完成")
            return False
        if success_count > 0:
            logger.info(f"✅ 回调日志 {callback_log_id} 找到 {success_count} 条成功推送记录，开始标记为完成")
        else:
            logger.info(f"✅ 回调日志 {callback_log_id} 失败记录({failure_count}条)已达到最大重试次数({settings.external_api_retry_max}条)，开始标记为完成")
        # 更新指定日志记录为已完成
        result = await db.execute(
            text("""
                UPDATE callback_failure_logs 
                SET is_completed = true 
                WHERE id = :log_id 
                AND is_completed = false
            """),
            {"log_id": callback_log_id}
        )
        # 检查是否真的更新了记录
        if result.rowcount == 0:
            logger.warning(f"⚠️ 回调日志 {callback_log_id} 已标记为完成或不存在")
            return False
        await db.commit()
        logger.info(f"✅ 回调日志 {callback_log_id} 已标记为完成")
        return True
    except Exception as e:
        logger.error(f"❌ 标记回调日志 {callback_log_id} 完成失败: {e}", exc_info=True)
        await db.rollback()
        return False
 async def get_related_records_by_unique_data_list(
    db: AsyncSession,
    unique_data_list: list,
    callback_log_id: int,
    limit_count: int
 ) -> Tuple[bool, list]:
    """根据unique_data_list中的数据查询相关记录，按创建时间排序
    Args:
        db: 异步数据库会话
        unique_data_list: 包含手机号、task_id、user_id的数据项列表
        callback_log_id: 回调日志ID，作为过滤条件
        limit_count: 获取记录数量限制
    Returns:
        Tuple[bool, list]: 
            - 第一个值表示查询是否成功（True表示成功，False表示失败）
            - 第二个值为查询到的相关记录列表，失败时返回空列表
    """
    from sqlalchemy import text
    related_records = []
    try:
        # 遍历unique_data_list查询相关记录
        for item in unique_data_list:
            # 提取手机号
            number_data = item.get('number_data', {})
            phone_number = number_data.get('number')
            # 提取task_id
            task = item.get('task', {})
            task_id = task.get('id', '')
            # 提取user_id
            user_id = item.get('user_id', '')
            if phone_number and task_id and user_id:
                result = await db.execute(
                    text("""
                        SELECT phone_number, task_id, user_id, created_at
                        FROM callback_failure_data 
                        WHERE phone_number = :phone_number 
                        AND task_id = :task_id 
                        AND user_id = :user_id
                        AND callback_failure_log_id = :callback_log_id
                        ORDER BY created_at DESC 
                        LIMIT :limit_count
                    """),
                    {
                        "phone_number": phone_number,
                        "task_id": task_id,
                        "user_id": user_id,
                        "callback_log_id": callback_log_id,
                        "limit_count": limit_count
                    }
                )
                records = result.fetchall()
                if records:
                    related_records.extend([{
                        "phone_number": record[0],
                        "task_id": record[1], 
                        "user_id": record[2],
                        "created_at": record[3]
                    } for record in records])
        logger.info(f"📊 查询到相关记录数量: {len(related_records)} (阈值: {limit_count})")
        if related_records:
            for i, record in enumerate(related_records):
                logger.info(f"📋 记录{i+1}: 手机号={record['phone_number']}, task_id={record['task_id']}, 用户ID={record['user_id']}, 创建时间={record['created_at']}")
        return True, related_records
    except Exception as e:
        logger.error(f"❌ 查询相关记录失败: {e}", exc_info=True)
        return False, []
 async def _log_dtc_push_call(
    db: AsyncSession,
    callback_failure_log_id: int,
    request_url: str,
    request_headers: Dict[str, Any],
    request_body: Dict[str, Any],
    response_status: int,
    response_headers: Dict[str, str],
    response_body: str,
    retry_count: int
 ) -> bool:
    """记录DTC推送调用日志"""
    from sqlalchemy import text
    try:
        # 提取手机号从请求体中
        phone_number = None
        if 'data' in request_body and isinstance(request_body['data'], list) and len(request_body['data']) > 0:
            first_item = request_body['data'][0]
            if 'number_data' in first_item and 'number' in first_item['number_data']:
                phone_number = first_item['number_data']['number']
        # 记录到external_api_logs表
        await db.execute(
            text("""
                INSERT INTO external_api_logs (
                    callback_failure_log_id,
                    phone_number,
                    request_url,
                    request_headers,
                    request_body,
                    response_status,
                    response_headers,
                    response_body,
                    retry_count,
                    created_at
                ) VALUES (
                    :callback_failure_log_id,
                    :phone_number,
                    :request_url,
                    :request_headers,
                    :request_body,
                    :response_status,
                    :response_headers,
                    :response_body,
                    :retry_count,
                    datetime('now')
                )
            """),
            {
                "callback_failure_log_id": callback_failure_log_id,
                "phone_number": phone_number,
                "request_url": request_url,
                "request_headers": json.dumps(request_headers, ensure_ascii=False),
                "request_body": json.dumps(request_body, ensure_ascii=False),
                "response_status": response_status,
                "response_headers": json.dumps(response_headers, ensure_ascii=False),
                "response_body": response_body,
                "retry_count": retry_count
            }
        )
        await db.commit()
        logger.debug(f"📝 DTC推送调用日志记录成功，状态码: {response_status}")
        return True
    except Exception as e:
        logger.error(f"❌ 记录DTC推送调用日志失败: {e}", exc_info=True)
        await db.rollback()
        return False
--- a/app/celery_app.py
+++ b/app/celery_app.py
@@ -0,0 +1,69 @@
 """
 Celery应用配置
 """
 from celery import Celery
 from celery.schedules import crontab
 from app.config import settings
 from app.logger import (
    get_celery_logger,
    get_celery_beat_logger,
    get_celery_worker_logger,
    LoggerManager,
 )
 # 确保日志系统初始化
 LoggerManager.setup_logging()
 logger = get_celery_logger()
 beat_logger = get_celery_beat_logger()
 worker_logger = get_celery_worker_logger()
 # 创建Celery应用实例
 celery_app = Celery(
    "ai_talk_callback",
    broker=settings.celery_broker_url,
    backend=settings.celery_result_backend,
    include=["app.celery_tasks"],
 )
 # Celery配置
 celery_app.conf.update(
    task_serializer=settings.celery_task_serializer,
    result_serializer=settings.celery_result_serializer,
    accept_content=settings.celery_accept_content,
    timezone=settings.celery_timezone,
    enable_utc=settings.celery_enable_utc,
    task_track_started=True,
    task_time_limit=30 * 60,  # 30分钟超时
    task_soft_time_limit=25 * 60,  # 25分钟软超时
    worker_prefetch_multiplier=1,
    worker_max_tasks_per_child=1000,
    # Worker日志配置
    worker_log_format="[%(asctime)s: %(levelname)s/%(processName)s] %(message)s",
    worker_task_log_format="[%(asctime)s: %(levelname)s/%(processName)s][%(task_name)s(%(task_id)s)] %(message)s",
    # Beat 调度配置
    # crontab(hour=9, minute=0)  # 每天上午9点执行
    beat_schedule={
        "daily-morning-task": {
            "task": "call_api",
            "schedule": crontab(hour=9, minute=0),  # 每天上午9点执行
        },
    },
 )
 logger.info("🌿 Celery应用配置完成")
 beat_logger.info("📅 Celery Beat 调度器配置完成")
 beat_logger.info("📋 定时任务列表:")
 for task_name, task_config in celery_app.conf.beat_schedule.items():
    beat_logger.info(f"  - {task_name}: {task_config['schedule']}秒")
 # Worker 配置日志
 worker_logger.info("🔧 Celery Worker 配置完成")
 worker_logger.info("⚙️ Worker 配置参数:")
 worker_logger.info(f"  - 任务超时: {celery_app.conf.task_time_limit}秒")
 worker_logger.info(f"  - 软超时: {celery_app.conf.task_soft_time_limit}秒")
 worker_logger.info(f"  - 预取倍数: {celery_app.conf.worker_prefetch_multiplier}")
 worker_logger.info(
    f"  - 每个子进程最大任务数: {celery_app.conf.worker_max_tasks_per_child}"
 )
 worker_logger.info(f"  - 包含模块: {celery_app.conf.include}")
--- a/app/celery_tasks.py
+++ b/app/celery_tasks.py
@@ -0,0 +1,389 @@
 """
 Celery任务定义
 """
 from datetime import datetime
 from functools import wraps
 import json
 from app.celery_app import celery_app
 from app.config import settings
 from app.logger import get_celery_tasks_logger
 from app.callback_service import (
    save_callback_data_items,
    get_uncompleted_callback_log,
    call_external_api_with_retry,
    mark_callback_log_completed,
    get_related_records_by_unique_data_list
 )
 from app.redis_lock import async_redis_manager, distributed_lock
 import requests
 from app.api_config import API_CONFIG, RETRY_CONFIG
 logger = get_celery_tasks_logger()
 def conditional_task(enabled=True):
    def decorator(task_func):
        @wraps(task_func)
        def wrapper(*args, **kwargs):
            if not enabled:
                logger.info(f"Task {task_func.__name__} is disabled")
                return None
            return task_func(*args, **kwargs)
        return wrapper
    return decorator
 # 创建异步数据库连接用于Celery任务
 from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
 # 使用异步数据库连接
 async_engine = create_async_engine(
    settings.database_url,
    echo=settings.debug,
    future=True,
 )
 AsyncSessionLocal = async_sessionmaker(
    async_engine, class_=AsyncSession, expire_on_commit=False
 )
@celery_app.task(bind=True, name='push_data_to_dtc')
@conditional_task(enabled=settings.enabled_push_data_to_dtc_task)
 def push_data_to_dtc_task(self):
    """
    推送数据给DTC的Celery任务
    自动获取一条未完成的回调请求进行处理
    """
    import asyncio
    async def async_task():
        task_name = 'push_data_to_dtc'
        logger.info(f"🌿 开始推送数据给DTC任务")
        try:
            # 连接异步Redis
            connection_success = await async_redis_manager.connect()
            if not connection_success:
                logger.error(f"❌ Redis管理器连接失败，任务停止执行")
                raise Exception("Redis连接失败")
            redis_pool = await async_redis_manager.get_redis_pool()
            # 使用异步分布式锁
            async with distributed_lock(redis_pool, f"celery_task:{task_name}", timeout=300) as lock_acquired:
                if not lock_acquired:
                    logger.warning(f"⚠️ 任务 {task_name} 正在执行中，跳过本次执行")
                    return {"status": "skipped", "message": f"任务 {task_name} 正在执行中，跳过本次执行"}
                logger.info(f"🔒 成功获取任务 {task_name} 的分布式锁")
                async with AsyncSessionLocal() as db:
                    # 更新任务状态
                    self.update_state(
                        state='PROGRESS',
                        meta={'current': 0, 'total': 100, 'status': f'正在获取下一条回调请求日志...'}
                    )
                    # 获取一条未完成的回调请求（按创建时间取最小值）
                    query_success, callback_log_data = await get_uncompleted_callback_log(db)
                    if not query_success:
                        logger.error("❌ 查询未完成的回调请求失败，尝试再查一次")
                        query_success, callback_log_data = await get_uncompleted_callback_log(db)
                        if not query_success:
                            logger.error("❌ 再次查询未完成的回调请求失败，停止任务执行")
                            raise Exception("查询未完成的回调请求失败，任务停止执行")
                    if not callback_log_data:
                        logger.info("📋 没有找到未完成的回调请求")
                        return {"status": "skipped", "message": "没有找到未完成的回调请求"}
                    callback_log_id, site_id, request_headers_json, request_body_json = callback_log_data
                    logger.info(f"📋 获取到未完成的回调请求: ID={callback_log_id}, site_id={site_id}")
                    # 更新任务状态
                    self.update_state(
                        state='PROGRESS',
                        meta={'current': 25, 'total': 100, 'status': f'获取回调请求成功(ID={callback_log_id}, site_id={site_id})，开始分析处理...'}
                    )
                    # 解析请求头和请求体
                    request_headers = json.loads(request_headers_json)
                    request_body = json.loads(request_body_json)
                    # 保存callback_data.data中的数据
                    data_list = request_body.get('data', [])
                    if data_list and len(data_list) > 0:
                        save_success = await save_callback_data_items(db, data_list, callback_log_id)
                        if not save_success:
                            logger.error(f"❌ 保存callback_data:{callback_log_id}失败，停止任务执行")
                            raise Exception(f"保存callback_data:{callback_log_id}失败，任务停止执行")
                    # 更新任务状态
                    self.update_state(
                        state='PROGRESS',
                        meta={'current': 50, 'total': 100, 'status': f'请求体中通话明细处理成功(ID={callback_log_id}, site_id={site_id})，开始过滤需要转发的通话记录...'}
                    )
                    # 根据手机号、task_id、user_id给data_list去重
                    data_list = request_body.get('data', [])
                    unique_data_list = []
                    seen_records = set()
                    original_count = len(data_list)
                    for item in data_list:
                        # 提取手机号
                        number_data = item.get('number_data', {})
                        phone_number = number_data.get('number', '')
                        # 提取task_id
                        task = item.get('task', {})
                        task_id = task.get('id', '')
                        # 提取user_id
                        user_id = item.get('user_id', '')
                        # 创建唯一标识
                        unique_key = (phone_number, task_id, user_id)
                        # 如果这个组合没见过，则添加到去重列表中
                        if unique_key not in seen_records:
                            seen_records.add(unique_key)
                            unique_data_list.append(item)
                    logger.info(f"🔄 数据去重完成: 原始数据 {original_count} 条，去重后 {len(unique_data_list)} 条 - callback_log_id: {callback_log_id}, site_id: {site_id}")
                    if len(unique_data_list) < original_count:
                        logger.info(f"🗑️ 移除了 {original_count - len(unique_data_list)} 条重复数据 - callback_log_id: {callback_log_id}, site_id: {site_id}")
                    # 查询相关数据：根据手机号、task_id、user_id作为条件，按创建时间排序获取前三条记录
                    query_success, related_records = await get_related_records_by_unique_data_list(
                        db, unique_data_list, callback_log_id, settings.count_threshold
                    )
                    if not query_success:
                        logger.error("❌ 查询相关记录失败，停止任务执行")
                        raise Exception("查询相关记录失败，任务停止执行")
                    # 没有有效通过记录的情况
                    if not related_records:
                        logger.info(f"📋 没有有效的通过记录需要处理，直接返回 - callback_log_id: {callback_log_id}, site_id={site_id}")
                        return {"status": "completed", "message": "没有有效的通过记录需要处理"}
                    # 更新任务状态
                    self.update_state(
                        state='PROGRESS',
                        meta={'current': 70, 'total': 100, 'status': f'需要推送的通过记录已获取成功(ID={callback_log_id}, site_id={site_id})，准备转发...'}
                    )
                    # 由于已经确认有记录，直接进入推送逻辑
                    # 创建只包含有效数据项的请求体
                    filtered_request_body = request_body.copy()
                    filtered_request_body['data'] = related_records
                    # 更新任务状态
                    self.update_state(
                        state='PROGRESS',
                        meta={'current': 85, 'total': 100, 'status': f'开始推送数据给DTC(ID={callback_log_id}, site_id={site_id})...'}
                    )
                    success, retry_count = await call_external_api_with_retry(
                        db=db,
                        request_body=filtered_request_body,
                        request_headers=request_headers,
                        max_retries=settings.external_api_retry_max,
                        callback_failure_log_id=callback_log_id
                    )
                    if success:
                        logger.info(f"✅ 推送数据给DTC成功，处理的数据项数量: {len(related_records)}, 重试次数: {retry_count} - callback_log_id: {callback_log_id}, site_id: {site_id}")
                        logger.info(f"🎉 推送数据给DTC处理完成 - callback_log_id: {callback_log_id}, site_id: {site_id}")
                    else:
                        logger.error(f"❌ 推送数据给DTC失败，重试次数: {retry_count} - callback_log_id: {callback_log_id}, site_id: {site_id}")
                    # 标记CallbackFailureLog为已完成
                    mark_success = await mark_callback_log_completed(db, callback_log_id)
                    if not mark_success:
                        logger.error(f"❌ 标记回调请求日志状态失败，任务终止执行")
                        raise Exception("标记回调请求日志状态失败，任务终止执行")
                    # 更新任务状态
                    self.update_state(
                        state='PROGRESS',
                        meta={'current': 100, 'total': 100, 'status': f'标记回调请求日志状态(ID={callback_log_id}, site_id={site_id})'}
                    )
                    logger.info(f"✅ 回调日志 {callback_log_id} 已成功标记为完成")
                    return {
                        "status": "completed",
                        "message": "任务完成，回调日志已标记为完成"
                    }
        except Exception as e:
            logger.error(f"❌ 任务执行失败: {e}", exc_info=True)
            return {"status": "error", "message": str(e)}
        finally:
            # 分布式锁会通过上下文管理器自动释放
            logger.debug(f"🔓 任务 {task_name} 的分布式锁已通过上下文管理器处理")
            # 清理Redis连接
            try:
                await async_redis_manager.close()
                logger.debug(f"🔌 Redis连接已清理")
            except Exception as e:
                logger.warning(f"⚠️ 清理Redis连接时出现警告: {e}")
    # 在同步的Celery任务中运行异步代码
    loop = None
    try:
        # 创建新的事件循环
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        return loop.run_until_complete(async_task())
    except Exception as e:
        logger.error(f"❌ 异步任务执行失败: {e}", exc_info=True)
        return {"status": "error", "message": str(e)}
    finally:
        # 确保所有异步任务完成后再关闭事件循环
        if loop and not loop.is_closed():
            try:
                # 等待所有待处理的任务完成
                pending = asyncio.all_tasks(loop)
                if pending:
                    logger.debug(f"⏳ 等待 {len(pending)} 个异步任务完成...")
                    loop.run_until_complete(asyncio.gather(*pending, return_exceptions=True))
                # 关闭事件循环
                loop.close()
                logger.debug(f"🔌 事件循环已正确关闭")
            except Exception as e:
                logger.warning(f"⚠️ 关闭事件循环时出现警告: {e}")
        else:
            logger.debug(f"🔌 事件循环已关闭或未创建")
@celery_app.task(bind=True, name='call_api', max_retries=None)
@conditional_task(enabled=settings.enabled_call_api_task)
 def execute_call_api_task(self):
    """遍历API配置文件中的所有API信息并调用接口"""
    current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    logger.info(f"开始遍历API配置文件执行任务 - 当前时间: {current_time}")
    # 获取重试配置
    max_retries = RETRY_CONFIG['max_retries']
    retry_delay = RETRY_CONFIG['retry_delay']
    retry_on_status = RETRY_CONFIG['retry_on_status']
    # 遍历所有API信息
    results = []
    success_count = 0
    for api_key, call_config in API_CONFIG.items():
        logger.info(f"开始处理API配置: {api_key}")
        call_url = call_config['url']
        method = call_config['method'].upper()
        headers = call_config['headers'].copy()  # 复制headers避免修改原配置
        timeout = call_config['timeout']
        payload = call_config['body']
        # 从环境变量获取Authorization并添加到headers
        if settings.api_authorization_token:
            headers['Authorization'] = settings.api_authorization_token
        try:
            logger.info(f"正在调用任务API: {method} {call_url}")
            # 根据method选择请求方式
            if method == 'POST':
                response = requests.post(call_url, headers=headers, json=payload, timeout=timeout)
            elif method == 'GET':
                response = requests.get(call_url, headers=headers, params=payload, timeout=timeout)
            elif method == 'PUT':
                response = requests.put(call_url, headers=headers, json=payload, timeout=timeout)
            elif method == 'DELETE':
                response = requests.delete(call_url, headers=headers, timeout=timeout)
            else:
                raise ValueError(f"不支持的HTTP方法: {method}")
            # 检查响应状态
            if response.status_code in [200, 201, 204]:  # 成功状态码
                try:
                    result_data = response.json()
                except:
                    result_data = {'response': response.text}
                logger.info(f"任务API调用成功，响应数据: {result_data}")
                print(f"[{current_time}] ✅ 任务API成功 ({api_key}) - {method} {call_url}")
                success_count += 1
                results.append({
                    'call_api_key': api_key,
                    'status': 'success',
                    'message': f'任务API调用成功 ({method})',
                    'response': result_data,
                    'timestamp': current_time,
                    'retry_count': getattr(self.request, 'retries', 0)
                })
            elif response.status_code in retry_on_status:
                # 需要重试的状态码
                current_retry = getattr(self.request, 'retries', 0)
                if current_retry < max_retries:
                    logger.warning(f"任务API需要重试，当前重试次数: {current_retry + 1}/{max_retries}")
                    print(f"[{current_time}] 🔄 任务API失败，正在重试 ({current_retry + 1}/{max_retries}) - {api_key} - 状态码: {response.status_code}")
                    raise self.retry(countdown=retry_delay, exc=Exception(f"任务API失败，状态码: {response.status_code}"))
                else:
                    logger.error(f"任务API失败，已达到最大重试次数: {max_retries}")
                    print(f"[{current_time}] ❌ 任务API失败，已达到最大重试次数 - {api_key} - 状态码: {response.status_code}")
                    results.append({
                        'call_api_key': api_key,
                        'status': 'failed',
                        'message': f'任务API失败，已达到最大重试次数: {max_retries}',
                        'response': response.text,
                        'timestamp': current_time,
                        'retry_count': current_retry + 1
                    })
            else:
                # 其他失败状态码，不重试
                logger.error(f"任务API失败，状态码: {response.status_code}, 响应: {response.text}")
                print(f"[{current_time}] ❌ 任务API失败 - {api_key} - 状态码: {response.status_code}")
                results.append({
                    'call_api_key': api_key,
                    'status': 'failed',
                    'message': f'任务API失败，状态码: {response.status_code}',
                    'response': response.text,
                    'timestamp': current_time,
                    'retry_count': getattr(self.request, 'retries', 0)
                })
        except requests.exceptions.RequestException as e:
            # 网络异常，需要重试
            current_retry = getattr(self.request, 'retries', 0)
            if current_retry < max_retries:
                logger.warning(f"任务API异常，需要重试，当前重试次数: {current_retry + 1}/{max_retries}")
                print(f"[{current_time}] 🔄 任务API异常，正在重试 ({current_retry + 1}/{max_retries}) - {api_key} - 异常: {str(e)}")
                raise self.retry(countdown=retry_delay, exc=e)
            else:
                logger.error(f"任务API异常，已达到最大重试次数: {max_retries}")
                print(f"[{current_time}] ❌ 任务API异常，已达到最大重试次数 - {api_key} - 异常: {str(e)}")
                results.append({
                    'call_api_key': api_key,
                    'status': 'error',
                    'message': f'任务API异常，已达到最大重试次数: {max_retries}',
                    'error': str(e),
                    'timestamp': current_time,
                    'retry_count': current_retry + 1
                })
    # 汇总结果
    total_count = len(results)
    print(f"[{current_time}] 📊 任务汇总 - 成功: {success_count}/{total_count}")
    return {
        'status': 'completed',
        'total_apis': total_count,
        'success_count': success_count,
        'failed_count': total_count - success_count,
        'results': results,
        'timestamp': current_time
    }
--- a/app/config.py
+++ b/app/config.py
@@ -1,21 +1,50 @@
 from pydantic_settings import BaseSettings
 class Settings(BaseSettings):
    @property
    def redis_url(self) -> str:
        """从Celery配置获取Redis URL"""
        return self.celery_broker_url
    # 数据库配置
    database_url: str = ""
-    # Redis配置
+    # Celery配置
-    redis_url: str = "redis://localhost:6379/0"
+    celery_broker_url: str = "redis://localhost:6379/0"
-    redis_lock_timeout: int = 30  # 锁超时时间（秒）
+    celery_result_backend: str = "redis://localhost:6379/1"
-    redis_lock_retry_delay: float = 0.1  # 锁重试延迟（秒）
+    celery_task_serializer: str = "json"
-    redis_lock_max_retries: int = 100  # 最大重试次数
+    celery_result_serializer: str = "json"
    celery_accept_content: list = ["json"]
    celery_timezone: str = "Asia/Shanghai"
    celery_enable_utc: bool = False
    # 任务配置
    enabled_push_data_to_dtc_task: bool = False  # 是否启用推送数据给DTC的任务
    enabled_call_api_task: bool = False  # 是否启用调用API的任务
    # Flower监控配置
    flower_enabled: bool = True  # 是否启用Flower监控
    flower_url: str = "http://localhost:5555"  # Flower访问URL
    flower_port: int = 5555  # Flower服务端口
    flower_url_prefix: str = ""  # Flower URL前缀
    flower_basic_auth: str = "admin:admin123"  # Flower基础认证，格式：username:password
    # Redis配置 (从Celery配置获取)
    redis_password: str = ""
    redis_max_connections: int = 20
    redis_timeout: int = 5
    # 业务配置
    count_threshold: int = 3  # count阈值，大于等于此值直接返回
    external_api_enabled: bool = False  # 是否启用外部API调用
    external_api_retry_max: int = 3  # 外部API最大重试次数
    external_api_url: str = ""
    # Redis锁配置 (使用默认值)
    redis_lock_timeout: int = 300  # 锁超时时间（秒）
    redis_lock_max_retries: int = 10  # 最大重试次数
    redis_lock_retry_delay: float = 0.5  # 重试延迟（秒）
    # 日志配置
    log_level: str = "INFO"  # DEBUG, INFO, WARNING, ERROR, CRITICAL
    log_file: str = "logs/app.log"  # 日志文件路径
@@ -24,9 +53,14 @@ class Settings(BaseSettings):
    log_format: str = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
    log_date_format: str = "%Y-%m-%d %H:%M:%S"
    # API配置
    api_authorization_token: str = ""  # API Authorization Token
    # 应用配置
    app_name: str = "AI Talk Callback API"
    environment: str = "production"  # 环境: production, test, development
    debug: bool = False
    disable_docs: bool = False
    class Config:
        env_file = ".env"
--- a/app/database.py
+++ b/app/database.py
@@ -1,59 +1,70 @@
 from datetime import datetime
 from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
 from sqlalchemy.orm import DeclarativeBase
-from sqlalchemy import Column, String, Integer, DateTime, Text, JSON
+from sqlalchemy import Column, String, Integer, DateTime, Text, JSON, Index, text, Boolean
 from datetime import datetime
 from app.config import settings
-from app.logger import get_logger
+from app.logger import db_logger
 from sqlalchemy.sql.expression import func
-logger = get_logger("database")
+logger = db_logger
 class Base(DeclarativeBase):
    pass
-class CallbackLog(Base):
+class CallbackFailureLog(Base):
-    __tablename__ = "callback_logs"
+    __tablename__ = "callback_failure_logs"
-    id = Column(Integer, primary_key=True, autoincrement=True)
+    id = Column(Integer, primary_key=True, autoincrement=True, comment="日志ID")
    site_id = Column(String(100), nullable=False, comment="站点ID")  # 新增siteId字段
    remote_address = Column(String(45), nullable=True, comment="客户端IP地址")
    server_ip = Column(String(45), nullable=True, comment="服务器IP地址")
-    request_url = Column(String(500), nullable=False)
+    request_url = Column(String(500), nullable=False, comment="请求URL")
-    request_headers = Column(JSON, nullable=False)
+    request_headers = Column(JSON, nullable=False, comment="请求头")
-    request_body = Column(JSON, nullable=False)
+    request_body = Column(JSON, nullable=False, comment="请求体")
-    created_at = Column(DateTime, default=datetime.utcnow)
+    is_completed = Column(Boolean, default=False, comment="是否已完成")
    created_at = Column(DateTime, default=datetime.now(), comment="创建时间")
 class CallbackFailureData(Base):
    __tablename__ = "callback_failure_data"
    id = Column(Integer, primary_key=True, autoincrement=True, comment="数据ID")
    callback_failure_log_id = Column(Integer, nullable=False, comment="回调失败日志ID")
    phone_number = Column(String(20), nullable=False, comment="手机号")
    user_id = Column(String(100), nullable=True, comment="用户ID")
    task_id = Column(String(100), nullable=False, comment="任务ID")
    status = Column(Integer, nullable=False, comment="状态")
    status_description = Column(String(200), nullable=False, comment="状态描述")
    raw_data = Column(Text, nullable=False, comment="原始回调数据JSON字符串")
    calldate = Column(DateTime, nullable=True, comment="通话日期")
    created_at = Column(DateTime, default=datetime.now(), comment="创建时间")
 class ExternalApiLog(Base):
    __tablename__ = "external_api_logs"  # 重命名表，更通用
-    id = Column(Integer, primary_key=True, autoincrement=True)
+    id = Column(Integer, primary_key=True, autoincrement=True, comment="日志ID")
-    request_url = Column(String(500), nullable=False)
+    callback_failure_log_id = Column(Integer, nullable=False, comment="回调失败日志ID")
-    request_headers = Column(JSON, nullable=False)
+    request_url = Column(String(500), nullable=False, comment="外部接口请求URL")
-    request_body = Column(JSON, nullable=False)
+    request_headers = Column(JSON, nullable=False, comment="外部接口请求头")
-    response_status = Column(Integer)
+    request_body = Column(JSON, nullable=False, comment="外部接口请求体")
-    response_headers = Column(JSON)
+    response_status = Column(Integer, nullable=False, comment="外部接口响应状态码")
-    response_body = Column(Text)
+    response_headers = Column(JSON, nullable=False, comment="外部接口响应头")
-    retry_count = Column(Integer, default=0)  # 重试次数
+    response_body = Column(Text, nullable=False, comment="外部接口响应体")
-    created_at = Column(DateTime, default=datetime.utcnow)
+    retry_count = Column(Integer, default=0, comment="重试次数")
    created_at = Column(DateTime, default=datetime.now(), comment="创建时间")
 # 创建数据库引擎
-engine = create_async_engine(
+engine = create_async_engine(settings.database_url, echo=settings.debug, future=True)
    settings.database_url,
    echo=settings.debug,
    future=True
 )
 # 创建会话工厂
 AsyncSessionLocal = async_sessionmaker(
-    engine,
+    engine, class_=AsyncSession, expire_on_commit=False
    class_=AsyncSession,
    expire_on_commit=False
 )
 async def get_db():
    async with AsyncSessionLocal() as session:
        try:
@@ -61,12 +72,22 @@ async def get_db():
        finally:
            await session.close()
 async def init_db():
    logger.info("📊 初始化数据库表结构...")
    try:
        async with engine.begin() as conn:
            await conn.run_sync(Base.metadata.create_all)
            # 检查并创建手机号索引
            await conn.execute(
                text(
                    """
                CREATE INDEX IF NOT EXISTS idx_phone_number 
                ON callback_failure_data (phone_number)
            """
                )
            )
        logger.info("✅ 数据库表结构初始化完成")
    except Exception as e:
        logger.error(f"❌ 数据库初始化失败: {e}")
--- a/app/logger.py
+++ b/app/logger.py
@@ -1,12 +1,58 @@
 import logging
 import logging.handlers
 import os
 from pathlib import Path
 from typing import Optional
 import time
 from datetime import datetime, timedelta
 from app.config import settings
 class CustomTimedRotatingFileHandler(logging.handlers.TimedRotatingFileHandler):
    """自定义时间轮转文件处理器，支持指定轮转时间"""
    def __init__(self, filename, hour=0, minute=0, when='midnight', interval=1, backupCount=7, encoding='utf-8'):
        self.target_hour = hour
        self.target_minute = minute
        self.target_when = when
        self.target_interval = interval
        # 先调用父类初始化
        super().__init__(filename, when=when, interval=interval, backupCount=backupCount, encoding=encoding)
        # 重新计算轮转时间
        self.computeRollover()
    def computeRollover(self, currentTime=None):
        """计算下一次轮转时间，设置为每天的指定时间"""
        if currentTime is None:
            currentTime = time.time()
        # 获取当前时间
        current_time = datetime.fromtimestamp(currentTime)
        # 创建今天的目标时间
        target_time = current_time.replace(hour=self.target_hour, minute=self.target_minute, second=0, microsecond=0)
        # 如果今天的目标时间已过，设置为明天
        if current_time >= target_time:
            target_time += timedelta(days=1)
        # 转换为时间戳
        self.rolloverAt = target_time.timestamp()
    def doRollover(self):
        # 调用父类轮转方法
        super().doRollover()
        # 重新计算下次轮转时间
        self.computeRollover()
        print(f"✅ 日志轮转完成，下次: {datetime.fromtimestamp(self.rolloverAt).strftime('%Y-%m-%d %H:%M:%S')}")
 class LoggerManager:
    """日志管理器"""
@@ -41,10 +87,12 @@ class LoggerManager:
        )
        console_handler.setFormatter(console_formatter)
-        # 创建文件处理器（按时间轮转）
+        # 创建文件处理器（按时间轮转，每天0点）
-        file_handler = logging.handlers.TimedRotatingFileHandler(
+        file_handler = CustomTimedRotatingFileHandler(
            filename=settings.log_file,
-            when='midnight',  # 每天午夜轮转
+            hour=0,           # 0点
            minute=0,         # 0分
            when='midnight',  # 每天轮转
            interval=1,       # 每天一次
            backupCount=settings.log_backup_count,
            encoding='utf-8',
@@ -106,5 +154,95 @@ def get_logger(name: Optional[str] = None) -> logging.Logger:
 app_logger = get_logger("app")
 api_logger = get_logger("api")
 db_logger = get_logger("database")
 redis_logger = get_logger("redis")
 external_api_logger = get_logger("external_api")
 # 各模块专用日志器
 main_logger = get_logger("main")
 routes_logger = get_logger("routes")
 celery_logger = get_logger("celery")
 celery_tasks_logger = get_logger("celery_tasks")
 callback_service_logger = get_logger("callback_service")
 redis_logger = get_logger("redis")
 # 通用文件日志器设置函数
 def setup_celery_file_logger(logger_name: str, log_file: str, level: str = "INFO"):
    """为Celery组件设置专用的文件日志器，同时输出到控制台和文件"""
    # 创建日志目录
    log_dir = Path(log_file).parent
    log_dir.mkdir(parents=True, exist_ok=True)
    # 获取日志器
    logger = get_logger(logger_name)
    logger.setLevel(getattr(logging, level.upper()))
    # 清除现有处理器（避免重复添加）
    logger.handlers.clear()
    # 创建控制台处理器
    console_handler = logging.StreamHandler()
    console_handler.setLevel(getattr(logging, level.upper()))
    # 控制台格式化器
    console_formatter = logging.Formatter(
        fmt='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
        datefmt='%H:%M:%S'
    )
    console_handler.setFormatter(console_formatter)
    # 创建文件处理器
    file_handler = CustomTimedRotatingFileHandler(
        filename=log_file,
        hour=0,           # 0点
        minute=0,         # 0分
        when='midnight',
        interval=1,
        backupCount=7,
        encoding='utf-8',
    )
    file_handler.suffix = "%Y%m%d"
    file_handler.setLevel(getattr(logging, level.upper()))
    # 文件格式化器
    file_formatter = logging.Formatter(
        fmt='%(asctime)s - %(name)s - %(levelname)s - [%(filename)s:%(lineno)d] - %(message)s',
        datefmt='%Y-%m-%d %H:%M:%S'
    )
    file_handler.setFormatter(file_formatter)
    # 添加处理器（先控制台，后文件）
    logger.addHandler(console_handler)
    logger.addHandler(file_handler)
    logger.propagate = False  # 防止重复输出到父日志器
    return logger
 # 各模块专用日志器映射
 CELERY_LOGGERS = {
    "celery_tasks": ("logs/celery_tasks.log", "INFO"),
    "celery": ("logs/celery_app.log", "INFO"),
    "celery.beat": ("logs/celery_app.log", "INFO"),  # Beat日志输出到celery_app.log
    "celery.worker": ("logs/celery_app.log", "INFO"),  # Worker日志输出到celery_app.log
 }
 # 初始化所有Celery文件日志器
 _celery_file_loggers = {}
 for logger_name, (log_file, level) in CELERY_LOGGERS.items():
    _celery_file_loggers[logger_name] = setup_celery_file_logger(logger_name, log_file, level)
 # 向后兼容的默认日志器
 logger = app_logger
 # 通用获取器函数
 def get_celery_file_logger(logger_name: str):
    """获取指定名称的Celery文件日志器"""
    return _celery_file_loggers.get(logger_name, get_logger(logger_name))
 # 简化的便捷函数
 def get_main_logger(): return main_logger
 def get_routes_logger(): return routes_logger
 def get_celery_logger(): return get_celery_file_logger("celery")
 def get_celery_tasks_logger(): return get_celery_file_logger("celery_tasks")
 def get_callback_service_logger(): return callback_service_logger
 def get_redis_logger(): return redis_logger
 def get_celery_beat_logger(): return get_celery_file_logger("celery.beat")
 def get_celery_worker_logger(): return get_celery_file_logger("celery.worker")
--- a/app/models.py
+++ b/app/models.py
@@ -1,60 +1,6 @@
 from pydantic import BaseModel, Field
 from typing import List, Optional
 class NumberData(BaseModel):
    number: str
    province: str
    city: str
    operator: str
 class Group(BaseModel):
    id: int
    name: str
 class Task(BaseModel):
    id: str
    name: str
 class User(BaseModel):
    id: str
    name: str
 class CustomerData(BaseModel):
    name: str
    email: str
    company: Optional[str] = None
    extra: Optional[str] = None
 class CallbackItem(BaseModel):
    bill: int
    duration: int
    callid: str
    calldate: str
    number: str
    numberid: str
    customer_id: str
    status: int
    status_str: str
    user_id: str
    type: int
    number_data: NumberData
    group: Group
    task: Task
    user: User
    customer_data: CustomerData
 class CallbackRequest(BaseModel):
    count: int = Field(..., ge=0, description="通话失败个数")
    data: List[CallbackItem]
 class CallbackResponse(BaseModel):
    success: bool
    message: str
--- a/app/redis_lock.py
+++ b/app/redis_lock.py
@@ -1,22 +1,223 @@
 import redis.asyncio as redis
 import asyncio
 import uuid
 from typing import Optional
 from contextlib import asynccontextmanager
 import aioredis
 from app.config import settings
-from app.logger import get_logger
+from app.logger import get_redis_logger
-logger = get_logger("redis")
+logger = get_redis_logger()
 class AsyncRedisLock:
    """异步Redis分布式锁"""
    def __init__(self, redis_pool, key: str, timeout: int = None):
        self.redis_pool = redis_pool
        self.key = f"lock:{key}"
        self.timeout = timeout or settings.redis_lock_timeout
        self.identifier = str(uuid.uuid4())
        self.acquired = False
    async def acquire(self, blocking: bool = False) -> bool:
        """获取分布式锁"""
        logger.debug(f"🔒 尝试获取Redis锁: {self.key}")
        try:
            # 使用SET命令的NX和EX选项原子性地获取锁
            result = await self.redis_pool.set(
                self.key, 
                self.identifier, 
                expire=self.timeout, 
                exist=self.redis_pool.SET_IF_NOT_EXIST
            )
            self.acquired = result
            if self.acquired:
                logger.debug(f"✅ Redis锁获取成功: {self.key}")
            else:
                logger.debug(f"❌ Redis锁获取失败: {self.key}")
            return self.acquired
        except Exception as e:
            logger.error(f"❌ 获取Redis锁失败: {e}")
            return False
    async def release(self) -> bool:
        """释放分布式锁"""
        if not self.acquired:
            return False
        try:
            # 使用Lua脚本确保只有锁的持有者才能释放锁
            lua_script = """
            if redis.call("GET", KEYS[1]) == ARGV[1] then
                return redis.call("DEL", KEYS[1])
            else
                return 0
            end
            """
            result = await self.redis_pool.eval(
                lua_script,
                1,
                self.key,
                self.identifier
            )
            self.acquired = False
            released = bool(result)
            if released:
                logger.debug(f"🔓 Redis锁释放成功: {self.key}")
            else:
                logger.warning(f"⚠️ Redis锁释放失败，可能已过期: {self.key}")
            return released
        except Exception as e:
            logger.error(f"❌ 释放Redis锁失败: {e}")
            return False
    async def __aenter__(self):
        """异步上下文管理器入口"""
        retries = 0
        while retries < settings.redis_lock_max_retries:
            if await self.acquire(blocking=False):
                return self
            await asyncio.sleep(settings.redis_lock_retry_delay)
            retries += 1
        raise TimeoutError(f"Failed to acquire lock {self.key} after {retries} retries")
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        """异步上下文管理器出口"""
        await self.release()
@asynccontextmanager
 async def distributed_lock(redis_pool, lock_key: str, timeout: int = None):
    """分布式锁上下文管理器
    Args:
        redis_pool: Redis连接池
        lock_key: 锁键名
        timeout: 锁超时时间（秒）
    Yields:
        bool: 是否成功获取锁
    """
    full_lock_key = f"lock:{lock_key}"
    lock_timeout = timeout or settings.redis_lock_timeout
    try:
        # 尝试获取锁，使用setnx命令，并设置过期时间
        identifier = str(uuid.uuid4())
        lock_acquired = await redis_pool.set(
            full_lock_key, 
            identifier, 
            expire=lock_timeout, 
            exist=redis_pool.SET_IF_NOT_EXIST
        )
        logger.debug(f"🔒 尝试获取分布式锁: {full_lock_key}, 结果: {lock_acquired}")
        if lock_acquired:
            try:
                yield True
            finally:
                # 使用Lua脚本安全释放锁，确保只有锁的持有者才能释放
                lua_script = """
                if redis.call("GET", KEYS[1]) == ARGV[1] then
                    return redis.call("DEL", KEYS[1])
                else
                    return 0
                end
                """
                result = await redis_pool.eval(
                    lua_script,
                    1,
                    full_lock_key,
                    identifier
                )
                if result:
                    logger.debug(f"🔓 分布式锁释放成功: {full_lock_key}")
                else:
                    logger.warning(f"⚠️ 分布式锁释放失败，可能已过期: {full_lock_key}")
        else:
            yield False
    except Exception as e:
        logger.error(f"❌ 分布式锁操作失败: {e}")
        yield False
 class AsyncRedisManager:
    def __init__(self):
        self.redis_pool: Optional[aioredis.Redis] = None
    async def connect(self) -> bool:
        """连接Redis
        Returns:
            bool: 连接是否成功
        """
        try:
            logger.info(f"🔴 正在连接Redis: {settings.redis_url}")
            self.redis_pool = await aioredis.create_redis_pool(
                settings.redis_url,
                encoding="utf-8",
                minsize=1,
                maxsize=settings.redis_max_connections,
                timeout=settings.redis_timeout
            )
            # 测试连接
            result = await self.redis_pool.ping()
            if result:
                logger.info("✅ Redis连接成功")
                return True
            else:
                logger.error("❌ Redis ping失败")
                return False
        except Exception as e:
            logger.error(f"❌ Redis连接失败: {e}")
            return False
    async def disconnect(self):
        """断开Redis连接"""
        if self.redis_pool:
            self.redis_pool.close()
            await self.redis_pool.wait_closed()
            logger.info("🔴 Redis连接已关闭")
    async def close(self):
        """断开Redis连接 (disconnect方法的别名)"""
        await self.disconnect()
    async def create_lock(self, key: str, timeout: int = None) -> AsyncRedisLock:
        """创建分布式锁"""
        if not self.redis_pool:
            raise RuntimeError("Redis client not connected")
        return AsyncRedisLock(self.redis_pool, key, timeout)
    async def get_redis_pool(self):
        """获取Redis连接池"""
        if not self.redis_pool:
            await self.connect()
        return self.redis_pool
 # 为了保持向后兼容，保留同步版本但标记为废弃
 class RedisLock:
-    def __init__(self, redis_client: redis.Redis, key: str, timeout: int = None):
+    """Redis分布式锁 (已废弃，请使用AsyncRedisLock)"""
    def __init__(self, redis_client, key: str, timeout: int = None):
        self.redis_client = redis_client
        self.key = f"lock:{key}"
        self.timeout = timeout or settings.redis_lock_timeout
        self.identifier = str(uuid.uuid4())
        self.acquired = False
-    async def acquire(self) -> bool:
+    def acquire(self, blocking: bool = True, timeout: int = None) -> bool:
        """获取分布式锁"""
        logger.debug(f"🔒 尝试获取Redis锁: {self.key}")
@@ -28,7 +229,7 @@ class RedisLock:
        end
        """
-        result = await self.redis_client.eval(
+        result = self.redis_client.eval(
            lua_script,
            1,
            self.key,
@@ -44,7 +245,7 @@ class RedisLock:
        return self.acquired
-    async def release(self) -> bool:
+    def release(self) -> bool:
        """释放分布式锁"""
        if not self.acquired:
            return False
@@ -57,7 +258,7 @@ class RedisLock:
        end
        """
-        result = await self.redis_client.eval(
+        result = self.redis_client.eval(
            lua_script,
            1,
            self.key,
@@ -67,39 +268,59 @@ class RedisLock:
        self.acquired = False
        return bool(result)
-    async def __aenter__(self):
+    def __enter__(self):
-        """异步上下文管理器入口"""
+        """同步上下文管理器入口"""
        retries = 0
        while retries < settings.redis_lock_max_retries:
-            if await self.acquire():
+            if self.acquire(blocking=False):
                return self
-            await asyncio.sleep(settings.redis_lock_retry_delay)
+            import time
            time.sleep(settings.redis_lock_retry_delay)
            retries += 1
        raise TimeoutError(f"Failed to acquire lock {self.key} after {retries} retries")
-    async def __aexit__(self, exc_type, exc_val, exc_tb):
+    def __exit__(self, exc_type, exc_val, exc_tb):
-        """异步上下文管理器出口"""
+        """同步上下文管理器出口"""
-        await self.release()
+        self.release()
 class RedisManager:
    def __init__(self):
-        self.redis_client: Optional[redis.Redis] = None
+        self.redis_client = None
-    async def connect(self):
+    def connect(self) -> bool:
-        """连接Redis"""
+        """连接Redis (已废弃，请使用AsyncRedisManager)
        Returns:
            bool: 连接是否成功
        """
        import redis
        try:
            logger.info(f"🔴 正在连接Redis: {settings.redis_url}")
            self.redis_client = redis.from_url(
                settings.redis_url,
                encoding="utf-8",
-            decode_responses=True
+                decode_responses=True,
                socket_connect_timeout=5,
                socket_timeout=5,
                retry_on_timeout=True
            )
-        await self.redis_client.ping()
+            result = self.redis_client.ping()
            if result:
                logger.info("✅ Redis连接成功")
                return True
            else:
                logger.error("❌ Redis ping失败")
                return False
        except Exception as e:
            logger.error(f"❌ Redis连接失败: {e}")
            return False
-    async def disconnect(self):
+    def disconnect(self):
        """断开Redis连接"""
        if self.redis_client:
-            await self.redis_client.close()
+            self.redis_client.close()
    def create_lock(self, key: str, timeout: int = None) -> RedisLock:
        """创建分布式锁"""
@@ -108,5 +329,6 @@ class RedisManager:
        return RedisLock(self.redis_client, key, timeout)
-# 全局Redis管理器实例
+# 全局Redis管理器实例（保持向后兼容）
 redis_manager = RedisManager()
 async_redis_manager = AsyncRedisManager()
--- a/app/routes.py
+++ b/app/routes.py
@@ -1,283 +1,42 @@
-from fastapi import APIRouter, Request, HTTPException, Depends, Path
+from fastapi import APIRouter, Request, Body, HTTPException, Depends, Path
 from sqlalchemy.ext.asyncio import AsyncSession
-import httpx
+from app.database import get_db
-import asyncio
+from app.models import CallbackResponse
 from typing import Dict, Any, Optional
 from app.database import get_db, CallbackLog, ExternalApiLog
 from app.models import CallbackRequest, CallbackResponse
 from app.redis_lock import redis_manager
 from app.config import settings
-from app.logger import get_logger
+from app.logger import get_routes_logger
 from app.callback_service import log_callback_request
-logger = get_logger("routes")
+logger = get_routes_logger()
 router = APIRouter()
 async def log_callback_request(
    db: AsyncSession,
    request: Request,
    site_id: str,
    callback_data: CallbackRequest
 ):
    """记录回调请求到数据库"""
    import json
    # 获取客户端IP地址
    client_ip = request.client.host if request.client else None
    client_port = request.client.port if request.client else None
    # 获取服务器IP地址
    server_ip = None
    server_port = None
    if hasattr(request, 'scope') and 'server' in request.scope:
        server_host, server_port_info = request.scope['server']
        server_ip = server_host
        server_port = server_port_info
    # 准备请求头信息（过滤敏感信息）
    request_headers = dict(request.headers)
    safe_headers = {}
    sensitive_headers = {'authorization', 'token', 'api-key', 'x-api-key', 'cookie'}
    for key, value in request_headers.items():
        if key.lower() in sensitive_headers:
            safe_headers[key] = "***REDACTED***"
        else:
            safe_headers[key] = value
    # 准备请求体信息
    request_body = callback_data.model_dump()
    # 记录site_id（JSON格式）
    logger.info(f"📝 site_id: {json.dumps(site_id, ensure_ascii=False)}")
    # 记录请求头（JSON格式）
    logger.info(f"📋 请求头: {json.dumps(safe_headers, ensure_ascii=False, indent=2)}")
    # 记录请求体（JSON格式）
    logger.info(f"📄 请求体: {json.dumps(request_body, ensure_ascii=False, indent=2)}")
    # 记录server_ip（JSON格式）
    server_info = {
        "ip": server_ip,
        "port": server_port
    }
    logger.info(f"🏠 server_ip: {json.dumps(server_info, ensure_ascii=False)}")
    # 记录client_ip（JSON格式）
    client_info = {
        "ip": client_ip,
        "port": client_port
    }
    logger.info(f"🖥️ client_ip: {json.dumps(client_info, ensure_ascii=False)}")
    # 记录callback_data（JSON格式）
    callback_data_json = {
        "count": callback_data.count,
        "data_count": len(callback_data.data) if callback_data.data else 0,
        "data_sample": callback_data.data[0].model_dump() if callback_data.data else None
    }
    logger.info(f"📦 callback_data: {json.dumps(callback_data_json, ensure_ascii=False, indent=2)}")
    try:
        # 保存到数据库
        callback_log = CallbackLog(
            site_id=site_id,
            remote_address=f"{client_ip}:{client_port}" if client_ip and client_port else client_ip,
            server_ip=f"{server_ip}:{server_port}" if server_ip and server_port else server_ip,
            request_url=str(request.url),
            request_headers=safe_headers,  # 保存过滤后的请求头
            request_body=request_body
        )
        db.add(callback_log)
        await db.commit()
        logger.info(f"✅ 回调请求记录成功保存到数据库，ID: {callback_log.id}")
    except Exception as e:
        logger.error(f"❌ 保存回调请求到数据库失败: {e}", exc_info=True)
        # 不重新抛出异常，避免影响主业务流程
        pass
 async def log_external_api_request(
    db: AsyncSession,
    request_url: str,
    request_headers: Dict[str, Any],
    request_body: Dict[str, Any],
    response_status: Optional[int],
    response_headers: Optional[Dict[str, Any]],
    response_body: Optional[str],
    retry_count: int = 0
 ):
    """记录外部API请求到数据库"""
    external_api_log = ExternalApiLog(
        request_url=request_url,
        request_headers=request_headers,
        request_body=request_body,
        response_status=response_status,
        response_headers=response_headers,
        response_body=response_body,
        retry_count=retry_count
    )
    db.add(external_api_log)
    await db.commit()
 async def call_external_api_with_retry(
    db: AsyncSession,
    request_body: Dict[str, Any],
    max_retries: int = None
 ) -> tuple[bool, int]:
    """
    调用外部API并支持重试机制
    Args:
        db: 数据库会话
        request_body: 请求体
        max_retries: 最大重试次数
    Returns:
        tuple[是否成功, 实际重试次数]
    """
    if max_retries is None:
        max_retries = settings.external_api_retry_max
    logger.info(f"🌐 开始调用外部API: {settings.external_api_url}, 最大重试次数: {max_retries}")
    headers = {
        "Content-Type": "application/json",
        "User-Agent": "AITalkCallbackService/1.0"
    }
    for attempt in range(max_retries + 1):  # +1 因为第一次不算重试
        try:
            logger.debug(f"📤 第{attempt + 1}次尝试调用外部API")
            async with httpx.AsyncClient(timeout=30.0) as client:
                response = await client.post(
                    settings.external_api_url,
                    headers=headers,
                    json=request_body
                )
                logger.debug(f"📥 外部API响应: status={response.status_code}")
                # 记录每次尝试的结果
                await log_external_api_request(
                    db=db,
                    request_url=settings.external_api_url,
                    request_headers=headers,
                    request_body=request_body,
                    response_status=response.status_code,
                    response_headers=dict(response.headers),
                    response_body=response.text,
                    retry_count=attempt
                )
                # 检查响应状态
                if response.status_code < 400:
                    logger.info(f"✅ 外部API调用成功，状态码: {response.status_code}")
                    return True, attempt
                else:
                    logger.warning(f"⚠️ 外部API返回错误状态码: {response.status_code}")
                    # 如果是最后一次尝试，直接返回失败
                    if attempt == max_retries:
                        logger.error(f"❌ 外部API调用最终失败，状态码: {response.status_code}")
                        return False, attempt
                    # 否则等待一段时间后重试
                    wait_time = 1 * (attempt + 1)  # 递增延迟
                    logger.info(f"⏳ 等待 {wait_time}秒 后重试...")
                    await asyncio.sleep(wait_time)
        except httpx.RequestError as e:
            logger.warning(f"⚠️ 外部API网络错误 (第{attempt + 1}次尝试): {e}")
            # 记录网络错误
            await log_external_api_request(
                db=db,
                request_url=settings.external_api_url,
                request_headers=headers,
                request_body=request_body,
                response_status=None,
                response_headers=None,
                response_body=f"RequestError: {str(e)}",
                retry_count=attempt
            )
            # 如果是最后一次尝试，直接返回失败
            if attempt == max_retries:
                logger.error(f"❌ 外部API调用最终失败: {e}")
                return False, attempt
            # 否则等待一段时间后重试
            wait_time = 1 * (attempt + 1)  # 递增延迟
            logger.info(f"⏳ 等待 {wait_time}秒 后重试...")
            await asyncio.sleep(wait_time)
    return False, max_retries
@router.post("/ai-talk/callback/{siteId}/failure", response_model=CallbackResponse)
 async def ai_talk_callback(
    callback_data: CallbackRequest,
    request: Request,
    callback_data: dict = Body(...),
    siteId: str = Path(..., description="站点ID"),
    db: AsyncSession = Depends(get_db)
 ):
    """
    AI Talk回调接口处理
    """
    logger.info(f"🔥 收到AI Talk回调请求: siteId={siteId}, count={callback_data.count}, data_count={len(callback_data.data)}")
    try:
        # 记录回调请求（包含siteId）
-        await log_callback_request(db, request, siteId, callback_data)
+        success = await log_callback_request(db, request, siteId, callback_data)
-        # 判断count是否大于等于阈值，如果是直接返回
+        if not success:
-        if callback_data.count >= settings.count_threshold:
+            logger.error("❌ 回调请求日志保存失败")
-            logger.info(f"✅ count={callback_data.count} >= {settings.count_threshold}，直接返回")
+            raise HTTPException(status_code=500, detail="回调请求日志保存失败")
        # 立即返回成功响应
        return CallbackResponse(
            success=True,
-                message=f"count={callback_data.count} >= {settings.count_threshold}，直接返回",
+            message="成功",
            processed=False,
            retry_count=0,
            site_id=siteId
        )
        logger.info(f"📞 count={callback_data.count} < {settings.count_threshold}，调用外部API")
        # count < 3，调用外部API，使用分布式锁防止并发调用
        async with redis_manager.create_lock(f"external_api_call_{siteId}_{callback_data.count}"):
            logger.debug(f"🔒 获取Redis锁成功: external_api_call_{siteId}_{callback_data.count}")
            # 调用外部API并支持重试
            request_body = callback_data.model_dump()
            success, retry_count = await call_external_api_with_retry(
                db=db,
                request_body=request_body,
                max_retries=settings.external_api_retry_max
            )
            if success:
                logger.info(f"✅ 外部API调用成功，重试次数: {retry_count}")
                return CallbackResponse(
                    success=True,
                    message="外部API调用成功",
                    processed=True,
                    retry_count=retry_count,
                    site_id=siteId
                )
            else:
                logger.error(f"❌ 外部API调用失败，已重试{retry_count}次")
                raise HTTPException(
                    status_code=500,
                    detail=f"外部API调用失败，已重试{retry_count}次"
                )
    except HTTPException:
        raise
    except Exception as e:
--- a/conftest.py
+++ b/conftest.py
@@ -5,7 +5,7 @@ from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
 from sqlalchemy.orm import sessionmaker
 # 测试数据库配置
-TEST_DATABASE_URL = "sqlite+aiosqlite:///./test.db"
+TEST_DATABASE_URL = "postgresql+asyncpg://postgres:12345@localhost:5432/test_ai_talk_callback"
@pytest.fixture(scope="session")
 def event_loop():
--- a/main.py
+++ b/main.py
@@ -1,33 +1,92 @@
 from datetime import datetime
 import os
 import traceback
 import signal
 import tempfile
 import time
 import subprocess
 import sys
 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
-from contextlib import asynccontextmanager
+import redis.asyncio as redis
-
+from sqlalchemy import text
 from app.config import settings
 from app.database import init_db
 from app.redis_lock import redis_manager
 from app.routes import router
 from app.logger import get_logger, LoggerManager
 # 初始化日志系统
 LoggerManager.setup_logging()
-logger = get_logger("main")
+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("🚀 应用启动中...")
-    try:
+    # Redis 连接对象
-        # 初始化数据库
+    redis_client = None
        logger.info("📊 初始化数据库连接...")
        await init_db()
        logger.info("✅ 数据库连接成功")
-        # 初始化Redis
+    try:
-        logger.info("🔴 初始化Redis连接...")
+        # 验证数据库连接
-        await redis_manager.connect()
+        logger.info("📊 验证数据库连接...")
-        logger.info("✅ Redis连接成功")
+        try:
            async with engine.begin() as conn:
                # 执行简单的查询来验证连接
                result = await conn.execute(text("SELECT 1"))
                connection_test = result.scalar()
                if connection_test == 1:
                    logger.info("✅ 数据库连接验证成功")
                else:
                    raise Exception("数据库连接测试失败")
        except Exception as db_error:
            logger.error(f"❌ 数据库连接验证失败: {db_error}")
            raise
        # 验证 Redis 连接
        logger.info("🔴 验证 Redis 连接...")
        try:
            redis_client = redis.from_url(settings.celery_broker_url)
            # 执行 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
        logger.info("✅ 服务初始化完成，FastAPI应用启动中...")
        logger.info(f"🎉 {settings.app_name} 启动完成!")
        yield
@@ -39,11 +98,14 @@ async def lifespan(app: FastAPI):
    finally:
        # 关闭时清理
        logger.info("🛑 应用关闭中...")
        # 关闭 Redis 连接
        if redis_client:
            try:
-            await redis_manager.disconnect()
+                await redis_client.close()
-            logger.info("✅ Redis连接已关闭")
+                logger.info("🔴 Redis 连接已关闭")
            except Exception as e:
-            logger.error(f"❌ Redis关闭时出错: {e}")
+                logger.warning(f"⚠️ 关闭 Redis 连接时出现警告: {e}")
        logger.info("👋 应用已关闭")
@@ -51,7 +113,18 @@ async def lifespan(app: FastAPI):
 app = FastAPI(
    title=settings.app_name,
    version="1.0.0",
-    lifespan=lifespan
+    lifespan=lifespan,
    docs_url=(
        "/docs"
        if not settings.disable_docs and settings.environment != "production"
        else None
    ),
    redoc_url=(
        "/redoc"
        if not settings.disable_docs and settings.environment != "production"
        else None
    ),
    openapi_url="/openapi.json" if not settings.disable_docs else None,
 )
 # 添加CORS中间件
@@ -69,21 +142,179 @@ app.include_router(router)
@app.get("/")
 async def root():
    # 检查环境，生产环境下禁用根接口
    if settings.environment == "production":
        logger.warning("🚫 生产环境下禁止访问根接口")
        from fastapi import HTTPException
        raise HTTPException(status_code=404, detail="Not Found")
    logger.info("📝 根接口被访问")
    return {"message": f"Welcome to {settings.app_name}"}
@app.get("/health")
 async def health_check():
    # 检查环境，生产环境下禁用健康检查接口
    if settings.environment == "production":
        logger.warning("🚫 生产环境下禁止访问健康检查接口")
        from fastapi import HTTPException
        raise HTTPException(status_code=404, detail="Not Found")
    logger.debug("💓 健康检查接口被访问")
    return {"status": "healthy"}
-if __name__ == "__main__":
+# 全局变量存储进程
-    import uvicorn
+processes = []
-    uvicorn.run(
+
-        "main:app",
+
-        host="0.0.0.0",
+def signal_handler(signum, frame):
-        port=8000,
+    """信号处理器，用于优雅关闭所有服务"""
-        reload=settings.debug
+    print(f"\n🛑 接收到信号 {signum}，正在关闭所有服务...")
    # 逆序关闭进程（最后启动的最先关闭）
    for i, process in enumerate(reversed(processes)):
        if process and process.poll() is None:  # 进程仍在运行
            print(f"🔄 正在关闭进程 {len(processes) - i}...")
            try:
                process.terminate()  # 发送 SIGTERM 信号
                process.wait(timeout=10)  # 等待最多10秒
                print(f"✅ 进程已关闭")
            except subprocess.TimeoutExpired:
                print(f"⚠️ 进程未在10秒内响应，强制关闭...")
                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)  # 终止信号
    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分钟软超时
            ]
        )
        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")}',
            ]
        )
        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",
                f"--broker={settings.celery_broker_url}",
                "flower",
                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",
            "main:app",
            "--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🌐 服务地址:")
        print("  🚀 API 服务: http://localhost:8000")
        if settings.environment != "production" and not settings.disable_docs:
            print("  📖 API 文档: http://localhost:8000/docs")
        if settings.flower_enabled:
            print(f"  📊 监控界面: {settings.flower_url}")
        print("\n💡 使用 Ctrl+C 可以优雅关闭所有服务")
        print("=" * 60)
        # 等待所有进程
        while True:
            # 检查是否有进程异常退出
            for i, process in enumerate(processes):
                if process and process.poll() is not None:
                    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()
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,15 +1,16 @@
-fastapi==0.104.1
+fastapi>=0.123.0
-uvicorn[standard]==0.24.0
+uvicorn[standard]>=0.38.0
-sqlalchemy==2.0.23
+sqlalchemy>=2.0.44
-asyncpg==0.29.0
+asyncpg>=0.31.0
-alembic==1.12.1
+alembic>=1.17.2
-redis==5.0.1
+celery>=5.6.0
-aioredis==2.0.1
+redis>=7.1.0
-pydantic==2.5.0
+aioredis==1.3.1
-pydantic-settings==2.1.0
+flower>=2.0.1
-python-multipart==0.0.6
+pydantic>=2.12.5
-httpx==0.25.2
+pydantic-settings>=2.12.0
-python-dotenv==1.0.0
+python-multipart>=0.0.20
-pytest==7.4.3
+httpx>=0.28.1
-pytest-asyncio==0.21.1
+python-dotenv>=1.2.1
-aiosqlite==0.19.0
+pytest>=9.0.2
 requests==2.32.5
--- a/setup.py
+++ b/setup.py
@@ -81,7 +81,7 @@ async def init_tables():
            print(f"📋 已创建的表: {', '.join(tables)}")
            # 检查必要的表
-            required_tables = ['callback_logs', 'external_api_logs']
+            required_tables = ['callback_failure_logs', 'callback_failure_data', 'external_api_logs']
            missing_tables = [table for table in required_tables if table not in tables]
            if missing_tables:
@@ -90,6 +90,20 @@ async def init_tables():
            else:
                print("✅ 所有必要的表都已创建")
            # 检查手机号索引
            result = await conn.execute(text("""
                SELECT indexname 
                FROM pg_indexes 
                WHERE tablename = 'callback_failure_data' 
                AND indexname = 'idx_phone_number'
            """))
            phone_index = result.fetchone()
            if phone_index:
                print("✅ 手机号索引 'idx_phone_number' 已创建")
            else:
                print("⚠️ 手机号索引 'idx_phone_number' 未找到")
        return True
    except Exception as e:
@@ -125,7 +139,7 @@ async def main():
    print("🚀 开始数据库初始化...")
    print(f"📋 配置信息:")
    print(f"   - 数据库URL: {settings.database_url}")
-    print(f"   - Redis URL: {settings.redis_url}")
+    print(f"   - Celery Broker URL: {settings.celery_broker_url}")
    print(f"   - 应用名称: {settings.app_name}")
    print(f"   - 外部API URL: {settings.external_api_url}")
    print()
@@ -150,10 +164,11 @@ async def main():
    print("🎉 数据库初始化完成！")
    print("\n📝 下一步:")
    print("   1. 配置 .env 文件中的数据库连接信息")
-    print("   2. 确保 Redis 服务正在运行")
+    print("   2. 确保 Redis 服务正在运行（用于Celery）")
-    print("   3. 运行应用: python main.py")
+    print("   3. 启动 Celery Worker: python celery_worker.py")
-    print(f"   4. 访问API文档: http://localhost:8000/docs")
+    print("   4. 运行应用: python main.py")
-    print(f"   5. 接口地址: POST /ai-talk/callback/{{siteId}}/failure")
+    print(f"   5. 访问API文档: http://localhost:8000/docs")
    print(f"   6. 接口地址: POST /ai-talk/callback/{{siteId}}/failure")
 if __name__ == "__main__":
--- a/test_callback.py
+++ b/test_callback.py
@@ -88,7 +88,7 @@ class TestAITalkCallback:
    @patch('app.routes.get_db')
    @patch('app.routes.redis_manager')
-    @patch('app.routes.log_callback_request')
+    @patch('app.callback_service.log_callback_request')
    def test_callback_count_below_threshold_success(
        self, 
        mock_log_callback, 
@@ -126,7 +126,7 @@ class TestAITalkCallback:
    @patch('app.routes.get_db')
    @patch('app.routes.redis_manager')
-    @patch('app.routes.log_callback_request')
+    @patch('app.callback_service.log_callback_request')
    def test_callback_count_above_threshold_direct_return(
        self, 
        mock_log_callback, 
@@ -154,7 +154,7 @@ class TestAITalkCallback:
    @patch('app.routes.get_db')
    @patch('app.routes.redis_manager')
-    @patch('app.routes.log_callback_request')
+    @patch('app.callback_service.log_callback_request')
    def test_callback_external_api_failure(
        self, 
        mock_log_callback, 
@@ -215,7 +215,7 @@ class TestAITalkCallback:
    @patch('app.routes.get_db')
    @patch('app.routes.redis_manager')
-    @patch('app.routes.log_callback_request')
+    @patch('app.callback_service.log_callback_request')
    def test_callback_empty_data_list(
        self, 
        mock_log_callback, 
@@ -251,7 +251,7 @@ class TestAITalkCallback:
    @patch('app.routes.get_db')
    @patch('app.routes.redis_manager')
-    @patch('app.routes.log_callback_request')
+    @patch('app.callback_service.log_callback_request')
    def test_callback_database_error(
        self, 
        mock_log_callback, 
@@ -273,7 +273,7 @@ class TestAITalkCallback:
    @patch('app.routes.get_db')
    @patch('app.routes.redis_manager')
-    @patch('app.routes.log_callback_request')
+    @patch('app.callback_service.log_callback_request')
    def test_callback_redis_lock_error(
        self, 
        mock_log_callback, 
@@ -306,7 +306,7 @@ class TestAITalkCallback:
        sample_callback_request
    ):
        """测试remote_address和server_ip字段是否正确保存"""
-        from app.database import CallbackLog
+        from app.database import CallbackFailureLog
        # 设置模拟对象
        mock_db = AsyncMock()
@@ -333,9 +333,9 @@ class TestAITalkCallback:
            assert mock_db.add.called
            assert mock_db.commit.called
-            # 获取传递给add的CallbackLog对象
+            # 获取传递给add的CallbackFailureLog对象
            call_args = mock_db.add.call_args[0][0]
-            assert isinstance(call_args, CallbackLog)
+            assert isinstance(call_args, CallbackFailureLog)
            # 验证新字段存在（可能为None，因为测试环境）
            assert hasattr(call_args, 'remote_address')
@@ -356,7 +356,7 @@ class TestAITalkCallback:
        sample_callback_request
    ):
        """测试JSON格式日志记录功能"""
-        from app.database import CallbackLog
+        from app.database import CallbackFailureLog
        # 设置模拟对象
        mock_db = AsyncMock()
@@ -382,9 +382,9 @@ class TestAITalkCallback:
            # 验证log_callback_request被调用
            assert mock_db.add.called
-            # 获取传递给add的CallbackLog对象
+            # 获取传递给add的CallbackFailureLog对象
            call_args = mock_db.add.call_args[0][0]
-            assert isinstance(call_args, CallbackLog)
+            assert isinstance(call_args, CallbackFailureLog)
            # 验证数据结构适合JSON序列化
            import json
--- a/test_redis.py
+++ b/test_redis.py
@@ -0,0 +1,105 @@
 #!/usr/bin/env python3
 """
 Redis 连接测试脚本
 """
 import os
 import sys
 import asyncio
 # 添加项目根目录到Python路径
 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
 from app.config import settings
 from app.logger import get_logger
 logger = get_logger("redis_test")
 async def test_redis_connection():
    """测试 Redis 连接"""
    logger.info("🔴 测试 Redis 连接...")
    try:
        import redis.asyncio as redis
        # 创建 Redis 客户端
        redis_client = redis.from_url(settings.celery_broker_url)
        # 测试连接
        await redis_client.ping()
        logger.info("✅ Redis 连接成功")
        # 测试基本操作
        test_key = "test_connection"
        test_value = "Hello Redis!"
        await redis_client.set(test_key, test_value, ex=60)  # 设置60秒过期
        retrieved_value = await redis_client.get(test_key)
        if retrieved_value == test_value:
            logger.info("✅ Redis 读写操作正常")
        else:
            logger.warning("⚠️ Redis 读写操作异常")
        # 清理测试数据
        await redis_client.delete(test_key)
        # 关闭连接
        await redis_client.close()
        return True
    except Exception as e:
        logger.error(f"❌ Redis 连接失败: {e}")
        return False
 async def test_celery_redis():
    """测试 Celery 相关的 Redis 操作"""
    logger.info("🌿 测试 Celery Redis 配置...")
    try:
        import redis.asyncio as redis
        # 测试 Broker 连接
        broker_client = redis.from_url(settings.celery_broker_url)
        await broker_client.ping()
        logger.info(f"✅ Celery Broker (Redis) 连接成功: {settings.celery_broker_url}")
        # 测试 Backend 连接
        backend_client = redis.from_url(settings.celery_result_backend)
        await backend_client.ping()
        logger.info(f"✅ Celery Backend (Redis) 连接成功: {settings.celery_result_backend}")
        # 关闭连接
        await broker_client.close()
        await backend_client.close()
        return True
    except Exception as e:
        logger.error(f"❌ Celery Redis 测试失败: {e}")
        return False
 async def main():
    """主函数"""
    logger.info("🧪 开始 Redis 连接测试...")
    # 显示配置信息
    logger.info(f"📡 Celery Broker: {settings.celery_broker_url}")
    logger.info(f"💾 Celery Backend: {settings.celery_result_backend}")
    # 测试基础连接
    basic_ok = await test_redis_connection()
    # 测试 Celery 连接
    celery_ok = await test_celery_redis()
    if basic_ok and celery_ok:
        logger.info("🎉 所有 Redis 测试通过！")
        return 0
    else:
        logger.error("❌ Redis 测试失败")
        return 1
 if __name__ == "__main__":
    sys.exit(asyncio.run(main()))
--- a/test_redis_manager_connection.py
+++ b/test_redis_manager_connection.py
@@ -0,0 +1,126 @@
 #!/usr/bin/env python3
 """
 测试RedisManager连接功能
 验证connect方法的返回值
 """
 from app.redis_lock import redis_manager
 from app.logger import get_redis_logger
 logger = get_redis_logger()
 def test_connection():
    """测试连接"""
    logger.info("🧪 测试Redis连接...")
    # 测试连接
    success = redis_manager.connect()
    if success:
        logger.info("✅ 连接成功")
        # 测试创建锁
        try:
            lock = redis_manager.create_lock("test:lock", timeout=10)
            logger.info("✅ 锁创建成功")
            # 尝试获取锁
            acquired = lock.acquire(blocking=False)
            if acquired:
                logger.info("✅ 锁获取成功")
                # 释放锁
                released = lock.release()
                if released:
                    logger.info("✅ 锁释放成功")
                else:
                    logger.error("❌ 锁释放失败")
                    return False
            else:
                logger.error("❌ 锁获取失败")
                return False
        except Exception as e:
            logger.error(f"❌ 锁操作失败: {e}")
            return False
    else:
        logger.error("❌ 连接失败")
        return False
    return True
 def test_connection_failures():
    """测试连接失败场景"""
    logger.info("🧪 测试连接失败处理...")
    # 直接测试连接一个不存在的主机
    from app.config import settings
    original_url = settings.redis_url
    try:
        # 创建一个新的RedisManager实例用于测试失败场景
        from app.redis_lock import RedisManager
        test_manager = RedisManager()
        # 模拟连接失败 - 直接使用无效的URL
        import redis
        invalid_redis_client = redis.from_url(
            "redis://invalid_host:6379/0",
            encoding="utf-8",
            decode_responses=True,
            socket_connect_timeout=1,
            socket_timeout=1
        )
        test_manager.redis_client = invalid_redis_client
        # 尝试ping，这应该会失败
        try:
            result = test_manager.redis_client.ping()
            logger.error("❌ 连接应该失败但成功了")
            return False
        except Exception:
            logger.info("✅ 连接失败处理正确")
            return True
    except Exception as e:
        logger.info("✅ 连接失败处理正确")
        return True
 def main():
    """主函数"""
    logger.info("🚀 开始RedisManager连接测试")
    # 测试1: 连接
    logger.info("\n" + "="*50)
    logger.info("测试1: 连接")
    logger.info("="*50)
    connection_success = test_connection()
    # 测试2: 连接失败处理
    logger.info("\n" + "="*50)
    logger.info("测试2: 连接失败处理")
    logger.info("="*50)
    failure_success = test_connection_failures()
    # 断开连接
    redis_manager.disconnect()
    logger.info("🔌 Redis连接已断开")
    # 总结
    logger.info("\n" + "="*50)
    logger.info("测试结果总结")
    logger.info("="*50)
    if connection_success and failure_success:
        logger.info("🎉 所有测试通过！RedisManager连接功能正常")
        return True
    else:
        logger.error("❌ 部分测试失败，请检查实现")
        return False
 if __name__ == "__main__":
    success = main()
    exit(0 if success else 1)
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
liangtianyu	6e9dc7f3ca	删除无效任务	2025-12-13 09:27:44 +08:00
liangtianyu	4d3c72a696	注释目前不使用的任务	2025-12-13 09:22:19 +08:00
mark.tian	495515ea8f	改为每天上午9点执行	2025-12-12 06:14:26 +08:00
mark.tian	936b08ceb5	日志轮转改为0点	2025-12-11 18:10:10 +08:00
mark.tian	1edae9f5f5	优化日志和配置	2025-12-11 18:08:20 +08:00
mark.tian	06aa1cbc17	优化日志输出	2025-12-11 16:18:38 +08:00
mark.tian	ba3c93c896	测试通过	2025-12-11 12:45:58 +08:00
mark.tian	b2320cdfb7	保存测试配置	2025-12-11 12:30:28 +08:00
mark.tian	710e16d454	api配置测试代码	2025-12-11 12:26:47 +08:00
mark.tian	5616d32024	改造配置文件，去掉敏感信息	2025-12-11 11:23:26 +08:00
mark.tian	851e92ad3a	优化回调请求处理为已完成	2025-12-10 12:18:28 +08:00
mark.tian	0533a455cd	推送通话记录到DTC调试通过	2025-12-10 11:56:05 +08:00
mark.tian	572cc3113f	celery任务改为异步访问数据库	2025-12-10 11:40:56 +08:00
mark.tian	d3d3502f5a	增加忽略文件	2025-12-10 10:17:35 +08:00
mark.tian	1a46b55e42	改造启动方式为一键启动所有服务	2025-12-10 09:50:31 +08:00
mark.tian	32e1e8fd24	优化启动代码	2025-12-10 08:57:36 +08:00
mark.tian	fdb7e1f801	去掉无用的记录文件	2025-12-10 08:25:17 +08:00
mark.tian	573aa29f36	给每个定时任务增加一个开关	2025-12-10 08:19:04 +08:00
mark.tian	37c8503f40	Merge branch 'main' of https://git.shiliulizi.cn/mark.tian/ai-talk-callback	2025-12-10 08:11:40 +08:00
mark.tian	3fbbf1285c	禁用外呼task启动	2025-12-10 08:11:24 +08:00
liangtianyu	8d58ab1c74	更新依赖	2025-12-09 22:33:09 +08:00
mark.tian	34520d4113	调接口的任务改为每天上午9点执行一次	2025-12-09 21:58:49 +08:00
mark.tian	ae9b548311	增加执行外部api的定时任务	2025-12-09 21:32:07 +08:00
mark.tian	d00bb5e475	测试	2025-12-09 10:01:00 +08:00
mark.tian	ad1ae1a54a	测试	2025-12-09 09:35:38 +08:00
mark.tian	6853624bd2	定时任务增加redis连接	2025-12-09 08:05:45 +08:00
mark.tian	e146dfaadf	格式化	2025-12-08 15:17:37 +08:00
mark.tian	5cf3b008d3	celery worker在windows上启动失败，增加参数--pool=solo	2025-12-08 15:16:17 +08:00
mark.tian	f9c74a1dda	统一日志类的定义	2025-12-08 13:37:04 +08:00
mark.tian	a55c321fce	增加停止服务开关	2025-12-08 13:03:02 +08:00
mark.tian	4b0ab3260a	梳理main.py中的服务	2025-12-08 12:31:05 +08:00
mark.tian	43ff4aa5f9	增加组装推送消息的逻辑	2025-12-08 09:49:36 +08:00
mark.tian	038363a17e	优化推送任务	2025-12-05 06:20:14 +08:00
mark.tian	d3bbe5f649	推送数据给DTC改为独立执行的任务	2025-12-04 15:05:56 +08:00
mark.tian	929f00b34e	通话详情增加user_id	2025-12-04 10:21:04 +08:00
liangtianyu	b2f2b9dbe9	修改model默认时间为服务器时间	2025-12-04 08:49:34 +08:00
mark.tian	da56319f85	修复参数错误	2025-12-04 08:40:32 +08:00
mark.tian	7aeb84df37	去掉redis 启动时只检查数据库连接不更新表结构	2025-12-04 08:35:33 +08:00
mark.tian	6ac998ebdf	保存请求失败返回500 调整表结构	2025-12-04 08:14:21 +08:00
mark.tian	f1f694bd63	独立出调用外部接口	2025-12-03 21:58:16 +08:00
mark.tian	703408c808	接口请求体改用dict处理保存每次通话信息到库	2025-12-03 21:56:06 +08:00
mark.tian	70444d5665	记录未打通通过记录的保存	2025-12-03 18:17:58 +08:00
mark.tian	3b3b44889b	测试不用sqlite，改为postgresql	2025-12-03 12:26:41 +08:00
mark.tian	f3561c5605	删掉.env文件	2025-12-03 12:18:25 +08:00
mark.tian	7d9c7ef573	请求头和请求体改为用原始请求里面的	2025-12-03 12:13:01 +08:00
liangtianyu	f4b40e7ff3	增加接口文档禁用	2025-12-03 10:24:55 +08:00
liangtianyu	099f7f44a2	调整依赖版本	2025-12-03 09:19:17 +08:00
mark.tian	05395f2688	加日志	2025-12-03 08:58:44 +08:00
mark.tian	4c544c296c	增加外部接口调用开关	2025-12-03 08:52:13 +08:00
mark.tian	5fc68ddb58	建立回调日志表与外部接口日志表之间的关系	2025-12-02 22:32:05 +08:00