habit-tracker/app/services/journal_service.py

import calendar
import mimetypes
import random
import uuid
from datetime import date
from pathlib import Path

from fastapi import UploadFile
from PIL import Image
from sqlalchemy import func, select
from sqlalchemy.orm import Session

from app.config import settings
from app.models.journal import (
    JournalAttachment,
    JournalAttachmentType,
    JournalCategory,
    JournalEntry,
    JournalEntryMood,
    JournalPrompt,
    JournalTag,
)
from app.schemas.journal import (
    JournalAttachmentOut,
    JournalCalendarDay,
    JournalCategoryCreate,
    JournalDayDetailItem,
    JournalEntryCreate,
    JournalEntryUpdate,
    JournalOnThisDayItem,
)

ALLOWED_IMAGE_TYPES = {"image/jpeg", "image/png", "image/webp", "image/gif"}
ALLOWED_VIDEO_TYPES = {"video/mp4", "video/quicktime"}
THUMBNAIL_WIDTH = 400
DEFAULT_CATEGORY_NAME = "일상"
DEFAULT_CATEGORY_TEMPLATE = """**1. Story : 오늘 무슨 일이 있었나요?**

-

**2. Feelings : 오늘 들었던 생각과 나의 감정은?**

-

**3. Decisions : 오늘 내가 내린 결정이 있나요?**

-

**4. Insights : 나에 대해 새롭게 알게된 사실이 있나요?**

-

→ 나에 대한 인사이트는 메타인지를 키워주는 **소중한 기록**입니다. 꼭 보관하세요.

**5. Actions : 내가 다음에 할 행동은 무엇인가요?**

- """


# ---- 카테고리 ----


def list_categories(db: Session, user_id: int) -> list[JournalCategory]:
    stmt = select(JournalCategory).where(JournalCategory.user_id == user_id)
    stmt = stmt.order_by(
        JournalCategory.sort_order.is_(None), JournalCategory.sort_order, JournalCategory.created_at
    )
    return list(db.scalars(stmt))


def get_category(db: Session, category_id: int, user_id: int) -> JournalCategory | None:
    return db.scalar(
        select(JournalCategory).where(JournalCategory.id == category_id, JournalCategory.user_id == user_id)
    )


def create_category(db: Session, user_id: int, data: JournalCategoryCreate) -> JournalCategory:
    category = JournalCategory(
        user_id=user_id, name=data.name, color=data.color, content_template=data.content_template
    )
    db.add(category)
    db.commit()
    db.refresh(category)
    return category


def update_category(db: Session, category: JournalCategory, data: JournalCategoryCreate) -> JournalCategory:
    category.name = data.name
    category.color = data.color
    category.content_template = data.content_template
    db.commit()
    db.refresh(category)
    return category


def ensure_default_category(db: Session, user_id: int) -> JournalCategory:
    """유저가 카테고리를 하나도 만든 적 없으면 기본 카테고리를 자동 생성한다."""
    existing = db.scalar(
        select(JournalCategory)
        .where(JournalCategory.user_id == user_id)
        .order_by(JournalCategory.sort_order.is_(None), JournalCategory.sort_order, JournalCategory.created_at)
        .limit(1)
    )
    if existing is not None:
        return existing
    return create_category(
        db, user_id, JournalCategoryCreate(name=DEFAULT_CATEGORY_NAME, content_template=DEFAULT_CATEGORY_TEMPLATE)
    )


def reorder_categories(db: Session, user_id: int, ordered_ids: list[int]) -> None:
    categories = db.scalars(
        select(JournalCategory).where(JournalCategory.id.in_(ordered_ids), JournalCategory.user_id == user_id)
    ).all()
    category_map = {c.id: c for c in categories}
    for index, category_id in enumerate(ordered_ids):
        category = category_map.get(category_id)
        if category is not None:
            category.sort_order = index
    db.commit()


def count_entries_by_category(db: Session, user_id: int) -> dict[int, int]:
    rows = db.execute(
        select(JournalEntry.category_id, func.count(JournalEntry.id))
        .where(JournalEntry.user_id == user_id)
        .group_by(JournalEntry.category_id)
    ).all()
    return {row[0]: row[1] for row in rows}


def delete_category(db: Session, category: JournalCategory) -> None:
    """카테고리를 지우면 안의 엔트리도 함께 지워진다 — 첨부파일 디스크 삭제까지 하려면
    ORM cascade에만 맡기지 않고 delete_entry를 하나씩 거쳐야 한다."""
    entries = list(db.scalars(select(JournalEntry).where(JournalEntry.category_id == category.id)))
    for entry in entries:
        delete_entry(db, entry)
    db.delete(category)
    db.commit()


# ---- 태그 ----


