Class RoverSign

RoverSign/utils/database/models.py:295–401

class RoverSign(BaseIDModel, table=True):
    __table_args__: Dict[str, Any] = {"extend_existing": True}
    uid: str = Field(title="鸣潮UID")
    pgr_uid: Optional[str] = Field(default=None, title="战双UID")
    game_sign: int = Field(default=0, title="游戏签到（鸣潮）")
    pgr_game_sign: int = Field(default=0, title="游戏签到（战双）")
    bbs_sign: int = Field(default=0, title="社区签到")
    bbs_detail: int = Field(default=0, title="社区浏览")
    bbs_like: int = Field(default=0, title="社区点赞")
    bbs_share: int = Field(default=0, title="社区分享")
    date: str = Field(default=get_today_date(), title="签到日期")

    @classmethod
    async def _find_sign_record(
        cls: Type[T_RoverSign],
        session: AsyncSession,
        uid: str,
        date: str,
    ) -> Optional[T_RoverSign]:
        """查找指定UID和日期的签到记录（内部方法）"""
        query = select(cls).where(cls.uid == uid).where(cls.date == date)
        result = await session.execute(query)
        return result.scalars().first()

    @classmethod
    @with_lock
    @with_session
    async def upsert_rover_sign(
        cls: Type[T_RoverSign],
        session: AsyncSession,
        rover_sign_data: RoverSignData,
    ) -> Optional[T_RoverSign]:
        """
        插入或更新签到数据
        返回更新后的记录或新插入的记录
        """
        if not rover_sign_data.uid:
            return None

        # 确保日期有值
        rover_sign_data.date = rover_sign_data.date or get_today_date()

        # 查询是否存在记录
        record = await cls._find_sign_record(
            session, rover_sign_data.uid, rover_sign_data.date
        )

        if record:
            # 更新已有记录
            for field in [
                "game_sign",
                "pgr_game_sign",
                "bbs_sign",
                "bbs_detail",
                "bbs_like",
                "bbs_share",
            ]:
                value = getattr(rover_sign_data, field)