#2023年最新Ajax爬取技术教程

import httpx

async def fetch_api_data(url: str, method: str = "GET", params: dict = None, headers: dict = None, data: dict = None):
    """
    通用异步模拟请求函数
    :param url: 目标接口地址
    :param method: 请求方法（GET 或 POST）
    :param params: GET 请求的查询参数
    :param headers: 完整请求头（建议从 curlconverter 复制过来）
    :param data: POST 请求的表单数据
    """
    async with httpx.AsyncClient(http2=True, follow_redirects=True) as client:
        try:
            resp = await client.request(
                method=method,
                url=url,
                params=params,
                headers=headers,
                data=data,
                timeout=10.0  # 设置超时，防止卡死
            )
            resp.raise_for_status()  # 遇到 4xx/5xx 自动抛出异常
            return resp.json()       # 大多数接口返回 JSON，直接解析
        except httpx.HTTPStatusError as e:
            print(f"请求失败，状态码：{e.response.status_code}")
        except Exception as e:
            print(f"其他错误：{e}")

import execjs
import time

# 补全依赖后的 JS 代码
js_code = """
// 假设这是从 Sources 找到的生成 token 的函数
function getRandomStr() {
    return Math.random().toString(36).substr(2, 9);
}
function getToken(t) {
    return getRandomStr() + t.toString().substr(-6);
}
"""

# 编译 JS
ctx = execjs.compile(js_code)
# 调用函数，传入当前时间戳（单位：秒）
timestamp = int(time.time())
token = ctx.call("getToken", timestamp)
print(f"生成的动态 token：{token}")

import httpx
from parsel import Selector
import asyncio

# ---------------------- 配置参数 ----------------------
UID = "2830678474"          # 目标博主 ID
MAX_PAGE = 2                # 最大爬取页数
HEADERS = {
    "User-Agent": "Mozilla/5.0 (iPhone; CPU iPhone OS 16_6 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/16.6 Mobile/15E148 Safari/604.1",
    "X-Requested-With": "XMLHttpRequest",
    "Referer": f"https://m.weibo.cn/u/{UID}",
}

# ---------------------- 核心函数 ----------------------
async def fetch_weibo_page(uid: str, page: int = 1):
    """爬取单页微博数据"""
    async with httpx.AsyncClient(http2=True, follow_redirects=True) as client:
        try:
            resp = await client.get(
                url="https://m.weibo.cn/api/feed/profile",
                params={"uid": uid, "page": page},
                headers=HEADERS,
                timeout=10.0
            )
            resp.raise_for_status()
            return resp.json()
        except Exception as e:
            print(f"第{page}页爬取失败：{e}")
            return None

def parse_weibo_html(html: str):
    """解析微博 HTML 片段，提取有用信息"""
    selector = Selector(text=html)
    weibo_list = []
    # 遍历每个微博卡片
    for card in selector.css(".card-wrap"):
        # 过滤掉广告卡片（包含 mblog-tag--ad 类名）
        if card.css(".mblog-tag--ad"):
            continue
        weibo_list.append({
            "微博ID": card.css(".card::attr(mid)").get(),
            "发布时间": card.css(".time::text").get(),
            "微博正文": "".join(card.css(".weibo-text::text, .weibo-text a::text").getall()).strip(),
            "点赞数": card.css(".like-count::text").get("0"),
            "评论数": card.css(".comment-count::text").get("0"),
            "转发数": card.css(".repost-count::text").get("0"),
        })
    return weibo_list

async def main():
    """主函数：多页爬取"""
    all_weibos = []
    for page in range(1, MAX_PAGE + 1):
        print(f"正在爬取第{page}页...")
        data = await fetch_weibo_page(UID, page)
        if not data or not data.get("data", {}).get("cards"):
            print(f"第{page}页没有数据，停止爬取")
            break
        # 2023 年 12 月时，接口返回的 cards 中第一个元素是置顶微博（isTop:1），后续才是普通微博
        # 这里把所有卡片中的 HTML 片段拼接起来，统一解析
        full_html = "".join([
            card.get("mblog", {}).get("text", "")
            for card in data["data"]["cards"]
            if card.get("mblog")
        ])
        page_weibos = parse_weibo_html(full_html)
        all_weibos.extend(page_weibos)
        # 加上 3 秒延迟，避免请求过快被封
        await asyncio.sleep(3)

    # 打印最终结果
    print(f"\n爬取完成，共获取 {len(all_weibos)} 条有效微博：")
    for idx, weibo in enumerate(all_weibos, 1):
        print(f"\n【第{idx}条】")
        for key, value in weibo.items():
            print(f"{key}：{value}")

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