Initial commit: habit tracker PWA with Google OAuth, push notifications
FastAPI + SQLAlchemy/Alembic + MariaDB backend with Jinja2/htmx/Alpine server-rendered frontend. Multi-user via Google OAuth, daily habit tracking, monthly/weekly history views, Web Push reminders via APScheduler, and PWA support (manifest, service worker, offline caching). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Claude Sonnet 5
@@ -0,0 +1,254 @@
|
||||
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
|
||||
from app.schemas.habit_log import HabitStats, MonthlySummaryDay, TodayItem, WeeklyMatrixRow
|
||||
from app.services import habit_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,
|
||||
)
|
||||
|
||||
|
||||
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)
|
||||
)
|
||||
)
|
||||
)
|
||||
|
||||
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 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 toggle_check_and_celebrate(db: Session, habit: Habit, log_date: date) -> tuple[bool, int | None]:
|
||||
"""체크를 토글하고, 새로 체크되어 스트릭이 마일스톤에 도달했으면 축하 푸시를 보낸다.
|
||||
|
||||
반환값: (checked, milestone_streak). milestone_streak은 이번 토글로 막 달성한 마일스톤 값이면 그 값,
|
||||
체크 해제거나 마일스톤이 아니면 None.
|
||||
"""
|
||||
checked = toggle_check(db, habit.id, log_date)
|
||||
if not checked:
|
||||
return checked, None
|
||||
|
||||
streak = get_habit_stats(db, habit).current_streak
|
||||
if streak not in MILESTONE_STREAKS:
|
||||
return checked, None
|
||||
|
||||
if habit.user_id is not None:
|
||||
push_service.send_to_user(
|
||||
db, habit.user_id, title=f"🔥 {habit.name}", body=f"{streak}일 연속 달성했어요!", url="/today"
|
||||
)
|
||||
return checked, streak
|
||||
|
||||
|
||||
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))
|
||||
.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:
|
||||
"""습관 생성일부터 오늘까지의 완료율과, 오늘(또는 어제)부터 거슬러 올라간 연속 달성일을 계산한다.
|
||||
|
||||
요일 스케줄은 현재 습관의 weekdays_mask를 과거에도 그대로 적용한 것으로 간주한다
|
||||
(과거 요일 변경 이력은 추적하지 않음 — 월별/주별 집계와 같은 단순화).
|
||||
"""
|
||||
today = date.today()
|
||||
start = habit.created_at.date()
|
||||
|
||||
checked_dates = set(db.scalars(select(HabitLog.log_date).where(HabitLog.habit_id == habit.id)))
|
||||
|
||||
scheduled_days = 0
|
||||
checked_days = 0
|
||||
d = start
|
||||
while d <= today:
|
||||
if habit.is_scheduled_on(d.weekday()):
|
||||
scheduled_days += 1
|
||||
if d in checked_dates:
|
||||
checked_days += 1
|
||||
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,
|
||||
)
|
||||
|
||||
|
||||
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])
|
||||
)
|
||||
).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
|
||||
Reference in New Issue
Block a user