habit-tracker/app/services/log_service.py

import calendar
from datetime import date, timedelta

from sqlalchemy import func, select
from sqlalchemy.orm import Session

from app.models.habit import Habit, HabitStatus, HabitType
from app.models.habit_log import HabitLog, HabitLogStatus
from app.schemas.habit_log import HabitStats, MonthlySummaryDay, TodayItem, WeeklyMatrixRow
from app.schemas.level import LevelInfo
from app.services import habit_service, level_service, push_service

# 체크 시 축하 푸시/배지를 트리거하는 연속 달성일 마일스톤.
MILESTONE_STREAKS = {7, 30, 66, 100, 200, 365}


def _to_today_item(db: Session, habit: Habit, checked: bool) -> TodayItem:
    stats = get_habit_stats(db, habit)
    return TodayItem(
        habit_id=habit.id,
        name=habit.name,
        habit_type=habit.habit_type.value,
        condition_text=habit.condition_text,
        reminder_time=habit.reminder_time.strftime("%H:%M") if habit.reminder_time else None,
        checked=checked,
        completion_rate=stats.completion_rate,
        current_streak=stats.current_streak,
        scheduled_days=stats.scheduled_days,
        level_info=stats.level_info,
    )


def get_today_items(db: Session, user_id: int, target_date: date) -> tuple[list[TodayItem], list[TodayItem]]:
    """오늘 요일에 예정된 active 습관을 형성/중단으로 나누어 체크 여부와 함께 반환한다."""
    weekday = target_date.weekday()
    active_habits = habit_service.list_habits(db, user_id, status=HabitStatus.ACTIVE)
    scheduled = [h for h in active_habits if h.is_scheduled_on(weekday)]

    checked_ids: set[int] = set()
    habit_ids = [h.id for h in scheduled]
    if habit_ids:
        checked_ids = set(
            db.scalars(
                select(HabitLog.habit_id).where(
                    HabitLog.log_date == target_date,
                    HabitLog.habit_id.in_(habit_ids),
                    HabitLog.status == HabitLogStatus.DONE,
                )
            )
        )

    build_items = [_to_today_item(db, h, h.id in checked_ids) for h in scheduled if h.habit_type == HabitType.BUILD]
    quit_items = [_to_today_item(db, h, h.id in checked_ids) for h in scheduled if h.habit_type == HabitType.QUIT]
    return build_items, quit_items


def get_yesterday_missed_items(db: Session, user_id: int) -> list[TodayItem]:
    """어제 예정되어 있었지만 아직 완료/실패가 결정되지 않은 습관 목록

    (형성+중단 합쳐서, /today 화면의 소급 체크·실패 확정용). 완료로 체크했거나 실패로 확정한
    습관은 둘 다 "결정됨"으로 보고 목록에서 뺀다.
    """
    yesterday = date.today() - timedelta(days=1)
    build_items, quit_items = get_today_items(db, user_id, yesterday)
    all_items = build_items + quit_items

    habit_ids = [item.habit_id for item in all_items]
    failed_ids: set[int] = set()
    if habit_ids:
        failed_ids = set(
            db.scalars(
                select(HabitLog.habit_id).where(
                    HabitLog.log_date == yesterday,
                    HabitLog.habit_id.in_(habit_ids),
                    HabitLog.status == HabitLogStatus.FAILED,
                )
            )
        )
    return [item for item in all_items if not item.checked and item.habit_id not in failed_ids]


def toggle_check(db: Session, habit_id: int, log_date: date) -> bool:
    """체크 상태를 반전시키고 토글 후의 체크 여부를 반환한다.

    호출측에서 이미 habit_service.get_habit(db, habit_id, user_id)로 소유권을 검증한 뒤에만 불러야 한다.
    """
    existing = db.scalar(select(HabitLog).where(HabitLog.habit_id == habit_id, HabitLog.log_date == log_date))
    if existing:
        db.delete(existing)
        db.commit()
        return False
    db.add(HabitLog(habit_id=habit_id, log_date=log_date))
    db.commit()
    return True


