44 lines
1.0 KiB
Python
44 lines
1.0 KiB
Python
# Create your tasks here
|
|
from __future__ import absolute_import, unicode_literals
|
|
from datetime import timedelta
|
|
from apps.utils.tasks import CustomTask
|
|
from apps.third.models import Tlog
|
|
from celery import shared_task
|
|
from django.utils import timezone
|
|
from apps.third.dahua import dhClient
|
|
from apps.third.xunxi import xxClient
|
|
from apps.third.speaker import spClient
|
|
from apps.third.king.k import kingClient
|
|
|
|
|
|
@shared_task(base=CustomTask)
|
|
def clear_tlog(days: int = 7):
|
|
"""清除N天前的日志记录,默认七天
|
|
|
|
清除N天前的日志记录
|
|
"""
|
|
now = timezone.now()
|
|
days7_ago = now - timedelta(days=days)
|
|
Tlog.objects.filter(create_time__lte=days7_ago).delete()
|
|
|
|
|
|
@shared_task(base=CustomTask)
|
|
def get_third_token():
|
|
xxClient._get_token()
|
|
dhClient._get_token()
|
|
spClient._get_token()
|
|
|
|
|
|
@shared_task(base=CustomTask)
|
|
def get_king_token():
|
|
kingClient._get_token()
|
|
|
|
# @shared_task(base=CustomTask)
|
|
# def get_dh_token():
|
|
# dhClient._get_token()
|
|
|
|
|
|
# @shared_task(base=CustomTask)
|
|
# def get_sp_token():
|
|
# spClient._get_token()
|