Write a crawler to download nice wallpapers from official accounts

Foreword

Many years ago when I was still in university, I once wrote a similar article, though at the time it involved scraping good-looking wallpapers from a game's official website.

Recently, my WeChat official account keeps recommending various wallpapers to me, many of which are quite beautiful. However, saving them one by one is too troublesome, so I decided to write a crawler (爬虫) to download them automatically.

Features of This Web Crawler

Let me briefly list the key features involved in this project, though not all of them will be covered in this article—the main focus will be on the web crawler portion.

If anyone is interested in the other features, I’ll share them later.

• Retrieve all articles from a specified WeChat public account.
• Download wallpapers in the articles that match specific rules.
• Filter out irrelevant images, such as follow-us icons.
• Data persistence (using an async ORM and a lightweight NoSQL).
• Image analysis (size information, perceptual hashing, file MD5).
• All processes display progress bars for a user-friendly experience.

Web Crawler Related Articles

Over the past few years, I have written quite a few articles related to web crawlers.

• Writing a Crawler to Automatically Download Attractive Wallpapers from a Game's Official Website
• Selenium Crawler Practice: Capturing Images from Web Pages
• Selenium Crawler Practice (Pitfall Records): AJAX Request Packet Capture and Browser Exit
• Crawler Notes: Improving Data Collection Efficiency! Using Proxy Pools and Thread Pools
• Summary of C# Crawler Development
• Running a Crawler on Your Phone! Initial Exploration of the Flutter Crawler Framework

Project Structure

We still use the pdm tool for dependency management.

The dependencies used in this project are as follows:

dependencies = [
    "requests>=2.32.3",
    "bs4>=0.0.2",
    "loguru>=0.7.3",
    "tqdm>=4.67.1",
    "tinydb>=4.8.2",
    "pony>=0.7.19",
    "tortoise-orm[aiosqlite]>=0.23.0",
    "orjson>=3.10.14",
    "aerich[toml]>=0.8.1",
    "pillow>=11.1.0",
    "imagehash>=4.3.1",
]

