feat: 新增采集日更排行榜记录工作流

jobs/__init__.py

@@ -4,8 +4,11 @@
 from prefect.client.schemas.schedules import CronSchedule
 from prefect.deployments.runner import RunnerDeployment
 
-from jobs.fetch_article_earning_rank_records import (
-    fetch_article_earning_rank_records_job,
+from jobs.fetch_article_earning_ranking_records import (
+    fetch_article_earning_ranking_records_job,
+)
+from jobs.fetch_daily_update_ranking_records import (
+    fetch_daily_update_ranking_records_job,
 )
 from utils.job_model import Job
 
@@ -35,7 +38,10 @@ def create_deployment(job: Job, flow: FlowType) -> DeploymentType:
     )
 
 
-JOBS: Tuple[Job, ...] = (fetch_article_earning_rank_records_job,)
+JOBS: Tuple[Job, ...] = (
+    fetch_article_earning_ranking_records_job,
+    fetch_daily_update_ranking_records_job,
+)
 
 FLOWS: Tuple[FlowType, ...] = tuple(map(create_flow, JOBS))
 DEPLOYMENTS: Tuple[DeploymentType, ...] = tuple(

jobs/fetch_article_earning_ranking_records.py

@@ -36,7 +36,7 @@ class ArticleEarningRankingRecordModel(Document):
     earning: EarningField
 
     class Settings:
-        name = "article_earning_ranking_record"
+        name = "article_earning_ranking_records"
         indexes = ("date", "ranking")
 
 

jobs/fetch_daily_update_ranking_records.py

@@ -0,0 +1,61 @@
+from datetime import date, datetime
+from typing import List
+
+from beanie import Document
+from jkit.ranking.daily_update import DailyUpdateRanking, DailyUpdateRankingRecord
+from prefect import flow, get_run_logger
+from pydantic import BaseModel, PositiveInt
+
+from utils.db import init_db
+from utils.job_model import Job
+
+
+class UserInfoField(BaseModel):
+    slug: str
+    name: str
+
+
+class DailyUpdateRankingRecordModel(Document):
+    date: date
+    ranking: PositiveInt
+    days: PositiveInt
+    user_info: UserInfoField
+
+    class Settings:
+        name = "daily_update_ranking_records"
+        indexes = ("date", "ranking", "days")
+
+
+def process_item(
+    item: DailyUpdateRankingRecord, /, *, current_date: date
+) -> DailyUpdateRankingRecordModel:
+    return DailyUpdateRankingRecordModel(
+        date=current_date,
+        ranking=item.ranking,
+        days=item.days,
+        user_info=UserInfoField(slug=item.user_info.slug, name=item.user_info.name),
+    )
+
+
+@flow
+async def main() -> None:
+    logger = get_run_logger()
+
+    current_date = datetime.now().date()
+
+    await init_db([DailyUpdateRankingRecordModel])
+    logger.info("初始化 ODM 模型成功")
+
+    data: List[DailyUpdateRankingRecordModel] = []
+    async for item in DailyUpdateRanking():
+        processed_item = process_item(item, current_date=current_date)
+        data.append(processed_item)
+
+    await DailyUpdateRankingRecordModel.insert_many(data)
+
+
+fetch_daily_update_ranking_records_job = Job(
+    func=main,
+    name="采集日更排行榜记录",
+    cron="0 1 * * *",
+)