How to Schedule MediaCrawler Tasks

MediaCrawler uses a lightweight, in-process scheduler built on Python's asyncio library through the ExpiringLocalCache class, enabling you to run recurring background tasks like crawling jobs at fixed intervals without external dependencies.

MediaCrawler is an open-source framework for scraping content from platforms like Zhihu, Xiaohongshu, and Weibo. While the repository does not ship with a dedicated cron service, it implements a reusable asynchronous scheduling pattern in its cache layer. This guide explains how to leverage the existing asyncio-based mechanism in cache/local_cache.py to schedule MediaCrawler tasks.

The Asyncio-Based Scheduling Architecture

The scheduling mechanism centers on the ExpiringLocalCache class located in cache/local_cache.py. This class provides a generic pattern for running background coroutines that can be adapted for any recurring work, including crawl jobs.

The Core Mechanism in ExpiringLocalCache

The scheduler initializes inside the cache constructor via self._schedule_clear(), which creates a persistent background task:

# From cache/local_cache.py

def _schedule_clear(self):
    try:
        loop = asyncio.get_event_loop()
    except RuntimeError:
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
    
    self._cron_task = loop.create_task(self._start_clear_cron())

The _start_clear_cron() method runs an infinite loop that executes work and sleeps for a specified interval:

async def _start_clear_cron(self):
    while True:
        self._clear()
        await asyncio.sleep(self._cron_interval)

By default, _clear() removes expired cache entries, but you can replace this logic with any async function to create a custom scheduler.

Handling the Event Loop

The scheduler ensures a running event loop even when instantiated outside an async context. If asyncio.get_event_loop() raises a RuntimeError, the code creates a new loop via asyncio.new_event_loop() and sets it as the current loop. This guarantees that loop.create_task() can always spawn the background coroutine.

Implementing a Periodic Crawl Job

To schedule recurring crawls, wrap your crawling logic in a class that replicates the pattern from ExpiringLocalCache. The following example shows how to run a crawl every 5 minutes:

import asyncio
from cache.local_cache import ExpiringLocalCache

async def crawl_job():
    """Your media platform crawling logic goes here."""
    print("Executing scheduled crawl...")
    # await fetch_zhihu_posts()

    # await process_and_store()

class PeriodicCrawler:
    def __init__(self, interval: int = 300):
        self._interval = interval  # seconds

        self._task: asyncio.Task | None = None
        self._start_scheduler()
    
    def _start_scheduler(self):
        try:
            loop = asyncio.get_event_loop()
        except RuntimeError:
            loop = asyncio.new_event_loop()
            asyncio.set_event_loop(loop)
        
        self._task = loop.create_task(self._run())
    
    async def _run(self):
        while True:
            await crawl_job()
            await asyncio.sleep(self._interval)
    
    def stop(self):
        if self._task:
            self._task.cancel()

# Initialize from main.py or your entry point

if __name__ == "__main__":
    # Keep the cache for temporary data during crawling

    cache = ExpiringLocalCache(cron_interval=60)  # Clean cache every minute

    
    # Start the crawler scheduler

    crawler = PeriodicCrawler(interval=300)     # Crawl every 5 minutes

    
    try:
        asyncio.get_event_loop().run_forever()
    finally:
        crawler.stop()

Integration Points in the Codebase

Initialize your scheduler in main/main.py, which serves as the application entry point and already creates the ExpiringLocalCache instance and launches the FastAPI server. For service-oriented architectures, hook your periodic task into api/services/crawler_manager.py, where crawler logic is exposed as manageable services.

When to Use External Schedulers

While the built-in pattern works for simple, in-process scheduling, certain scenarios require external tools:

• Celery: For distributed workers or tasks that must survive process restarts. The repository's .gitignore already excludes celerybeat.pid and celerybeat-schedule, indicating Celery integration is a common extension.
• APScheduler: For complex scheduling requirements like cron expressions or one-off jobs at specific dates.
• Asyncio Primitives: For simple delays, use loop.call_at() or loop.call_later() to schedule single executions without the infinite loop pattern.

The abstract base class in cache/abs_cache.py provides a foundation for implementing custom cache objects that also run scheduled jobs, should you need to extend the scheduling behavior while maintaining the existing architecture.

Summary

• MediaCrawler's scheduling mechanism lives in cache/local_cache.py within the ExpiringLocalCache class.
• The scheduler uses asyncio.get_event_loop() and create_task() to run background coroutines at fixed intervals.
• You can reuse this pattern by creating a class that spawns an infinite loop of asyncio.sleep() calls interleaved with your crawling logic.
• Initialize schedulers in main/main.py alongside the existing cache and server setup.
• For production-grade distributed scheduling, integrate Celery or APScheduler rather than using the in-process pattern.

Frequently Asked Questions

Does MediaCrawler have a built-in task scheduler?

Yes, MediaCrawler includes a lightweight, pure-Python scheduler implemented in the ExpiringLocalCache class. It uses Python's asyncio library to run background tasks at fixed intervals, making it suitable for simple recurring crawl jobs without adding external dependencies.

How do I run a crawl every 10 minutes in MediaCrawler?

Create a wrapper class that follows the pattern in cache/local_cache.py. Initialize it with an interval of 600 seconds (10 minutes), and have it call your crawling logic in an infinite loop using await asyncio.sleep(600). Start this scheduler from your entry point (such as main/main.py) before running the event loop.

Can I use Celery with MediaCrawler?

Yes. While the built-in scheduler handles in-process tasks, the repository's .gitignore references Celery files, suggesting the codebase is designed to work with external task queues. For distributed crawling or tasks requiring persistence across restarts, deploy Celery workers and trigger your crawling functions from Celery tasks rather than using the internal asyncio scheduler.

Where should I initialize the scheduler in the codebase?

The natural location is main/main.py, which already initializes the ExpiringLocalCache and starts the FastAPI server. You can instantiate your custom scheduler class here, ensuring it runs alongside the existing application infrastructure. For modular architectures, consider integrating with api/services/crawler_manager.py, which manages crawler service lifecycles.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}