定时调外部任务开启api

test_beat_schedule.py

@@ -0,0 +1,53 @@
+#!/usr/bin/env python3
+"""测试新的调度配置 - 运行2次观察执行情况"""
+
+import subprocess
+import time
+import sys
+import threading
+from datetime import datetime
+
+def run_beat():
+    """运行Celery Beat调度器"""
+    try:
+        print("🚀 启动Celery Beat调度器...")
+        process = subprocess.Popen([
+            sys.executable, '-m', 'celery', '-A', 'celery_app', 'beat', '--loglevel=info'
+        ], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1)
+        
+        # 实时输出日志
+        for line in iter(process.stdout.readline, ''):
+            print(f"[BEAT] {line.strip()}")
+            
+    except KeyboardInterrupt:
+        print("\n🛑 停止Beat调度器")
+        process.terminate()
+    except Exception as e:
+        print(f"❌ Beat启动失败: {e}")
+
+def test_schedule():
+    """测试新的调度配置"""
+    print("=" * 60)
+    print(f"测试新的调度配置 - 每2分钟执行一次")
+    print(f"开始时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
+    print("=" * 60)
+    print()
+    
+    print("⚠️  注意: 这是测试模式，将运行大约4分钟来观察2次任务执行")
+    print("💡 如果Redis没有运行，请先启动Redis服务")
+    print()
+    
+    # 启动Beat调度器
+    beat_thread = threading.Thread(target=run_beat)
+    beat_thread.daemon = True
+    beat_thread.start()
+    
+    # 等待4分钟观察执行
+    try:
+        time.sleep(240)  # 4分钟，应该能看到2次任务执行
+        print("\n✅ 测试完成！")
+    except KeyboardInterrupt:
+        print("\n🛑 手动停止测试")
+
+if __name__ == "__main__":
+    test_schedule()