.

bili.ts

@@ -0,0 +1,112 @@
+import * as crypto from 'crypto'
+import axios from 'axios'
+import { getEncodeHeader } from './src/utils/crypto'
+import * as WebSocket from 'ws'
+import { Proto } from './src/danmu/models/proto.model'
+
+// axios 拦截器
+const api = axios.create({
+    baseURL: 'https://live-open.biliapi.com'
+    // baseURL: "http://test-live-open.biliapi.net" //test
+})
+
+// 鉴权加密处理headers，下次请求自动带上
+api.interceptors.request.use((config) => {
+    const headers = getEncodeHeader(config.data, 'Cwgk7TsPkMtKtGO8rY0LndiF', 'bV9HiQ03OQ0Hk00l90N1QWyAFSI8qR')
+    Object.keys(headers).forEach((key) => {
+        config.headers[key] = headers[key]
+    })
+    return config
+})
+
+let gameInfo, websocketInfo, anchorInfo
+async function start() {
+    const { data } = await api.post('/v2/app/start', { app_id: 1700371901242, code: 'CYCPQYCZ0LJ43' })
+    if (data.code === 0) {
+        console.log('开始成功')
+        console.log(data.data)
+        gameInfo = data.data.game_info
+        websocketInfo = data.data.websocket_info
+        anchorInfo = data.data.anchor_info
+    } else {
+        console.log('开始失败')
+        console.log(data)
+    }
+}
+
+async function end(gameId) {
+    const { data } = await api.post('/v2/app/end', {
+        app_id: 1700371901242,
+        game_id: gameId
+    })
+    if (data.code === 0) {
+        console.log('结束成功')
+    } else {
+        console.log('结束失败')
+        console.log(data)
+    }
+}
+
+let heartbeatTimer
+function heartbeat(gameId) {
+    heartbeatTimer = setInterval(async () => {
+        const { data } = await api.post('/v2/app/batchHeartbeat', {
+            game_ids: [gameId]
+        })
+        if (data.code === 0) {
+            console.log('心跳成功')
+        } else {
+            console.log('心跳失败')
+            console.log(data)
+        }
+    }, 1000)
+}
+
+// heartbeat('83dc18da-c141-439e-8421-081a11045f5d')
+// end( '80cb8117-fd90-48d2-a382-e609219879a4')
+let ws = null
+start().then(() => {
+    heartbeat(gameInfo.game_id)
+    try {
+        ws = new WebSocket(websocketInfo.wss_link[1])
+
+        ws.on('error', console.error)
+
+        ws.on('open', function open() {
+            const proto = new Proto()
+            proto.op = 7
+            proto.body = websocketInfo.auth_body
+            ws.send(proto.pack())
+
+            setTimeout(() => {
+                const proto = new Proto()
+                proto.op = 2
+
+                ws.send(proto.pack())
+            }, 1000)
+        })
+
+        ws.on('message', function message(data, isBinary) {
+            console.log('received: %s', data)
+            try {
+                const proto = new Proto()
+                proto.unpack(data)
+                console.log(proto.body)
+            } catch (error) {
+                console.log(error.stack)
+            }
+        })
+    } catch (error) {
+        console.log(error.stack)
+    }
+})
+
+setTimeout(() => {
+    end(gameInfo.game_id)
+    if (heartbeatTimer) {
+        clearInterval(heartbeatTimer)
+    }
+    if (ws) {
+        ws.close()
+    }
+}, 60000)

package.json

@@ -39,6 +39,7 @@
     "@nestjs/passport": "^9.0.3",
     "@nestjs/platform-express": "^9.3.3",
     "@nestjs/platform-socket.io": "^9.4.3",
+    "@nestjs/schedule": "^3.0.4",
     "@nestjs/swagger": "^6.2.1",
     "@nestjs/throttler": "^4.0.0",
     "@nestjs/typeorm": "^9.0.1",
@@ -84,6 +85,7 @@
     "tnwx": "^2.5.6",
     "typeorm": "^0.3.12",
     "uuid": "^9.0.0",
