feat: add scheduler jobs and content filters

README.md

@@ -161,15 +161,26 @@ uv run playwright install
 # 从配置文件中读取关键词搜索相关的帖子并爬取帖子信息与评论
 uv run main.py --platform xhs --lt qrcode --type search
 
+# 按发布时间和互动指标过滤爬取目标，例如只保存 2024-07-01 之后且点赞数不低于 1000 的内容
+uv run main.py --platform xhs --lt qrcode --type search --content_filters '{"publish_time":{"min":"2024-07-01"},"liked_count":{"min":1000}}'
+
+# 也可以只按互动指标过滤
+uv run main.py --platform xhs --lt qrcode --type search --content_filters '{"liked_count":{"min":1000}}'
+
 # 从配置文件中读取指定的帖子ID列表获取指定帖子的信息与评论信息
 uv run main.py --platform xhs --lt qrcode --type detail
 
+# 只维护登录态，不进入抓取流程
+uv run main.py --platform xhs --lt qrcode --type login
+
 # 打开对应APP扫二维码登录
 
 # 其他平台爬虫使用示例，执行下面的命令查看
 uv run main.py --help
 ```
 
+内容过滤支持不同平台的发布时间、点赞、收藏、转发、评论等字段，详见 [内容过滤使用指南](docs/内容过滤使用指南.md)。
+
 <details>
 <summary>🖥️ <strong>WebUI 可视化操作界面</strong></summary>
 
@@ -211,6 +222,18 @@ uv run uvicorn api.main:app --port 8080 --reload
 
 然后访问 `http://localhost:8080` 即可。
 
+#### 多实例调度器
+
+启动同一个 API 服务后，访问 `http://localhost:8080/scheduler` 可打开多实例调度器。调度器支持创建多个独立作业，每个作业拥有独立浏览器 Profile、CDP 端口、登录态、爬取目标和爬取参数。作业参数、内容过滤、Cookie 和代理都可以在页面中通过输入框、下拉菜单配置，无需手写 JSON。
+
+调度器运行数据默认保存在 `data/scheduler/`：
+
+- `scheduler.db`：作业配置、运行记录、日志和产物索引
+- `profiles/{job_id}/`：作业独立浏览器 Profile
+- `artifacts/{job_id}/{task_id}/`：单次运行抓取产物
+
+详细说明请查看：[多实例调度器使用指南](docs/多实例调度器使用指南.md)
+
 #### WebUI 功能特性
 
 - 可视化配置爬虫参数（平台、登录方式、爬取类型等）

api/main.py

@@ -32,7 +32,7 @@
 from fastapi.staticfiles import StaticFiles
 from fastapi.responses import FileResponse
 
-from .routers import crawler_router, data_router, websocket_router
+from .routers import crawler_router, data_router, scheduler_router, websocket_router
 
 # Project root directory (used for running subprocesses like uv run main.py)
 PROJECT_ROOT = Path(__file__).parent.parent
@@ -45,6 +45,7 @@
 
 # Get webui static files directory
 WEBUI_DIR = os.path.join(os.path.dirname(__file__), "webui")
