95 lines
2.7 KiB
Python
95 lines
2.7 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
|
|
|
|
try:
|
|
profile = user.profile
|
|
if profile.trends_disabled:
|
|
logger.info("User %s (%d) has trends disabled, skipping", user, user_id)
|
|
return
|
|
disabled = set(profile.disabled_trends or [])
|
|
except Exception:
|
|
disabled = set()
|
|
|
|
active_slugs = [s for s in TREND_REGISTRY if s not in disabled]
|
|
total = len(active_slugs)
|
|
logger.info(
|
|
"Computing %d trends for user %s (%d)",
|
|
total,
|
|
user,
|
|
user_id,
|
|
)
|
|
|
|
for idx, slug in enumerate(active_slugs, 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
|
|
|
|
try:
|
|
profile = user.profile
|
|
if profile.trends_disabled:
|
|
logger.info(
|
|
"User %d has trends disabled, skipping '%s'", user_id, slug
|
|
)
|
|
return
|
|
if slug in (profile.disabled_trends or []):
|
|
logger.info(
|
|
"User %d has trend '%s' disabled, skipping", user_id, slug
|
|
)
|
|
return
|
|
except Exception:
|
|
pass
|
|
|
|
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)
|