+    "ws": "^8.14.2",
     "yup": "^1.0.0",
     "zod": "^3.22.2"
   },

src/app.module.ts

@@ -13,12 +13,14 @@ import { FileModule } from './file/file.module'
 import { SysConfigModule } from './sys-config/sys-config.module'
 import { AllExceptionsFilter } from './filters/all-exceptions-filter.filter'
 import { DownloadModule } from './download/download.module'
-import { GameModule } from './game/game.module';
-import { TwitchModule } from './twitch/twitch.module';
-import { EventsModule } from './events/events.module';
-import { DanmuModule } from './danmu/danmu.module';
-import { PromptModule } from './prompt/prompt.module';
-import { RoomModule } from './room/room.module';
+import { GameModule } from './game/game.module'
+import { TwitchModule } from './twitch/twitch.module'
+import { EventsModule } from './events/events.module'
+import { DanmuModule } from './danmu/danmu.module'
+import { PromptModule } from './prompt/prompt.module'
+import { RoomModule } from './room/room.module'
+import { ScheduleModule } from '@nestjs/schedule'
+
 @Module({
     imports: [
         DevtoolsModule.register({
@@ -64,6 +66,7 @@ import { RoomModule } from './room/room.module';
                 autoLoadEntities: true
             })
         }),
+        ScheduleModule.forRoot(),
         AliyunModule,
         SmsModule,
         UsersModule,
@@ -76,7 +79,7 @@ import { RoomModule } from './room/room.module';
         EventsModule,
         DanmuModule,
         PromptModule,
-        RoomModule,
+        RoomModule
     ],
     controllers: [],
     providers: [

src/danmu/danmu.service.ts

@@ -1,8 +1,8 @@
-import { Inject, Injectable, Logger, OnModuleInit, forwardRef } from '@nestjs/common'
+import { Inject, Injectable, InternalServerErrorException, Logger, OnModuleInit, forwardRef } from '@nestjs/common'
 import { InjectRepository } from '@nestjs/typeorm'
 import { In, Like, MoreThan, Repository } from 'typeorm'
 import { Danmu } from './entities/danmu.enitity'
-import axios from 'axios'
+import axios, { AxiosInstance } from 'axios'
 import { Client } from 'tmi.js'
 import { AccessToken, AccessTokenType } from './entities/access-token.entity'
 import { StreamPlatform } from '../common/enums/stream-platform.enum'
@@ -14,12 +14,17 @@ import { FileService } from 'src/file/file.service'
 import path = require('path')
 import { HttpsProxyAgent } from 'https-proxy-agent'
 import { SysConfigService } from '../sys-config/sys-config.service'
+import { getEncodeHeader } from '../utils/crypto'
+import * as WebSocket from 'ws'
+import { Proto } from './models/proto.model'
+import { Interval } from '@nestjs/schedule'
 
 @Injectable()
 export class DanmuService implements OnModuleInit {
     private readonly logger = new Logger(DanmuService.name)
     client: Client
-
+    biliApi: AxiosInstance
+    biliWs: { [roomId: number]: WebSocket } = {}
     constructor(
         @InjectRepository(Danmu)
         private readonly danmuRepository: Repository<Danmu>,
@@ -31,7 +36,22 @@ export class DanmuService implements OnModuleInit {
         private readonly roomService: RoomService,
         private readonly fileService: FileService,
         private readonly sysConfigService: SysConfigService
-    ) {}
+    ) {
+        // axios 拦截器
+        this.biliApi = axios.create({
+            baseURL: 'https://live-open.biliapi.com'
+            // baseURL: "http://test-live-open.biliapi.net" //test
+        })
+
+        // 鉴权加密处理headers，下次请求自动带上
+        this.biliApi.interceptors.request.use((config) => {
+            const headers = getEncodeHeader(config.data, 'Cwgk7TsPkMtKtGO8rY0LndiF', 'bV9HiQ03OQ0Hk00l90N1QWyAFSI8qR')
+            Object.keys(headers).forEach((key) => {
+                config.headers[key] = headers[key]
+            })
+            return config
+        })
+    }
 
     async onModuleInit() {
         const channels = (await this.roomService.findAllActiveRoom())
@@ -141,6 +161,71 @@ export class DanmuService implements OnModuleInit {
         }
     }
 
+    async handleBiliMessage(data: any) {
+        try {
+            const proto = new Proto()
+            proto.unpack(data)
+            this.logger.debug('received message from bilibili op=' + proto.op)
+            switch (proto.op) {
+                case 3:
+                    this.logger.debug('收到心跳回复')
+                    break
+                case 5:
+                    this.logger.debug('收到弹幕')
+                    this.logger.debug(proto.body)
+                    const data = JSON.parse(proto.body)
+                    if (data.cmd === 'LIVE_OPEN_PLATFORM_DM') {
+                        const channel = `${data.data.room_id}`
+                        const platformUserId = `${data.data.uid}`
+                        const name = data.data.uname
+                        const avatar = data.data.uface
+                        const msgId = data.data.msg_id
+                        const room = await this.roomService.findActiveRoom(channel)
+                        if (room) {
+                            let danmuUser = await this.danmuUserRepository.findOne({
+                                where: { roomId: room.id, platform: StreamPlatform.Bilibili, platformUserId }
+                            })
+                            if (!danmuUser) {
+                                danmuUser = await this.danmuUserRepository.save(
+                                    new DanmuUser({
+                                        roomId: room.id,
+                                        platform: StreamPlatform.Bilibili,
+                                        platformUserId,
+                                        name,
+                                        avatar
+                                    })
+                                )
+                            }
+                            await this.danmuRepository.save(
+                                new Danmu({
+                                    platform: StreamPlatform.Bilibili,
+                                    danmuUserId: danmuUser.id,
+                                    channel,
+                                    platformRoomId: channel,
+                                    platformUserId,
+                                    content: data.data.msg,
+                                    roomId: room.id,
+                                    gameId: room.currentGameId,
+                                    msgId
+                                })
+                            )
+                        }
+                    }
+                    break
+                case 8:
+                    this.logger.debug('收到鉴权回复')
+                    if (JSON.parse(proto.body).code !== 0) {
+                        this.logger.error('鉴权失败' + proto.body)
+                    } else {
+                        this.logger.debug('鉴权成功')
+                    }
+                    break
+            }
+        } catch (error) {
+            this.logger.error(error.stack)
+        }
+    }
+
     async startDanmu(room: Room) {
         if (room.platform === StreamPlatform.Twitch) {
             try {
@@ -148,6 +233,34 @@ export class DanmuService implements OnModuleInit {
             } catch (error) {
                 this.logger.error(`failed to join twitch channel ${room.channelId}, ${error}`)
             }
+        } else if (room.platform === StreamPlatform.Bilibili) {
+            if (!room.code) {
+                throw new InternalServerErrorException('缺少身份码')
+            }
+            const { data } = await this.biliApi.post('/v2/app/start', { app_id: 1700371901242, code: room.code })
+            if (data.code === 0) {
+                this.logger.log(`开启直播间成功`)
+                room.config = data.data
+                await this.roomService.updateRoom(room.id, {
+                    config: data.data
+                })
+                const ws = new WebSocket(room.config.websocket_info.wss_link[0])
+                ws.on('error', this.logger.error)
+
+                ws.on('open', function open() {
+                    const proto = new Proto()
+                    proto.op = 7
+                    proto.body = room.config.websocket_info.auth_body
+                    ws.send(proto.pack())
+                })
+
+                ws.on('message', this.handleBiliMessage.bind(this))
+
+                this.biliWs[room.id] = ws
+            } else {
+                this.logger.error(`开启直播间失败: ${JSON.stringify(data)}`)
+                throw new InternalServerErrorException(data.message)
+            }
         }
     }
 
@@ -158,6 +271,19 @@ export class DanmuService implements OnModuleInit {
             } catch (error) {
                 this.logger.error(`failed to leave twitch channel ${room.channelId}, ${error}`)
             }
+        } else if (room.platform === StreamPlatform.Bilibili) {
+            const { data } = await this.biliApi.post('/v2/app/end', {
+                app_id: 1700371901242,
+                game_id: room.config.game_info.game_id
+            })
+            if (data.code !== 0) {
+                this.logger.error(`结束直播间失败: ${JSON.stringify(data)}`)
+            }
+            this.logger.log(`结束直播间成功`)
+            if (this.biliWs[room.id]) {
+                this.biliWs[room.id].close()
+                delete this.biliWs[room.id]
+            }
         }
     }
 
@@ -232,4 +358,27 @@ export class DanmuService implements OnModuleInit {
         danmuUser.name = user.display_name
         return await this.danmuUserRepository.save(danmuUser)
     }
+
+    @Interval(10000)
+    async heartbeat() {
+        Object.keys(this.biliWs).forEach(async (roomId) => {
+            try {
+                const room = await this.roomService.findById(Number(roomId))
+                const { data } = await this.biliApi.post('/v2/app/batchHeartbeat', {
+                    game_ids: [room.config.game_info.game_id]
+                })
+                if (data.code === 0) {
+                    this.logger.debug('心跳发送成功')
+                } else {
+                    this.logger.error(`心跳发送失败: ${JSON.stringify(data)}`)
+                }
+                const ws = this.biliWs[roomId]
+                const proto = new Proto()
+                proto.op = 2
+                ws.send(proto.pack())
+            } catch (error) {
+                this.logger.error(error)
+            }
+        })
+    }
 }

