Việc thu thập dữ liệu từ các trang web đọc truyện yêu cầu sự kết hợp giữa kỹ thuật xử lý HTTP request, phân tích mã nguồn HTML và tối ưu hóa tốc độ tải xuống. Trong hướng dẫn này, chúng ta sẽ xây dựng một công cụ bằng Python sử dụng thư viện aiohttp để tải truyện bất đồng bộ, giúp tăng hiệu suất đáng kể so với phương pháp tuần tự thông thường.

1. Khởi tạo phiên đăng nhập

Để truy cập vào các nội dung hạn chế hoặc để đồng bộ cài đặt cá nhân, bước đầu tiên là thực hiện xác thực với hệ thống. Chúng ta sẽ sử dụng requests để gửi dữ liệu đăng nhập và nhận JSON phản hồi.

import requests

def perform_login(user, pwd):
    login_url = "https://passport.17k.com/ck/user/login"
    payload = {
        "loginName": user,
        "password": pwd
    }
    response = requests.post(login_url, data=payload).json()
    
    if response.get('data', {}).get('nickname'):
        print(f"Xác thực thành công: {response['data']['nickname']}")
        return True
    return False

2. Lọc danh sách truyện theo điều kiện

Tiếp theo, chúng ta cần tìm kiếm các tác phẩm phù hợp (ví dụ: đã hoàn thành và miễn phí). Việc gửi các tham số filter giúp server trả về đúng danh mục mong muốn trước khi chúng ta bóc tách URL.

import re

def fetch_novel_list():
    base_url = 'https://www.17k.com/all/book/2_0_0_0_3_0_1_0_1.html'
    html_content = requests.get(base_url).text
    novel_map = {}
    
    # Sử dụng Regex để trích xuất ID và tên truyện
    patterns = re.findall('<a target="_blank" href="//www.17k.com/book/(.*?)</a>', html_content)
    for item in patterns:
        parts = item.split('" >')
        if len(parts) == 2:
            novel_id, title = parts[0], parts[1]
            novel_map[title] = novel_id
            
    return novel_map

def apply_search_filters():
    filter_configs = {
        "Completed": "https://static.17k.com/js/mod/WXShare.js?v=0911",
        "Free": "https://static.17k.com/js/mod/WXShare.js?v=0911"
    }
    for label, api_url in filter_configs.items():
        resp = requests.get(api_url, params={"v": "0911"}).text
        if "成功" in resp:
            print(f"Kích hoạt bộ lọc {label} thành công.")
    return fetch_novel_list()

3. Trích xuất danh sách chương

Sau khi có ID truyện, chúng ta cần truy cập vào trang chi tiết để lấy link của tất cả các chương hiện có.

def get_chapter_manifest(book_id, headers):
    index_url = f"https://www.17k.com/list/{book_id}.html"
    res = requests.get(index_url, headers=headers)
    res.encoding = 'utf-8'
    
    titles = re.findall('title="(.*?)&', res.text)
    links = re.findall('href="/chapter(.*?)"', res.text)
    
    return dict(zip(titles, links))

4. Tải dữ liệu bất đồng bộ với Asyncio

Để tối ưu tốc độ, thay vì tải từng chương một, chúng ta sử dụng aiohttp để thực hiện nhiều request cùng lúc.

import asyncio
    import aiohttp
    import os

async def save_chapter(session, chapter_name, path_suffix, storage_path, headers):
    target_url = f"https://www.17k.com/chapter{path_suffix}"
    try:
        async with session.get(target_url, headers=headers) as response:
            content = await response.text()
            # Bóc tách nội dung chính trong thẻ div
            body_match = re.findall(r'<div class="p">(.*?)</div>', content, re.DOTALL)
            
            if body_match:
                clean_text = "\n".join(re.findall('<p>(.*?)</p>', body_match[0].strip()))
                file_name = os.path.join(storage_path, f"{chapter_name}.txt")
                
                with open(file_name, "w", encoding="utf-8") as f:
                    chapter_title = re.findall('<h1>(.*?)</h1>', content)[0]
                    f.write(chapter_title + "\n\n")
                    f.write(clean_text)
    except Exception as e:
        print(f"Lỗi khi tải chương {chapter_name}: {e}")

async def batch_download(chapters_data, storage_path, headers):
    async with aiohttp.ClientSession() as session:
        queue = []
        for name, link in chapters_data.items():
            task = asyncio.create_task(save_chapter(session, name, link, storage_path, headers))
            queue.append(task)
        await asyncio.gather(*queue)

5. Quy trình điều phối chính

Hàm điều phối sẽ kết nối tất cả các thành phần: từ tạo thư mục, đăng nhập đến việc khởi chạy vòng lặp sự kiện của Asyncio.

def run_crawler():
    base_dir = "Crawler_Output"
    if not os.path.exists(base_dir):
        os.makedirs(base_dir)

    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/118.0.0.0 Safari/537.36"
    }

    if perform_login("username_example", "password_example"):
        books = apply_search_filters()
        for title, b_id in books.items():
            book_path = os.path.join(base_dir, title)
            if not os.path.exists(book_path):
                os.makedirs(book_path)
            
            # Lấy danh sách chương
            chapters = get_chapter_manifest(b_id, headers)
            if chapters:
                print(f"Bắt đầu tải: {title}")
                asyncio.run(batch_download(chapters, book_path, headers))

if __name__ == '__main__':
    run_crawler()