From f0f561d003a3aa403ba562fc281afede5147b1d2 Mon Sep 17 00:00:00 2001
From: xingc <xingcweb@vip.qq.com>
Date: Mon, 27 Oct 2025 17:24:03 +0800
Subject: [PATCH] =?UTF-8?q?feat(tasks):=20=E6=96=B0=E5=A2=9E=E8=BE=85?=
 =?UTF-8?q?=E5=8A=A9=E5=AE=9A=E6=97=B6=E4=BB=BB=E5=8A=A1=EF=BC=8C=E4=BC=9A?=
 =?UTF-8?q?=E5=91=98=E7=A7=AF=E5=88=86=E8=B5=A0=E9=80=81=E3=80=81=E6=90=9C?=
 =?UTF-8?q?=E5=8D=A1=E7=BC=93=E5=AD=98=E8=BF=87=E6=9C=9F=E3=80=81redis?=
 =?UTF-8?q?=E9=94=AE=E8=BF=87=E6=9C=9F=E5=AE=9A=E4=B9=89?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 tasks.py | 161 +++++++++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 161 insertions(+)
 create mode 100644 tasks.py

diff --git a/tasks.py b/tasks.py
new file mode 100644
index 0000000..244f8b6
--- /dev/null
+++ b/tasks.py
@@ -0,0 +1,161 @@
+# -*- coding: utf-8 -*-
+# @File: tasks.py
+# @Date: 2025/8/14 02:47
+# @Software: PyCharm
+# @Author: xingc
+# @Desc: 辅助定时任务
+import asyncio
+import logging
+from datetime import datetime, timedelta
+
+from apscheduler.schedulers.asyncio import AsyncIOScheduler
+from apscheduler.triggers.cron import CronTrigger
+from sqlalchemy import func, desc, select, update
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from core.db.crud import CRUD
+from core.db.engine import get_cache, async_db_session
+from core.db.models import User, SearchCardCache, PointsTransaction
+from core.utils.helper import generate_keyname
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s | %(name)s | %(levelname)s - %(message)s",  # 日志格式
+)
+logger = logging.getLogger(__name__)
+
+
+async def check_cache_key_expire():
+    """检测需要定义过期的键"""
+    redis_conn = await get_cache()
+    current_date = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
+    second = ((current_date + timedelta(hours=1)) - current_date).total_seconds()
+    expire_keynames = set()
+    async with redis_conn:
+        keynames = await redis_conn.keys(generate_keyname('*', use_today=True))
+        logger.info(keynames)
+        for keyname in keynames:
+            if await redis_conn.ttl(keyname) == -1:
+                expire_keynames.add(keyname)
+                await redis_conn.expire(keyname, int(second))
+    logger.info('【任务】：检测需要定义过期的键\t'
+                f'键名：{expire_keynames}\t'
+                f'数量：{len(expire_keynames)}')
+
+
+@async_db_session()
+async def daily_member_points(db_session: AsyncSession, batch_step: int = 50):
+    """会员每日积分赠送"""
+    member_total_smt = select(
+        func.count(User.user_id),
+    ).where(
+        User.vip_expire_date > User.created_at
+    )
+    member_total_result = await db_session.execute(member_total_smt)
+    member_total = member_total_result.scalar()
+    if member_total < 1:
+        logger.info("【任务】：会员每日积分赠送\t没有需要更新积分的会员用户")
+        return
+
+    page = 0
+    success_total = 0
+    while True:
+        stmt = (select(User).filter(
+            User.vip_expire_date > User.created_at
+        ).with_for_update().offset(page * batch_step)
+                .limit(batch_step)
+                .order_by(User.created_at.asc()))
+        result = await db_session.execute(stmt)
+        users = result.scalars().all()
+        if not users:
+            break
+
+        user_ids = {user.user_id for user in users}
+        try:
+            update_stmt = update(
+                User
+            ).values(
+                points=User.points + 1
+            ).where(
+                User.user_id.in_(user_ids),
+            )
+            await db_session.execute(update_stmt)
+
+            current_date = datetime.now()
+            points_records = [
+                PointsTransaction(
+                    user_id=user.user_id,
+                    transaction_type=3,
+                    points=1,
+                    balance=user.points + 1,
+                    description="会员每日积分赠送",
+                    created_at=current_date
+                )
+                for user in users
+            ]
+            db_session.add_all(points_records)
+            await db_session.commit()
+            success_total += len(users)
+
+        except Exception as e:
+            logger.info(f"【任务】：会员每日积分赠送\tuser_ids -> {user_ids}\terror -> {e}")
+            await db_session.rollback()
+            continue
+
+        finally:
+            page += 1
+
+    logger.info(f"【任务】：会员每日积分赠送\t成功为 {success_total} 个会员用户赠送积分")
+
+
+@async_db_session()
+async def clear_expire_cache_card(db_session: AsyncSession):
+    """清除过期卡片缓存信息"""
+    expire_date = datetime.now() + timedelta(days=30)
+    db_crud = CRUD(db_session, SearchCardCache)
+    result = await db_crud.deletes(
+        filters={
+            'created_at__lt': expire_date
+        }
+    )
+    logger.info('【任务】：清除过期卡片缓存信息\t'
+                f'过期时间：{expire_date}\t'
+                f'结果：{result}')
+
+
+async def main():
+    scheduler = AsyncIOScheduler()
+    # crontab
+    scheduler.add_job(
+        check_cache_key_expire,
+        trigger=CronTrigger(hour='*/1'),
+        id='check_cache_key_expire'
+    )
+
+    scheduler.add_job(
+        daily_member_points,
+        trigger=CronTrigger(day='*/1', hour='17', minute='*/1'),
+        id='daily_member_points'
+    )
+
+    scheduler.add_job(
+        clear_expire_cache_card,
+        trigger=CronTrigger(day='*/1', hour='3'),
+        id='clear_expire_cache_card'
+    )
+
+    scheduler.start()
+
+    try:
+        while True:
+            await asyncio.sleep(1)
+    except (KeyboardInterrupt, SystemExit):
+        scheduler.shutdown()
+        # await engine.dispose()
+
+
+if __name__ == '__main__':
+    try:
+        asyncio.run(main())
+    except KeyboardInterrupt:
+        logger.info("程序已停止")