feat: 添加 update_video_stats 方法以更新视频统计信息

DB.py

@@ -586,3 +586,31 @@ class DBSA:
                 cls.push_record_many(payloads)
             except Exception as re:
                 print("[Redis 回退失败]", re)
+
+    @classmethod
+    def update_video_stats(cls, locator:dict, stats:dict) -> int:
+        """
+        立即更新 sh_dm_video_v2 表中的统计字段。
+
+        :param locator: 用于定位行的字典，必须包含: v_xid, rn
+        :param stats: 需要更新的统计字段，如 {"fans": 633, "videos": 10090, "view": 1678408}
+        :return: 受影响的行数
+        """
+        v_xid = locator.get("v_xid")
+        rn = locator.get("rn")
+        if not v_xid or not rn:
+            raise ValueError("locator 必须包含 'v_xid' 和 'rn'")
+
+        params = dict(stats)
+        params["updatetime"] = int(time.time())
+
+        # 使用 Core Table 的 update() 方法
+        stmt = (
+            video
+            .update()
+            .where(video.c.v_xid == v_xid, video.c.rn == rn)
+            .values(**params)
+        )
+        with _engine.begin() as conn:
+            result = conn.execute(stmt)
+        return result.rowcount