"""每日心情叶服务""" 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()] async def get_grove(self, user_id: str) -> dict: """获取情绪共鸣林:自己 + 好友今日的心情叶,聚合成俯瞰森林""" from app.models.friend import Friend today = date.today() # 自己 + 好友 ID friends_result = await self.db.execute( select(Friend.friend_user_id).where(Friend.user_id == user_id) ) visible_ids = [user_id] + [r[0] for r in friends_result.all()] # 查今日所有人的叶子 result = await self.db.execute( select(DailyMoodLeaf).where( DailyMoodLeaf.user_id.in_(visible_ids), DailyMoodLeaf.leaf_date == today, ) ) leaves = result.scalars().all() # 排布:自己在中心(position 0),好友按确定性环绕 import math positioned = [] others = [l for l in leaves if l.user_id != user_id] my_leaf = next((l for l in leaves if l.user_id == user_id), None) if my_leaf: positioned.append({ "is_self": True, "user_id": my_leaf.user_id, "mood": my_leaf.mood, "leaf_seed": my_leaf.leaf_seed, "angle": 0, "radius": 0, }) n = len(others) for i, l in enumerate(others): angle = (2 * math.pi * i / n) if n > 0 else 0 positioned.append({ "is_self": False, "user_id": l.user_id, "mood": l.mood, "leaf_seed": l.leaf_seed, "angle": angle, "radius": 1, }) # 聚合情绪天气 mood_counts: dict[str, int] = {} for l in leaves: mood_counts[l.mood or "unknown"] = mood_counts.get(l.mood or "unknown", 0) + 1 return { "leaves": positioned, "total": len(leaves), "mood_counts": mood_counts, "date": today.isoformat(), } 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, }