如何读取URL的文件和Web用多线程刮擦它们

Question

我正在python中实现网络刮擦脚本，该脚本读取JSON文件，并获取一个网址列表以刮擦每个文件。 该文件包含超过60k行，其中约50k是唯一的（因此首先我要删除重复项）。

要执行此过程，我将有下一个：

import contextlib
from bs4 import BeautifulSoup
import feedparser
import pandas
import requests
import time

BASE_URL = 'https://www.iso.org'

def create_iso_details_json(p_merged_iso_df):
    merged_iso_details_df = p_merged_iso_df.drop_duplicates(subset=['Link']).drop(columns=['TC', 'ICS'], axis=1)
    iso_details_dfs = [parse_iso_details(iso, stage, link) 
                       for iso, stage, link in zip(merged_iso_details_df['Standard and/or project'], merged_iso_details_df['Stage'], merged_iso_details_df['Link']) 
                       if link != '']
    merged_iso_details_df = pandas.concat(iso_details_dfs)
    print('Total rows retrieved: ', len(merged_iso_details_df.index))
    merged_iso_details_df.to_json('iso_details.json', orient="records")
    
def parse_iso_details(p_iso, p_stage, p_url):
    print('URL: ', p_url)
    soup = BeautifulSoup(requests.get(p_url).text, 'html.parser')
    try:
        feed_details_url = BASE_URL + soup.find('section', {'id': 'product-details'}).find('a', {'class': 'ss-icon ss-social-circle text-warning text-sm'})['href']
    except AttributeError:
        print('Could not find feed data for URL: ', p_url)
    print(feed_details_url)
    iso_details_dfs = []
    if feed_details_url is not None:
        iso_details_dfs.append(read_iso_details(feed_details_url, p_iso, p_stage))
    with contextlib.suppress(ValueError):
        return pandas.concat(iso_details_dfs)
    
def read_iso_details(p_feed_details_url, p_iso, p_stage):
    data = {'Standard and/or project': p_iso, 'Stage': p_stage}
    df = pandas.DataFrame(data, index=[0])
    feed = feedparser.parse(p_feed_details_url)
    df['Publication date'] = [entry.published for entry in feed.entries]
    return df

def main():
    start_time = time.time()
    merged_iso_df = pandas.read_json('input_file.json', dtype={"Stage": str})
    create_iso_details_json(merged_iso_df)
    print(f"--- {time.time() - start_time} seconds ---")

if __name__ == "__main__":
    main()

我将pandas DataFrame的结果合并为以后将其写入另一个JSON文件。

现在，这需要花费很多时间，因为该过程每个输入URL的请求提出，并且持续0.5到1秒。

我想通过多线程（不是多处理）来实现此过程，以便处理时间显着减少。

实现这一目标的最佳方法是什么？将输入JSON文件分为多个部分，作为要创建处理的线程数量的数量？如何将每个线程的结果合并到一个编写输出JSON文件中？

先感谢您。

Answer 1

此网站 解释了多读的多读。您可以做的是将URL分为相等的部分并同时运行它们。问题是，您基本上只是将所花费的时间除以所使用的线程数。但据我所知，这是您可以做的最好的事情，而不会过度复杂化。

Answer 2

我终于设法通过@yui在答案中发布的多线程来实现该过程。

真正的问题是将每个线程的结果合并到一个线程中，因此我决定将每个线程结果写入附录模式为CSV的文件。然后，当所有线程完成后，我会读取CSV并将结果写入Requierd JSON文件。

BASE_URL = 'https://www.iso.org'
NUM_THREADS = 4
q = Queue()
INPUT_FILE = 'iso_tc_ics.json'
ISO_DETAILS_CSV_FILE = 'iso_details.csv'
OUTPUT_FILE = 'iso_details.json'

def create_iso_details_json(p_queue):
    iso_details_df = p_queue.get()
    iso_details_dfs = [parse_iso_details(iso, stage, link) 
                       for iso, stage, link in zip(iso_details_df['Standard and/or project'], iso_details_df['Stage'], iso_details_df['Link']) 
                       if link != '']
    iso_details_df = pandas.concat(iso_details_dfs)
    print('Rows retrieved: ', len(iso_details_df.index))
    return iso_details_df
    
def parse_iso_details(p_iso, p_stage, p_url):
    print('URL: ', p_url)
    soup = BeautifulSoup(requests.get(p_url).text, 'html.parser')
    try:
        feed_details_url = BASE_URL + soup.find('section', {'id': 'product-details'}).find('a', {'class': 'ss-icon ss-social-circle text-warning text-sm'})['href']
    except AttributeError:
        print('Could not find feed data for URL: ', p_url)
    print(feed_details_url)
    iso_details_dfs = []
    if feed_details_url is not None:
        iso_details_dfs.append(read_iso_details(feed_details_url, p_iso, p_stage))
    with contextlib.suppress(ValueError):
        return pandas.concat(iso_details_dfs)
    
def read_iso_details(p_feed_details_url, p_iso, p_stage):
    data = {'Standard and/or project': p_iso, 'Stage': p_stage}
    df = pandas.DataFrame(data, index=[0])
    feed = feedparser.parse(p_feed_details_url)
    df['Publication date'] = [entry.published for entry in feed.entries]
    return df

def main():
    global q
    result_df = create_iso_details_json(q)
    with open(ISO_DETAILS_CSV_FILE, 'a') as f:
        result_df.to_csv(f, mode='a', index=False, header=not f.tell(), encoding='ISO-8859-1')
    q.task_done()

def init():
    merged_iso_df = pandas.read_json(INPUT_FILE, dtype={"Stage": str})
    merged_iso_details_df = merged_iso_df.drop_duplicates(subset=['Link']).drop(columns=['TC', 'ICS'], axis=1)
    iso_details_df_chunks = numpy.array_split(merged_iso_details_df, NUM_THREADS)

    for iso_details_df in iso_details_df_chunks:
        q.put(iso_details_df)

    for _ in range(NUM_THREADS):
        worker = Thread(target=main)
        worker.daemon = True
        worker.start()

def end():
    q.join()
    result_iso_details_df = pandas.read_csv(ISO_DETAILS_CSV_FILE, dtype={"Stage": str}, encoding='ISO-8859-1')
    print('Total rows retrieved: ', len(result_iso_details_df.index))
    result_iso_details_df.to_json(OUTPUT_FILE, orient="records")
    with contextlib.suppress(OSError):
        os.remove(ISO_DETAILS_CSV_FILE)

if __name__ == "__main__":
    start_time = time.time()

    init()

    end()
    
    print(f"--- {time.time() - start_time} seconds ---")

Answer 3

我会选择Asyncio和Aiohttp，这是一个完整的示例，说明如何同时执行多个请求并最终获得结果，

import aiohttp
import asyncio

async def geturl(url, session):
    async with session.get(url) as resp:
        if resp.status == 200:
            return (await resp.json())['name']
        else:
            return "ERROR"

async def main():
    urls = [f'https://pokeapi.co/api/v2/pokemon/{i}' for i in range(1,10)]
    async with aiohttp.ClientSession() as session:
        tasks = [geturl(url, session) for url in urls]
        # asyncio.gather will run all the tasks concurrently
        # and return their results once all tasks have returned
        all_results = await asyncio.gather(*tasks)
        print(all_results)

asyncio.run(main())

这将按照方式打印出前10个口袋妖怪名称，您可以根据需要调整