feat: 恢复全量同步功能

Author: z0z0r4

mcim_sync/checker/curseforge.py

@@ -165,4 +165,4 @@ def check_newest_search_result(classId: int) -> List[int]:
 
         time.sleep(CURSEFORGE_DELAY)
 
-    return new_modids
+    return new_modids

mcim_sync/config.py

@@ -44,6 +44,7 @@ def validate_bool(cls, v):
 class JobInterval(BaseModel):
     curseforge_refresh: int = 60 * 60 * 2  # 2 hours
     modrinth_refresh: int = 60 * 60 * 2  # 2 hours
+    curseforge_refresh_full: int = 60 * 60 * 48  # 48 hours
     sync_curseforge_by_queue: int = 60 * 5  # 5 minutes
     sync_curseforge_by_search: int = 60 * 60 * 2 # 2 hours
     sync_modrinth_by_queue: int = 60 * 5  # 5 minutes

mcim_sync/fetcher/curseforge.py

@@ -1,4 +1,4 @@
-from typing import Union, List, Set
+from typing import Union, List, Set, Optional
 from odmantic import query
 import datetime
 import time
@@ -53,4 +53,3 @@ def fetch_expired_curseforge_data() -> List[int]:
         time.sleep(CURSEFORGE_DELAY)
         log.debug(f"Delay {CURSEFORGE_DELAY} seconds")
     return list(expired_modids)
-

mcim_sync/sync/curseforge.py

@@ -1,8 +1,7 @@
-from typing import List, Optional, Union
+from typing import List, Optional
 from tenacity import retry, stop_after_attempt, wait_fixed
 from odmantic import query
 from enum import Enum
-import time
 
 from mcim_sync.models.database.curseforge import (
     File,
@@ -20,10 +19,12 @@
     get_mutil_mods_info,
     get_search_result,
 )
+
 # from mcim_sync.models.database.file_cdn import File as FileCDN
 from mcim_sync.utils.constans import ProjectDetail
 from mcim_sync.utils.loger import log
 from mcim_sync.utils.model_submitter import ModelSubmitter
+
 # from mcim_sync.utils import find_hash_in_curseforge_hashes
 from mcim_sync.database.mongodb import sync_mongo_engine as mongodb_engine
 from mcim_sync.config import Config