def mark_failed(db: Session, habit_id: int, log_date: date) -> None:
    """해당 날짜를 명시적으로 실패로 확정한다. 이미 완료/실패가 결정된 날짜면 아무 것도 하지 않는다."""
    existing = db.scalar(select(HabitLog).where(HabitLog.habit_id == habit_id, HabitLog.log_date == log_date))
    if existing:
        return
    db.add(HabitLog(habit_id=habit_id, log_date=log_date, status=HabitLogStatus.FAILED))
    db.commit()


def toggle_check_and_celebrate(db: Session, habit: Habit, log_date: date) -> tuple[bool, int | None, int | None]:
    """체크를 토글하고, 새로 체크되어 스트릭 마일스톤이나 레벨업에 도달했으면 축하 푸시를 보낸다.

    반환값: (checked, milestone_streak, level_up).
    - milestone_streak: 이번 토글로 막 달성한 연속 달성일 마일스톤 값, 아니면 None.
    - level_up: 이번 토글로 막 오른 새 레벨, 아니면 None.
    체크 해제인 경우 둘 다 None.
    """
    level_before = get_habit_stats(db, habit).level_info.level
    checked = toggle_check(db, habit.id, log_date)
    if not checked:
        return checked, None, None

    stats = get_habit_stats(db, habit)
    streak = stats.current_streak
    milestone_streak = streak if streak in MILESTONE_STREAKS else None
    level_up = stats.level_info.level if stats.level_info.level > level_before else None

    if habit.user_id is not None:
        if milestone_streak is not None:
            push_service.send_to_user(
                db, habit.user_id, title=f"🔥 {habit.name}", body=f"{streak}일 연속 달성했어요!", url="/today"
            )
        if level_up is not None:
            push_service.send_to_user(
                db, habit.user_id, title=f"⭐ {habit.name}", body=f"레벨 {level_up}로 올랐어요!", url="/today"
            )
    return checked, milestone_streak, level_up


def list_logs(
    db: Session, user_id: int, habit_id: int | None = None, start: date | None = None, end: date | None = None
) -> list[HabitLog]:
    stmt = select(HabitLog).join(Habit, HabitLog.habit_id == Habit.id).where(Habit.user_id == user_id)
    if habit_id is not None:
        stmt = stmt.where(HabitLog.habit_id == habit_id)
    if start is not None:
        stmt = stmt.where(HabitLog.log_date >= start)
    if end is not None:
        stmt = stmt.where(HabitLog.log_date <= end)
    stmt = stmt.order_by(HabitLog.log_date)
    return list(db.scalars(stmt))


def _checked_counts_by_date(db: Session, habit_ids: list[int], start: date, end: date) -> dict[date, int]:
    if not habit_ids:
        return {}
    rows = db.execute(
        select(HabitLog.log_date, func.count(HabitLog.id))
        .where(
            HabitLog.habit_id.in_(habit_ids),
            HabitLog.log_date.between(start, end),
            HabitLog.status == HabitLogStatus.DONE,
        )
        .group_by(HabitLog.log_date)
    ).all()
    return {row[0]: row[1] for row in rows}


def get_monthly_summary(db: Session, user_id: int, year: int, month: int) -> list[MonthlySummaryDay]:
    """해당 월의 날짜별 예정 습관 수 / 체크된 습관 수를 집계한다 (현재 active 습관 기준)."""
    days_in_month = calendar.monthrange(year, month)[1]
    first_day = date(year, month, 1)
    last_day = date(year, month, days_in_month)

    active_habits = habit_service.list_habits(db, user_id, status=HabitStatus.ACTIVE)
    checked_counts = _checked_counts_by_date(db, [h.id for h in active_habits], first_day, last_day)

    summaries = []
    for day_num in range(1, days_in_month + 1):
        d = date(year, month, day_num)
        # 습관이 생성되기 전 날짜는 "예정되었지만 안 함"으로 잘못 잡히지 않도록 제외한다.
        scheduled = sum(1 for h in active_habits if h.created_at.date() <= d and h.is_scheduled_on(d.weekday()))
        summaries.append(
            MonthlySummaryDay(log_date=d, scheduled_count=scheduled, checked_count=checked_counts.get(d, 0))
        )
    return summaries


