每周必看爬虫

venv

*.db

cookie.txt

cookies/

__pycache__

Weekly Popular API 数据结构说明

## 整体结构

```json

{

  "code": 0,           // 状态码，0表示成功

  "message": "0",      // 状态信息

  "ttl": 1,           // TTL值

  "data": {           // 核心数据对象

    "config": {...},   // 页面配置信息

    "reminder": "...", // 提醒文字

    "list": [...]      // 视频列表数组

  }

}

```

## config 配置对象

包含页面元数据：

- 期次信息：id, number, name, label

- 主题信息：subject, color, cover

- 时间信息：stime, etime

- 分享信息：share_title, share_subtitle

- 媒体ID：media_id

## list 视频列表

包含多个视频对象，每个视频对象包含：

### 基础信息

- `aid`: 视频ID

- `bvid`: B站视频ID

- `title`: 视频标题

- `desc`: 视频描述

- `pic`: 封面图URL

### 分类信息

- `tid`, `tname`: 主分类

- `tidv2`, `tnamev2`: 新版分类

- `pid_v2`, `pid_name_v2`: 父级分类

### 时间信息

- `pubdate`: 发布时间戳

- `ctime`: 创建时间戳

- `duration`: 视频时长(秒)

### 作者信息

- `owner`: 包含mid(用户ID), name, face(头像)

### 统计数据

- `stat`: 包含view(播放), danmaku(弹幕), reply(评论), favorite(收藏), coin(投币), share(分享), like(点赞)等

### 其他属性

- `rights`: 权限信息

- `dimension`: 视频分辨率

- `dynamic`: 动态内容

- `rcmd_reason`: 推荐理由

这个结构清晰地展示了B站"每周必看"页面的完整数据模型，从页面配置到具体的视频内容都有详细的字段描述。

 No newline at end of file

requests==2.32.5

toml==0.10.2

tqdm==4.67.1

from functools import reduce

from hashlib import md5

import urllib.parse

from http.cookiejar import MozillaCookieJar

import time

import requests

headers = {

	'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/142.0.0.0 Safari/537.36',

	'Referer': 'https://www.bilibili.com/'

}

class WbiKeys:

	img_key: str

	sub_key: str

	def __init__(self, img_key: str, sub_key: str):

		self.img_key = img_key

		self.sub_key = sub_key

	_mixinKeyEncTab = [

		46, 47, 18, 2, 53, 8, 23, 32, 15, 50, 10, 31, 58, 3, 45, 35, 27, 43, 5, 49,

		33, 9, 42, 19, 29, 28, 14, 39, 12, 38, 41, 13, 37, 48, 7, 16, 24, 55, 40,

		61, 26, 17, 0, 1, 60, 51, 30, 4, 22, 25, 54, 21, 56, 59, 6, 63, 57, 62, 11,

		36, 20, 34, 44, 52

	]

	def _getMixinKey(self, orig: str):

		'对 imgKey 和 subKey 进行字符顺序打乱编码'

		return reduce(lambda s, i: s + orig[i], WbiKeys._mixinKeyEncTab, '')[:32]

	def sign(self, params: dict, img_key: str, sub_key: str):

		'为请求参数进行 wbi 签名'

		mixin_key = self._getMixinKey(img_key + sub_key)

		curr_time = round(time.time())

		params['wts'] = curr_time

		params = dict(sorted(params.items()))

		# 过滤 value 中的 "!'()*" 字符

		params = {

			k : ''.join(filter(lambda chr: chr not in "!'()*", str(v)))

			for k, v 

			in params.items()

		}

		query = urllib.parse.urlencode(params)                      # 序列化参数

		wbi_sign = md5((query + mixin_key).encode()).hexdigest()    # 计算 w_rid

		params['w_rid'] = wbi_sign

		return params

	@staticmethod

	def getWbiKeys(cookiejar=None):

		"""获取最新的 img_key 和 sub_key

		可选参数:

			cookiejar: requests-compatible 的 cookiejar，用于携带登录态或风控 Cookie

		"""

		resp = requests.get('https://api.bilibili.com/x/web-interface/nav', headers=headers, cookies=cookiejar)

		resp.raise_for_status()

		json_content = resp.json()

		img_url: str = json_content['data']['wbi_img']['img_url']

		sub_url: str = json_content['data']['wbi_img']['sub_url']

		img_key = img_url.rsplit('/', 1)[1].split('.')[0]

		sub_key = sub_url.rsplit('/', 1)[1].split('.')[0]

		return WbiKeys(img_key, sub_key)

