话说学习编码最好的方式就是读优秀的代码和写代码。读书破万卷，下笔如有神!，俗话还说：熟读唐诗三百首，不会作诗也会吟!，那我把大牛们的代码读几遍，是不是也会写出牛光闪闪的代码呢？
又扯远了，言归正传，昨天给大家介绍了一下SteemData Notify，觉得是个挺有意思的产品：
今天呢，就来学习一下它的代码，看看具体是如何实现相关功能的。
代码功能模块 代码托管在github上，地址是： https://github.com/SteemData/notify.steemdata.com
通过简单分析，可以知晓代码分为应用(网站)端和后端
应用(网站)端 src/app.py
后端 src/worker.py
应用端： 使用Python + Flask + MongoDB
我们这节着重分析后端的代码
后端代码 源码在这里https://github.com/SteemData/notify.steemdata.com/blob/master/src/worker.py
通过阅读main函数
blockchain worker
confirmation worker
notifier worker
顾名思义，分别区块链工作进程、确认进程、通知进程， 分别通过不同的命令行参数启动。
这篇文章我们主要学习三者之一的Blockchain Worker
Blockchain Worker 函数名： run_blockchain_worker()
1 2 3 4 5 6 7 8 9 10 11 12 13 14 b = Blockchain() types = [ 'account_update', 'change_recovery_account', 'request_account_recovery', 'transfer', 'transfer_from_savings', 'set_withdraw_vesting_route', 'withdraw_vesting', 'fill_order', 'fill_convert_request', 'fill_transfer_from_savings', 'fill_vesting_withdraw', ]
b = Blockchain() 定义了Blockchain 实例stream(self, filter_by: Union[str, list] = list(), *args, **kwargs)
types = [ xxxx]
有了上述对Blockchain以及filter_by的讲解，下面的代码就很好理解了
1 2 3 4 5 6 7 8 9 10 11 12 try: block = db.last_processed_block.find_one() start_block = int(block['block_num']) - 1 except Exception: start_block = None for op in b.stream(filter_by=types, start_block=start_block): processed = db.processed_blockchains.find({'_id': op['_id']}).count() if not processed: if parse_blockchain(op): db.processed_blockchains.insert_one(op) db.last_processed_block.delete_many({}) db.last_processed_block.insert_one(op)
大致就是，把我们关心的操作读取回来，并且写到数据库中。
processed = db.processed_blockchains.find({'_id': op['_id']}).count()
parse_blockchain(op)
db.processed_blockchains.insert_one(op)
db.last_processed_block.delete_many({})db.last_processed_block.insert_one(op)
parse_blockchain(op) 前文我们说过，这个来获取我们关心的操作，这个是如何实现的呢？
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 def parse_blockchain(op): settings = None message = None if op['type'] == 'account_update': settings = find_user_settings(op['account']) if settings and settings['account_update']: message = 'Received event: account_update (%s)' % op['account'] elif op['type'] in ['transfer', 'transfer_from_savings']: settings = find_user_settings(op['from']) if settings and settings[op['type']]: message = 'Received event: %s\nEvent detail: %s -> %s (%s)' % ( op['type'], op['from'], op['to'], op['amount'], ) .... if settings and message: db.notifications.insert_one({ 'username': settings['username'], 'email': settings['email'], 'telegram_channel_id': settings['telegram_channel_id'], 'message': message, 'email_sent': False, 'telegram_sent': False, 'created_at': datetime.utcnow(), }) return True
嗯，我好像发现了一个了不起的BUG，一会再说。
1 2 3 4 5 6 def find_user_settings(username): try: rows = db.settings.find({'username': username, 'confirmed': True}).sort('created_at', -1) return rows[0] except Exception: return dict()
这个函数的功能，就是按用户名，读取用户（已确认)的设置并返回。
1 2 3 4 if op['type'] == 'account_update': settings = find_user_settings(op['account']) if settings and settings['account_update']: message = 'Received event: account_update (%s)' % op['account']
这样，这段代码就好理解了，按白话文翻译就是
因为操作只能是这些操作中的一种，所以要逐一判断一下，直到发现或者判断完毕没有发现为止。
1 2 3 4 5 6 7 8 9 10 11 12 if settings and message: db.notifications.insert_one({ 'username': settings['username'], 'email': settings['email'], 'telegram_channel_id': settings['telegram_channel_id'], 'message': message, 'email_sent': False, 'telegram_sent': False, 'created_at': datetime.utcnow(), }) return True
如果发现了需要处理的操作，将其插入notifications 表。
BUG，BUG
你问我后果是啥？貌似没啥后果，浪费一些数据库空间和CPU运算能力而已。
总结
Blockchain Worker 将blockchain上和账户有关的动态抓取进来
判断是否是SteemData Notify 注册(并确认)用户相关的操作以及是否是用户关心的操作
如果是，写入数据库通知表notifications
咦，一总结好像也没啥内容呢，那我为啥写了这么多呢？晕，这啰嗦的毛病要改一改了。
感谢阅读 / Thank you for reading.
This page is synchronized from the post: SteemData Notify 代码学习一： Blockchain Worker / Code Study of SteemData Notify: Part one
