162 lines
4.9 KiB
Python
162 lines
4.9 KiB
Python
# -*- 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='8'),
|
|
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("程序已停止")
|