70 lines
2.0 KiB
Python
70 lines
2.0 KiB
Python
import logging
|
|
|
|
from celery import shared_task
|
|
from django.contrib.auth import get_user_model
|
|
from django.utils import timezone
|
|
from trends.trends import TREND_REGISTRY
|
|
from trends.utils import compute_and_save_trend, get_supported_periods
|
|
|
|
logger = logging.getLogger(__name__)
|
|
User = get_user_model()
|
|
|
|
|
|
@shared_task
|
|
def compute_all_trends():
|
|
user_ids = list(User.objects.filter(is_active=True).values_list("id", flat=True))
|
|
logger.info("Dispatching trend computation for %d users", len(user_ids))
|
|
for uid in user_ids:
|
|
compute_user_trends.delay(uid)
|
|
|
|
|
|
@shared_task
|
|
def compute_user_trends(user_id):
|
|
try:
|
|
user = User.objects.get(id=user_id)
|
|
except User.DoesNotExist:
|
|
logger.warning("User %s not found, skipping trends", user_id)
|
|
return
|
|
|
|
total = len(TREND_REGISTRY)
|
|
logger.info(
|
|
"Computing %d trends for user %s (%d)",
|
|
total,
|
|
user,
|
|
user_id,
|
|
)
|
|
|
|
for idx, (slug, _) in enumerate(TREND_REGISTRY.items(), start=1):
|
|
compute_single_trend.delay(user_id, slug)
|
|
|
|
logger.info("Dispatched all %d trends for user %s (%d)", total, user, user_id)
|
|
|
|
|
|
@shared_task
|
|
def compute_single_trend(user_id, slug):
|
|
try:
|
|
user = User.objects.get(id=user_id)
|
|
except User.DoesNotExist:
|
|
logger.warning("User %d not found for trend '%s', skipping", user_id, slug)
|
|
return
|
|
|
|
if slug not in TREND_REGISTRY:
|
|
logger.warning("Unknown trend slug '%s' for user %d", slug, user_id)
|
|
return
|
|
|
|
periods = get_supported_periods(slug)
|
|
|
|
for period in periods:
|
|
logger.info("[%s/%s] Computing for user %d...", slug, period, user_id)
|
|
try:
|
|
elapsed = compute_and_save_trend(user, slug, period)
|
|
logger.info(
|
|
"[%s/%s] Completed for user %d in %.1fs",
|
|
slug,
|
|
period,
|
|
user_id,
|
|
elapsed,
|
|
)
|
|
except Exception:
|
|
logger.exception("[%s/%s] Failed for user %d", slug, period, user_id)
|