def weeklyPopular(number: int, keys: WbiKeys, cookiejar=None):

	"""获取某期每周热门视频列表

	可选参数:

		cookiejar: requests-compatible 的 cookiejar，用于携带登录态或风控 Cookie

	"""

	params = {

		'number': number,

	}

	signed_params = keys.sign(params, keys.img_key, keys.sub_key)

	base_url = 'https://api.bilibili.com/x/web-interface/popular/series/one'

	resp = requests.get(base_url, headers=headers, params=signed_params, cookies=cookiejar)

	resp.raise_for_status()

	return resp.json()

def userInfo(mid: int, keys: WbiKeys, cookiejar=None):

	"""获取用户信息（需要 wbi 签名和 Cookie）

	参数:

		mid: 用户的 mid

		keys: WbiKeys 实例，用于 wbi 签名

		cookiejar: requests-compatible 的 cookiejar（例如 requests.cookies.RequestsCookieJar），可为 None

	返回:

		接口返回的 JSON 解析结果（dict）

	"""

	params = {

		'mid': mid,

	}

	signed_params = keys.sign(params, keys.img_key, keys.sub_key)

	base_url = 'https://api.bilibili.com/x/space/wbi/acc/info'

	resp = requests.get(base_url, headers=headers, params=signed_params, cookies=cookiejar)

	resp.raise_for_status()

	return resp.json()

if __name__ == '__main__':

	cookiejar = MozillaCookieJar('cookie.txt')

	keys = WbiKeys.getWbiKeys(cookiejar=cookiejar)

	print('weeklyPopular(17) =>')

	print(weeklyPopular(17, keys, cookiejar=cookiejar))

	print('\nuserInfo(mid=1087540090) =>')

	print(userInfo(1087540090, keys, cookiejar=None))

#!/usr/bin/env python3

"""抓取第一期每周必看数据并写入 data.db（在一个事务中完成）。

行为：

- 若目标数据库中无表，先执行同目录下的 `init.sql` 创建表。

- 使用 `biliapi.WbiKeys.getWbiKeys()` 与 `biliapi.weeklyPopular(1, keys)` 获取数据。

- 将 `weekly_popular`（期次/时间）写入 `weekly_popular` 表并把完整数据存入 `json_data` 列。

- 对每个视频：去重/插入作者到 `authors`，再插入视频到 `videos`；将未映射字段打包到 `json_data`。

"""

import json

import os

import sqlite3

import sys

from pathlib import Path

from typing import Any, Dict

# 确保能导入同目录下的 biliapi.py

HERE = Path(__file__).resolve().parent

sys.path.insert(0, str(HERE))

import biliapi

import time

import logging

# 模块级全局 WBI keys：只获取一次并复用

WBI_KEYS = None

def init_wbi_keys(cookiejar=None):

    """初始化并返回模块级的 WBI keys；若已初始化则直接返回。

    可选参数:

        cookiejar: requests-compatible 的 cookiejar（传递给 getWbiKeys）

    """

    global WBI_KEYS

    if WBI_KEYS is None:

        WBI_KEYS = biliapi.WbiKeys.getWbiKeys(cookiejar=cookiejar)

    return WBI_KEYS

def ensure_schema(conn: sqlite3.Connection, init_sql_path: Path):

    """如果表不存在，则执行 init.sql 创建表。"""

    cur = conn.cursor()

    # 简单检查 weekly_popular 表是否存在

    cur.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='weekly_popular'")

    if cur.fetchone():

        return

    if not init_sql_path.exists():

        raise FileNotFoundError(f"init.sql not found at {init_sql_path}")

    sql = init_sql_path.read_text(encoding='utf-8')

    conn.executescript(sql)

def get_author_id(conn: sqlite3.Connection, owner: Dict[str, Any]) -> int:

    """根据 owner 对象插入或查询 authors，返回作者表 id。"""

    mid = owner.get('mid')

    name = owner.get('name') or ''

    json_text = json.dumps(owner, ensure_ascii=False)

    cur = conn.cursor()

    # 尝试按 mid 查找（mid 应该存在），若不存在则插入

    cur.execute('SELECT id FROM authors WHERE mid = ?', (mid,))

    row = cur.fetchone()

    if row:

        return row[0]

    # 若 mid 为 None（极少发生），生成一个负数 hash 作为占位 mid，保证 NOT NULL 且尽量唯一

    if mid is None:

        mid = -abs(hash(name)) & 0x7FFFFFFF

    try:

        cur.execute(

            'INSERT INTO authors (name, mid, json_data) VALUES (?, ?, ?)',

            (name, mid, json_text),

        )

        return cur.lastrowid

    except sqlite3.IntegrityError:

        # 可能存在并发/重复插入导致的唯一约束，重新查询一次返回 id

        cur.execute('SELECT id FROM authors WHERE mid = ?', (mid,))

        row = cur.fetchone()

        if row:

            return row[0]

        raise