def get_or_create_tags(db: Session, user_id: int, names: list[str]) -> list[JournalTag]:
    tags = []
    for name in names:
        tag = db.scalar(select(JournalTag).where(JournalTag.user_id == user_id, JournalTag.name == name))
        if tag is None:
            tag = JournalTag(user_id=user_id, name=name)
            db.add(tag)
            db.flush()
        tags.append(tag)
    return tags


# ---- 엔트리 ----


def list_entries(
    db: Session,
    user_id: int,
    category_id: int | None = None,
    start: date | None = None,
    end: date | None = None,
) -> list[JournalEntry]:
    stmt = select(JournalEntry).where(JournalEntry.user_id == user_id)
    if category_id is not None:
        stmt = stmt.where(JournalEntry.category_id == category_id)
    if start is not None:
        stmt = stmt.where(JournalEntry.entry_date >= start)
    if end is not None:
        stmt = stmt.where(JournalEntry.entry_date <= end)
    stmt = stmt.order_by(JournalEntry.entry_date.desc(), JournalEntry.created_at.desc())
    return list(db.scalars(stmt))


def get_entry(db: Session, entry_id: int, user_id: int) -> JournalEntry | None:
    return db.scalar(select(JournalEntry).where(JournalEntry.id == entry_id, JournalEntry.user_id == user_id))


def create_entry(db: Session, user_id: int, data: JournalEntryCreate) -> JournalEntry:
    tags = get_or_create_tags(db, user_id, data.tags)
    entry = JournalEntry(
        user_id=user_id,
        category_id=data.category_id,
        entry_date=data.entry_date,
        title=data.title,
        content=data.content,
        tags=tags,
        moods=[JournalEntryMood(mood=m) for m in data.moods],
    )
    db.add(entry)
    db.commit()
    db.refresh(entry)
    return entry


def update_entry(db: Session, entry: JournalEntry, data: JournalEntryUpdate) -> JournalEntry:
    tags = get_or_create_tags(db, entry.user_id, data.tags)
    entry.category_id = data.category_id
    entry.entry_date = data.entry_date
    entry.title = data.title
    entry.content = data.content
    entry.tags = tags
    entry.moods = [JournalEntryMood(mood=m) for m in data.moods]
    db.commit()
    db.refresh(entry)
    return entry


def delete_entry(db: Session, entry: JournalEntry) -> None:
    for attachment in list(entry.attachments):
        _delete_attachment_files(attachment)
    db.delete(entry)
    db.commit()


# ---- 첨부파일 ----


def _media_dir(user_id: int, entry_id: int) -> Path:
    return Path(settings.journal_media_root) / str(user_id) / str(entry_id)


def _delete_attachment_files(attachment: JournalAttachment) -> None:
    for path_str in (attachment.file_path, attachment.thumbnail_path):
        if path_str:
            Path(path_str).unlink(missing_ok=True)


def save_attachment(db: Session, entry: JournalEntry, upload_file: UploadFile) -> JournalAttachment:
    content_type = upload_file.content_type or ""
    if content_type in ALLOWED_IMAGE_TYPES:
        media_type = JournalAttachmentType.IMAGE
    elif content_type in ALLOWED_VIDEO_TYPES:
        media_type = JournalAttachmentType.VIDEO
    else:
        raise ValueError("지원하지 않는 파일 형식이에요 (사진: jpg/png/webp/gif, 영상: mp4/mov)")

    data = upload_file.file.read()
    max_bytes = settings.journal_max_upload_mb * 1024 * 1024
    if len(data) > max_bytes:
        raise ValueError(f"파일 용량은 {settings.journal_max_upload_mb}MB를 넘을 수 없어요")

    target_dir = _media_dir(entry.user_id, entry.id)
    target_dir.mkdir(parents=True, exist_ok=True)

    ext = mimetypes.guess_extension(content_type) or Path(upload_file.filename or "").suffix or ""
    stored_name = f"{uuid.uuid4().hex}{ext}"
    file_path = target_dir / stored_name
    file_path.write_bytes(data)

    thumbnail_path: Path | None = None
    if media_type == JournalAttachmentType.IMAGE:
        thumbnail_path = target_dir / f"{uuid.uuid4().hex}_thumb.jpg"
        try:
            with Image.open(file_path) as img:
                img = img.convert("RGB")
                w, h = img.size
                if w > THUMBNAIL_WIDTH:
                    img = img.resize((THUMBNAIL_WIDTH, round(h * THUMBNAIL_WIDTH / w)))
                img.save(thumbnail_path, "JPEG", quality=85)
        except Exception:
            # 손상되었거나 Pillow가 못 읽는 이미지여도 원본 업로드 자체는 실패시키지 않는다.
            thumbnail_path.unlink(missing_ok=True)
            thumbnail_path = None

    attachment = JournalAttachment(
        entry_id=entry.id,
        media_type=media_type,
        file_path=str(file_path),
        thumbnail_path=str(thumbnail_path) if thumbnail_path else None,
        original_filename=upload_file.filename or stored_name,
        file_size=len(data),
    )
    db.add(attachment)
    db.commit()
    db.refresh(attachment)
    return attachment


