哔哩哔哩漫畫簽到與福利券兌換

此文由 Mix Space 同步更新至 xLog
為獲得最佳瀏覽體驗，建議訪問原始鏈接
https://www.do1e.cn/posts/code/bilibili-manga

前置說明#

既然都搭建博客了，感覺可以在這裡把寫過的一些雜七雜八的簡單代碼放上來，都比較簡單就不打算在Github上給每一個都單獨建一個倉庫了。
這裡先更新一個 “哔哩哔哩漫畫簽到與福利券兌換”，未來有空還可以再寫一些。

動機#

哔哩哔哩漫畫每天簽到能領取一定的積分，積分可以兌換成可購買漫畫的福利券。
不過最優惠的 100 積分只在每天 0 點發放一定數量。

為此我基於NoneBot2和bilibili-api，並參考SocialSisterYi/bilibili-API-collect寫了兩個插件，分別用於每天簽到領積分和每天 0 點搶購福利券，目前用下來穩定性極高。

NoneBot2用起來還是比較簡單的，我也在上面開發了一些插件，有空也可以分享出來。

實現的效果如下：

~~似乎成功率太高了完全不需要推送~~
以防萬一還是推送一下吧

代碼#

get_sessdata.py#

用於獲取並刷新登錄信息

from bilibili_api import Credential, sync
import json
import logging

def get_sessdata(cookie_path: str) -> str:
    with open(cookie_path, 'r') as f:
        cookie = json.load(f)
    credential = Credential(**cookie)
    if sync(credential.check_refresh()):
        logging.info('Cookie needs to be refreshed.')
        sync(credential.refresh())
        cookie = credential.get_cookies()
        cookie = {k.lower(): v for k, v in cookie.items()}
        with open(cookie_path, 'w') as f:
            json.dump(cookie, f, indent=4)
        logging.info(f'Cookie refreshed.\n\t{cookie}')
    return cookie['sessdata']

其中 cookie_path 是一個 json 文件，包含以下 5 個 key：

sessdata
buvid3
bili_jct
ac_time_value
dedeuserid

可以使用瀏覽器 InPrivate 模式登錄後在 cookie 中獲取，詳見bilibili-api 文檔

checkin.py#

用於簡單的NoneBot2插件，可以在給 QQ 機器人發送/漫畫簽到手動執行，也會在每天指定的時間定時運行並將結果反饋。
我這裡沒有開發成通用的形式，讓所有的 QQ 機器人好友使用，因此僅支持發送給 QQ 機器人管理員。
當然你要用的話也可以考慮改為直接使用cron定時運行，再想別的辦法進行消息推送或者不推送。

from nonebot import get_driver, on_command, get_bot, require
from nonebot.log import logger
from nonebot.adapters.onebot.v11 import Bot, Event
from nonebot.typing import T_State
from nonebot.permission import SUPERUSER
require("nonebot_plugin_apscheduler")
from nonebot_plugin_apscheduler import scheduler
import requests
import time
import traceback

from .get_sessdata import get_sessdata

checkin = on_command("漫畫簽到", priority=5, permission=SUPERUSER) # 指定僅允許管理員使用

async def checkin_func():
    url = "https://manga.bilibili.com/twirp/activity.v1.Activity/ClockIn?platform=android"
    headers = {
        'user-agent': "Dalvik/2.1.0 (Linux; U; Android 14; 23046RP50C Build/UKQ1.230804.001) 6.8.5 os/android model/23046RP50C mobi_app/android_comic build/36608060 channel/xiaomi innerVer/36608060 osVer/14 network/2",
        'accept': 'application/json, text/plain, */*',
        'accept-encoding': 'gzip, deflate, br',
    }
    # 此處隱去我的cookie_path
    headers.update({'cookie': f'SESSDATA={get_sessdata(config.cookie_path)}'})
    logger.info(f'headers: {headers}')

    try:
        resp = requests.post(url, headers=headers)
        resp.raise_for_status()
        retdata = resp.json()
        if retdata['code'] == 0:
            msg = 'Bilibili漫畫簽到成功'
            logger.success(msg)
        else:
            msg = 'Bilibili漫畫簽到失敗，' + retdata['msg']
            logger.error(msg)
    except Exception as e:
        msg = 'Bilibili漫畫簽到失敗，' + str(e)
        logger.error(traceback.format_exc())
    return msg

@checkin.handle()
async def handle_checkin(bot: Bot, event: Event, state: T_State):
    msg = await checkin_func()
    await checkin.finish(msg)

# 此處設置每天簽到的時間
@scheduler.scheduled_job('cron', hour=14, minute=57, id='manga_checkin')
async def handle_checkin_scheduler():
    msg = await checkin_func()
    bot = get_bot()
    # 定時消息僅發送給管理員
    await bot.send_private_msg(user_id=next(iter(bot.config.superusers)), message=msg)

exchange.py#

