Next Flow 官方文档学习 从Your first script开始params.str = 'Hello world!' process splitLetters { output: path 'chunk_*' """ printf '${params.str}' | split -b 6 - chunk_ """ } process convertToUpper { input: path x output: stdout """ cat $x | tr '[a-z]' '[A-Z]' """ #bash中的tr命令 ，执行字符串转换（替换） #简单的查找替换 #如该句指令，指将小写字母转化为大写字母 #语法：管道输入 | tr '[SET1]' '[SET2]' #这里，SET1 代表要在文本中找到的字母，SET2 代表要替换的字母。 #tr -d 用于删除指定的字符 #tr -c 用于将连续重复的指定字符变唯一 } workflow { splitLetters | flatten | convertToUpper | view { it.trim() } }''是针对字符的，比如a；而""是针对字符串的，比如"hello"。而shell呢，主要是用c/c++实现的，所以在shell里，""和''是有区别的。单引号限制最大，任何的字符都会原样输出，变量是无效的双引号比较友好，可以引用，也可以有转义字符例子：当您需要访问脚本中的系统环境变量时，您有两个选择。如果不需要访问任何 Nextflow 变量，则可以使用单引号定义脚本块：process printPath { ''' echo The path is: $PATH ''' }否则，您可以使用双引号定义脚本，并通过以下方式转义系统环境变量 用反斜杠字符作为前缀，如以下示例所示：\process doOtherThings { """ blastp -db \$DB -query query.fa -outfmt 6 > blast_result cat blast_result | head -n $MAX | cut -f 2 > top_hits blastdbcmd -db \$DB -entry_batch top_hits > sequences """ }一句话解释：单引号直接传送给bash，双引号先尝试进行nextflow进行解析，获取类似$的关键字，完成替换（或识别完转义）后输送给bash解释器：shebang#!/usr/bin/perlNextflow脚本语言是Groovy编程语言的扩展。 Groovy是Java虚拟机的强大编程语言。下一个流 语法专门用于以声明方式简化计算管道的编写。Nextflow可以执行任何一段Groovy代码或使用JVM平台的任何库。{dotted startColor="#ff6c6c" endColor="#1989fa"/}注意看一下隐式变量其中含有很多脚本自带的常用变量{dotted startColor="#ff6c6c" endColor="#1989fa"/}脚本自身也有一个解释器，以java为基础可以完成基本的编程操作如条件判断：if () {} else {}字符串操作与python 支持 + 连接两个字符串多行字符串使用"""string_line1string_line2"""如果在多行字符串内的行末加上 \ 反斜杠会取消换行"""string_line1 \string_line2"""这些都和基本的编程语言一致{dotted startColor="#ff6c6c" endColor="#1989fa"/}

回调 _ 以python下的项目blivedm为例 调用的三种方式 （参考原文）同步调用同步调用是最基本并且最简单的一种调用方式，类A的方法a()调用类B的方法b(),一直等待b()方法执行完毕，a()方法再继续往下走。这种调用方式适用于方法b()执行时间不长的情况，因为b()方法执行时间一长或者直接阻塞的话，a()方法的余下代码是无法执行下去的，这样会造成整个流程的阻塞。异步调用是一种类似消息或事件的机制，是为了解决同步调用可能出现阻塞，导致整个流程卡住而产生的一种调用方式。类A的方法a()通过新起线程的方式调用类B的方法b(),代码接着直接往下执行，这样无论方法b()执行时间多久，都不会阻塞方法a()的执行。但是这种方式，由于方法a()不等待方法b()执行完成，在方法a()需要方法b()执行结果的情况下(视具体业务而定，有些业务比如启动异步线程发个微信通知、刷新一个缓存这种就没有必要)，必须通过一定的方法对方法b()的执行结果进行监听。在Java中，可以使用Future+Callable的方式做到这一点。回调回调的思想是: 类A的a()方法调用了类B的b()方法 类B的b方法执行完毕主动调用类A的callback()方法这样一种调用方式组成了上图，也就是一种双向的调用方式回调函数是一个函数或过程，不过它是一个由调用方自己实现，供被调用方使用的特殊函数。在面向对象的语言中，回调则是通过接口或抽象类来实现的，我们把实现这种接口的类称为回调类，回调类的对象称为回调对象。回调回调函数按照目前比较通行的定义，只要把函数以参数的形式传给其它函数，由其它函数调用，那就是回调函数。回调有三个部分组成：回调函数、中间函数和起始函数举个住旅馆叫醒服务的例子：1.旅馆提供叫醒服务是起始函数2.有三种叫醒旅客的方式：打电话叫、服务员敲门叫，在你头上浇盆水，这三种叫醒方式属于三种回调函数3.旅客通知旅馆使用何种叫醒方式的动作，属于中间函数一、创建三种回调函数：1.morning_call(me)2.knocking_door(me)3.watering_head(me)二、创建中间函数：def ask_for_wakeup(people, call_function)return call_function(people)三、用起始函数（主函数）来做这件事情def main():return ask_for_wakeup('Antony', watering_head)酱紫，最后一部调用回调函数，又返回回调函数，最终得到结果阻塞式回调 和 延迟式回调 也可以叫做 同步回调 和 异步回调在阻塞式回调里，回调函数的调用一定发生在主函数返回之前在延迟式回调里，回调函数的调用有可能是在起始函数返回之后你到一个商店买东西，刚好你要的东西没有货，于是你在店员那里留下了你的电话，过了几天店里有货了，店员就打了你的电话，然后你接到电话后就到店里去取了货。在这个例子里，你的电话号码就叫回调函数，你把电话留给店员就叫登记回调函数，店里后来有货了叫做 触发回调事件，店员给你打电话叫做 调用回调函数，你到店里去取货叫做 响应回调事件。

blivedm 项目阅读 之 models.py 这个文件应该主要是数据结构# -*- coding: utf-8 -*- import json from typing import * __all__ = ( 'HeartbeatMessage', 'DanmakuMessage', 'GiftMessage', 'GuardBuyMessage', 'SuperChatMessage', 'SuperChatDeleteMessage', ) class HeartbeatMessage: """ 心跳消息 :param popularity: 人气值 """ def __init__( self, popularity: int = None, ): self.popularity: int = popularity @classmethod def from_command(cls, data: dict): return cls( popularity=data['popularity'], ) class DanmakuMessage: """ 弹幕消息 :param mode: 弹幕显示模式（滚动、顶部、底部） :param font_size: 字体尺寸 :param color: 颜色 :param timestamp: 时间戳（毫秒） :param rnd: 随机数，前端叫作弹幕ID，可能是去重用的 :param uid_crc32: 用户ID文本的CRC32 :param msg_type: 是否礼物弹幕（节奏风暴） :param bubble: 右侧评论栏气泡 :param dm_type: 弹幕类型，0文本，1表情，2语音 :param emoticon_options: 表情参数 :param voice_config: 语音参数 :param mode_info: 一些附加参数 :param msg: 弹幕内容 :param uid: 用户ID :param uname: 用户名 :param admin: 是否房管 :param vip: 是否月费老爷 :param svip: 是否年费老爷 :param urank: 用户身份，用来判断是否正式会员，猜测非正式会员为5000，正式会员为10000 :param mobile_verify: 是否绑定手机 :param uname_color: 用户名颜色 :param medal_level: 勋章等级 :param medal_name: 勋章名 :param runame: 勋章房间主播名 :param medal_room_id: 勋章房间ID :param mcolor: 勋章颜色 :param special_medal: 特殊勋章 :param user_level: 用户等级 :param ulevel_color: 用户等级颜色 :param ulevel_rank: 用户等级排名，>50000时为'>50000' :param old_title: 旧头衔 :param title: 头衔 :param privilege_type: 舰队类型，0非舰队，1总督，2提督，3舰长 """ def __init__( self, mode: int = None, font_size: int = None, color: int = None, timestamp: int = None, rnd: int = None, uid_crc32: str = None, msg_type: int = None, bubble: int = None, dm_type: int = None, emoticon_options: Union[dict, str] = None, voice_config: Union[dict, str] = None, mode_info: dict = None, msg: str = None, uid: int = None, uname: str = None, admin: int = None, vip: int = None, svip: int = None, urank: int = None, mobile_verify: int = None, uname_color: str = None, medal_level: str = None, medal_name: str = None, runame: str = None, medal_room_id: int = None, mcolor: int = None, special_medal: str = None, user_level: int = None, ulevel_color: int = None, ulevel_rank: str = None, old_title: str = None, title: str = None, privilege_type: int = None, ): self.mode: int = mode self.font_size: int = font_size self.color: int = color self.timestamp: int = timestamp self.rnd: int = rnd self.uid_crc32: str = uid_crc32 self.msg_type: int = msg_type self.bubble: int = bubble self.dm_type: int = dm_type self.emoticon_options: Union[dict, str] = emoticon_options self.voice_config: Union[dict, str] = voice_config self.mode_info: dict = mode_info self.msg: str = msg self.uid: int = uid self.uname: str = uname self.admin: int = admin self.vip: int = vip self.svip: int = svip self.urank: int = urank self.mobile_verify: int = mobile_verify self.uname_color: str = uname_color self.medal_level: str = medal_level self.medal_name: str = medal_name self.runame: str = runame self.medal_room_id: int = medal_room_id self.mcolor: int = mcolor self.special_medal: str = special_medal self.user_level: int = user_level self.ulevel_color: int = ulevel_color self.ulevel_rank: str = ulevel_rank self.old_title: str = old_title self.title: str = title self.privilege_type: int = privilege_type @classmethod def from_command(cls, info: dict): if len(info[3]) != 0: medal_level = info[3][0] medal_name = info[3][1] runame = info[3][2] room_id = info[3][3] mcolor = info[3][4] special_medal = info[3][5] else: medal_level = 0 medal_name = '' runame = '' room_id = 0 mcolor = 0 special_medal = 0 return cls( mode=info[0][1], font_size=info[0][2], color=info[0][3], timestamp=info[0][4], rnd=info[0][5], uid_crc32=info[0][7], msg_type=info[0][9], bubble=info[0][10], dm_type=info[0][12], emoticon_options=info[0][13], voice_config=info[0][14], mode_info=info[0][15], msg=info[1], uid=info[2][0], uname=info[2][1], admin=info[2][2], vip=info[2][3], svip=info[2][4], urank=info[2][5], mobile_verify=info[2][6], uname_color=info[2][7], medal_level=medal_level, medal_name=medal_name, runame=runame, medal_room_id=room_id, mcolor=mcolor, special_medal=special_medal, user_level=info[4][0], ulevel_color=info[4][2], ulevel_rank=info[4][3], old_title=info[5][0], title=info[5][1], privilege_type=info[7], ) @property def emoticon_options_dict(self) -> dict: """ 示例： {'bulge_display': 0, 'emoticon_unique': 'official_13', 'height': 60, 'in_player_area': 1, 'is_dynamic': 1, 'url': 'https://i0.hdslb.com/bfs/live/a98e35996545509188fe4d24bd1a56518ea5af48.png', 'width': 183} """ if isinstance(self.emoticon_options, dict): return self.emoticon_options try: return json.loads(self.emoticon_options) except (json.JSONDecodeError, TypeError): return {} @property def voice_config_dict(self) -> dict: """ 示例： {'voice_url': 'https%3A%2F%2Fboss.hdslb.com%2Flive-dm-voice%2Fb5b26e48b556915cbf3312a59d3bb2561627725945.wav %3FX-Amz-Algorithm%3DAWS4-HMAC-SHA256%26X-Amz-Credential%3D2663ba902868f12f%252F20210731%252Fshjd%252Fs3%25 2Faws4_request%26X-Amz-Date%3D20210731T100545Z%26X-Amz-Expires%3D600000%26X-Amz-SignedHeaders%3Dhost%26 X-Amz-Signature%3D114e7cb5ac91c72e231c26d8ca211e53914722f36309b861a6409ffb20f07ab8', 'file_format': 'wav', 'text': '汤，下午好。', 'file_duration': 1} """ if isinstance(self.voice_config, dict): return self.voice_config try: return json.loads(self.voice_config) except (json.JSONDecodeError, TypeError): return {} class GiftMessage: """ 礼物消息 :param gift_name: 礼物名 :param num: 数量 :param uname: 用户名 :param face: 用户头像URL :param guard_level: 舰队等级，0非舰队，1总督，2提督，3舰长 :param uid: 用户ID :param timestamp: 时间戳 :param gift_id: 礼物ID :param gift_type: 礼物类型（未知） :param action: 目前遇到的有'喂食'、'赠送' :param price: 礼物单价瓜子数 :param rnd: 随机数，可能是去重用的。有时是时间戳+去重ID，有时是UUID :param coin_type: 瓜子类型，'silver'或'gold'，1000金瓜子 = 1元 :param total_coin: 总瓜子数 :param tid: 可能是事务ID，有时和rnd相同 """ def __init__( self, gift_name: str = None, num: int = None, uname: str = None, face: str = None, guard_level: int = None, uid: int = None, timestamp: int = None, gift_id: int = None, gift_type: int = None, action: str = None, price: int = None, rnd: str = None, coin_type: str = None, total_coin: int = None, tid: str = None, ): self.gift_name = gift_name self.num = num self.uname = uname self.face = face self.guard_level = guard_level self.uid = uid self.timestamp = timestamp self.gift_id = gift_id self.gift_type = gift_type self.action = action self.price = price self.rnd = rnd self.coin_type = coin_type self.total_coin = total_coin self.tid = tid @classmethod def from_command(cls, data: dict): return cls( gift_name=data['giftName'], num=data['num'], uname=data['uname'], face=data['face'], guard_level=data['guard_level'], uid=data['uid'], timestamp=data['timestamp'], gift_id=data['giftId'], gift_type=data['giftType'], action=data['action'], price=data['price'], rnd=data['rnd'], coin_type=data['coin_type'], total_coin=data['total_coin'], tid=data['tid'], ) class GuardBuyMessage: """ 上舰消息 :param uid: 用户ID :param username: 用户名 :param guard_level: 舰队等级，0非舰队，1总督，2提督，3舰长 :param num: 数量 :param price: 单价金瓜子数 :param gift_id: 礼物ID :param gift_name: 礼物名 :param start_time: 开始时间戳，和结束时间戳相同 :param end_time: 结束时间戳，和开始时间戳相同 """ def __init__( self, uid: int = None, username: str = None, guard_level: int = None, num: int = None, price: int = None, gift_id: int = None, gift_name: str = None, start_time: int = None, end_time: int = None, ): self.uid: int = uid self.username: str = username self.guard_level: int = guard_level self.num: int = num self.price: int = price self.gift_id: int = gift_id self.gift_name: str = gift_name self.start_time: int = start_time self.end_time: int = end_time @classmethod def from_command(cls, data: dict): return cls( uid=data['uid'], username=data['username'], guard_level=data['guard_level'], num=data['num'], price=data['price'], gift_id=data['gift_id'], gift_name=data['gift_name'], start_time=data['start_time'], end_time=data['end_time'], ) class SuperChatMessage: """ 醒目留言消息 :param price: 价格（人民币） :param message: 消息 :param message_trans: 消息日文翻译（目前只出现在SUPER_CHAT_MESSAGE_JPN） :param start_time: 开始时间戳 :param end_time: 结束时间戳 :param time: 剩余时间（约等于 结束时间戳 - 开始时间戳） :param id_: str，醒目留言ID，删除时用 :param gift_id: 礼物ID :param gift_name: 礼物名 :param uid: 用户ID :param uname: 用户名 :param face: 用户头像URL :param guard_level: 舰队等级，0非舰队，1总督，2提督，3舰长 :param user_level: 用户等级 :param background_bottom_color: 底部背景色，'#rrggbb' :param background_color: 背景色，'#rrggbb' :param background_icon: 背景图标 :param background_image: 背景图URL :param background_price_color: 背景价格颜色，'#rrggbb' """ def __init__( self, price: int = None, message: str = None, message_trans: str = None, start_time: int = None, end_time: int = None, time: int = None, id_: int = None, gift_id: int = None, gift_name: str = None, uid: int = None, uname: str = None, face: str = None, guard_level: int = None, user_level: int = None, background_bottom_color: str = None, background_color: str = None, background_icon: str = None, background_image: str = None, background_price_color: str = None, ): self.price: int = price self.message: str = message self.message_trans: str = message_trans self.start_time: int = start_time self.end_time: int = end_time self.time: int = time self.id: int = id_ self.gift_id: int = gift_id self.gift_name: str = gift_name self.uid: int = uid self.uname: str = uname self.face: str = face self.guard_level: int = guard_level self.user_level: int = user_level self.background_bottom_color: str = background_bottom_color self.background_color: str = background_color self.background_icon: str = background_icon self.background_image: str = background_image self.background_price_color: str = background_price_color @classmethod def from_command(cls, data: dict): return cls( price=data['price'], message=data['message'], message_trans=data['message_trans'], start_time=data['start_time'], end_time=data['end_time'], time=data['time'], id_=data['id'], gift_id=data['gift']['gift_id'], gift_name=data['gift']['gift_name'], uid=data['uid'], uname=data['user_info']['uname'], face=data['user_info']['face'], guard_level=data['user_info']['guard_level'], user_level=data['user_info']['user_level'], background_bottom_color=data['background_bottom_color'], background_color=data['background_color'], background_icon=data['background_icon'], background_image=data['background_image'], background_price_color=data['background_price_color'], ) class SuperChatDeleteMessage: """ 删除醒目留言消息 :param ids: 醒目留言ID数组 """ def __init__( self, ids: List[int] = None, ): self.ids: List[int] = ids @classmethod def from_command(cls, data: dict): return cls( ids=data['ids'], )

blivedm 项目阅读 之 handlers.py Logging 官方文档# -*- coding: utf-8 -*- import logging #Python中的logging模块，是用于实现日志的功能。 #此日志的功能，简单的说，可以控制需要输出的信息，输出（显示）到哪里；相关要显示的信息，有很多种等级，比如info，warning，error等 # """最常见的应用是： 把info，warning，error同时输出到cmd窗口（显示）和（写入）log文件中； 其中info表示告诉用户，这个是普通的信息； warning和error分别提醒用户，有些警告，甚至是错误信息，需要用户注意； 把debug类信息，只输出（写入）到log文件中； """ from typing import * from . import client as client_ from . import models __all__ = ( 'HandlerInterface', 'BaseHandler', ) """以单下划线开头（_foo）的代表不能直接访问的类属性，需通过类提供的接口进行访问，不能用“from xxx import *”而导入；以双下划线开头的（__foo）代表类的私有成员；以双下划线开头和结尾的（__foo__）代表python里特殊方法专用的标识，如 __init__（）代表类的构造函数。""" # #Python模块__all__变量 #该变量的值是一个列表，存储的是当前模块中一些成员（变量、函数或者类）的名称。通过在模块文件中设置 __all__ 变量，当其它文件以“from 模块名 import *”的形式导入该模块时，该文件中只能使用 __all__ 列表中指定的成员。 #也就是说，只有以“from 模块名 import *”形式导入的模块，当该模块设有 __all__ 变量时，只能导入该变量指定的成员，未指定的成员是无法导入的。 logger = logging.getLogger('blivedm') #返回一个logger对象 # 常见可忽略的cmd IGNORED_CMDS = ( 'COMBO_SEND', 'ENTRY_EFFECT', 'HOT_RANK_CHANGED', 'HOT_RANK_CHANGED_V2', 'INTERACT_WORD', 'LIVE', 'LIVE_INTERACTIVE_GAME', 'NOTICE_MSG', 'ONLINE_RANK_COUNT', 'ONLINE_RANK_TOP3', 'ONLINE_RANK_V2', 'PK_BATTLE_END', 'PK_BATTLE_FINAL_PROCESS', 'PK_BATTLE_PROCESS', 'PK_BATTLE_PROCESS_NEW', 'PK_BATTLE_SETTLE', 'PK_BATTLE_SETTLE_USER', 'PK_BATTLE_SETTLE_V2', 'PREPARING', 'ROOM_REAL_TIME_MESSAGE_UPDATE', 'STOP_LIVE_ROOM_LIST', 'SUPER_CHAT_MESSAGE_JPN', 'WIDGET_BANNER', ) # 已打日志的未知cmd logged_unknown_cmds = set() class HandlerInterface: """ 直播消息处理器接口 """ async def handle(self, client: client_.BLiveClient, command: dict): raise NotImplementedError #raise 手动抛出异常 """ raise 语句的基本语法格式为： raise [exceptionName [(reason)]] 其中，用 [] 括起来的为可选参数，其作用是指定抛出的异常名称，以及异常信息的相关描述。如果可选参数全部省略，则 raise 会把当前错误原样抛出；如果仅省略 (reason)，则在抛出异常时，将不附带任何的异常描述信息。 也就是说，raise 语句有如下三种常用的用法： raise：单独一个 raise。该语句引发当前上下文中捕获的异常（比如在 except 块中），或默认引发 RuntimeError 异常。 raise 异常类名称：raise 后带一个异常类名称，表示引发执行类型的异常。 raise 异常类名称(描述信息)：在引发指定类型的异常的同时，附带异常的描述信息。 raise 语句引发的异常通常用 try except（else finally）异常处理结构来捕获并进行处理。 try: a = input("输入一个数：") #判断用户输入的是否为数字 if(not a.isdigit()): raise ValueError("a 必须是数字") except ValueError as e: print("引发异常：",repr(e)) """ ''' raise NotImplementedError 在面向对象编程中，可以先预留一个方法接口不实现，在其子类中实现。 如果要求其子类一定要实现，不实现的时候会导致问题，那么采用raise的方式就很好。 程序识别到了这个方法并没有在子类中实现却被调用了。且只有这个子类的实例化对象调用相应的方法的时候才会报错。 只要相应的方法接口进行了实现，在执行的时候未实施的错误便不会报出。 在此例中对应的是handle方法 ''' class BaseHandler(HandlerInterface): """ 一个简单的消息处理器实现，带消息分发和消息类型转换。继承并重写_on_xxx方法即可实现自己的处理器 """ def __heartbeat_callback(self, client: client_.BLiveClient, command: dict): return self._on_heartbeat(client, models.HeartbeatMessage.from_command(command['data'])) def __danmu_msg_callback(self, client: client_.BLiveClient, command: dict): return self._on_danmaku(client, models.DanmakuMessage.from_command(command['info'])) def __send_gift_callback(self, client: client_.BLiveClient, command: dict): return self._on_gift(client, models.GiftMessage.from_command(command['data'])) def __guard_buy_callback(self, client: client_.BLiveClient, command: dict): return self._on_buy_guard(client, models.GuardBuyMessage.from_command(command['data'])) def __super_chat_message_callback(self, client: client_.BLiveClient, command: dict): return self._on_super_chat(client, models.SuperChatMessage.from_command(command['data'])) def __super_chat_message_delete_callback(self, client: client_.BLiveClient, command: dict): return self._on_super_chat_delete(client, models.SuperChatDeleteMessage.from_command(command['data'])) #在定义函数时，变量默认值用=，变量默认类型用: """ Typing包中的Optional类 意思是可选类型，作用几乎和带默认值的参数等价，不同的是使用Optional会告诉你的IDE或者框架：这个参数除了给定的默认值外还可以是None，而且使用有些静态检查工具如mypy时，对 a: int =None这样类似的声明可能会提示报错，但使用a :Optional[int] = None不会。 Typing包中的Callable 是一个可调用对象类型,查看是否可调用 第一个用法：isinstance(对象, Callable) # 返回True或False 这里是第二个用法Callable 作为函数参数 Callable 作为函数参数使用，其实只是做一个类型检查的作用，检查传入的参数值 get_func 是否为可调用对象 第三个用法Callable 作为函数返回值使用，其实只是做一个类型检查的作用，看看返回值是否为可调用对象 #def XXX -> 数据类型 #标注返回值的数据类型 #def f(ham: str, eggs: str = 'eggs') -> str: def get_name_return() -> Callable[[str], None]: return print_name # 等价写法，相当于直接返回一个函数对象 def get_name_test(): return print_name """ """这里应该是定义了一个callback的数据结构(字典)，然后定义了字典的Optional默认参数，最后检查其可调用性""" # cmd -> 处理回调 _CMD_CALLBACK_DICT: Dict[ str, Optional[Callable[ ['BaseHandler', client_.BLiveClient, dict], Awaitable ]]# ] = { # 收到心跳包，这是blivedm自造的消息，原本的心跳包格式不一样 '_HEARTBEAT': __heartbeat_callback, # 收到弹幕 # go-common\app\service\live\live-dm\service\v1\send.go 'DANMU_MSG': __danmu_msg_callback, # 有人送礼 'SEND_GIFT': __send_gift_callback, # 有人上舰 'GUARD_BUY': __guard_buy_callback, # 醒目留言 'SUPER_CHAT_MESSAGE': __super_chat_message_callback, # 删除醒目留言 'SUPER_CHAT_MESSAGE_DELETE': __super_chat_message_delete_callback, }#大括号括出的是字典，中括号括出的是列表 # 忽略其他常见cmd for cmd in IGNORED_CMDS: _CMD_CALLBACK_DICT[cmd] = None del cmd async def handle(self, client: client_.BLiveClient, command: dict): cmd = command.get('cmd', '') pos = cmd.find(':') # 2019-5-29 B站弹幕升级新增了参数 if pos != -1: cmd = cmd[:pos] if cmd not in self._CMD_CALLBACK_DICT: # 只有第一次遇到未知cmd时打日志 if cmd not in logged_unknown_cmds: logger.warning('room=%d unknown cmd=%s, command=%s', client.room_id, cmd, command) logged_unknown_cmds.add(cmd) return callback = self._CMD_CALLBACK_DICT[cmd] if callback is not None: await callback(self, client, command) async def _on_heartbeat(self, client: client_.BLiveClient, message: models.HeartbeatMessage): """ 收到心跳包（人气值） """ async def _on_danmaku(self, client: client_.BLiveClient, message: models.DanmakuMessage): """ 收到弹幕 """ async def _on_gift(self, client: client_.BLiveClient, message: models.GiftMessage): """ 收到礼物 """ async def _on_buy_guard(self, client: client_.BLiveClient, message: models.GuardBuyMessage): """ 有人上舰 """ async def _on_super_chat(self, client: client_.BLiveClient, message: models.SuperChatMessage): """ 醒目留言 """ async def _on_super_chat_delete(self, client: client_.BLiveClient, message: models.SuperChatDeleteMessage): """ 删除醒目留言 """参考链接 Python - typing 模块 —— Callable Python:Optional和带默认值的参数 ython编程中NotImplementedError的使用方法[Python模块学习——logging](https://www.cnblogs.com/captain_jack/archive/2011/01/21/1941453.html) [Python日志库logging总结-可能是目前为止将logging库总结的最好的一篇文章](https://cloud.tencent.com/developer/article/1354396#:~:text=2%E3%80%81logging%20%E6%B5%81%E7%A8%8B%201%20%E5%88%A4%E6%96%AD%20Logger%20%E5%AF%B9%E8%B1%A1%E5%AF%B9%E4%BA%8E%E8%AE%BE%E7%BD%AE%E7%9A%84%E7%BA%A7%E5%88%AB%E6%98%AF%E5%90%A6%E5%8F%AF%E7%94%A8%EF%BC%8C%E5%A6%82%E6%9E%9C%E5%8F%AF%E7%94%A8%EF%BC%8C%E5%88%99%E5%BE%80%E4%B8%8B%E6%89%A7%E8%A1%8C%EF%BC%8C%E5%90%A6%E5%88%99%EF%BC%8C%E6%B5%81%E7%A8%8B%E7%BB%93%E6%9D%9F%E3%80%82%202%20%E5%88%9B%E5%BB%BA,%E5%A6%82%E6%9E%9C%E4%BC%A0%E5%85%A5%E7%9A%84%20Handler%20%E5%A4%A7%E4%BA%8E%20Logger%20%E4%B8%AD%E8%AE%BE%E7%BD%AE%E7%9A%84%E7%BA%A7%E5%88%AB%EF%BC%8C%E4%B9%9F%E5%8D%B3%20Handler%20%E6%9C%89%E6%95%88%EF%BC%8C%E5%88%99%E5%BE%80%E4%B8%8B%E6%89%A7%E8%A1%8C%EF%BC%8C%E5%90%A6%E5%88%99%EF%BC%8C%E6%B5%81%E7%A8%8B%E7%BB%93%E6%9D%9F%E3%80%82%20%E6%9B%B4%E5%A4%9A%E9%A1%B9%E7%9B%AE) [Python模块和包](http://c.biancheng.net/python/module_package/) Python 中下划线的 5 种含义

Python Async/Await入门 本文主要参照知乎： Python Async/Await入门指南简要解释await的意义一句话版本：asyncio.wait会通过事件循环串联起多个异步函数(协程)，在异步函数中会有需要await的对象，多个函数同时去调用await后的运算，一旦任意await得到结果，就去处理后面的逻辑这篇文章评论下的一位同志看了几篇文章，说一下拙见，用这个await的场景大概是你有一堆异步函数需要调用，比如washing1 ,washing2,washing3,三个里面都有await，在主函数里调用这三个异步函数，然后三个函数都启动了，哪个await等到结果了，线程就去处理后面的逻辑，他们一般会这样写这样实际执行时间是5秒多，而不是3 +2 +5。感觉还是要配合异步模块好一点# async def washing1(): await asyncio.sleep(3) # 使用 asyncio.sleep(), 它返回的是一个可等待的对象 print('washer1 finished') async def washing2(): await asyncio.sleep(2) print('washer2 finished') async def washing3(): await asyncio.sleep(5) print('washer3 finished') # 1. 创建一个事件循环 loop = asyncio.get_event_loop() # 2. 将异步函数加入事件队列 tasks = [ washing1(), washing2(), washing3(), ] # 3. 执行事件队列, 直到最晚的一个事件被处理完毕后结束 loop.run_until_complete(asyncio.wait(tasks))事件循环事件循环参考文档 事件循环是每个 asyncio 应用的核心。 事件循环会运行异步任务和回调，执行网络 IO 操作，以及运行子进程。 asyncio.get_running_loop()返回当前 OS 线程中正在运行的事件循环。 asyncio.get_event_loop()获取当前事件循环。 如果当前 OS 线程没有设置当前事件循环，该 OS 线程为主线程，并且 set_event_loop() 还没有被调用，则 asyncio 将创建一个新的事件循环并将其设为当前事件循环。 知识补充Python random模块和函数Python中的random模块用于生成随机数。import random random.random() print( random.randint(1,10) ) # 产生 1 到 10 的一个整数型随机数 print( random.random() ) # 产生 0 到 1 之间的随机浮点数 print( random.uniform(1.1,5.4) ) # 产生 1.1 到 5.4 之间的随机浮点数，区间可以不是整数 print( random.choice('tomorrow') ) # 从序列中随机选取一个元素 print( random.choice('abcdefghijklmnopqrstuvwxyz!@#$%^&*()')# 随机字符 print( random.sample('zyxwvutsrqponmlkjihgfedcba',5)# 多个字符中生成指定数量的随机字符： print( random.randrange(1,100,2) ) # 生成从1到100的间隔为2的随机整数 a=[1,3,5,6,7] # 将序列a中的元素顺序打乱 random.shuffle(a) # 将序列a中的元素顺序打乱