def summarize_completion_rate(summaries: list[MonthlySummaryDay], up_to: date) -> float:
    """월별 요약에서 up_to(보통 오늘)까지 지난 날짜만 모아 전체 완료율(%)을 계산한다.

    아직 지나지 않은 미래 날짜는 scheduled_count는 있어도 checked_count가 항상 0이라
    포함시키면 완료율이 부당하게 낮아지므로 제외한다.
    """
    past = [s for s in summaries if s.log_date <= up_to]
    total_scheduled = sum(s.scheduled_count for s in past)
    total_checked = sum(s.checked_count for s in past)
    return round(total_checked / total_scheduled * 100, 1) if total_scheduled else 0.0


def get_period_completion_rate(db: Session, user_id: int, start: date, end: date) -> tuple[float, int, int]:
    """[start, end] 구간(양끝 포함)의 예정/체크 수를 집계해 완료율(%)과 함께 반환한다.

    get_monthly_summary와 같은 규칙(현재 active 습관 기준, 습관 생성일 이전 제외)을 임의 기간에
    적용한 버전 — 주간 요약 알림처럼 달력 월 경계에 안 맞는 기간을 집계할 때 쓴다.
    """
    active_habits = habit_service.list_habits(db, user_id, status=HabitStatus.ACTIVE)
    checked_counts = _checked_counts_by_date(db, [h.id for h in active_habits], start, end)

    total_scheduled = 0
    total_checked = 0
    d = start
    while d <= end:
        total_scheduled += sum(
            1 for h in active_habits if h.created_at.date() <= d and h.is_scheduled_on(d.weekday())
        )
        total_checked += checked_counts.get(d, 0)
        d += timedelta(days=1)

    rate = round(total_checked / total_scheduled * 100, 1) if total_scheduled else 0.0
    return rate, total_scheduled, total_checked


def get_habit_stats(db: Session, habit: Habit) -> HabitStats:
    """습관 생성일부터 오늘까지의 완료율과, 오늘(또는 어제)부터 거슬러 올라간 연속 달성일을 계산한다.
    같은 순회에서 레벨/경험치(XP)도 함께 누적한다.

    요일 스케줄은 현재 습관의 weekdays_mask를 과거에도 그대로 적용한 것으로 간주한다
    (과거 요일 변경 이력은 추적하지 않음 — 월별/주별 집계와 같은 단순화).
    """
    today = date.today()
    start = habit.created_at.date()

    # 완료/포기된 습관은 그 시점 이후로 XP가 더 이상 변하지 않는다(동결) — completion_rate/current_streak
    # 계산 범위는 건드리지 않고 XP 누적에만 이 경계를 적용한다.
    freeze_date: date | None = None
    if habit.status != HabitStatus.ACTIVE and (habit.completed_at or habit.abandoned_at):
        freeze_date = (habit.completed_at or habit.abandoned_at).date()

    checked_dates = set(
        db.scalars(
            select(HabitLog.log_date).where(
                HabitLog.habit_id == habit.id, HabitLog.status == HabitLogStatus.DONE
            )
        )
    )

    scheduled_days = 0
    checked_days = 0
    xp = 0
    hit_streak = 0
    miss_streak = 0
    d = start
    while d <= today:
        if habit.is_scheduled_on(d.weekday()):
            scheduled_days += 1
            is_checked = d in checked_dates
            if is_checked:
                checked_days += 1
            if freeze_date is None or d <= freeze_date:
                if d == today and not is_checked:
                    pass  # 오늘은 아직 안 지났으니 XP 변동 없음 (current_streak과 같은 규칙)
                elif is_checked:
                    hit_streak += 1
                    miss_streak = 0
                    xp += level_service.XP_PER_CHECK
                    xp += level_service.STREAK_MILESTONE_BONUS_XP.get(hit_streak, 0)
                else:
                    hit_streak = 0
                    miss_streak += 1
                    penalty = min(level_service.MISS_PENALTY_BASE * miss_streak, level_service.MISS_PENALTY_CAP)
                    xp = max(0, xp - penalty)
        d += timedelta(days=1)

    completion_rate = round(checked_days / scheduled_days * 100, 1) if scheduled_days else 0.0

    streak = 0
    d = today
    while d >= start:
        if habit.is_scheduled_on(d.weekday()):
            if d in checked_dates:
                streak += 1
            elif d != today:
                break
            # d가 오늘이고 아직 체크 전이면: 하루가 아직 안 끝났으니 스트릭을 끊지 않고 계속 거슬러 올라간다.
        d -= timedelta(days=1)

    return HabitStats(
        completion_rate=completion_rate,
        current_streak=streak,
        scheduled_days=scheduled_days,
        checked_days=checked_days,
        level_info=level_service.level_from_xp(xp),
    )


