前言
有段时间没写博客了，这段时间主要在做以下几件事情：
1、锻炼，周末的时候约了高中的好友去爬山。太久没锻炼了，累趴下了😅，还是要多练练。
2、无意间看到了雷公的视频，对我触动很大，不论是一级还是二级市场，都需要注重心态、技术。
3、趁着Gas低，交互了几个项目。
4、上班😄不上班没饭吃，肚子饿
正题
最开始有这个想法来源于：想把tg上订阅的Foresight News的bot消息转发到微信，这样有几个好处：
1、上班可以光明正大的看快讯
2、微信相当于聚合器，接收多路的消息。只要打开微信，就能收到我关注的消息，而不是群聊的垃圾信息。
…
但是目前还没搞定，tg的教程没怎么看明白。网上的教程基本都是微信转发到tg的，因为tg的服务器上文件是永久保存的，比化腾良心多了。
刚好推特上关注了@Whale Alert，这是一个产品，专门监控链上巨鲸大额转账的，刚好有开放的API（一定是特别的缘分），思路就是：申请免费api（虽然有限制，但是要啥自行车）—— 将返回的结果发送到企业微信（个人微信消息推送挺严格的，所以另辟蹊径，反正两者是可以绑定的）
我关注了很多大V，当看一个工具好不好用的时候，我通常会看共同关注，这些大v消息肯定是比我灵敏多了。
@Whale Alert API文档，其中免费账号的限制的内容。
https://docs.whale-alert.io/#introduction
直接上代码吧。都写注释了
先pip安装依赖，其他都自带了。就这两个
pip install ntplib
pip install requests
main.py
# -*- coding : utf-8 -*-
import os
import time
import ntplib
import json
import configparser
import requests
from send_msgs_to_wechat import SendWeiXinWork
config_path = os.path.join("properties.conf")
# 获取配置文件对象
cf = configparser.ConfigParser()
# 读取配置
cf.read(config_path, encoding='utf-8')
# 本地测试，开代理
os.environ["http_proxy"] = "http://127.0.0.1:10810"
os.environ["https_proxy"] = "http://127.0.0.1:10810"
# 创建发送消息对象
SendWeiXinWork = SendWeiXinWork()
# 从ntp服务器获取时间。防止运行的机器时间不对
def get_timestamp_from_ntp():
ntp_url = 'pool.ntp.org'
# noinspection PyBroadException
try:
ntp_client = ntplib.NTPClient()
response_outside = ntp_client.request(ntp_url).tx_time
start_timestamp = round(response_outside)
return start_timestamp
except Exception as exc:
return round(time.time())
def main():
# 发送警报的数组
msgs = []
# 时间阈值
time_windows = 60
# e.g https://api.whale-alert.io/v1/transactions?api_key='申请的key'&min_value=500000&start=1655707296
# 警报的最小价值
min_value = 5000000
api_key = cf.get("Whale Alert", "api_key")
base_url = 'https://api.whale-alert.io/v1/transactions?'
base_scan_url = 'https://whale-alert.io/transaction'
# 从ntp服务器获取时间。防止运行的机器时间不对
# 减去窗口时间time_windows，相当于从前一分钟的时间作为开始时间
start_timestamp = get_timestamp_from_ntp() - time_windows
print(start_timestamp)
# 拼接需要访问的url和参数。
url = base_url + 'api_key' + '=' + api_key + '&' + 'min_value' + '=' + str(min_value) + '&' + 'start' + '=' + str(
start_timestamp)
response = requests.get(url)
resp_dict = json.loads(response.text)
# 判断状态是否是success并且数据条数要大于0
if 'success'.__eq__(resp_dict['result']) and resp_dict['count'] > 0:
transactions = resp_dict['transactions']
if len(transactions) > 0:
for transaction in transactions:
# 在哪条链上
blockchain = transaction['blockchain']
# 代币名称
symbol = transaction['symbol']
# 交易类型
transaction_type = transaction['transaction_type']
# 交易hash
transaction_hash = transaction['hash']
# 发送地址
from_address = transaction['from']['address']
# 发送地址所属
from_owner = transaction['from'].get('owner', 'Unknown')
# 发送地址所属类型
from_owner_type = transaction['from'].get('owner_type', 'Unknown')
# 接收地址
to_address = transaction['to']['address']
# 接收地址所属
to_owner = transaction['to'].get('owner', 'Unknown')
# 接收地址所属类型
to_owner_type = transaction['to'].get('owner_type', 'Unknown')
# 交易时间戳
transaction_timestamp = transaction['timestamp']
transaction_date = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(transaction_timestamp))
# 币本位总数
transaction_amount = str(transaction['amount']) + ' ' + symbol
# 以usd计算价值
transaction_amount_usd = str(transaction['amount_usd']) + ' ' + 'usd'
# 转账次数
transaction_count = transaction['transaction_count']
msg = '区块链名称' + ': ' + blockchain + '\n' + '符号' + ': ' + symbol + '\n' + '交易类型' + ': ' + transaction_type + '\n' + '交易hash' + ': ' + transaction_hash + '\n' + '交易发起者地址' + ': ' + from_address + '\n' + '交易发起者地址所属' + ': ' + from_owner + '\n' + '交易发起者地址所属类型' + ': ' + from_owner_type + '\n' + '交易接收地址' + ': ' + to_address + '\n' + '交易接收地址所属' + ': ' + to_owner + '\n' + '交易接收地址所属类型' + ': ' + to_owner_type + '\n' + '交易发生时间' + ': ' + transaction_date + '\n' + '币本位总额' + ': ' + str(transaction_amount) + '\n' + '币本位价值usd总额' + ': ' + str(transaction_amount_usd) + '\n' + '交易转账次数' + ': ' + str(transaction_count) + '\n' + '区块链浏览器详情' + ': ' + base_scan_url + '/' + blockchain + '/' + transaction_hash
msgs.append(msg)
if len(msgs) > 0:
# 发送到微信
for msg in msgs:
# 1 代表往企业总群发
SendWeiXinWork.send_message("1", str(msg))
print(msg)
if __name__ == '__main__':
main()
注：代码里面当转账数额 > 500K 才会预警，要修改的话，改# 警报的最小价值 min_value = 5000000 这个参数。免费账号最小是500K。
send_msgs_to_wechat.py
# -*- coding : utf-8 -*-
# @DATE:2022/6/20 16:47
import requests, configparser, os
config_path = os.path.join("properties.conf")
# 获取配置文件对象
cf = configparser.ConfigParser()
# 读取配置
cf.read(config_path, encoding='utf-8')
# os.environ["http_proxy"] = "http://127.0.0.1:10810"
# os.environ["https_proxy"] = "http://127.0.0.1:10810"
class SendWeiXinWork():
def __init__(self):
self.CORP_ID = cf.get("Wei Xin", "CORP_ID") # 企业号的标识
self.SECRET = cf.get("Wei Xin", "SECRET") # 管理组凭证密钥
self.AGENT_ID = cf.get("Wei Xin", "AGENT_ID") # 应用ID
self.token = self.get_token()
def get_token(self):
url = "https://qyapi.weixin.qq.com/cgi-bin/gettoken"
data = {
"corpid": self.CORP_ID,
"corpsecret": self.SECRET
}
req = requests.get(url=url, params=data)
res = req.json()
if res["errmsg"] == "ok":
return res["access_token"]
else:
return res
def send_message(self, to_user, content):
url = "https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token=%s" % self.token
data = {
# "touser": to_user, # 发送个人就填用户账号
"toparty": to_user, # 发送组内成员就填部门ID
"msgtype": "text",
"agentid": self.AGENT_ID,
"text": {"content": content},
"safe": "0"
}
req = requests.post(url=url, json=data)
res = req.json()
if res["errmsg"] == "ok":
print("send message sucessed")
return "send message sucessed"
else:
return res
# if __name__ == '__main__':
# # 创建发送消息对象
# SendWeiXinWork = SendWeiXinWork()
# SendWeiXinWork.send_message("1", 'a' + '\n' + 'b')
properties.conf
这里面主要放申请的key什么的。
[Whale Alert]
api_key = '你申请的key'
[Wei Xin]
XXX = ''
注：关于企业微信的申请和发送消息。我就不写了，也是参考下面这位大佬的。
https://www.jb51.net/article/244955.htm
效果图
后面就是部署到服务器上，然后定时调度就行了。这个教程后面出，今天还有其他事情要忙。
需要提的是：
1、CentOS7 上默认安装的python版本是2.X的。本篇中用的是3.7，所以需要重新软链接到python3。切记不要把2.X删掉，要不然yum就用不了了
2、调度器用自带的crontab 就行。