src/danmu/entities/danmu.enitity.ts

@@ -34,8 +34,11 @@ export class Danmu {
 
     @Column({ nullable: true })
     @Index()
-    gameId: number
+    gameId?: number
 
     @Column({ nullable: true })
-    roomId: number
+    roomId?: number
+
+    @Column({ nullable: true })
+    msgId?: string
 }

src/danmu/models/proto.model.ts

@@ -0,0 +1,47 @@
+export class Proto {
+    packetLen = 0
+    headerLen = 16
+    ver = 0
+    public op = 0
+    seq = 0
+    public body: string = ''
+    maxBody = 2048
+
+    constructor() {}
+
+    pack() {
+        this.packetLen = this.body.length + this.headerLen
+        const buf = Buffer.alloc(this.packetLen) // 创建一个包含指定长度的 Buffer
+
+        buf.writeInt32BE(this.packetLen, 0) // 写入一个32位大端整数到偏移量0的位置
+        buf.writeUInt16BE(this.headerLen, 4) // 写入一个16位大端无符号整数到偏移量4的位置
+        buf.writeUInt16BE(this.ver, 6) // 写入一个16位大端无符号整数到偏移量6的位置
+        buf.writeInt32BE(this.op, 8) // 写入一个32位大端整数到偏移量8的位置
+        buf.writeInt32BE(this.seq, 12) // 写入一个32位大端整数到偏移量12的位置
+
+        // 将字符串编码为字节序列并写入 Buffer
+        const bodyBuffer = Buffer.from(this.body, 'utf-8')
+        bodyBuffer.copy(buf, this.headerLen)
+
+        return buf
+    }
+
+    unpack(buf) {
+        if (buf.length < this.headerLen) {
+            throw new Error('包头不够')
+        }
+        this.packetLen = buf.readInt32BE(0) // 读取32位大端整数，偏移量0
+        this.headerLen = buf.readUInt16BE(4) // 读取16位大端无符号整数，偏移量4
+        this.ver = buf.readUInt16BE(6) // 读取16位大端无符号整数，偏移量6
+        this.op = buf.readInt32BE(8) // 读取32位大端整数，偏移量8
+        this.seq = buf.readInt32BE(12) // 读取32位大端整数，偏移量12
+
+        if (this.packetLen < 0 || this.packetLen > this.maxBody) {
+            throw new Error(`包体长不对 packetLen=${this.packetLen} maxBody=${this.maxBody}`)
+        }
+        if (this.headerLen !== 16) {
+            throw new Error('包头长度不对')
+        }
+        this.body = buf.slice(16, this.packetLen).toString('utf-8') // 从偏移量16开始读取指定长度的数据并解码成字符串
+    }
+}

