Homework2_102301509/crawler.py

import argparse
import time
import logging
import requests
from lxml import etree
from pathlib import Path
from typing import List, Dict
from tqdm import tqdm
import json
import random

logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')

SESSION = requests.Session()
SESSION.headers.update({
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 '
                  '(KHTML, like Gecko) Chrome/120.0 Safari/537.36'
})
# 请求头，可以降低被拦截概率
SESSION.headers.update({
    'Referer': 'https://www.bilibili.com',
    'Accept': 'application/json, text/javascript, */*; q=0.01'
})

SEARCH_URL = 'https://api.bilibili.com/x/web-interface/search/type'
VIEW_URL = 'https://api.bilibili.com/x/web-interface/view'
COMMENT_XML = 'https://comment.bilibili.com/{cid}.xml'


def search_videos(keyword: str, max_videos: int = 360) -> List[Dict]:
    """通过 B 站搜索 API 获取视频列表（综合排序）
    返回的视频项包含 'bvid' 和 'title'
    """
    videos = []
    page = 1
    page_size = 20
    while len(videos) < max_videos:
        params = {
            'keyword': keyword,
            'search_type': 'video',
            'page': page,
            'page_size': page_size
        }
    # 简单重试与抖动，降低 412/403 风险
        resp = None
        for attempt in range(3):
            try:
                resp = SESSION.get(SEARCH_URL, params=params, timeout=10)
                if resp.status_code == 200:
                    break
                logging.warning('搜索重试 %d 次，状态码=%s', attempt, resp.status_code)
            except Exception as e:
                logging.exception('搜索第 %d 次出现异常: %s', attempt, e)
            time.sleep(0.5 + attempt * 0.5)
        if not resp or resp.status_code != 200:
            logging.warning('搜索失败，状态码=%s', getattr(resp, 'status_code', None))
            break
        data = resp.json()
        try:
            vlist = data.get('data', {}).get('result', [])
            if not vlist:
                break
            for item in vlist:
                title = item.get('title', '')
                bvid = item.get('bvid') or item.get('aid')
                if bvid:
                    videos.append({'bvid': bvid, 'title': title})
                if len(videos) >= max_videos:
                    break
        except Exception as e:
            logging.exception('搜索结果失败: %s', e)
            break
        page += 1
        time.sleep(0.5)
    logging.info('关键词「%s」共找到 %d 个视频,目标 %d', keyword, len(videos), max_videos)
    return videos


def get_cid(bvid: str) -> int:
    #通过接口获取
    params = {'bvid': bvid} if isinstance(bvid, str) and bvid.startswith('BV') else {'aid': bvid}
    for attempt in range(3):
        try:
            resp = SESSION.get(VIEW_URL, params=params, timeout=10)
            if resp.status_code != 200:
                logging.warning('获取 CID 失败：%s，状态码=%s', bvid, resp.status_code)
                time.sleep(1 + attempt)
                continue
            data = resp.json()
            return data.get('data', {}).get('cid')
        except Exception as e:
            logging.exception('第 %d 次获取 CID 异常：%s，错误=%s', attempt, bvid, e)
            time.sleep(1 + attempt)
    return None


def fetch_danmaku(cid: int) -> List[str]:
    #请求弹幕并解析出文本列表
    url = COMMENT_XML.format(cid=cid)
    for attempt in range(3):
        try:
            resp = SESSION.get(url, timeout=10)
            if resp.status_code != 200:
                logging.warning('弹幕获取失败：cid=%s 状态码=%s', cid, resp.status_code)
                time.sleep(1 + attempt)
                continue
            xml = resp.content
            root = etree.fromstring(xml)
            texts = [d.text or '' for d in root.findall('d')]
            return texts
        except Exception as e:
            logging.exception('第%d次获取弹幕异常：cid=%s 错误=%s', attempt, cid, e)
            time.sleep(1 + attempt)
    return []


def save_raw_texts(texts: List[str], out_file: Path):
    out_file.parent.mkdir(parents=True, exist_ok=True)
    with out_file.open('w', encoding='utf-8') as f:
        for t in texts:
            f.write(t.replace('\n', ' ') + '\n')


def load_progress(progress_path: Path) -> Dict:
    if not progress_path.exists():
        return {'done': []}
    try:
        with progress_path.open('r', encoding='utf-8') as f:
            return json.load(f)
    except Exception:
        return {'done': []}


def save_progress(progress_path: Path, data: Dict):
    progress_path.parent.mkdir(parents=True, exist_ok=True)
    with progress_path.open('w', encoding='utf-8') as f:
        json.dump(data, f, ensure_ascii=False, indent=2)


def main(keywords: List[str], max_videos_per_kw: int, out_dir: Path, delay: float = 0.5, progress_file: str = '进度.json'):
    out_dir = Path(out_dir)
    progress_path = out_dir / progress_file
    progress = load_progress(progress_path)
    done = set(progress.get('done', []))
    all_texts = []
    for kw in keywords:
        logging.info('开始处理关键词：%s', kw)
        videos = search_videos(kw, max_videos_per_kw)
        for v in tqdm(videos, desc=f'关键词 {kw} 的视频'):
            bvid = v.get('bvid')
            if not bvid:
                continue
            if bvid in done:
                logging.debug('跳过已处理视频：%s', bvid)
                continue
            try:
                cid = get_cid(bvid)
                if not cid:
                    continue
                texts = fetch_danmaku(cid)
                vid_dir = out_dir / '原始'
                vid_dir.mkdir(parents=True, exist_ok=True)
                save_raw_texts(texts, vid_dir / f'{bvid}_{cid}.txt')
                all_texts.extend(texts)
                done.add(bvid)
                #周期写入进度
                if len(done) % 20 == 0:
                    save_progress(progress_path, {'done': list(done)})
                # 随机延迟
                time.sleep(delay + random.random() * 0.3)
            except Exception:
                logging.exception('处理视频失败：%s', bvid)
    #保存聚合的原始文本
        save_raw_texts(all_texts, out_dir / f'原始弹幕_{kw}.txt')

    # 最终写入进度展示
    save_progress(progress_path, {'done': list(done)})

    #无数据时提示
    if not all_texts:
        logging.warning('未采集到任何弹幕——请检查网络、接口变更或反爬（HTTP 412/403）。')
        return

    #最后
    logging.info('抓取完成：共处理 %d 个视频，采集弹幕 %d 行。', len(done), len(all_texts))


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='B站弹幕抓取器')
    parser.add_argument('-k', '--keywords', nargs='+', default=['大语言模型'], help='关键词')
    parser.add_argument('-n', '--max-per-kw', type=int, default=120, help='每个关键词抓取的视频数')
    parser.add_argument('-o', '--out', default='输出', help='输出目录')
    parser.add_argument('--delay', type=float, default=0.5, help='请求间基础延迟')
    parser.add_argument('--progress-file', default='进度.json', help='断点续爬进度文件名（默认 进度.json）')
    parser.add_argument('--cookie', default=None, help='Cookie 字符串')
    parser.add_argument('--proxy', default=None, help='代理URL（例如 http://127.0.0.1:7890 ）')
    args = parser.parse_args()
    # apply optional cookie and proxy
    if args.cookie:
        SESSION.headers.update({'Cookie': args.cookie})
    if args.proxy:
        SESSION.proxies.update({'http': args.proxy, 'https': args.proxy})
    main(args.keywords, args.max_per_kw, Path(args.out))