python 监控微博用户更新+钉钉提醒
背景说明：1.近期关注某微博博主，但又不想频繁开微博；
2.日常办公软件为钉钉；
于是思考，能否关注博主更新， 一旦有更新即发送到钉钉（一开始这个项目起名叫上班摸鱼工具。。）
因为自己刚刚起步，能力有限，希望各位前辈多指点（里面也引用其他人的轮子，自己其实做的并不多）。
言归正传。分三部分介绍：
1.钉钉设置；
2.微博监控；
3.项目打包；
1.钉钉设置：
首先需要新建一个钉钉机器人用于信息推送；
钉钉-新建群-设置-智能群助手-自定义机器人（获取Webhook+secret）
2.微博监控：
市面上有几种方案，一种是直接通过不登录访问博主主页，将每次访问后的微博ID 存储到本地，定时访问查看是否有更新，由于我获得的数据比较乱，后面有其他方案我就放弃了；
第二种方案，模拟登录---获取关注用户列表----选择想要关注的博主--- 定时刷新，每次记录更新的ID，如果有则提示发布到钉钉（这里我引用的是Charles
的项目进行修改，整个监控+判断逻辑基本都是在这上面进行修改（我忘了 原项目地址，只知道微信公众号：Charles的皮卡丘，在此感谢） 代码功能说明如下：
3.打包执行文件：
为了能够直接双击运行省事点， 研究了市面上py - exe 的方案；最后找到 pyinstaller 这个模块；
需要注意，py3.8 安装的pyinstaller 打包会报错缺失模块，我换3.6.5版本就好了。
具体下载安装过程就不细说了， 到pyinstaller 目录下
把你的项目放到当前文件下
打开cmd终端，cd 到当前目录
执行pyinstaller -F xx.py 最后的xx.py 就是你项目 名字应该会看到一堆提示语，最后成功，在dist文件夹下就可以看到你的执行程序；
实际运行效果如下：
总结：
这个小项目还是有很多不好的地方，例如输入用户名、密码，让人觉得不好接受，如果可以做到不需要登录就监控应该会放心一些。第一个项目，请各位轻拍。
如下是项目源码：
'''
Function:
微博监控、钉钉提醒
业务流程说明：
登录微博--> 获取关注的用户列表--> 选择要关注的用户 --> session 记录关注用户的微博Id--> 定时刷新获取微博Id --> 如有新的则将最新的微博发布到钉钉
Author:
YQY（微信公众号：产品石匠），基于Charles的项目修改（微信公众号：Charles的皮卡丘）
'''
import re
from DecryptLogin import login
import hmac
import hashlib
import base64
import urllib.parse
import requests
import json,time,datetime
''' 钉钉监控 '''
def _dingding(token,sec,data):
data = data#微博数据
#钉钉设置
robot_id = token# token
timestamp = str(round(time.time() * 1000))
secret = sec #密钥
secret_enc = secret.encode('utf-8')
string_to_sign = '{}\n{}'.format(timestamp, secret)
string_to_sign_enc = string_to_sign.encode('utf-8')
hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
#调用钉钉API
url = 'https://oapi.dingtalk.com/robot/send?access_token=' + robot_id + '×tamp=' + timestamp + '&sign=' + sign
headers = {"Content-Type": "application/json ;charset=utf-8 "}
status = requests.post(url, json.dumps(data), headers=headers)
'''微博监控'''
class wbMonitor():
def __init__(self, username, password, time_interval, **kwargs):
_, self.session = login.Login().weibo(username, password, 'mobile')
self.headers = {
'Accept': 'application/json, text/plain, */*',
'Accept-Encoding': 'gzip, deflate, br',
'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
'Connection': 'keep-alive',
'Host': 'm.weibo.cn',
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/73.0.3683.86 Safari/537.36'
}
self.api_url = 'https://m.weibo.cn/api/container/getIndex?uid={}&luicode=10000011&lfid=231093_-_selffollowed&type=uid&value={}&containerid={}'
self.format_profile_url = 'https://m.weibo.cn/u/{}?uid={}&luicode=10000011&lfid=231093_-_selffollowed'
self.time_interval = time_interval
'''开始监控'''
def start(self, user_id=None):
if not user_id:
followed = self.__getFollowed()
print('未指定想要监控的用户ID, 您关注的用户有:\n(是否想选择其中一位进行监控?)')
print('-' * 40)
for idx, each in enumerate(sorted(followed.keys())):
print('[%d]. %s' % (idx+1, each))
print('-' * 40)
while True:
user_choice = input('请选择您想要监控的用户编号(例如1):')
try:
profile_url = followed]
user_id = re.findall(r'uid=(\d+)&', profile_url)
break
except:
print('您的输入有误, 请重新输入.')
else:
profile_url = self.format_profile_url.format(user_id, user_id)
self.__monitor(user_id, profile_url)
'''监控用户主页'''
def __monitor(self, user_id, profile_url):
user_name, containerid = self.__getContainerid(user_id, profile_url)
res = self.session.get(self.api_url.format(user_id, user_id, containerid))
weibo_ids = []
cards = res.json()['data']['cards']
for card in cards:
if card['card_type'] == 9:
weibo_ids.append(str(card['mblog']['id']))
while True:
weibo_ids = self.__checkUpdate(user_id, profile_url, weibo_ids)
time.sleep(self.time_interval)
'''检查用户是否有新的微博'''
def __checkUpdate(self, user_id, profile_url, weibo_ids):
user_name, containerid = self.__getContainerid(user_id, profile_url)
res = self.session.get(self.api_url.format(user_id, user_id, containerid))
cards = res.json()['data']['cards']
flag = False
for card in cards:
if card['card_type'] == 9:
if str(card['mblog']['id']) not in weibo_ids:
flag = True
weibo_ids.append(str(card['mblog']['id']))
print(str(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))) + ': 用户<%s>发布了新微博' % user_name)
pics = []
if card['mblog'].get('pics'):
for i in card['mblog']['pics']:
pics.append(i['url'])
pics = '||'.join(pics)
# print('[时间]: %s\n[来源]: %s\n[原文作者]: %s\n[内容]: %s\n[图片链接]： %s\n' %
# (card['mblog']['created_at'], card['mblog']['source'], card['mblog']['user']['screen_name'], card['mblog']['text'], pics))
text = "更新时间："+ card['mblog']['created_at']+ "\n 作者:"+ card['mblog']['user']['screen_name'] + "\n 内容:"+ card['mblog']['text'] + " \n 图片：" + pics
print(text)
''' 发送到钉钉 '''
data = {
"msgtype": "text",
"text": {
"content": text
},
"at": {
"atMobiles": [
"131XXX",
"189xxxx8325"
],
"isAtAll": 0
}
}
_dingding(token, sec, data)
if not flag:
print(str(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))) + ': 用户<%s>未发布新微博' % user_name)
return weibo_ids
'''获取containerid'''
def __getContainerid(self, user_id, profile_url):
self.session.get(profile_url)
containerid = re.findall(r'fid%3D(\d+)%26', str(self.session.cookies))
res = self.session.get(self.api_url.format(user_id, user_id, containerid))
user_name = self.__decode(re.findall(r'"screen_name":"(.*?)"', res.text))
for i in res.json()['data']['tabsInfo']['tabs']:
if i['tab_type'] == 'weibo':
containerid = i['containerid']
return user_name, containerid
'''获取关注列表'''
def __getFollowed(self):
data = {}
page = 0
while True:
page += 1
res = self.session.get('https://m.weibo.cn/api/container/getIndex?containerid=231093_-_selffollowed&page={}'.format(page), headers=self.headers)
profile_urls = re.findall(r'"profile_url":"(.*?)"', res.text)
screen_names = re.findall(r'"screen_name":"(.*?)"', res.text)
if len(profile_urls) == 0:
break
for screen_name, profile_url in zip(screen_names, profile_urls):
data = profile_url.replace('\\', '')
return data
'''解码'''
def __decode(self, content):
return content.encode('latin-1').decode('unicode_escape')
'''run'''
print("监控之前，我们需要做一些简单的设置！")
time.sleep(3)
if __name__ == '__main__':
import argparse
parser = argparse.ArgumentParser(description="微博监控")
parser.add_argument('-u', dest='username', help='用户名')
parser.add_argument('-p', dest='password', help='密码')
parser.add_argument('-i', dest='id', help='待监控用户id')
parser.add_argument('-t', dest='time_interval', default=1800, type=int, help='监控的时间间隔')
args = parser.parse_args()
username = input("输入微博用户名：")
psw = input("微博密码（没有后门，放心）：")
token = input("钉钉机器人的token id 是：")
sec = input("钉钉机器人密钥：")
print('开始采集...接下来将会加载所有关注的用户，请选择一个想要监控的用户')
time.sleep(3)
wb = wbMonitor(username=username, password=psw, time_interval=args.time_interval)
wb.start(args.id);
这个可以和腾讯云结合实现定时提醒
具体方法可参考我的这篇帖子（将bdtb的代码换为楼主的代码，并自定义运行时间，其他操作类似）：https://www.52pojie.cn/forum.php?mod=viewthread&tid=1155287&extra=page%3D1%26filter%3Dtypeid%26typeid%3D29 开始采集...接下来将会加载所有关注的用户，请选择一个想要监控的用户
Traceback (most recent call last):
File "c:\Users\Dell\AppData\Local\Programs\Python\Python39\Scripts\cs.py", line 170, in <module>
wb = wbMonitor(username=username, password=psw, time_interval=args.time_interval)
File "c:\Users\Dell\AppData\Local\Programs\Python\Python39\Scripts\cs.py", line 39, in __init__
_, self.session = login.Login().weibo(username, password, 'mobile')
File "C:\Users\Dell\AppData\Local\Programs\Python\Python39\lib\site-packages\DecryptLogin\core\weibo.py", line 348, in login
return selected_api.login(**args)
File "C:\Users\Dell\AppData\Local\Programs\Python\Python39\lib\site-packages\DecryptLogin\core\weibo.py", line 204, in login
raise RuntimeError(response_json.get('msg'))
RuntimeError: 系统错误，请稍后再试
请问这怎么解决 思路可以 没看代码，你说下频率是多久呢 昨天打包才失败, 不过是个好帖子, 以后应该用得到, 收藏了 感谢分享 用BARK提醒不香吗 本帖最后由 resu 于 2020-5-15 10:29 编辑
有参考价值！{:1_921:}
可以用TK做个简单GUI,设置存放于json就更方便了。 不登陆微博也可以的，微博有个访客系统，百度了解一下。
非常感谢楼主分享 这个操作困难吗