Add continuous model discovery and queued submission

.env.example

@@ -1,7 +1,14 @@
 STRATEGY_ID=replace-with-platform-strategy-id
+PIPELINE_MODE=continuous
 PORT=8080
 PIPELINE_RUN_MODE=resume
 PIPELINE_START_STEP=1
 PIPELINE_FORCE_RERUN_STEPS=
 PIPELINE_RUN_INTERVAL_SECONDS=3600
 PIPELINE_RETRY_INTERVAL_SECONDS=60
+PIPELINE_SCAN_INTERVAL_SECONDS=1800
+HF_START_PAGE=1
+HF_END_PAGE=0
+HF_MAX_PAGES=0
+HF_PER_PAGE=100
+PIPELINE_RETRY_FAILED=true

Dockerfile

@@ -7,7 +7,8 @@ ENV PYTHONDONTWRITEBYTECODE=1 \
 WORKDIR /app
 
 COPY requirements.txt .
-RUN pip install --no-cache-dir -r requirements.txt
+COPY vendor/ ./vendor/
+RUN pip install --no-cache-dir --no-index --find-links=/app/vendor -r requirements.txt
 
 COPY . .
 RUN mkdir -p /app/pipeline_outputs

README.md

@@ -1,5 +1,7 @@
 # new-pipeline
 
+当前版本：`v1.1.0`
+
 将 Hugging Face 镜像中的 GGUF 模型筛选、查重后，提交到 ModelHub
 执行适配验证。项目已按 ModelHub 智能体策略运行规范进行容器化。
 
@@ -9,12 +11,20 @@
 
 - `STRATEGY_ID`：当前策略 ID，提交任务时写入 `strategyId`。
 
+默认模式会持续扫描所选用户的全部模型。发现符合条件且未入库的模型后立即加入提交队列；提交端遇到平台任务上限时等待，扫描端继续工作。
+
 ModelHub 登录账号和密码按照当前部署约定固定在
 `modelhub_pipeline_resumable.py` 中，不需要额外配置环境变量。
 
 可选变量：
 
 - `PORT`：健康检查端口，默认 `8080`。
+- `PIPELINE_MODE`：`continuous`（默认，持续扫描）或 `batch`（兼容旧的四步批处理）。
+- `PIPELINE_SCAN_INTERVAL_SECONDS`：持续模式扫描间隔，默认 `1800` 秒。
+- `HF_START_PAGE` / `HF_END_PAGE`：扫描页范围；结束页为 `0` 时持续翻页直到没有数据。
+- `HF_MAX_PAGES`：单次扫描最大页数，默认 `0` 表示不限制。
+- `HF_PER_PAGE`：每页模型数，默认 `100`。
+- `PIPELINE_RETRY_FAILED`：是否重试历史失败模型，默认 `true`。
 - `PIPELINE_RUN_MODE`：`resume` 或 `fresh`，默认 `resume`。
 - `PIPELINE_START_STEP`：从第 1–4 步中的哪一步开始，默认 `1`。
 - `PIPELINE_FORCE_RERUN_STEPS`：强制重跑的步骤，例如 `1,2,3`。
@@ -50,6 +60,9 @@ docker run --rm -p 8080:8080 \
 镜像监听 `8080` 端口，并处理 `SIGTERM`。收到停机信号后，后台等待和轮询
 会立即停止；进行中的 HTTP 请求最长等待 20 秒，随后关闭会话并退出。
 
+`vendor/` 保存了 Linux x86_64 / Python 3.11 的离线 wheel。镜像构建不会访问
+PyPI，避免平台构建环境网络较慢时发生依赖下载超时。
+
 运行时产生的断点文件位于 `/app/pipeline_outputs`。如需跨 Pod 保留断点，
 应为该目录挂载持久卷。
 

main.py

@@ -11,6 +11,7 @@ import modelhub_pipeline_resumable as pipeline
 
 
 HOST = "0.0.0.0"
+VERSION = "1.1.0"
 PORT = int(os.getenv("PORT", "8080"))
 RUN_INTERVAL_SECONDS = int(os.getenv("PIPELINE_RUN_INTERVAL_SECONDS", "3600"))
 RETRY_INTERVAL_SECONDS = int(os.getenv("PIPELINE_RETRY_INTERVAL_SECONDS", "60"))
