chuni: add missing net battle funcs

titles/chuni/schema/profile.py

@@ -412,6 +412,18 @@ rating = Table(
     mysql_charset="utf8mb4",
 )
 
+net_battle = Table(
+    "chuni_profile_net_battle",
+    metadata,
+    Column("id", Integer, primary_key=True, nullable=False),
+    Column("user", Integer, ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"), nullable=False, unique=True),
+    Column("isRankUpChallengeFailed", Boolean),
+    Column("highestBattleRankId", Integer),
+    Column("battleIconId", Integer),
+    Column("battleIconNum", Integer),
+    Column("avatarEffectPoint", Integer),
+    mysql_charset="utf8mb4",
+)
 
 class ChuniProfileData(BaseData):
     async def update_name(self, user_id: int, new_name: str) -> bool:
@@ -779,4 +791,32 @@ class ChuniProfileData(BaseData):
         else:
             versions_raw = result.fetchall()
             versions = [row[0] for row in versions_raw]
-        return sorted(versions, reverse=True)
+        return sorted(versions, reverse=True)
+
+    async def put_net_battle(self, user_id: int, net_battle_data: Dict) -> Optional[int]:
+        sql = insert(net_battle).values(
+            user=user_id,
+            isRankUpChallengeFailed=net_battle_data['isRankUpChallengeFailed'],
+            highestBattleRankId=net_battle_data['highestBattleRankId'],
+            battleIconId=net_battle_data['battleIconId'],
+            battleIconNum=net_battle_data['battleIconNum'],
+            avatarEffectPoint=net_battle_data['avatarEffectPoint'],
+        )
+
+        conflict = sql.on_duplicate_key_update(
+            isRankUpChallengeFailed=net_battle_data['isRankUpChallengeFailed'],
+            highestBattleRankId=net_battle_data['highestBattleRankId'],
+            battleIconId=net_battle_data['battleIconId'],
+            battleIconNum=net_battle_data['battleIconNum'],
+            avatarEffectPoint=net_battle_data['avatarEffectPoint'],
+        )
+
+        result = await self.execute(conflict)
+        if result:
+            return result.inserted_primary_key['id']
+        self.logger.error(f"Failed to put net battle data for user {user_id}")
+
+    async def get_net_battle(self, user_id: int, net_battle_data: Dict) -> Optional[Row]:
+        result = await self.execute(net_battle.select(net_battle.c.user == user_id))
+        if result:
+            return result.fetchone()