哔哩哔哩漫画のサインインと特典券の交換

この文は Mix Space によって xLog に同期更新されています
最適なブラウジング体験を得るために、元のリンクを訪れることをお勧めします
https://www.do1e.cn/posts/code/bilibili-manga

前置きの説明#

ブログを立ち上げたので、ここに書いたいくつかの雑多な簡単なコードを載せてみようと思います。どれも比較的簡単なので、Github でそれぞれに個別のリポジトリを作るつもりはありません。
まずは「哔哩哔哩漫画のチェックインと福利券の交換」を更新します。将来的に時間があれば、さらに書くこともできます。

動機#

哔哩哔哩漫画では、毎日チェックインすることで一定のポイントを獲得でき、そのポイントは漫画を購入するための福利券に交換できます。
ただし、最もお得な 100 ポイントは毎日 0 時に一定数が配布されます。

そのため、NoneBot2 と bilibili-api を基に、SocialSisterYi/bilibili-API-collect を参考にして、毎日チェックインしてポイントを獲得するためのプラグインと、毎日 0 時に福利券を購入するためのプラグインの 2 つを作成しました。現在のところ、非常に高い安定性を持っています。

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 は、以下の 5 つのキーを含む json ファイルです：

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('商品リストの取得に失敗しました。')
            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)