@@ -67,6 +68,7 @@ class Handler(BaseHTTPRequestHandler):
                 200,
                 {
                     "name": "new-pipeline",
+                    "version": VERSION,
                     "status": "running",
                     "strategy_id": STRATEGY_ID,
                     "pipeline_status": state["pipeline_status"],

modelhub_pipeline_resumable.py

@@ -1,9 +1,10 @@
-import os
 import argparse
+import os
 import re
 import json
 import random
 import threading
+import queue
 from typing import Dict, Tuple, List, Optional, Set
 
 import requests
@@ -42,9 +43,9 @@ STRATEGY_ID = os.environ["STRATEGY_ID"]
 ORG_NAME = "mradermacher"
 HF_API_URL = "https://hf-mirror.com/api/models"
 
-PER_PAGE = 100
+PER_PAGE = int(os.getenv("HF_PER_PAGE", "100"))
-START_PAGE = 6  # 起始页码
+START_PAGE = int(os.getenv("HF_START_PAGE", "1"))
-END_PAGE = 6    # 结束页码
+END_PAGE = int(os.getenv("HF_END_PAGE", "0"))  # 0 表示持续翻页直到末页
 REQUEST_DELAY = 1  # 每次请求的基础延迟（秒），可根据情况调整
 RANDOM_DELAY_RANGE = (0.5, 2)  # 随机延迟范围（秒），避免固定间隔被识别
 RETRY_TIMES = 3  # 429错误重试次数
@@ -66,6 +67,14 @@ TASK_LIMIT_CODE = 60007
 POLL_INTERVAL_SECONDS = 120
 MAX_LIMIT_RETRY_TIMES = None  # None 表示一直等到有空位
 
+PIPELINE_MODE = os.getenv("PIPELINE_MODE", "continuous").strip().lower()
+SCAN_INTERVAL_SECONDS = int(os.getenv("PIPELINE_SCAN_INTERVAL_SECONDS", "1800"))
+MAX_SCAN_PAGES = int(os.getenv("HF_MAX_PAGES", "0"))  # 0 表示不限制
+RETRY_FAILED = os.getenv("PIPELINE_RETRY_FAILED", "true").strip().lower() in {
+    "1", "true", "yes", "on"
+}
+SHUTDOWN_TIMEOUT_SECONDS = 25
+
 DEBUG_PRINT_PAYLOAD = False
 
 # 是否开启断点续跑：已经成功提交过的模型，下次运行自动跳过
@@ -349,7 +358,11 @@ def get_model_filename(model_id: str) -> str:
 def get_org_models(session: requests.Session, org_name: str) -> List[str]:
     models: List[str] = []
 
-    for page in range(START_PAGE, END_PAGE + 1):
+    page = START_PAGE
+    pages_seen = 0
+    while END_PAGE == 0 or page <= END_PAGE:
+        if MAX_SCAN_PAGES and pages_seen >= MAX_SCAN_PAGES:
+            break
         try:
             interruptible_sleep(random.uniform(*RANDOM_DELAY_RANGE))
 
@@ -377,12 +390,15 @@ def get_org_models(session: requests.Session, org_name: str) -> List[str]:
                     models.append(model_id)
 
             print(f"成功获取第 {page} 页，共 {len(page_models)} 个模型")
+            pages_seen += 1
+            page += 1
 
         except ShutdownRequested:
             raise
         except Exception as e:
             print(f"获取第 {page} 页失败: {e}")
-            continue
+            pages_seen += 1
+            page += 1
 
     # 去重但保持顺序
     seen = set()
@@ -810,11 +826,140 @@ def submit_adapt_task_with_polling(token: str, model_id: str) -> Tuple[bool, str
         return False, current_token
 
 
+def iter_org_models(session: requests.Session, org_name: str):
+    """逐页获取模型，避免等待全量列表完成后才开始处理。"""
+    page = START_PAGE
+    pages_seen = 0
+    while END_PAGE == 0 or page <= END_PAGE:
+        if MAX_SCAN_PAGES and pages_seen >= MAX_SCAN_PAGES:
+            return
+        check_shutdown()
+        try:
+            interruptible_sleep(random.uniform(*RANDOM_DELAY_RANGE))
+            response = session.get(
+                HF_API_URL,
+                params={
+                    "author": org_name,
+                    "page": page,
+                    "perPage": PER_PAGE,
+                    "sort": "lastModified",
+                    "direction": "-1",
+                },
+                timeout=20,
+            )
+            response.raise_for_status()
+            data = response.json()
+            if not data:
+                return
+            page_models = [item.get("modelId") for item in data if item.get("modelId")]
+            print(f"扫描第 {page} 页，共 {len(page_models)} 个模型")
+            for model_id in page_models:
+                yield model_id
+            pages_seen += 1
+            page += 1
+        except ShutdownRequested:
+            raise
+        except Exception as exc:
+            print(f"扫描第 {page} 页失败: {exc}")
+            pages_seen += 1
+            page += 1
+
+
+def continuous_main() -> None:
+    """持续扫描新模型，并由独立消费者逐个提交适配任务。"""
+    ensure_output_dir()
+    pending: "queue.Queue[str]" = queue.Queue()
+    submitted = load_model_set(SUBMITTED_FILE)
+    failed = load_model_set(FAILED_FILE)
+    scheduled: Set[str] = set()
+    stats = {"scanned": 0, "queued": 0, "submitted": 0, "failed": 0}
+    stats_lock = threading.Lock()
+
+    def submit_worker() -> None:
+        token: Optional[str] = None
+        while not stop_event_is_set():
+            try:
+                model_id = pending.get(timeout=1)
+            except queue.Empty:
+                continue
+            try:
+                if token is None:
+                    token = get_token()
+                ok, token = submit_adapt_task_with_polling(token, model_id)
+                if ok:
+                    submitted.add(model_id)
+                    append_model(SUBMITTED_FILE, model_id)
+                    with stats_lock:
+                        stats["submitted"] += 1
+                else:
+                    failed.add(model_id)
+                    append_model(FAILED_FILE, model_id)
+                    with stats_lock:
+                        stats["failed"] += 1
+            except ShutdownRequested:
+                return
+            except Exception as exc:
+                print(f"提交线程异常 {model_id}: {exc}")
+                failed.add(model_id)
+                append_model(FAILED_FILE, model_id)
+                with stats_lock:
+                    stats["failed"] += 1
+            finally:
+                scheduled.discard(model_id)
+                pending.task_done()
+
+    worker = threading.Thread(target=submit_worker, name="modelhub-submitter", daemon=True)
+    worker.start()
+    try:
+        while not stop_event_is_set():
+            session = create_hf_session()
+            try:
+                for model_id in iter_org_models(session, ORG_NAME):
+                    check_shutdown()
+                    with stats_lock:
+                        stats["scanned"] += 1
+                    if model_id in submitted or model_id in scheduled:
+                        continue
+                    if model_id in failed and not RETRY_FAILED:
+                        continue
+                    target_filename = get_model_filename(model_id)
+                    size_gb = extract_gguf_file_size(session, model_id, target_filename)
+                    if size_gb < 0 or size_gb > MAX_FILE_SIZE_GB:
+                        continue
+                    in_db = check_model_in_modelhub_db(model_id)
+                    if in_db is not False:
+                        continue
+                    scheduled.add(model_id)
+                    pending.put(model_id)
+                    with stats_lock:
+                        stats["queued"] += 1
+                    print(f"已加入提交队列: {model_id} (队列长度 {pending.qsize()})")
+            finally:
+                session.close()
+            print(f"扫描周期完成: {stats}; 下一次扫描等待 {SCAN_INTERVAL_SECONDS} 秒")
+            interruptible_sleep(SCAN_INTERVAL_SECONDS)
+    except ShutdownRequested:
+        pass
+    finally:
+        request_shutdown()
+        worker.join(timeout=SHUTDOWN_TIMEOUT_SECONDS)
+
+
+def stop_event_is_set() -> bool:
+    return _shutdown_event.is_set()
+
+
 # ============================================================
 # 10. 主流程：筛选 -> 查重 -> 提交
 # ============================================================
 
 def main() -> None:
+    if PIPELINE_MODE not in {"continuous", "batch"}:
+        raise ValueError("PIPELINE_MODE 必须是 continuous 或 batch")
+    if PIPELINE_MODE == "continuous":
+        continuous_main()
+        return
+
     ensure_output_dir()
     args = parse_args()
     check_shutdown()

requirements.txt

@@ -1 +1 @@
-requests>=2.31.0,<3.0.0
+requests==2.34.2