src/room/entities/room.entity.ts

@@ -19,6 +19,9 @@ export class Room {
     @Column()
     channelId: string
 
+    @Column({ nullable: true })
+    code?: string
+
     @Column({ default: false })
     active: boolean
 

src/room/room.service.ts

@@ -51,15 +51,15 @@ export class RoomService {
     }
 
     async startRoom(id: number) {
-        await this.roomRepository.update({ id }, { active: true })
         const room = await this.roomRepository.findOne({ where: { id } })
         await this.danmuService.startDanmu(room)
+        await this.roomRepository.update({ id }, { active: true })
     }
 
     async stopRoom(id: number) {
-        await this.roomRepository.update({ id }, { active: false })
         const room = await this.roomRepository.findOne({ where: { id } })
         await this.danmuService.stopDanmu(room)
+        await this.roomRepository.update({ id }, { active: false })
     }
 
     async survivalRank(roomId: number) {

src/utils/crypto.ts

@@ -0,0 +1,40 @@
+import * as crypto from 'crypto'
+
+/**
+ * 鉴权加密
+ * @param {*} params
+ * @returns
+ */
+export function getEncodeHeader(params = {}, appKey, appSecret) {
+    const timestamp = parseInt(Date.now() / 1000 + '')
+    const nonce = parseInt(Math.random() * 100000 + '') + timestamp
+    const header = {
+        'x-bili-accesskeyid': appKey,
+        'x-bili-content-md5': getMd5Content(JSON.stringify(params)),
+        'x-bili-signature-method': 'HMAC-SHA256',
+        'x-bili-signature-nonce': nonce + '',
+        'x-bili-signature-version': '1.0',
+        'x-bili-timestamp': timestamp
+    }
+    const data = []
+    for (const key in header) {
+        data.push(`${key}:${header[key]}`)
+    }
+
+    const signature = crypto.createHmac('sha256', appSecret).update(data.join('\n')).digest('hex')
+    return {
+        Accept: 'application/json',
+        'Content-Type': 'application/json',
+        ...header,
+        Authorization: signature
+    }
+}
+
+/**
+ * MD5加密
+ * @param {*} str
+ * @returns
+ */
+export function getMd5Content(str) {
+    return crypto.createHash('md5').update(str).digest('hex')
+}

yarn.lock

@@ -1053,6 +1053,14 @@
     socket.io "4.6.2"
     tslib "2.5.3"
 
+"@nestjs/schedule@^3.0.4":
+  version "3.0.4"
+  resolved "https://registry.npmmirror.com/@nestjs/schedule/-/schedule-3.0.4.tgz#c2a3d96bb2cb7d562349f030823080b8513b34c3"
+  integrity sha512-uFJpuZsXfpvgx2y7/KrIZW9e1L68TLiwRodZ6+Gc8xqQiHSUzAVn+9F4YMxWFlHITZvvkjWziUFgRNCitDcTZQ==
+  dependencies:
+    cron "2.4.3"
+    uuid "9.0.1"
+
 "@nestjs/schematics@^9.0.0", "@nestjs/schematics@^9.0.4":
   version "9.0.4"
   resolved "https://registry.npmjs.org/@nestjs/schematics/-/schematics-9.0.4.tgz"
@@ -1427,6 +1435,11 @@
   dependencies:
     "@types/node" "*"
 
+"@types/luxon@~3.3.0":
+  version "3.3.3"
+  resolved "https://registry.npmmirror.com/@types/luxon/-/luxon-3.3.3.tgz#b2e20a9536f91ab3e6e7895c91883e1a7ad49a6e"
+  integrity sha512-/BJF3NT0pRMuxrenr42emRUF67sXwcZCd+S1ksG/Fcf9O7C3kKCY4uJSbKBE4KDUIYr3WMsvfmWD8hRjXExBJQ==
+
 "@types/mime@*":
   version "3.0.1"
   resolved "https://registry.npmjs.org/@types/mime/-/mime-3.0.1.tgz"
@@ -2766,6 +2779,14 @@ create-require@^1.1.0:
   resolved "https://registry.npmjs.org/create-require/-/create-require-1.1.1.tgz"
   integrity sha512-dcKFX3jn0MpIaXjisoRvexIJVEKzaq7z2rZKxf+MSr9TkdmHmsU4m2lcLojrj/FHl8mk5VxMmYA+ftRkP/3oKQ==
 
+cron@2.4.3:
+  version "2.4.3"
+  resolved "https://registry.npmmirror.com/cron/-/cron-2.4.3.tgz#4e43d8d9a6373b8f28d876c4e9a47c14422d8652"
+  integrity sha512-YBvExkQYF7w0PxyeFLRyr817YVDhGxaCi5/uRRMqa4aWD3IFKRd+uNbpW1VWMdqQy8PZ7CElc+accXJcauPKzQ==
+  dependencies:
+    "@types/luxon" "~3.3.0"
+    luxon "~3.3.0"
+
 cross-spawn@^7.0.0, cross-spawn@^7.0.2, cross-spawn@^7.0.3:
   version "7.0.3"
   resolved "https://registry.npmjs.org/cross-spawn/-/cross-spawn-7.0.3.tgz"
@@ -5133,6 +5154,11 @@ lru-cache@^9.0.0:
   resolved "https://registry.npmmirror.com/lru-cache/-/lru-cache-9.1.1.tgz#c58a93de58630b688de39ad04ef02ef26f1902f1"
   integrity sha512-65/Jky17UwSb0BuB9V+MyDpsOtXKmYwzhyl+cOa9XUiI4uV2Ouy/2voFP3+al0BjZbJgMBD8FojMpAf+Z+qn4A==
 
+luxon@~3.3.0:
+  version "3.3.0"
+  resolved "https://registry.npmmirror.com/luxon/-/luxon-3.3.0.tgz#d73ab5b5d2b49a461c47cedbc7e73309b4805b48"
+  integrity sha512-An0UCfG/rSiqtAIiBPO0Y9/zAnHUZxAMiCpTd5h2smgsj7GGmcenvrvww2cqNA8/4A5ZrD1gJpHN2mIHZQF+Mg==
+
 macos-release@^2.5.0:
   version "2.5.0"
   resolved "https://registry.npmjs.org/macos-release/-/macos-release-2.5.0.tgz"
@@ -7357,6 +7383,11 @@ uuid@9.0.0, uuid@^9.0.0:
   resolved "https://registry.npmjs.org/uuid/-/uuid-9.0.0.tgz"
   integrity sha512-MXcSTerfPa4uqyzStbRoTgt5XIe3x5+42+q1sDuy3R5MDk66URdLMOZe5aPX/SQd+kuYAh0FdP/pO28IkQyTeg==
 
+uuid@9.0.1:
+  version "9.0.1"
+  resolved "https://registry.npmmirror.com/uuid/-/uuid-9.0.1.tgz#e188d4c8853cc722220392c424cd637f32293f30"
+  integrity sha512-b+1eJOlsR9K8HJpow9Ok3fiWOWSIcIzXodvv0rQjVoOVNpWMpxf1wZNpt4y9h10odCNrqnYp1OBzRktckBe3sA==
+
 uuid@^3.3.2, uuid@^3.3.3:
   version "3.4.0"
   resolved "https://registry.npmmirror.com/uuid/-/uuid-3.4.0.tgz#b23e4358afa8a202fe7a100af1f5f883f02007ee"
@@ -7577,7 +7608,7 @@ write-file-atomic@^4.0.2:
     imurmurhash "^0.1.4"
     signal-exit "^3.0.7"
 
-ws@^8.2.0:
+ws@^8.14.2, ws@^8.2.0:
   version "8.14.2"
   resolved "https://registry.npmmirror.com/ws/-/ws-8.14.2.tgz#6c249a806eb2db7a20d26d51e7709eab7b2e6c7f"
   integrity sha512-wEBG1ftX4jcglPxgFCMJmZ2PLtSbJ2Peg6TmpJFTbe9GZYOQCDPdMYu/Tm0/bGZkw8paZnJY45J4K2PZrLYq8g==