@@ -87,6 +88,8 @@ def sync_mod_all_files(
     params = {"index": 0, "pageSize": 50}
     file_id_list = []
 
+    original_files_count = mongodb_engine.count(File, File.modId == modId)
+
     while True:
         res = get_mod_files(modId, params["index"], params["pageSize"])
         append_model_from_files_res(
@@ -108,7 +111,7 @@ def sync_mod_all_files(
         File, File.modId == modId, query.not_in(File.id, file_id_list)
     )
     log.info(
-        f"Finished sync mod {modId}, total {page.totalCount} files, removed {removed_count} files"
+        f"Finished sync mod {modId}, total {page.totalCount} files, removed {removed_count} files, original files {original_files_count}"
     )
 
     return page.totalCount
@@ -118,7 +121,7 @@ def sync_mod_all_files_at_once(
     modId: int, latestFiles: List[dict], need_to_cache: bool = True
 ) -> Optional[int]:
     max_retries = 3
-    page_size= 10000
+    page_size = 10000
     for i in range(max_retries):
         res = get_mod_files(modId, index=0, pageSize=page_size)
 
@@ -128,17 +131,21 @@ def sync_mod_all_files_at_once(
 
         if page.resultCount != page.totalCount or len(file_id_list) != page.resultCount:
             log.warning(
-                f"ResultCount {page.resultCount} != TotalCount {page.totalCount} for mod {modId}, or the count of files != resultCount, response maybe incomplete, passing sync, retrying {i+1}/{max_retries}"
+                f"ResultCount {page.resultCount} != TotalCount {page.totalCount} for mod {modId}, or the count of files != resultCount, response maybe incomplete, passing sync, retrying {i + 1}/{max_retries}"
             )
             # time.sleep(1)
             page_size -= 1
             continue
         else:
             break
     else:
-        log.error(f"Failed to get all files for mod {modId} after {max_retries} retries")
+        log.error(
+            f"Failed to get all files for mod {modId} after {max_retries} retries"
+        )
         return None
 
+    original_files_count = mongodb_engine.count(File, File.modId == modId)
+
     append_model_from_files_res(
         res, latestFiles=latestFiles, need_to_cache=need_to_cache
     )
@@ -147,7 +154,7 @@ def sync_mod_all_files_at_once(
         File, File.modId == modId, query.not_in(File.id, file_id_list)
     )
     log.info(
-        f"Finished sync mod {modId}, total {page.totalCount} files, resultCount {page.resultCount}, removed {removed_count} files"
+        f"Finished sync mod {modId}, total {page.totalCount} files, resultCount {page.resultCount}, removed {removed_count} files, existing files {original_files_count}"
     )
 
     return page.totalCount

mcim_sync/tasks/curseforge.py

@@ -17,7 +17,10 @@
     check_new_modids,
     check_newest_search_result,
 )
-from mcim_sync.fetcher.curseforge import fetch_expired_curseforge_data
+from mcim_sync.fetcher.curseforge import (
+    fetch_expired_curseforge_data,
+    fetch_all_curseforge_data,
+)
 from mcim_sync.queues.curseforge import clear_curseforge_all_queues
 from mcim_sync.tasks import create_tasks_pool, curseforge_pause_event
 
@@ -57,7 +60,9 @@ def refresh_curseforge_with_modify_date() -> bool:
 
     failed_count = len(curseforge_expired_modids) - len(projects_detail_info)
     failed_modids = [
-        modid for modid in curseforge_expired_modids if modid not in projects_detail_info
+        modid
+        for modid in curseforge_expired_modids
+        if modid not in projects_detail_info
     ]
     log.info(
         f"CurseForge expired data sync finished, total: {len(curseforge_expired_modids)}, "
@@ -218,30 +223,39 @@ def sync_curseforge_by_search():
     return True
 
 
-# def sync_curseforge_full():
-#     log.info("Start fetching all data.")
-#     total_data = {
-#         "curseforge": 0,
-#     }
+def sync_curseforge_full():
+    log.info("Start fetching curseforge all data.")
 
-#     if SYNC_CURSEFORGE:
-#         curseforge_data = fetch_all_curseforge_data()
-#         log.info(f"Curseforge data totally fetched: {len(curseforge_data)}")
-#         total_data["curseforge"] = len(curseforge_data)
+    curseforge_data = fetch_all_curseforge_data()
+    log.info(f"Curseforge data totally fetched: {len(curseforge_data)}")
 
-#     # 允许请求
-#     curseforge_pause_event.set()
+    curseforge_pause_event.set()
 
-#     curseforge_pool, curseforge_futures = create_tasks_pool(
-#         sync_mod_all_files, curseforge_data, MAX_WORKERS, "curseforge"
-#     )
+    curseforge_pool, curseforge_futures = create_tasks_pool(
+        sync_mod, curseforge_data, MAX_WORKERS, "curseforge"
+    )
 
-#     log.info(
-#         f"All {len(curseforge_futures)} tasks submitted, waiting for completion..."
-#     )
+    log.info(
+        f"All {len(curseforge_futures)} tasks submitted, waiting for completion..."
+    )
 
-#     for future in as_completed(curseforge_futures):
-#         # 不需要返回值
-#         pass
+    projects_detail_info = []
+    for future in as_completed(curseforge_futures):
+        result = future.result()
+        if result:
+            projects_detail_info.append(result)
+    else:
+        curseforge_pool.shutdown()
+
+    failed_count = len(curseforge_data) - len(projects_detail_info)
+    failed_modids = [
+        modid for modid in curseforge_data if modid not in projects_detail_info
+    ]
+
+    log.info(
+        f"CurseForge full sync finished, total: {len(curseforge_data)}, "
+        f"success: {len(projects_detail_info)}, failed: {failed_count}, "
+        f"failed modids: {failed_modids if failed_modids else 'None'}"
+    )
 
-#     curseforge_pool.shutdown()
+    curseforge_pool.shutdown()

start.py

@@ -13,12 +13,14 @@
     refresh_modrinth_with_modify_date,
     refresh_modrinth_tags,
     sync_modrinth_by_search,
+    
 )
 from mcim_sync.tasks.curseforge import (
     sync_curseforge_queue,
     refresh_curseforge_with_modify_date,
     refresh_curseforge_categories,
     sync_curseforge_by_search,
+    sync_curseforge_full
 )
 from mcim_sync.tasks.misc import send_statistics_to_telegram
 
@@ -47,6 +49,14 @@ def main():
             trigger=IntervalTrigger(seconds=config.interval.modrinth_refresh),
             name="modrinth_refresh",
         )
+    
+    if config.job_config.curseforge_refresh_full:
+        # 添加全量刷新任务，每 48 小时执行一次
+        curseforge_full_refresh_job = scheduler.add_job(
+            sync_curseforge_full,
+            trigger=IntervalTrigger(seconds=config.interval.curseforge_refresh_full),
+            name="curseforge_full_refresh",
+        )
 
     if config.job_config.sync_curseforge_by_queue:
         # 添加定时同步任务，用于检查 api 未找到的请求数据