def get_weekly_matrix(db: Session, user_id: int, week_start: date) -> list[WeeklyMatrixRow]:
    """week_start(호출측에서 정한 주 시작일, 현재 /history는 일요일을 사용)부터 7일간,
    active 습관별 요일 체크 매트릭스를 반환한다. 이 함수 자체는 week_start가 어떤 요일이든 상관없다."""
    week_days = [week_start + timedelta(days=i) for i in range(7)]
    active_habits = habit_service.list_habits(db, user_id, status=HabitStatus.ACTIVE)
    habit_ids = [h.id for h in active_habits]

    checked_pairs: set[tuple[int, date]] = set()
    if habit_ids:
        rows = db.execute(
            select(HabitLog.habit_id, HabitLog.log_date).where(
                HabitLog.habit_id.in_(habit_ids),
                HabitLog.log_date.between(week_days[0], week_days[-1]),
                HabitLog.status == HabitLogStatus.DONE,
            )
        ).all()
        checked_pairs = {(r[0], r[1]) for r in rows}

    today = date.today()
    result = []
    for h in active_habits:
        checks: dict[str, bool | None] = {}
        scheduled_past = 0
        checked_past = 0
        for d in week_days:
            # 습관이 생성되기 전 날짜는 요일이 맞아도 "예정 없음"으로 취급한다.
            if d < h.created_at.date() or not h.is_scheduled_on(d.weekday()):
                checks[d.isoformat()] = None
                continue
            is_checked = (h.id, d) in checked_pairs
            checks[d.isoformat()] = is_checked
            if d <= today:
                scheduled_past += 1
                if is_checked:
                    checked_past += 1
        completion_rate = round(checked_past / scheduled_past * 100, 1) if scheduled_past else 0.0
        result.append(
            WeeklyMatrixRow(
                habit_id=h.id,
                name=h.name,
                habit_type=h.habit_type.value,
                checks=checks,
                completion_rate=completion_rate,
            )
        )
    return result


def get_account_level(db: Session, user_id: int) -> LevelInfo:
    """유저가 가진 모든 습관(활성+완료+포기)의 XP를 합산해 계정 전체 레벨을 계산한다.

    완료/포기된 습관도 (동결된 상태로) 합산에 포함한다 — 완료해도 그동안 쌓은 계정 레벨은 유지되어야 한다.
    습관별 레벨과 같은 곡선(level_service.level_from_xp)을 그대로 재사용한다.
    """
    habits = habit_service.list_habits(db, user_id)
    total_xp = sum(get_habit_stats(db, h).level_info.xp for h in habits)
    return level_service.level_from_xp(total_xp)