新增任务,手动触发定时任务更新
This commit is contained in:
@@ -44,6 +44,9 @@ _registered_tasks: dict = {} # task_id -> cron_expression
|
|||||||
SIGNIN_LOG_RETAIN_DAYS = int(os.getenv("SIGNIN_LOG_RETAIN_DAYS", "30"))
|
SIGNIN_LOG_RETAIN_DAYS = int(os.getenv("SIGNIN_LOG_RETAIN_DAYS", "30"))
|
||||||
CLEANUP_BATCH_SIZE = 1000
|
CLEANUP_BATCH_SIZE = 1000
|
||||||
|
|
||||||
|
# Redis 订阅线程是否运行
|
||||||
|
_redis_listener_running = False
|
||||||
|
|
||||||
WEIBO_HEADERS = {
|
WEIBO_HEADERS = {
|
||||||
"User-Agent": (
|
"User-Agent": (
|
||||||
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
|
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
|
||||||
@@ -384,10 +387,49 @@ async def _async_cleanup():
|
|||||||
return {"deleted": total_deleted, "cutoff": cutoff.isoformat()}
|
return {"deleted": total_deleted, "cutoff": cutoff.isoformat()}
|
||||||
|
|
||||||
|
|
||||||
|
# =============== Redis 订阅(实时同步任务变更) ===============
|
||||||
|
|
||||||
|
def _start_redis_listener():
|
||||||
|
"""启动 Redis pub/sub 监听线程,收到 task_updates 消息后立即同步。"""
|
||||||
|
global _redis_listener_running
|
||||||
|
if not shared_settings.USE_REDIS or not shared_settings.REDIS_URL:
|
||||||
|
logger.info("⚠️ Redis 未启用,任务变更仅靠 5 分钟轮询同步")
|
||||||
|
return
|
||||||
|
|
||||||
|
import threading
|
||||||
|
import redis
|
||||||
|
|
||||||
|
def _listen():
|
||||||
|
global _redis_listener_running
|
||||||
|
_redis_listener_running = True
|
||||||
|
logger.info("📡 Redis 订阅线程启动,监听 task_updates 频道")
|
||||||
|
while _redis_listener_running:
|
||||||
|
try:
|
||||||
|
r = redis.from_url(shared_settings.REDIS_URL, decode_responses=True)
|
||||||
|
pubsub = r.pubsub()
|
||||||
|
pubsub.subscribe("task_updates")
|
||||||
|
for message in pubsub.listen():
|
||||||
|
if not _redis_listener_running:
|
||||||
|
break
|
||||||
|
if message["type"] == "message":
|
||||||
|
logger.info(f"📡 收到任务变更通知: {message['data'][:200]}")
|
||||||
|
sync_db_tasks()
|
||||||
|
pubsub.close()
|
||||||
|
r.close()
|
||||||
|
except Exception as e:
|
||||||
|
logger.warning(f"Redis 订阅异常,5 秒后重连: {e}")
|
||||||
|
_time.sleep(5)
|
||||||
|
|
||||||
|
t = threading.Thread(target=_listen, daemon=True, name="redis-listener")
|
||||||
|
t.start()
|
||||||
|
|
||||||
|
|
||||||
# =============== 启动入口 ===============
|
# =============== 启动入口 ===============
|
||||||
|
|
||||||
def _shutdown(signum, frame):
|
def _shutdown(signum, frame):
|
||||||
|
global _redis_listener_running
|
||||||
logger.info(f"收到信号 {signum},正在关闭调度器...")
|
logger.info(f"收到信号 {signum},正在关闭调度器...")
|
||||||
|
_redis_listener_running = False
|
||||||
scheduler.shutdown(wait=False)
|
scheduler.shutdown(wait=False)
|
||||||
sys.exit(0)
|
sys.exit(0)
|
||||||
|
|
||||||
@@ -407,6 +449,9 @@ if __name__ == "__main__":
|
|||||||
# 首次同步 DB 任务
|
# 首次同步 DB 任务
|
||||||
sync_db_tasks()
|
sync_db_tasks()
|
||||||
|
|
||||||
|
# 启动 Redis 订阅,实时接收任务变更通知
|
||||||
|
_start_redis_listener()
|
||||||
|
|
||||||
# 每 5 分钟重新同步 DB(处理新增/删除/修改的任务)
|
# 每 5 分钟重新同步 DB(处理新增/删除/修改的任务)
|
||||||
scheduler.add_job(
|
scheduler.add_job(
|
||||||
sync_db_tasks,
|
sync_db_tasks,
|
||||||
|
|||||||
Reference in New Issue
Block a user