"""每日心情叶服务""" import hashlib import uuid from datetime import date from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.models.daily_leaf import DailyMoodLeaf class LeafService: def __init__(self, db: AsyncSession): self.db = db async def get_or_create_today(self, user_id: str) -> dict: """获取或创建今日心情叶""" today = date.today() result = await self.db.execute( select(DailyMoodLeaf).where( DailyMoodLeaf.user_id == user_id, DailyMoodLeaf.leaf_date == today, ) ) leaf = result.scalars().first() if not leaf: seed = hashlib.md5(f"{user_id}:{today.isoformat()}".encode()).hexdigest()[:16] leaf = DailyMoodLeaf( id=str(uuid.uuid4()), user_id=user_id, leaf_date=today, leaf_seed=seed, ) self.db.add(leaf) await self.db.flush() return self._leaf_to_dict(leaf) async def update_leaf(self, user_id: str, leaf_id: str, mood: str | None, note: str | None) -> dict: """更新今日叶子的心情和备注""" result = await self.db.execute( select(DailyMoodLeaf).where(DailyMoodLeaf.id == leaf_id) ) leaf = result.scalars().first() if not leaf: raise ValueError("叶子不存在") if leaf.user_id != user_id: raise ValueError("无权操作此叶子") if mood is not None: leaf.mood = mood if note is not None: leaf.note = note await self.db.flush() return self._leaf_to_dict(leaf) async def get_collection(self, user_id: str, limit: int = 60) -> list[dict]: """获取叶子收藏(历史)""" result = await self.db.execute( select(DailyMoodLeaf) .where(DailyMoodLeaf.user_id == user_id) .order_by(DailyMoodLeaf.leaf_date.desc()) .limit(limit) ) return [self._leaf_to_dict(l) for l in result.scalars().all()] def _leaf_to_dict(self, leaf: DailyMoodLeaf) -> dict: return { "id": leaf.id, "user_id": leaf.user_id, "leaf_date": leaf.leaf_date.isoformat() if leaf.leaf_date else None, "mood": leaf.mood, "note": leaf.note, "leaf_seed": leaf.leaf_seed, "created_at": leaf.created_at, }