def get_attachment(db: Session, attachment_id: int, user_id: int) -> JournalAttachment | None:
    return db.scalar(
        select(JournalAttachment)
        .join(JournalEntry, JournalAttachment.entry_id == JournalEntry.id)
        .where(JournalAttachment.id == attachment_id, JournalEntry.user_id == user_id)
    )


def delete_attachment(db: Session, attachment: JournalAttachment) -> None:
    _delete_attachment_files(attachment)
    db.delete(attachment)
    db.commit()


# ---- 캘린더 / day-detail / 회상 ----


def _to_day_detail_item(entry: JournalEntry) -> JournalDayDetailItem:
    return JournalDayDetailItem(
        id=entry.id,
        category_id=entry.category_id,
        category_name=entry.category.name,
        category_color=entry.category.color,
        title=entry.title,
        content=entry.content,
        moods=[m.mood for m in entry.moods],
        tags=[t.name for t in entry.tags],
        attachments=[
            JournalAttachmentOut(
                id=a.id,
                media_type=a.media_type,
                original_filename=a.original_filename,
                has_thumbnail=a.thumbnail_path is not None,
            )
            for a in entry.attachments
        ],
    )


def get_monthly_journal_summary(
    db: Session, user_id: int, year: int, month: int, category_id: int | None = None
) -> dict[date, JournalCalendarDay]:
    """해당 월의 날짜별 엔트리 개수와, 그날 등장한 카테고리 색상 목록(점 표시용)을 집계한다."""
    days_in_month = calendar.monthrange(year, month)[1]
    first_day = date(year, month, 1)
    last_day = date(year, month, days_in_month)

    stmt = (
        select(JournalEntry.entry_date, JournalCategory.color)
        .join(JournalCategory, JournalEntry.category_id == JournalCategory.id)
        .where(JournalEntry.user_id == user_id, JournalEntry.entry_date.between(first_day, last_day))
    )
    if category_id is not None:
        stmt = stmt.where(JournalEntry.category_id == category_id)
    rows = db.execute(stmt).all()

    counts: dict[date, int] = {}
    colors_by_date: dict[date, list[str]] = {}
    for entry_date, color in rows:
        counts[entry_date] = counts.get(entry_date, 0) + 1
        colors = colors_by_date.setdefault(entry_date, [])
        color = color or "var(--color-accent)"
        if color not in colors:
            colors.append(color)

    return {
        d: JournalCalendarDay(entry_date=d, total_count=counts[d], category_colors=colors_by_date[d])
        for d in counts
    }


def get_day_entries(
    db: Session, user_id: int, target_date: date, category_id: int | None = None
) -> list[JournalDayDetailItem]:
    stmt = select(JournalEntry).where(
        JournalEntry.user_id == user_id, JournalEntry.entry_date == target_date
    )
    if category_id is not None:
        stmt = stmt.where(JournalEntry.category_id == category_id)
    stmt = stmt.order_by(JournalEntry.created_at)
    return [_to_day_detail_item(e) for e in db.scalars(stmt)]


def get_on_this_day(db: Session, user_id: int, today: date) -> list[JournalOnThisDayItem]:
    """오늘과 월/일이 같은 과거 연도의 엔트리를 반환한다("1년 전 오늘" 회상 카드).

    MONTH()/DAY() 같은 DB 종속 함수 대신 파이썬에서 필터링해 SQLite(테스트)/MariaDB(운영)
    양쪽에서 동일하게 동작하게 한다 — 개인 규모 데이터라 전체 스캔 비용도 무시할 만하다.
    """
    stmt = select(JournalEntry).where(JournalEntry.user_id == user_id).order_by(JournalEntry.entry_date.desc())
    matches = [
        e
        for e in db.scalars(stmt)
        if e.entry_date.month == today.month
        and e.entry_date.day == today.day
        and e.entry_date.year != today.year
    ]
    items = []
    for e in matches:
        base = _to_day_detail_item(e)
        items.append(
            JournalOnThisDayItem(
                **base.model_dump(),
                entry_date=e.entry_date,
                years_ago=today.year - e.entry_date.year,
            )
        )
    return items


# ---- 프롬프트 ----


def get_random_prompt(db: Session) -> JournalPrompt | None:
    prompts = list(db.scalars(select(JournalPrompt)))
    return random.choice(prompts) if prompts else None