def map_video_to_row(video: Dict[str, Any], author_id: int, weekly_n: int) -> Dict[str, Any]:

    """把视频对象映射为 videos 表所需列，返回 dict 用于插入。"""

    # 基本列映射

    bvid = video.get('bvid')

    tid = video.get('tid')

    tidv2 = video.get('tidv2')

    pidv2 = video.get('pid_v2') if 'pid_v2' in video else video.get('pidv2')

    ctime = video.get('ctime') or video.get('pubdate')

    duration = video.get('duration')

    stat = video.get('stat', {}) or {}

    stat_view = stat.get('view') or 0

    stat_danmaku = stat.get('danmaku') or 0

    stat_reply = stat.get('reply') or 0

    stat_favorite = stat.get('favorite') or 0

    stat_coin = stat.get('coin') or 0

    stat_share = stat.get('share') or 0

    stat_like = stat.get('like') or 0

    # 准备 json_data：只保留指定的少量字段以缩减存储

    selected_keys = [

        'tname', 'pic', 'videos', 'title', 'desc', 'state', 'dynamic', 'cid', 'dimension',

        'short_link_v2', 'cover43', 'tnamev2', 'pid_name_v2', 'season_type', 'ai_rcmd',

        'ogv_info', 'rcmd_reason'

    ]

    extra = {k: video.get(k) for k in selected_keys if k in video}

    json_data = json.dumps(extra, ensure_ascii=False)

    return {

        'bvid': bvid,

        'tid': tid,

        'tidv2': tidv2,

        'pidv2': pidv2,

        'author_id': author_id,

        'weekly_popular_n': weekly_n,

        'ctime': ctime,

        'duration': duration,

        'stat_view': stat_view,

        'stat_danmaku': stat_danmaku,

        'stat_reply': stat_reply,

        'stat_favorite': stat_favorite,

        'stat_coin': stat_coin,

        'stat_share': stat_share,

        'stat_like': stat_like,

        'json_data': json_data,

    }

def crawl_first_issue(db_path: str = 'data.db', number: int = 1, cookiejar=None):

    """抓取指定期次（默认第1期）并写入数据库（在单个事务中）。"""

    # WBI 签名/网络有时会失败，使用模块级 keys（只获取一次），重试请求3次

    resp = None

    last_exc = None

    try:

        keys = init_wbi_keys(cookiejar=cookiejar)

    except Exception as e:

        raise RuntimeError(f"获取 WbiKeys 失败: {e}")

    for attempt in range(1, 4):

        try:

            resp = biliapi.weeklyPopular(number, keys, cookiejar=cookiejar)

            if resp and resp.get('code') == 0:

                break

            last_exc = RuntimeError(f"API returned code {resp.get('code')} message={resp.get('message')}")

        except Exception as e:

            last_exc = e

        # 简短等待后重试

        time.sleep(1)

    if resp is None or resp.get('code') != 0:

        raise RuntimeError(f"请求每周必看数据失败（重试 3 次），最后错误：{last_exc}")

    data = resp.get('data') or {}

    config = data.get('config') or {}

    n = config.get('number') or config.get('id') or number

    stime = config.get('stime')

    etime = config.get('etime')

    # 仅保存 data 中的 config 到 json_data，缩减存储量

    json_data_week = json.dumps(config, ensure_ascii=False)

    db_file = Path(db_path)

    conn = sqlite3.connect(str(db_file))

    try:

        # 开启外键支持

        conn.execute('PRAGMA foreign_keys = ON')

        # 若没有表则创建

        init_sql_path = HERE / 'init.sql'

        ensure_schema(conn, init_sql_path)

        cur = conn.cursor()

        # 在事务中插入所有数据

        conn.execute('BEGIN')

        # 插入或替换 weekly_popular

        cur.execute(

            'INSERT OR REPLACE INTO weekly_popular (n, stime, etime, json_data) VALUES (?, ?, ?, ?)',

            (n, stime, etime, json_data_week),

        )

        # 遍历视频列表

        videos = data.get('list') or []

        for v in videos:

            owner = v.get('owner') or {}

            author_id = get_author_id(conn, owner)

            # 调试输出：在插入视频前记录外键引用值，便于诊断 FOREIGN KEY 失败

            row = map_video_to_row(v, author_id, n)

            print(f"DEBUG: inserting video bvid={row.get('bvid')} author_id={author_id} weekly_n={n}")

            # 插入视频；使用 INSERT OR REPLACE 保证重复运行不会失败

            cur.execute(

                '''INSERT OR REPLACE INTO videos

                (bvid, tid, tidv2, pidv2, author_id, weekly_popular_n, ctime, duration,

                 stat_view, stat_danmaku, stat_reply, stat_favorite, stat_coin, stat_share, stat_like, json_data)

                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)''',

                (

                    row['bvid'], row['tid'], row['tidv2'], row['pidv2'], row['author_id'], row['weekly_popular_n'],

                    row['ctime'], row['duration'], row['stat_view'], row['stat_danmaku'], row['stat_reply'],

                    row['stat_favorite'], row['stat_coin'], row['stat_share'], row['stat_like'], row['json_data'],

                ),

            )

        conn.commit()

        print(f"成功写入期次 {n}（{len(videos)} 个视频）到数据库 {db_path}")

    except Exception:

        conn.rollback()

        raise

    finally:

        conn.close()