There is also a dev dependency for observing the database (we trialed a lightweight NoSQL database, which didn't have a visual method).

[dependency-groups]
dev = [
    "jupyterlab>=4.3.4",
]

Data Persistence

For projects like this, I always try out different data persistence solutions each time.

For relational databases, I previously used the peewee ORM.

The main issue I found later was its lack of support for automatic migrations (maybe it supports it now, but I used it years ago).

Everything else was decent, workable enough.

This time, I didn't implement persistence initially, but a few unexpected shutdowns led to progress loss, and having to write numerous matching rules was truly a hassle.

So I ended up refactoring everything completely.

I tried tinydb (a single-file document NoSQL), pony (a relational ORM), and tortoise-orm one after another.

In the end, I chose tortoise-orm because its syntax is very similar to Django ORM—I just didn't want to step out of my comfort zone.

Model Definition

from tortoise.models import Model
from tortoise import fields


class Article(Model):
    id = fields.IntField(primary_key=True)
    raw_id = fields.TextField()
    title = fields.TextField()
    url = fields.TextField()
    created_at = fields.DatetimeField()
    updated_at = fields.DatetimeField()
    html = fields.TextField()
    raw_json = fields.JSONField()

    def __str__(self):
        return self.title


class Image(Model):
    id = fields.IntField(primary_key=True)
    article = fields.ForeignKeyField('models.Article', related_name='images')
    url = fields.TextField()
    is_downloaded = fields.BooleanField(default=False)
    downloaded_at = fields.DatetimeField(null=True)
    local_file = fields.TextField(null=True)
    size = fields.IntField(null=True, description='unit: bytes')
    width = fields.IntField(null=True)
    height = fields.IntField(null=True)
    image_hash = fields.TextField(null=True)
    md5_hash = fields.TextField(null=True)

    def __str__(self):
        return self.url

These two models are sufficient to meet all the needs of this project. They can even further enable subsequent features, such as: similar image recognition, image classification, etc.

Get All Articles from a Specific WeChat Official Account

This method requires having a WeChat Official Account (公众号).

The article list is obtained through the "hyperlink" feature added within the Official Account.

For specific instructions, please refer to the reference materials.

Prerequisites

Here I'll only mention a few key points. After entering the hyperlink menu, press F12 to capture network packets.

Focus on the /cgi-bin/appmsg endpoint. You need to extract the following from it:

• Cookie
• token
• fakeid - Official Account ID, encoded in Base64

The first two change with each login. You might consider using Selenium with a local proxy to automate packet capture and update them automatically. For details, refer to my previous article: Selenium Scraping Practice (Pitfall Records) - Ajax Request Packet Capture & Browser Exit

Code Implementation

I have encapsulated the operations into a class

class ArticleCrawler:
    def __init__(self):
        self.url = "接口地址，根据抓包地址来"
        self.cookie = ""
        self.headers = {
            "Cookie": self.cookie,
            "User-Agent": "填写合适的UA",
        }
        self.payload_data = {} # 根据实际抓包拿到的数据来
        self.session = requests.Session()
        self.session.headers.update(self.headers)

    def fetch_html(self, url):
        """获取文章 HTML"""
        try:
            response = self.session.get(url, timeout=10)
            response.raise_for_status()
            return response.text
        except Exception as e:
            logger.error(f"Failed to fetch HTML for {url}: {e}")
            return None

    @property
    def total_count(self):
        """获取文章总数"""
        content_json = self.session.get(self.url, params=self.payload_data).json()
        try:
            count = int(content_json["app_msg_cnt"])
            return count
        except Exception as e:
            logger.error(e)
            logger.warning(f'response json: {content_json}')

        return None

    async def crawl_list(self, count, per_page=5):
        """获取文章列表并存入数据库"""
        logger.info(f'正在获取文章列表，total count: {count}')

        created_articles = []

        page = int(math.ceil(count / per_page))
        for i in tqdm(range(page), ncols=100, desc="获取文章列表"):
            payload = self.payload_data.copy()
            payload["begin"] = str(i * per_page)
            resp_json = self.session.get(self.url, params=payload).json()
            articles = resp_json["app_msg_list"]

            # 存入
            for item in articles:
                # 检查是否已经存在，避免重复插入
                if await Article.filter(raw_id=item['aid']).exists():
                    continue

                created_item = await Article.create(
                    raw_id=item['aid'],
                    title=item['title'],
                    url=item['link'],
                    created_at=datetime.fromtimestamp(item["create_time"]),
                    updated_at=datetime.fromtimestamp(item["update_time"]),
                    html='',
                    raw_json=item,
                )
                created_articles.append(created_item)

            time.sleep(random.uniform(3, 6))

        logger.info(f'created articles: {len(created_articles)}')

    async def crawl_all_list(self):
        return self.crawl_list(self.total_count)

    async def crawl_articles(self, fake=False):
        # 这里根据实际情况，筛选出壁纸文章
        qs = (
            Article.filter(title__icontains='壁纸')
            .filter(Q(html='') | Q(html__isnull=True))
        )

        count = await qs.count()

        logger.info(f'符合条件的没有HTML的文章数量: {count}')

        if fake: return

        with tqdm(
                total=count,
                ncols=100,
                desc="⬇ Downloading articles",
                # 可选颜色 [hex (#00ff00), BLACK, RED, GREEN, YELLOW, BLUE, MAGENTA, CYAN, WHITE]
                colour='green',
                unit="page",
                bar_format='{l_bar}{bar} | {n_fmt}/{total_fmt} pages [{rate_fmt}]',
        ) as pbar:
            async for article in qs:
                article: Article
                article.html = self.fetch_html(article.url)
                await article.save()
                pbar.update(1)
                time.sleep(random.uniform(2, 5))

What does this code do?

It should be said what functionality this class has.

• Get the total number of articles for a specified Official Account (公众号)
• Loop through and retrieve the Official Account's articles by page, including article title, URL, and content
• Store the articles in a database

Code Analysis

The key here is the crawl_list method

The code is actually quite rough, with no error handling, and it accesses the database in each loop, which certainly won't have good performance.

The correct approach would be to first read out the IDs of articles already present in the database, so it wouldn't need to query the database in every loop iteration.

However, since it's a simple crawler, optimization wasn't prioritized.

Then, each loop uses time.sleep(random.uniform(3, 6)) to pause for a random period of time.

Progress Bar

Here, the tqdm library is used to implement the progress bar (the Python ecosystem seems to have simpler progress bar libraries; I've used some before, though most are based on wrappers around tqdm).

Usage of the bar_format parameter: Use bar_format to customize the format of the progress bar, which can display the number of processed files, total file count, processing speed, etc.

• {l_bar} is the left part of the progress bar, including the description and percentage.
• {bar} is the actual progress bar.
• {n_fmt}/{total_fmt} displays the current progress and total count.
• {rate_fmt} displays the processing rate.

Parsing Web Pages

The previous step only downloaded the HTML of the article; we still need to extract the image URLs from the web page.

This is where we need to write a parsing method.

def parse_html(html: str) -> list:
    soup = BeautifulSoup(html, 'html.parser')
    img_elements = soup.select('img.wxw-img')

    images = []

    for img_element in img_elements:
        img_url = img_element['data-src']
        images.append(img_url)

    return images

This simply uses a css selector (CSS 选择器) to extract the images.

Extracting Images

Remember that the model has an Image field, right?

It hasn't been used so far.

This section is for extracting and storing images in the database.

async def extract_images_from_articles():
	# Write queries based on actual conditions
    qs = (
        Article.filter(title__icontains='壁纸')
        .exclude(Q(html='') | Q(html__isnull=True))
    )

    article_count = await qs.count()

    with tqdm(
            total=article_count,
            ncols=100,
            desc="⬇ extract images from articles",
            colour='green',
            unit="article",
            bar_format='{l_bar}{bar} | {n_fmt}/{total_fmt} articles [{rate_fmt}]',
    ) as pbar:
        async for article in qs:
            article: Article
            images = parse_html(article.html)
            for img_url in images:
                if await Image.filter(url=img_url).exists():
                    continue

                await Image.create(
                    article=article,
                    url=img_url,
                )

            pbar.update(1)

    logger.info(f'article count: {article_count}, image count: {await Image.all().count()}')

This method first reads the articles from the database, then extracts images from the articles' HTML, and finally stores all the images in the database.

There is also the issue of repeatedly querying the database in a loop here, but I’m too lazy to optimize it…

Download Images

Similarly, I wrote the ImageCrawler class

class ImageCrawler:
    def __init__(self):
        self.session = requests.Session()
        self.session.headers.update(headers)
        self.images_dir = os.path.join('output', 'images')
        os.makedirs(self.images_dir, exist_ok=True)

    def download_image(self, url):
        img_path = os.path.join(self.images_dir, f'{time.time()}.{extract_image_format_re(url)}')
        img_fullpath = os.path.join(os.getcwd(), img_path)

        try:
            response = self.session.get(url)
            with open(img_fullpath, 'wb') as f:
                f.write(response.content)

            return img_path
        except Exception as e:
            logger.error(e)

        return None

This code is much simpler, as it simply downloads images.

I used a timestamp for the image filename.

However, actually collecting the images is not that simple.

Next, write a method to download images.

async def download_images():
    images = await Image.filter(is_downloaded=False)

    if not images:
        logger.info(f'no images to download')
        return

    c = ImageCrawler()

    with tqdm(
            total=len(images),
            ncols=100,
            desc="⬇ download images",
            colour='green',
            unit="image",
            bar_format='{l_bar}{bar} | {n_fmt}/{total_fmt} images [{rate_fmt}]',
    ) as pbar:
        for image in images:
            image: Image
            img_path = c.download_image(image.url)
            if not img_path:
                continue

            image.is_downloaded = True
            image.local_file = img_path
            await image.save()

            pbar.update(1)
            time.sleep(random.uniform(1, 3))

Filter for undownloaded images, download them, and then update the database to store the local file path.

Running the Program

Finally, we need to string all parts of the program together, like making candied hawthorn on a stick.

This time, we're using asynchronous code, so things will be a bit different.

async def main():
    await init()
    await extract_images_from_articles()
    await download_images()

Finally, call it at the program's entry point:

if __name__ == '__main__':
    run_async(main())

The run_async method is provided by tortoise-orm, which waits for the asynchronous method to complete and reclaims the database connection.

Development Log

I exported the git commit logs and compiled them into this development log table.

How to export such records?

Use the git command to export commit history.

git log --pretty=format:"- %s (%ad)" --date=iso

This uses Markdown list format.

Once generated, you can adjust it into a table based on your needs.

Summary

Web scraping doesn't need much explanation—these straightforward tasks are easy to handle. Not to brag, but I can write scrapers in any language effortlessly, since scraping is often introductory content in many programming courses and not really difficult. The interesting part is pairing each scraper with new tools to experiment, or trying different tech stacks or even devices (like when I previously ran a scraper on my phone). Maybe in the future I could run scrapers on microcontrollers? (Seems impractical though—memory and storage are too limited. A Raspberry Pi would work, but that's essentially a small server.)

PS: I should try writing a scraper in Rust.

Following up on channels, I've continued writing async Python code. Async is indeed great, but unfortunately, Python's async development started relatively late, and the ecosystem isn't fully mature yet.

I really wish Django ORM would support async soon, so it can be seamlessly integrated with channels...