CHUNITHM X-VERSE support (#238)

Reviewed-on: https://gitea.tendokyu.moe/Hay1tsme/artemis/pulls/238
Co-authored-by: beerpsi <beerpsi@duck.com>
Co-committed-by: beerpsi <beerpsi@duck.com>

titles/chuni/schema/item.py

@@ -314,6 +314,37 @@ unlock_challenge = Table(
     mysql_charset="utf8mb4",
 )
 
+linked_verse: Table = Table(
+    "chuni_item_linked_verse",
+    metadata,
+    Column("id", Integer, primary_key=True, nullable=False),
+    Column(
+        "user",
+        Integer,
+        ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
+        nullable=False,
+    ),
+    Column("linkedVerseId", Integer, nullable=False),
+    Column("progress", String(255)),
+    Column("statusOpen", Integer),
+    Column("statusUnlock", Integer),
+    Column("isFirstClear", Integer),
+    Column("numClear", Integer),
+    Column("clearCourseId", Integer),
+    Column("clearCourseLevel", Integer),
+    Column("clearScore", Integer),
+    Column("clearDate", String(25)),
+    Column("clearUserId1", Integer),
+    Column("clearUserId2", Integer),
+    Column("clearUserId3", Integer),
+    Column("clearUserName0", String(20)),
+    Column("clearUserName1", String(20)),
+    Column("clearUserName2", String(20)),
+    Column("clearUserName3", String(20)),
+    UniqueConstraint("user", "linkedVerseId", name="chuni_item_linked_verse_uk"),
+    mysql_charset="utf8mb4",
+)
+
 
 class ChuniItemData(BaseData):
     async def get_oldest_free_matching(self, version: int) -> Optional[Row]:
@@ -394,7 +425,6 @@ class ChuniItemData(BaseData):
     async def is_favorite(
         self, user_id: int, version: int, fav_id: int, fav_kind: int = 1
     ) -> bool:
-
         sql = favorite.select(
             and_(
                 favorite.c.version == version,
@@ -849,3 +879,22 @@ class ChuniItemData(BaseData):
         if result is None:
             return None
         return result.fetchall()
+
+    async def get_linked_verse(self, aime_id: int) -> Optional[List[Row]]:
+        result = await self.execute(
+            linked_verse.select().where(linked_verse.c.user == aime_id)
+        )
+
+        if result:
+            return result.fetchall()
+
+    async def put_linked_verse(self, aime_id: int, linked_verse_data: Dict):
+        linked_verse_data = self.fix_bools(linked_verse_data)
+        sql = insert(linked_verse).values(user=aime_id, **linked_verse_data)
+        conflict = sql.on_duplicate_key_update(**linked_verse_data)
+        result = await self.execute(conflict)
+
+        if result:
+            return result.inserted_primary_key["id"]
+
+        self.logger.error("Failed to put Linked Verse data for user %s", aime_id)

titles/chuni/schema/profile.py

@@ -132,6 +132,9 @@ profile = Table(
     Column("avatarFront", Integer, server_default="0"),
     Column("avatarSkin", Integer, server_default="0"),
     Column("avatarHead", Integer, server_default="0"),
+    Column(
+        "stageId", Integer, server_default="99999", nullable=False
+    ),  # 99999 is the pseudo stage ID for unset stage
     UniqueConstraint("user", "version", name="chuni_profile_profile_uk"),
     mysql_charset="utf8mb4",
 )

titles/chuni/schema/static.py

@@ -310,6 +310,27 @@ unlock_challenge = Table(
 )
 
 
+linked_verse: Table = Table(
+    "chuni_static_linked_verse",
+    metadata,
+    Column("id", Integer, primary_key=True, nullable=False),
+    Column("version", Integer, nullable=False),
+    Column("linkedVerseId", Integer, nullable=False),
+    Column("name", String(255)),
+    Column("isEnabled", Boolean, server_default="1", nullable=False),
+    Column("startDate", TIMESTAMP, server_default=func.now()),
+    Column("courseId1", Integer),
+    Column("courseId2", Integer),
+    Column("courseId3", Integer),
+    Column("courseId4", Integer),
+    Column("courseId5", Integer),
+    UniqueConstraint(
+        "version", "linkedVerseId", name="chuni_static_linked_verse_pk"
+    ),
+    mysql_charset="utf8mb4",
+)
+
+
 class ChuniStaticData(BaseData):
     async def put_login_bonus(
         self,
@@ -1232,3 +1253,57 @@ class ChuniStaticData(BaseData):
         if result is None:
             return None
         return result.fetchall()
+        
+    async def put_linked_verse(
+        self,
+        version: int,
+        linked_verse_id: int,
+        name: str,
+        course_id1: Optional[int] = None,
+        course_id2: Optional[int] = None,
+        course_id3: Optional[int] = None,
+        course_id4: Optional[int] = None,
+        course_id5: Optional[int] = None,
+    ) -> Optional[int]:
+
+        sql = insert(linked_verse).values(
+            version=version,
+            linkedVerseId=linked_verse_id,
+            name=name,
+            courseId1=course_id1,
+            courseId2=course_id2,
+            courseId3=course_id3,
+            courseId4=course_id4,
+            courseId5=course_id5,
+        )
+
+        conflict = sql.on_duplicate_key_update(
+            name=name,
+            courseId1=course_id1,
+            courseId2=course_id2,
+            courseId3=course_id3,
+            courseId4=course_id4,
+            courseId5=course_id5,
+        )
+
+        result = await self.execute(conflict)
+        
+        if result is None:
+            return None
+        
+        return result.lastrowid
+
+    async def get_linked_verses(self, version: int) -> Optional[List[Dict]]:
+        sql = linked_verse.select(
+            and_(
+                linked_verse.c.version == version,
+                linked_verse.c.isEnabled == True,
+            )
+        ).order_by(linked_verse.c.startDate.asc())
+
+        result = await self.execute(sql)
+        
+        if result is None:
+            return None
+        
+        return result.fetchall()