vrobbler/vrobbler/apps/trends/trends/mood.py

from collections import Counter, defaultdict
from datetime import timedelta

from django.db.models import Count, Q
from django.db.models.functions import Extract
from django.utils import timezone
from scrobbles.models import Scrobble


def _mood_scrobbles(user, period="all_time"):
    from trends.utils import get_date_range

    start, end = get_date_range(period)
    filters = Q(user=user, media_type=Scrobble.MediaType.MOOD)
    if start:
        filters &= Q(timestamp__gte=start)
    if end:
        filters &= Q(timestamp__lte=end)
    return Scrobble.objects.filter(filters).select_related("mood")


def _avg_quality(values):
    if not values:
        return 0.0
    return round(sum(values) / len(values), 2)


def compute_mood_trajectory(user, period="all_time"):
    scrobbles = _mood_scrobbles(user, period).order_by("timestamp")
    by_date = defaultdict(list)
    for s in scrobbles:
        quality = s.log.get("mood_quality")
        if quality is not None:
            day_key = s.timestamp.strftime("%Y-%m-%d")
            by_date[day_key].append(quality)

    trajectory = []
    for date_key in sorted(by_date):
        values = by_date[date_key]
        trajectory.append(
            {
                "date": date_key,
                "avg_quality": _avg_quality(values),
                "count": len(values),
            }
        )

    return {"trajectory": trajectory}


def compute_mood_by_time(user, period="all_time"):
    scrobbles = _mood_scrobbles(user, period)
    by_hour = defaultdict(list)
    by_day = defaultdict(list)

    for s in scrobbles:
        quality = s.log.get("mood_quality")
        if quality is not None and s.timestamp:
            by_hour[s.timestamp.hour].append(quality)
            by_day[s.timestamp.isoweekday()].append(quality)

    hours = []
    for h in range(24):
        vals = by_hour.get(h, [])
        hours.append(
            {
                "hour": h,
                "avg_quality": _avg_quality(vals),
                "count": len(vals),
            }
        )

    DAY_NAMES = {
        1: "Monday",
        2: "Tuesday",
        3: "Wednesday",
        4: "Thursday",
        5: "Friday",
        6: "Saturday",
        7: "Sunday",
    }
    days = []
    for d in range(1, 8):
        vals = by_day.get(d, [])
        days.append(
            {
                "day_index": d,
                "day_name": DAY_NAMES[d],
                "avg_quality": _avg_quality(vals),
                "count": len(vals),
            }
        )

    return {"hours": hours, "days": days}


def compute_mood_distribution(user, period="all_time"):
    scrobbles = _mood_scrobbles(user, period)
    mood_counts = Counter()
    type_counts = Counter()

    for s in scrobbles:
        if s.mood and s.mood.title:
            mood_counts[s.mood.title] += 1
        mood_type = s.log.get("mood_type")
        if mood_type:
            type_counts[mood_type] += 1

    moods = [
        {"mood": mood, "count": count}
        for mood, count in mood_counts.most_common()
    ]
    total = sum(mood_counts.values())

    return {
        "moods": moods,
        "total": total,
        "positive_count": type_counts.get("positive", 0),
        "negative_count": type_counts.get("negative", 0),
    }


def compute_mood_streaks(user, period="all_time"):
    scrobbles = list(
        _mood_scrobbles(user, period).order_by("timestamp")
    )
    if not scrobbles:
        return {"streaks": [], "current_streak": None}

    streaks = []
    current_start = scrobbles[0].timestamp.date()
    current_type = scrobbles[0].log.get("mood_type") or "unknown"
    current_length = 1

    for s in scrobbles[1:]:
        mood_type = s.log.get("mood_type") or "unknown"
        if mood_type == current_type:
            current_length += 1
        else:
            streaks.append(
                {
                    "start_date": current_start.isoformat(),
                    "end_date": scrobbles[scrobbles.index(s) - 1].timestamp.date().isoformat(),
                    "mood_type": current_type,
                    "length": current_length,
                }
            )
            current_start = s.timestamp.date()
            current_type = mood_type
            current_length = 1

    streaks.append(
        {
            "start_date": current_start.isoformat(),
            "end_date": scrobbles[-1].timestamp.date().isoformat(),
            "mood_type": current_type,
            "length": current_length,
        }
    )

    streaks.sort(key=lambda x: x["length"], reverse=True)

    current_streak = {
        "mood_type": current_type,
        "length": current_length,
        "start_date": current_start.isoformat(),
    }

    return {"streaks": streaks[:10], "current_streak": current_streak}


def compute_mood_weather(user, period="all_time"):
    scrobbles = _mood_scrobbles(user, period)
    by_condition = defaultdict(list)
    by_temp_range = defaultdict(list)

    for s in scrobbles:
        quality = s.log.get("mood_quality")
        if quality is None:
            continue
        desc = s.log.get("weather_description")
        temp = s.log.get("weather_temp")
        if desc:
            by_condition[desc].append(quality)
        if temp is not None:
            bucket = f"{(int(temp) // 10) * 10}-{(int(temp) // 10) * 10 + 9}F"
            by_temp_range[bucket].append(quality)

    conditions = [
        {"condition": cond, "avg_quality": _avg_quality(vals), "count": len(vals)}
        for cond, vals in sorted(by_condition.items(), key=lambda x: len(x[1]), reverse=True)
    ]

    temp_ranges = [
        {"range": rng, "avg_quality": _avg_quality(vals), "count": len(vals)}
        for rng, vals in sorted(by_temp_range.items())
    ]

    return {"conditions": conditions, "temp_ranges": temp_ranges}