+SCHEDULER_WEBUI_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "scheduler_webui")
 
 # CORS configuration - allow frontend dev server access
 app.add_middleware(
@@ -63,6 +64,7 @@
 # Register routers
 app.include_router(crawler_router, prefix="/api")
 app.include_router(data_router, prefix="/api")
+app.include_router(scheduler_router, prefix="/api")
 app.include_router(websocket_router, prefix="/api")
 
 
@@ -85,6 +87,18 @@ async def health_check():
     return {"status": "ok"}
 
 
+@app.get("/scheduler")
+async def serve_scheduler_frontend():
+    """Return scheduler WebUI page."""
+    index_path = os.path.join(SCHEDULER_WEBUI_DIR, "index.html")
+    if os.path.exists(index_path):
+        return FileResponse(index_path)
+    return {
+        "message": "MediaCrawler Scheduler WebUI",
+        "note": "Scheduler WebUI source was not found",
+    }
+
+
 @app.get("/api/env/check")
 async def check_environment():
     """Check if MediaCrawler environment is configured correctly"""
@@ -175,6 +189,7 @@ async def get_config_options():
             {"value": "search", "label": "Search Mode"},
             {"value": "detail", "label": "Detail Mode"},
             {"value": "creator", "label": "Creator Mode"},
+            {"value": "login", "label": "Login Only"},
         ],
         "save_options": [
             {"value": "jsonl", "label": "JSONL File"},
@@ -200,6 +215,13 @@ async def get_config_options():
     # Mount other static files (e.g., vite.svg)
     app.mount("/static", StaticFiles(directory=WEBUI_DIR), name="webui-static")
 
+if os.path.exists(SCHEDULER_WEBUI_DIR):
+    app.mount(
+        "/scheduler/static",
+        StaticFiles(directory=SCHEDULER_WEBUI_DIR),
+        name="scheduler-webui-static",
+    )
+
 
 if __name__ == "__main__":
     uvicorn.run(app, host="0.0.0.0", port=8080)

api/routers/__init__.py

@@ -18,6 +18,7 @@
 
 from .crawler import router as crawler_router
 from .data import router as data_router
+from .scheduler import router as scheduler_router
 from .websocket import router as websocket_router
 
-__all__ = ["crawler_router", "data_router", "websocket_router"]
+__all__ = ["crawler_router", "data_router", "scheduler_router", "websocket_router"]

api/routers/scheduler.py

@@ -0,0 +1,251 @@
+# -*- coding: utf-8 -*-
+
+from fastapi import APIRouter, HTTPException, Query
+
+from api.scheduler.manager import scheduler_manager
+from api.scheduler.schemas import (
+    ArtifactSummaryResponse,
+    ArtifactResponse,
+    InstanceCreateRequest,
+    InstanceResponse,
+    InstanceUpdateRequest,
+    JobCreateRequest,
+    JobResponse,
+    JobUpdateRequest,
+    SchedulerStatusResponse,
+    TaskCreateRequest,
+    TaskLogResponse,
+    TaskResponse,
+)
+
+router = APIRouter(prefix="/scheduler", tags=["scheduler"])
+
+
+@router.get("/status", response_model=SchedulerStatusResponse)
+async def scheduler_status():
+    return scheduler_manager.status()
+
+
+@router.get("/jobs", response_model=list[JobResponse])
+async def list_jobs():
+    return scheduler_manager.list_jobs()
+
+
+@router.post("/jobs", response_model=JobResponse)
+async def create_job(request: JobCreateRequest):
+    return scheduler_manager.create_job(request)
+
+
+@router.get("/jobs/{job_id}", response_model=JobResponse)
+async def get_job(job_id: str):
+    job = scheduler_manager.get_job(job_id)
+    if not job:
+        raise HTTPException(status_code=404, detail="Job not found")
+    return job
+
+
+@router.patch("/jobs/{job_id}", response_model=JobResponse)
+async def update_job(job_id: str, request: JobUpdateRequest):
+    try:
+        job = await scheduler_manager.update_job(job_id, request)
+    except RuntimeError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+    if not job:
+        raise HTTPException(status_code=404, detail="Job not found")
+    return job
+
+
+@router.delete("/jobs/{job_id}")
+async def delete_job(job_id: str):
+    try:
+        deleted = await scheduler_manager.delete_job(job_id)
+    except RuntimeError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+    if not deleted:
+        raise HTTPException(status_code=404, detail="Job not found")
+    return {"status": "ok", "message": "Job deleted"}
+
+
+@router.post("/jobs/{job_id}/login", response_model=TaskResponse)
+async def login_job(job_id: str):
+    try:
+        return await scheduler_manager.login_job(job_id)
+    except KeyError as exc:
+        raise HTTPException(status_code=404, detail="Job not found") from exc
+    except RuntimeError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+
+
+@router.post("/jobs/{job_id}/run", response_model=TaskResponse)
+async def run_job(job_id: str):
+    try:
+        return await scheduler_manager.run_job(job_id)
+    except KeyError as exc:
+        raise HTTPException(status_code=404, detail="Job not found") from exc
+    except RuntimeError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+
+
+@router.post("/jobs/{job_id}/stop", response_model=JobResponse)
+async def stop_job(job_id: str):
+    job = await scheduler_manager.stop_job(job_id)
+    if not job:
+        raise HTTPException(status_code=404, detail="Job not found")
+    return job
+
+
+@router.get("/jobs/{job_id}/logs", response_model=list[TaskLogResponse])
+async def list_job_logs(job_id: str, limit: int = Query(default=300, ge=1, le=1000)):
+    try:
+        return scheduler_manager.list_job_logs(job_id, limit=limit)
+    except KeyError as exc:
+        raise HTTPException(status_code=404, detail="Job not found") from exc
+
+
+@router.get("/jobs/{job_id}/artifacts", response_model=list[ArtifactResponse])
+async def list_job_artifacts(job_id: str):
+    try:
+        return scheduler_manager.list_job_artifacts(job_id)
+    except KeyError as exc:
+        raise HTTPException(status_code=404, detail="Job not found") from exc
+
+
+@router.get("/jobs/{job_id}/artifact-summary", response_model=ArtifactSummaryResponse)
+async def list_job_artifact_summary(
+    job_id: str,
+    work_limit: int = Query(default=200, ge=1, le=500),
+    word_limit: int = Query(default=80, ge=1, le=200),
+):
+    try:
+        return scheduler_manager.list_job_artifact_summary(job_id, work_limit=work_limit, word_limit=word_limit)
+    except KeyError as exc:
+        raise HTTPException(status_code=404, detail="Job not found") from exc
+
+
+@router.post("/jobs/{job_id}/artifacts/{artifact_id}/open")
+async def open_job_artifact(job_id: str, artifact_id: str):
+    try:
+        return scheduler_manager.open_job_artifact(job_id, artifact_id)
+    except KeyError as exc:
+        raise HTTPException(status_code=404, detail="Artifact not found") from exc
+    except (RuntimeError, OSError) as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+
+
+@router.delete("/jobs/{job_id}/artifacts/{artifact_id}")
+async def delete_job_artifact(job_id: str, artifact_id: str):
+    try:
+        return scheduler_manager.delete_job_artifact(job_id, artifact_id)
+    except KeyError as exc:
+        raise HTTPException(status_code=404, detail="Artifact not found") from exc
+    except (RuntimeError, OSError) as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+
+
+@router.get("/instances", response_model=list[InstanceResponse])
+async def list_instances():
+    return scheduler_manager.list_instances()
+
+
+@router.post("/instances", response_model=InstanceResponse)
+async def create_instance(request: InstanceCreateRequest):
+    return scheduler_manager.create_instance(request)
+
+
+@router.get("/instances/{instance_id}", response_model=InstanceResponse)
+async def get_instance(instance_id: str):
+    instance = scheduler_manager.get_instance(instance_id)
+    if not instance:
+        raise HTTPException(status_code=404, detail="Instance not found")
+    return instance
+
+
+@router.patch("/instances/{instance_id}", response_model=InstanceResponse)
+async def update_instance(instance_id: str, request: InstanceUpdateRequest):
+    try:
+        instance = await scheduler_manager.update_instance(instance_id, request)
+    except RuntimeError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+    if not instance:
+        raise HTTPException(status_code=404, detail="Instance not found")
+    return instance
+
+
+@router.delete("/instances/{instance_id}")
+async def delete_instance(instance_id: str):
+    try:
+        deleted = await scheduler_manager.delete_instance(instance_id)
+    except RuntimeError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+    if not deleted:
+        raise HTTPException(status_code=404, detail="Instance not found")
+    return {"status": "ok", "message": "Instance deleted"}
+
+
+@router.post("/instances/{instance_id}/login", response_model=TaskResponse)
+async def login_instance(instance_id: str):
+    try:
+        return await scheduler_manager.create_login_task(instance_id)
+    except KeyError as exc:
+        raise HTTPException(status_code=404, detail="Instance not found") from exc
+    except RuntimeError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+
+
+@router.get("/tasks", response_model=list[TaskResponse])
+async def list_tasks(
+    instance_id: str | None = None,
+    limit: int = Query(default=100, ge=1, le=500),
+):
+    return scheduler_manager.list_tasks(instance_id=instance_id, limit=limit)
+
+
+@router.post("/tasks", response_model=TaskResponse)
+async def create_task(request: TaskCreateRequest):
+    try:
+        return await scheduler_manager.create_task(request)
+    except KeyError as exc:
+        raise HTTPException(status_code=404, detail="Instance not found") from exc
+    except RuntimeError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+
+
+@router.get("/tasks/{task_id}", response_model=TaskResponse)
+async def get_task(task_id: str):
+    task = scheduler_manager.get_task(task_id)
+    if not task:
+        raise HTTPException(status_code=404, detail="Task not found")
+    return task
+
+
+@router.post("/tasks/{task_id}/start", response_model=TaskResponse)
+async def start_task(task_id: str):
+    try:
+        task = await scheduler_manager.start_task(task_id)
+    except RuntimeError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+    if not task:
+        raise HTTPException(status_code=404, detail="Task not found")
+    return task
+
+
+@router.post("/tasks/{task_id}/cancel", response_model=TaskResponse)
+async def cancel_task(task_id: str):
+    task = await scheduler_manager.cancel_task(task_id)
+    if not task:
+        raise HTTPException(status_code=404, detail="Task not found")
+    return task
+
+
+@router.get("/tasks/{task_id}/logs", response_model=list[TaskLogResponse])
+async def list_task_logs(task_id: str, limit: int = Query(default=300, ge=1, le=1000)):
+    if not scheduler_manager.get_task(task_id):
+        raise HTTPException(status_code=404, detail="Task not found")
+    return scheduler_manager.list_logs(task_id, limit=limit)
+
+
+@router.get("/tasks/{task_id}/artifacts", response_model=list[ArtifactResponse])
+async def list_task_artifacts(task_id: str):
+    if not scheduler_manager.get_task(task_id):
+        raise HTTPException(status_code=404, detail="Task not found")
+    return scheduler_manager.list_artifacts(task_id)

api/scheduler/__init__.py

@@ -0,0 +1,5 @@
+# -*- coding: utf-8 -*-
+
+from .manager import SchedulerManager, scheduler_manager
+
+__all__ = ["SchedulerManager", "scheduler_manager"]