应对动态网页抓取挑战：通过内部API获取稳定HTML数据

pip install httpx trio pandas

import httpx
import trio
import re
import pandas as pd

# 模拟浏览器User-Agent，有助于避免被网站识别为爬虫
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，获取包含API路径的初始HTML
        initial_url_path = f'en/imdm/12.0'
        params_initial = {'topic': f'tables-{identifier}'} # 使用动态标识符构建参数

        try:
            r_initial = await client.get(initial_url_path, params=params_initial)
            r_initial.raise_for_status() # 检查HTTP状态码

            # 2. 从初始响应中提取实际内容API的路径
            # 注意：这里假设"oldUrl"模式是稳定的。实际应用中可能需要更复杂的解析
            match = re.search(r'"oldUrl":"(.*?)"', r_initial.text)
            if not match:
                print(f"警告: 未在 {identifier} 的初始响应中找到 'oldUrl'。跳过。")
                return pd.DataFrame() # 返回空DataFrame

            relative_api_path = match.group(1)

            # 3. 构建完整的API URL
            # 例如：api/v1/content/en/imdm/12.0?topic=t-accessdateval
            api_endpoint_path = "api/v1/content/" + relative_api_path

            # 4. 请求内部API获取真实内容
            params_api = {
                'parsebody': 'true', # 可能指示API返回完整HTML体
                'lang': 'en'         # 指定语言
            }
            r_api = await client.get(api_endpoint_path, params=params_api)
            r_api.raise_for_status() # 检查HTTP状态码

            # 5. 使用pandas直接解析API响应中的HTML表格
            # 寻找class为'defaultstyle'的表格
            df_list = pd.read_html(r_api.content, attrs={'class': 'defaultstyle'})
            if not df_list:
                print(f"警告: 未在 {identifier} 的API响应中找到指定表格。")
                return pd.DataFrame()

            df = df_list[0]

            # 假设我们需要从原始页面的结构中获取 Table 名称和 Description
            # 这部分可能需要重新评估，因为API响应可能直接包含这些信息或需要从其他地方获取
            # 对于本示例，我们简化为只获取表格数据

            # 如果需要 Table 名称和 Description，可能需要从 API 响应的其他部分解析
            # 或者在原始请求的BeautifulSoup解析中获取，但前提是原始请求返回的是预期HTML
            # 鉴于本教程主要解决HTML不一致问题，我们聚焦于稳定获取表格

            # 示例中原始代码有：
            # Table = soup.find("h1", class_="topictitle1").get_text(strip=True).strip()
            # description = soup.find('p', class_='shortdesc').get_text(strip=True)
            # 我们可以尝试从r_api.text中用BeautifulSoup解析这些信息
            soup_api = BeautifulSoup(r_api.text, "html.parser")
            table_name_element = soup_api.find("h1", class_="topictitle1")
            description_element = soup_api.find('p', class_='shortdesc')

            table_name = table_name_element.get_text(strip=True) if table_name_element else ""
            description = description_element.get_text(strip=True) if description_element else ""

            # 将这些信息添加到DataFrame中
            df.insert(0, "Table", table_name)
            df["Description"] = description

            # 原始代码中的 feature_list 处理，这里简化，如果需要可以进一步从soup_api中提取
            # div_element = soup_api.find('div', class_='p')
            # if div_element:
            #     a_elements = div_element.find_all('a')
            #     feature_list = [a.get_text(strip=True) for a in a_elements]
            #     for i, feature in enumerate(feature_list):
            #         df[f"Feature_{i+1}"] = feature

            return df

        except httpx.HTTPStatusError as e:
            print(f"HTTP错误获取 {identifier}: {e}")
            return pd.DataFrame()
        except httpx.RequestError as e:
            print(f"请求错误获取 {identifier}: {e}")
            return pd.DataFrame()
        except Exception as e:
            print(f"处理 {identifier} 时发生未知错误: {e}")
            return pd.DataFrame()

async def main():
    # 假设 'identifiers.csv' 包含一个名为 'Identifier' 的列
    # 例如：identifiers.csv 内容
    # Identifier
    # t-accessdateval
    # another_table_id
    df_identifiers = pd.read_csv('identifiers.csv')

    tasks = []
    for index, row in df_identifiers.iterrows():
        identifier = row['Identifier']
        tasks.append(fetch_table_data(identifier))

    results = await trio.gather(*tasks)

    # 合并所有结果
    df_combined = pd.concat(results, ignore_index=True)

    print(df_combined)
    # 保存到CSV
    df_combined.to_csv('combined_table_data_api.csv', index=False)

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