用於在每天 0 點兌換的NoneBot2插件。實際上在 23 點 59 分開始運行並進行預處理，之後檢測系統時間，于 0 點準時兌換，因此請確保系統時間準確，定時同步時間：

timedatectl set-ntp true

NoneBot2邏輯和上面的簽到插件類似，查看exchange_func函數了解兌換邏輯即可，代碼如下：

from nonebot import get_driver, on_command, get_bot, require
from nonebot.log import logger
from nonebot.adapters.onebot.v11 import Bot, Event
from nonebot.typing import T_State
from nonebot.permission import SUPERUSER
require("nonebot_plugin_apscheduler")
from nonebot_plugin_apscheduler import scheduler
import requests
import time
import traceback

from .get_sessdata import get_sessdata

exchange = on_command("漫畫兌換", priority=5, permission=SUPERUSER) # 指定僅允許管理員使用


async def exchange_func():
    point_of_target = 100 # 僅兌換100積分的福利券
    type_of_target = 7    # 僅兌換漫畫福利券，避免其他類型的商品
    get_point_url = 'https://manga.bilibili.com/twirp/pointshop.v1.Pointshop/GetUserPoint'
    list_url = 'https://manga.bilibili.com/twirp/pointshop.v1.Pointshop/ListProduct'
    exchange_url = 'https://manga.bilibili.com/twirp/pointshop.v1.Pointshop/Exchange'
    headers = {
        'user-agent': "Dalvik/2.1.0 (Linux; U; Android 14; 23046RP50C Build/UKQ1.230804.001) 6.8.5 os/android model/23046RP50C mobi_app/android_comic build/36608060 channel/xiaomi innerVer/36608060 osVer/14 network/2",
        'accept': 'application/json, text/plain, */*',
        'accept-encoding': 'gzip, deflate, br',
    }
    # 此處隱去我的cookie_path
    headers.update({'cookie': f'SESSDATA={get_sessdata(config.cookie_path)}'})
    logger.info(f'headers: {headers}')

    def get_point(session):
        try:
            resp = session.post(get_point_url)
            resp.raise_for_status()
            return resp.json()['data']['point']
        except Exception as e:
            logger.error(traceback.format_exc())
            return None

    def get_target_id(session):
        try:
            resp = session.post(list_url)
            resp.raise_for_status()
            if resp.json()['code'] != 0:
                raise RuntimeError('Failed to get product list.')
            products = resp.json()['data']
            for product in products:
                if product['real_cost'] == point_of_target and product['type'] == type_of_target:
                    return product['id']
            return -1
        except Exception as e:
            logger.error(traceback.format_exc())
        return None

    def exchange(session, product_id):
        try:
            resp = session.post(exchange_url, json={'product_id': product_id, 'product_num': 1, 'point': point_of_target})
            resp.raise_for_status()
            return resp.json()
        except Exception as e:
            logger.error(traceback.format_exc())
            return None

    session = requests.Session()
    session.headers.update(headers)

    # 獲取剩餘點數，小於目標點數則退出
    point, retry = None, 10
    while point is None and retry > 0:
        point = get_point(session)
        retry -= 1
    if point is None:
        msg = '獲取剩餘點數失敗'
        logger.error(msg)
        return msg
    point = int(point)
    logger.info(f'當前剩餘點數: {point}')
    if point < point_of_target:
        msg = f'剩餘{point}，點數不足'
        logger.error(msg)
        return msg

    # 獲取目標商品id
    product_id, retry = None, 10
    while product_id is None and retry > 0:
        product_id = get_target_id(session)
        retry -= 1
    if product_id is None:
        msg = '獲取目標商品id失敗'
        logger.error(msg)
        return msg
    logger.info(f'目標商品id: {product_id}')

    # 兌換
    retry, success_times = 100, 0
    while retry > 0:
        if time.strftime("%H", time.localtime()) == '23':
            while time.strftime("%H", time.localtime()) != '00':
                pass # 等待到0點
        resp = exchange(session, product_id)
        if resp is None:
            retry -= 1
            continue
        if resp['code'] == 2:
            msg = '兌換失敗，' + resp['msg']
            break
        if resp['code'] == 1 and resp['msg'] == '積分不足':
            msg = '兌換失敗，' + resp['msg']
            break
        if resp['code'] != 0:
            msg = '兌換失敗，' + resp['msg']
            retry -= 1
            continue
        else:
            success_times += 1
            msg = f'兌換成功{success_times}次'
        time.sleep(0.05)
    if success_times > 0:
        msg = f'兌換成功{success_times}次'
    logger.info(msg)
    return msg

@exchange.handle()
async def handle_exchange(bot: Bot, event: Event, state: T_State):
    msg = await exchange_func()
    await exchange.finish(msg)

@scheduler.scheduled_job('cron', hour='23', minute='59', id='manga_exchange')
async def handle_exchange_scheduler():
    msg = await exchange_func()
    bot = get_bot()
    await bot.send_private_msg(user_id=next(iter(bot.config.superusers)), message=msg)