get_cover接口优化

app/ab_test/get_cover.py

@@ -1,7 +1,5 @@
 from app.infra.shared.response import Response
-from app.infra.utils import fetch_channel_info
-from app.infra.utils import fetch_aigc_cover
-from app.infra.utils import fetch_long_video_cover
+from app.infra.mapper import LongVideoDatabaseMapper, AigcDatabaseMapper
 
 
 class GetCoverService(Response):
@@ -29,7 +27,7 @@ class GetCoverService(Response):
     async def fetch_cover_info(self, pool_name, channel_content_id: str):
         match pool_name:
             case "aigc":
-                fetch_response = await fetch_aigc_cover(self.pool, channel_content_id)
+                fetch_response = await AigcDatabaseMapper.fetch_aigc_cover(self.pool, channel_content_id)
                 if fetch_response:
                     image_oss = fetch_response[0]["oss_object_key"]
                     if image_oss:
@@ -45,7 +43,7 @@ class GetCoverService(Response):
                 else:
                     return await self.fetch_cover_info("long_video", channel_content_id)
             case "long_video":
-                fetch_response = await fetch_long_video_cover(
+                fetch_response = await LongVideoDatabaseMapper.fetch_long_video_cover(
                     self.pool, channel_content_id
                 )
                 if fetch_response:
@@ -68,7 +66,7 @@ class GetCoverService(Response):
         if video_index == 2:
             return await self.fetch_cover_info("long_video", seed_video_id)
 
-        channel_info = await fetch_channel_info(self.pool, content_id)
+        channel_info = await AigcDatabaseMapper.fetch_channel_info(self.pool, content_id)
         if not channel_info:
             return self.error_response(
                 error_code="402",

app/infra/mapper/__init__.py

@@ -0,0 +1,2 @@
+from .aigc_mapper import AigcDatabaseMapper
+from .long_video_mapper import LongVideoDatabaseMapper

app/infra/mapper/aigc_mapper.py

@@ -0,0 +1,33 @@
+from typing import List, Dict
+
+from app.core.database import DatabaseManager
+
+
+class AigcDatabaseMapper:
+    # 从 aigc 数据库查询 channel_content_id && channel 信息
+    @staticmethod
+    async def fetch_channel_info(pool: DatabaseManager, content_id: str) -> List[Dict]:
+        query = """
+            SELECT t1.channel_content_id, t2.channel
+            FROM produce_plan_exe_record t1 JOIN crawler_content t2 ON t1.channel_content_id = t2.channel_content_id
+            WHERE plan_exe_id = %s;
+        """
+        return await pool.async_fetch(query=query, db_name="aigc", params=(content_id,))
+
+    # 从 aigc 如何查询文章封面信息
+    @staticmethod
+    async def fetch_aigc_cover(
+        pool: DatabaseManager, channel_content_id: str
+    ) -> List[Dict]:
+        """
+        use channel_content_id to find article cover
+        """
+        COVER_TYPE = 2
+        query = """
+            select image_url, oss_object_key
+            from crawler_content_image
+            where channel_content_id = %s and image_type = %s;
+        """
+        return await pool.async_fetch(
+            query=query, db_name="aigc", params=(channel_content_id, COVER_TYPE)
+        )

app/infra/mapper/long_video_mapper.py

@@ -0,0 +1,20 @@
+from typing import List, Dict
+
+from app.core.database import DatabaseManager
+
+
+class LongVideoDatabaseMapper:
+    # 从 long_video 数据库查询 video_cover_snapshots 信息
+    @staticmethod
+    async def fetch_long_video_cover(
+        pool: DatabaseManager, video_id: str
+    ) -> List[Dict]:
+        """
+        use channel_content_id to find long video cover
+        """
+        query = """
+            select image_path from video_cover_snapshots where video_id = %s;
+        """
+        return await pool.async_fetch(
+            query=query, db_name="long_video", params=(video_id,)
+        )

app/infra/utils/__init__.py

@@ -1,3 +0,0 @@
-from .get_cover import fetch_channel_info
-from .get_cover import fetch_aigc_cover
-from .get_cover import fetch_long_video_cover

app/infra/utils/get_cover.py

@@ -1,46 +0,0 @@
-from aiomysql import DictCursor
-
-
-# fetch
-async def fetch_channel_info(pools, content_id):
-    """use content id to get channel_content_id && channel code"""
-    fetch_query = f"""
-        select t1.channel_content_id, t2.channel
-        from produce_plan_exe_record t1 join crawler_content t2 on t1.channel_content_id = t2.channel_content_id
-        where plan_exe_id = '{content_id}';
-    """
-    fetch_response = await pools.async_fetch(
-        query=fetch_query, db_name="aigc", cursor_type=DictCursor
-    )
-    return fetch_response
-
-
-async def fetch_aigc_cover(pools, channel_content_id):
-    """
-    use channel_content_id to find article cover
-    """
-    fetch_query = f"""
-        select image_url, oss_object_key
-        from crawler_content_image
-        where channel_content_id = '{channel_content_id}' and image_type = 2;
-    """
-    fetch_response = await pools.async_fetch(
-        query=fetch_query, db_name="aigc", cursor_type=DictCursor
-    )
-
-    return fetch_response
-
-
-async def fetch_long_video_cover(pools, channel_content_id):
-    """
-    use channel_content_id to find long video cover
-    """
-    fetch_query = f"""
-        select image_path
-        from video_cover_snapshots
-        where video_id = '{channel_content_id}';
-    """
-    fetch_response = await pools.async_fetch(
-        query=fetch_query, db_name="long_video", cursor_type=DictCursor
-    )
-    return fetch_response