def crawl_loop(db_path: str = 'data.db', start_number: int = None, delay: int = 30, cookiejar=None, max_retries_per_issue: int = 3):

    """自动循环爬取，从数据库记录的下一期或指定 start_number 开始。

    参数:

        db_path: sqlite 数据库路径

        start_number: 可选，若指定则从该期开始；否则从数据库中已存在最大期的下一期开始（若无记录则 1）

        delay: 每次请求间隔（秒）

        cookiejar: 可选的 requests cookiejar，传递给网络请求

        max_retries_per_issue: 每一期的最大重试次数，超过则跳到下一期

    """

    logging.info("启动 crawl_loop db=%s start=%s delay=%ss", db_path, start_number, delay)

    # 计算起始期次

    db_file = Path(db_path)

    if start_number is None:

        # 从数据库中读取当前已爬取的最大期次

        if not db_file.exists():

            current = 1

        else:

            conn = sqlite3.connect(str(db_file))

            try:

                cur = conn.cursor()

                cur.execute("SELECT MAX(n) FROM weekly_popular")

                row = cur.fetchone()

                if row and row[0]:

                    current = int(row[0]) + 1

                else:

                    current = 1

            except Exception:

                current = 1

            finally:

                conn.close()

    else:

        current = int(start_number)

    logging.info("从期次 %s 开始循环爬取", current)

    try:

        while True:

            attempt = 0

            success = False

            last_exc = None

            while attempt < max_retries_per_issue and not success:

                attempt += 1

                try:

                    logging.info("爬取期次 %s（尝试 %s/%s）", current, attempt, max_retries_per_issue)

                    crawl_first_issue(db_path=db_path, number=current, cookiejar=cookiejar)

                    success = True

                except Exception as e:

                    last_exc = e

                    logging.exception("爬取期次 %s 失败（尝试 %s）：%s", current, attempt, e)

                    # 指数退避：在失败后稍等一段时间再重试

                    backoff = min(60, 2 ** attempt)

                    logging.info("等待 %s 秒后重试...", backoff)

                    time.sleep(backoff)

            if not success:

                logging.error("期次 %s 达到最大重试次数 %s，程序终止", current, max_retries_per_issue)

                raise RuntimeError(f"期次 {current} 达到最大重试次数 {max_retries_per_issue}，最后错误: {last_exc}")

            # 成功后推进到下一期

            current += 1

            logging.info("等待 %s 秒准备爬取下一期 %s", delay, current)

            time.sleep(delay)

    except KeyboardInterrupt:

        logging.info("收到中断信号，退出 crawl_loop")

if __name__ == '__main__':

    # 允许通过环境变量或命令行参数指定 db 路径，默认项目根下 data.db

    import argparse

    parser = argparse.ArgumentParser(description='抓取每周必看并写入数据库')

    parser.add_argument('--db', '-d', default=str(HERE.parent / 'data.db'), help='目标 sqlite 数据库路径')

    parser.add_argument('--number', '-n', type=int, default=1, help='期次编号（默认 1）')

    parser.add_argument('--loop', action='store_true', help='启用循环爬取模式')

    parser.add_argument('--delay', type=int, default=30, help='循环模式下每次请求的延迟（秒），默认 30')

    parser.add_argument('--start', type=int, help='循环模式起始期次（若未指定则从数据库最大期次+1 开始）')

    parser.add_argument('--retries', type=int, default=3, help='每期的最大重试次数，超过则跳过到下一期，默认 3')

    args = parser.parse_args()

    # 配置简单日志

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

    if args.loop:

        crawl_loop(db_path=args.db, start_number=args.start, delay=args.delay, max_retries_per_issue=args.retries)

    else:

        crawl_first_issue(db_path=args.db, number=args.number)