添加最大数量限制

app/config/settings.py

@@ -38,6 +38,7 @@ class Settings(BaseSettings):
     clean_batch_size: int = 100
     clean_concurrency: int = 80
     clean_interval_seconds: int = 200
+    clean_total_limit: int = 0  # 累计清洗总数上限，达到后停止清洗任务；0 = 不限制
 
     # ──────────── 公司补充参数 ────────────
     company_batch_size: int = 20

app/services/job_clean_service.py

@@ -24,9 +24,23 @@ _id_gen = SnowflakeGenerator(instance=1)
 # 公司创建锁（防止并发重复插入同一公司）
 _company_lock = asyncio.Lock()
 
+# 累计已清洗总数（用于总数上限控制，进程内统计）
+_cleaned_total = 0
+
+
+def is_clean_limit_reached() -> bool:
+    """是否已达到累计清洗总数上限（clean_total_limit=0 表示不限制）"""
+    return 0 < settings.clean_total_limit <= _cleaned_total
+
 
 async def run_job_clean() -> None:
     """一次批量清洗任务"""
+    global _cleaned_total
+
+    # 总数上限：已达上限直接跳过
+    if is_clean_limit_reached():
+        return
+
     # 1. 从 PG 锁定一批待清洗数据
     async with PgSession() as pg:
         result = await pg.execute(
@@ -62,7 +76,11 @@ async def run_job_clean() -> None:
 
     # 汇总
     errors = sum(1 for r in results if isinstance(r, Exception))
-    log.info("岗位清洗：本批完成，共{}条，异常{}条", len(rows), errors)
+    _cleaned_total += len(rows)
+    log.info("岗位清洗：本批完成，共{}条，异常{}条，累计{}条", len(rows), errors, _cleaned_total)
+
+    if is_clean_limit_reached():
+        log.info("岗位清洗：已达累计上限 {} 条，停止清洗任务", settings.clean_total_limit)
 
 
 async def _clean_one(sem: asyncio.Semaphore, data: dict) -> None: