Bilibili Comics Sign-in and Voucher Redemption

This article is synchronized and updated to xLog by Mix Space
For the best browsing experience, it is recommended to visit the original link
https://www.do1e.cn/posts/code/bilibili-manga

Preliminary Notes#

Since I have set up a blog, I feel I can post some simple codes I have written here. They are all quite simple, so I don't plan to create a separate repository for each on Github.
Here I will first update a "Bilibili Manga Sign-in and Coupon Exchange", and I may write more in the future when I have time.

Motivation#

By signing in to Bilibili Manga every day, you can earn a certain number of points, which can be exchanged for coupons to purchase manga.
However, the best offer of 100 points is only available in a limited quantity at midnight every day.

To achieve this, I wrote two plugins based on NoneBot2 and bilibili-api, referring to SocialSisterYi/bilibili-API-collect, one for daily sign-in to earn points and the other for grabbing coupons at midnight. They have proven to be very stable.

Using NoneBot2 is relatively simple, and I have developed some plugins on it, which I can share when I have time.

The implemented effect is as follows:

~~It seems the success rate is so high that there's no need for notifications~~
Just in case, I will still send a notification.

Code#

get_sessdata.py#

Used to obtain and refresh login information

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']

Here, cookie_path is a JSON file containing the following 5 keys:

sessdata
buvid3
bili_jct
ac_time_value
dedeuserid

You can log in using the browser's InPrivate mode to obtain these from the cookies. For details, see the bilibili-api documentation.

checkin.py#

A simple NoneBot2 plugin for manual execution by sending /漫画签到 to the QQ bot. It will also run at a specified time every day and provide feedback on the results.
I haven't developed it into a general form for all QQ bot friends to use, so it only supports sending to the QQ bot administrator.
Of course, if you want to use it, you can consider changing it to run directly with cron and find another way to send messages or not send them.

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) # Specify that only administrators can use it

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',
    }
    # Here I have omitted my 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 Manga sign-in successful'
            logger.success(msg)
        else:
            msg = 'Bilibili Manga sign-in failed, ' + retdata['msg']
            logger.error(msg)
    except Exception as e:
        msg = 'Bilibili Manga sign-in failed, ' + 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)

# Here set the time for daily sign-in
@scheduler.scheduled_job('cron', hour=14, minute=57, id='manga_checkin')
async def handle_checkin_scheduler():
    msg = await checkin_func()
    bot = get_bot()
    # Scheduled messages are only sent to administrators
    await bot.send_private_msg(user_id=next(iter(bot.config.superusers)), message=msg)

exchange.py#

A NoneBot2 plugin for exchanging at midnight every day. In fact, it starts running at 23:59 for preprocessing, then checks the system time and exchanges precisely at midnight. Therefore, please ensure the system time is accurate and synchronize it regularly:

timedatectl set-ntp true

The logic of NoneBot2 is similar to the sign-in plugin above. Check the exchange_func function to understand the exchange logic. The code is as follows:

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) # Specify that only administrators can use it


async def exchange_func():
    point_of_target = 100 # Only exchange for coupons worth 100 points
    type_of_target = 7    # Only exchange for manga coupons to avoid other types of products
    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',
    }
    # Here I have omitted my 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)

    # Get remaining points; exit if less than target points
    point, retry = None, 10
    while point is None and retry > 0:
        point = get_point(session)
        retry -= 1
    if point is None:
        msg = 'Failed to get remaining points'
        logger.error(msg)
        return msg
    point = int(point)
    logger.info(f'Current remaining points: {point}')
    if point < point_of_target:
        msg = f'Remaining {point}, insufficient points'
        logger.error(msg)
        return msg

    # Get target product 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 = 'Failed to get target product ID'
        logger.error(msg)
        return msg
    logger.info(f'Target product ID: {product_id}')

    # Exchange
    retry, success_times = 100, 0
    while retry > 0:
        if time.strftime("%H", time.localtime()) == '23':
            while time.strftime("%H", time.localtime()) != '00':
                pass # Wait until midnight
        resp = exchange(session, product_id)
        if resp is None:
            retry -= 1
            continue
        if resp['code'] == 2:
            msg = 'Exchange failed, ' + resp['msg']
            break
        if resp['code'] == 1 and resp['msg'] == 'Insufficient points':
            msg = 'Exchange failed, ' + resp['msg']
            break
        if resp['code'] != 0:
            msg = 'Exchange failed, ' + resp['msg']
            retry -= 1
            continue
        else:
            success_times += 1
            msg = f'Exchange successful {success_times} times'
        time.sleep(0.05)
    if success_times > 0:
        msg = f'Exchange successful {success_times} 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)