解决Web Scraping中HTML结构不一致问题：IBM文档网站案例分析

import httpx  # 异步HTTP客户端
import trio   # 异步运行时
import re     # 正则表达式模块
import pandas as pd # 数据处理库

# 定义请求头，模拟浏览器
headers = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:109.0) Gecko/20100101 Firefox/113.0'
}

async def fetch_table_data(identifier: str):
    """
    异步函数：根据标识符获取表格数据
    """
    async with httpx.AsyncClient(headers=headers, base_url='https://www.ibm.com/docs') as client:
        # 1. 构造原始URL的参数
        initial_params = {
            'topic': f'tables-{identifier}' # 使用传入的标识符
        }
        # 2. 对原始URL发起首次请求，获取包含API线索的HTML
        # 注意：这里假设 identifier 最终会映射到类似 t-accessdateval 这样的 topic
        # 如果原始问题中的 URL 结构是固定的，如 "en/imdm/12.0?topic=t-accessdateval"
        # 那么 initial_params['topic'] 应该直接是 't-accessdateval' 或类似的固定值
        # 这里为了演示通用性，我们用 f'tables-{identifier}'

        # 原始问题中的URL是 "https://www.ibm.com/docs/en/imdm/12.0?topic=t-accessdateval"
        # 对应的path是 'en/imdm/12.0'，topic是 't-accessdateval'
        # 所以这里的 path 和 params 需要根据实际情况调整

        # 假设 identifier 已经是 't-accessdateval' 这样的形式
        initial_response = await client.get('en/imdm/12.0', params={'topic': identifier})

        # 3. 使用正则表达式从响应文本中提取 'oldUrl'
        # 'oldUrl' 通常指向一个内部API端点
        match = re.search(r'"oldUrl":"(.*?)"', initial_response.text)
        if not match:
            print(f"未能找到 {identifier} 的 oldUrl。")
            return pd.DataFrame() # 返回空DataFrame

        old_url_path = match.group(1)

        # 4. 构造API请求的URL路径
        # API路径通常是 'api/v1/content/' 加上提取到的 oldUrl
        api_url_path = "api/v1/content/" + old_url_path

        # 5. 定义API请求的参数
        api_params = {
            'parsebody': 'true', # 确保API返回解析后的内容
            'lang': 'en'         # 指定语言
        }

        # 6. 对API端点发起GET请求
        api_response = await client.get(api_url_path, params=api_params)

        # 7. 使用pandas.read_html直接解析API响应中的表格数据
        # attrs={'class': 'defaultstyle'} 用于精确匹配目标表格
        try:
            # api_response.content 包含的是HTML片段，pandas可以直接解析
            dataframes = pd.read_html(api_response.content, attrs={'class': 'defaultstyle'})
            if dataframes:
                df = dataframes[0]
                # 可以根据需要添加其他信息，例如原始的 identifier
                df.insert(0, "Source_Identifier", identifier)
                return df
            else:
                print(f"未能从 {identifier} 的API响应中找到表格。")
                return pd.DataFrame()
        except ValueError as e:
            print(f"解析 {identifier} 的HTML表格时出错: {e}")
            return pd.DataFrame()

async def main():
    # 假设 'identifiers.csv' 包含一列名为 'Identifier' 的数据
    # 例如：Identifier
    #      t-accessdateval
    #      t-another-table
    df_identifiers = pd.read_csv('identifiers.csv')

    all_dfs = []

    # 遍历所有标识符，并发地获取数据
    async with trio.TaskGroup() as tg:
        for index, row in df_identifiers.iterrows():
            identifier = row['Identifier']
            tg.start_soon(lambda id=identifier: all_dfs.append(trio.run(fetch_table_data, id))) # 包装为同步调用以添加到列表
            # 实际上，trio.run(fetch_table_data, id) 是一个阻塞调用，这里需要调整为异步收集结果
            # 正确的异步收集方式如下：
            # task_result = await fetch_table_data(identifier)
            # all_dfs.append(task_result)
            # 或者更优的，使用 tg.start_soon 来并行运行任务，并收集结果

            # 由于 trio.run 不能在另一个 trio.run 内部调用，我们需要调整收集结果的方式
            # 简单起见，这里先演示串行，如果需要并行，可以构建一个列表的 awaitables
            # 或者让 fetch_table_data 返回一个 Future/Deferred，然后在 TaskGroup 中等待
            # 对于本教程，我们先采用一个简化的并行/串行混合方式，或者直接在 main 中串行调用

            # 更直接的并行收集方式：
            # tasks = [tg.start_soon(fetch_table_data, identifier) for identifier in df_identifiers['Identifier']]
            # 然后需要一种机制来收集这些任务的结果。
            # 这里为了教程的简洁性，先展示一个可以运行的串行/伪并行结构，
            # 真正的并行收集需要更复杂的 TaskGroup 模式，例如使用 trio.Queue 或共享列表加锁。

            # 为了避免复杂性，我们在这里直接串行调用，或者使用一个简单的异步列表收集
            df = await fetch_table_data(identifier)
            if not df.empty:
                all_dfs.append(df)

    if all_dfs:
        combined_df = pd.concat(all_dfs, ignore_index=True)
        print(combined_df)
        combined_df.to_csv('combined_table_data_api.csv', index=False)
    else:
        print("未获取到任何数据。")

if __name__ == "__main__":
    trio.run(main)