feat: 实现新版数据库 users 表插入与更新机制

jobs/jianshu/article_earning_ranking.py

@@ -18,6 +18,7 @@
 from models.new.jianshu.article_earning_ranking_record import (
     ArticleEarningRankingRecord as NewDbArticleEarningRankingRecord,
 )
+from models.new.jianshu.user import User as NewDbUser
 from utils.log import (
     get_flow_run_name,
     log_flow_run_start,
@@ -75,6 +76,12 @@ async def process_item(
             name=author_info.name,
             avatar_url=author_info.avatar_url,
         )
+        await NewDbUser.upsert(
+            slug=author_slug,
+            id=author_info.id,
+            name=author_info.name,
+            avatar_url=author_info.avatar_url,
+        )
 
     return ArticleEarningRankingRecordDocument(
         date=date,

jobs/jianshu/daily_update_ranking.py

@@ -14,6 +14,7 @@
 from models.new.jianshu.daily_update_ranking_record import (
     DailyUpdateRankingRecord as NewDbDailyUpateRankingRecord,
 )
+from models.new.jianshu.user import User as NewDbUser
 from utils.log import log_flow_run_start, log_flow_run_success, logger
 from utils.prefect_helper import (
     generate_deployment_config,
@@ -29,6 +30,11 @@ async def process_item(
         name=item.user_info.name,
         avatar_url=item.user_info.avatar_url,
     )
+    await NewDbUser.upsert(
+        slug=item.user_info.slug,
+        name=item.user_info.name,
+        avatar_url=item.user_info.avatar_url,
+    )
 
     return DailyUpdateRankingRecordDocument(
         date=date,

jobs/jianshu/user_assets_ranking.py

@@ -14,6 +14,7 @@
     AssetsRankingRecordDocument,
 )
 from models.jianshu.user import UserDocument
+from models.new.jianshu.user import User as NewDbUser
 from models.new.jianshu.user_assets_ranking_record import (
     UserAssetsRankingRecord as NewDbAssetsRankingRecord,
 )
@@ -74,6 +75,12 @@ async def process_item(
             name=item.user_info.name,
             avatar_url=item.user_info.avatar_url,
         )
+        await NewDbUser.upsert(
+            slug=item.user_info.slug,
+            id=item.user_info.id,
+            name=item.user_info.name,
+            avatar_url=item.user_info.avatar_url,
+        )
 
     return AssetsRankingRecordDocument(
         date=date,

models/new/jianshu/user.py

@@ -59,3 +59,103 @@ async def create(self) -> None:
                 self.avatar_url,
             ),
         )
+
+    @classmethod
+    async def get_by_slug(cls, slug: str) -> Optional["User"]:
+        cursor = await jianshu_conn.execute(
+            "SELECT status, update_time, id, name, history_names, "
+            "avatar_url FROM users WHERE slug = %s;",
+            (slug,),
+        )
+
+        data = await cursor.fetchone()
+        if not data:
+            return None
+
+        return cls(
+            slug=slug,
+            status=data[0],
+            update_time=data[1],
+            id=data[2],
+            name=data[3],
+            history_names=data[4],
+            avatar_url=data[5],
+        )
+
+    @classmethod
+    async def upsert(
+        cls,
+        slug: str,
+        id: Optional[int] = None,  # noqa: A002
+        name: Optional[str] = None,
+        avatar_url: Optional[str] = None,
+    ) -> None:
+        user = await cls.get_by_slug(slug)
+        # 如果不存在，创建用户
+        if not user:
+            await cls(
+                slug=slug,
+                status=StatusEnum.NORMAL,
+                update_time=datetime.now(),
+                id=id,
+                name=name,
+                history_names=[],
+                avatar_url=avatar_url,
+            ).create()
+            return
+
+        # 如果当前数据不是最新，跳过更新
+        if user.update_time > datetime.now():
+            return
+
+        # 在一个事务中一次性完成全部字段的更新
+        async with jianshu_conn.transaction():
+            # 更新更新时间
+            await jianshu_conn.execute(
+                "UPDATE users SET update_time = %s WHERE slug = %s",
+                (datetime.now(), slug),
+            )
+
+            # ID 无法被修改，如果异常则抛出错误
+            if user.id and id and user.id != id:
+                raise ValueError(f"用户 ID 不一致：{user.id} != {id}")
+
+            # 如果没有存储 ID，进行添加
+            if not user.id and id:
+                await jianshu_conn.execute(
+                    "UPDATE users SET id = %s WHERE slug = %s",
+                    (id, slug),
+                )
+
+            # 如果没有存储昵称，进行添加
+            if not user.name and name:
+                await jianshu_conn.execute(
+                    "UPDATE users SET name = %s WHERE slug = %s",
+                    (name, slug),
+                )
+
+            # 更新昵称
+            if user.name and name and user.name != name:
+                await jianshu_conn.execute(
+                    "UPDATE users SET name = %s WHERE slug = %s",
+                    (name, slug),
+                )
+                await jianshu_conn.execute(
+                    "UPDATE users SET history_names = array_append(history_names, %s) "
+                    "WHERE slug = %s;",
+                    (user.name, slug),
+                )
+
+            # 如果没有存储头像链接，进行添加
+            if not user.avatar_url and avatar_url:
+                await jianshu_conn.execute(
+                    "UPDATE users SET avatar_url = %s WHERE slug = %s",
+                    (avatar_url, slug),
+                )
+
+            # 更新头像链接
+            if user.avatar_url and avatar_url and user.avatar_url != avatar_url:
+                await jianshu_conn.execute(
+                    "UPDATE users SET avatar_url = %s WHERE slug = %s",
+                    (avatar_url, slug),
+                )