您好,欢迎光临,本网站提供SEO关键词快速排名、网站优化服务。

互联网服务

网页制作长尾词核心词提升上词率平台

微博关键词监控器

作者:jcmp      发布时间:2021-05-05      浏览量:0
微博关键词监控器2020年4月2日阴雨连

微博关键词监控器

2020年4月2日阴雨连绵的下午,被疫情所困无法返校开学的我百无聊赖,突然想起之前抖音上刷到一个霓虹国程序员做了一个分手灯,就是一个监控社交网络,有人说分手就会亮一下的灯,我感觉十分有趣,于是决定自己也做一个。但是用计算机控制灯比较麻烦,比较之下我选择了较为方便的手机短信,于是一个沙雕脚本诞生了。

这个沙雕脚本使用爬虫技术监控微博的动态,当搜索关键词有新结果时便会以短信的方式予以通知,而且可以自行定义通知间隔,可以说是非常人(没)性(有)化(用)。

下面我将分模块介绍一下这个只有一百余行的python脚本。

短信模块

短信我们使用 https://www.twilio.com/ 提供的服务,他们家是有免费试用的,有钱的老爷们也可以使用其他供应商的服务,应该用起来都类似。首先在网站注册账号,然后绑定手机,将密钥信息填进API里就可以调用起来了,十分简洁方便。

# 配置auth_token = '**********************'account_sid = '***********************'client = Client(account_sid, auth_token)# 然后调用client.messages.create就可以了,下边有写成一个工具函数。

主程序部分

构建待访问池

def get_page(): # 先用requests构造请求,解析出关键词搜索出来的微博总页数 data_list = [] # @return,每次连接的具体参数 data = { 'containerid': '100103type=61&q={}'.format(kw), # 排序方式和搜索关键词 'page_type': 'searchall'} # 搜索方式 resp = requests.get(url=url, headers=headers, params=data) total_page = resp.json()['data']['cardlistInfo']['total'] # 微博总数 # 一页有10条微博,用总数对10整除,余数为0则页码为总数/10,余数不为0则页码为(总数/10)+1 if total_page % 10 == 0: page_num = int(total_page / 10) else: page_num = int(total_page / 10) + 1 # 页码为1,data为当前data,页码不为1,通过for循环构建每一页的data参数 if page_num == 1: data_list.append(data) return data_list else: for i in range(1, page_num + 1): data['page'] = i data_list.append(copy.deepcopy(data)) return data_list。

工具函数

def check_time(id, last_pool, time): # 检查是否被通知过 if time != "刚刚": # 我们只通知最近发出的微博(30s间隔刷新),所以所有不是“刚刚”的微博都不需要通知 return False for pool in last_pool: # 已经通知过的不用通知 if id in pool: return False return Truedef sent_message(text, phone_number): mes = client.messages.create( from_='*********', # 填写在active number处获得的号码 body=text, to=phone_number ) print("msg sent: {}".format(mes.sid))。

更新

def renew(last_pool): data_list = get_page() # 接收data参数列表 pool = [] # 本次连接的已访问池 for data in data_list: with get(url=url, headers=headers, params=data) as resp: # 携带参数发送请求 text = resp.text text_dict = json.loads(text)['data']['cards'] # 微博页面上的数据是以卡片的形式组织的 parse_dict = {} # 解析出的数据放在这个字典里边 for card in text_dict: if card['card_type'] == 9: # 只有这个类型的卡片是记录微博数据的,其他的设计渲染、布局等不用关心的东西 scheme = card['scheme'] # 卡片中提取信息 if card['mblog']['isLongText'] is False: text = card['mblog']['text'] text = re.sub(r'<.*?>|\n+', '', text) else: text = card['mblog']['longText']['longTextContent'] user = card['mblog']['user']['profile_url'] id = card['mblog']['id'] time = card['mblog']['created_at'] # 检查此条微博是否已经被通知 if not check_time(id, last_pool, time) or id in pool: text = "{} new wblog published recently\n".format(len(pool)) print(text) if len(pool) > 0: sent_message(text, "+86***********") # 调用发信息方法 return pool parse_dict['url'] = scheme parse_dict['text'] = text parse_dict['author'] = user parse_dict['time'] = time pool.append(id) # print('new weibo detected:\n' # 'author home page: {}\n' # 'content: {}\n' # 'time: {}\n'.format(parse_dict['author'], parse_dict['text'], parse_dict['time'])) text = "{} new wblog published recently\n".format(len(pool)) print(text) if len(pool) > 0: sent_message(text, "+86***********") return pool。

完整代码

import copyimport timeimport requestsimport reimport jsonfrom requests import getfrom twilio.rest import Clientauth_token = '**************************'account_sid = '************************'client = Client(account_sid, auth_token)def get_page(): """ 先用requests构造请求,解析出关键词搜索出来的微博总页数 :return: 返回每次请求需要的data参数 """ data_list = [] data = { 'containerid': '100103type=61&q={}'.format(kw), 'page_type': 'searchall'} resp = requests.get(url=url, headers=headers, params=data) total_page = resp.json()['data']['cardlistInfo']['total'] # 微博总数 # 一页有10条微博,用总数对10整除,余数为0则页码为总数/10,余数不为0则页码为(总数/10)+1 if total_page % 10 == 0: page_num = int(total_page / 10) else: page_num = int(total_page / 10) + 1 # 页码为1,data为当前data,页码不为1,通过for循环构建每一页的data参数 if page_num == 1: data_list.append(data) return data_list else: for i in range(1, page_num + 1): data['page'] = i data_list.append(copy.deepcopy(data)) return data_listdef check_time(id, last_pool, time): if time != "刚刚": return False for pool in last_pool: if id in pool: return False return Truedef sent_message(text, phone_number): mes = client.messages.create( from_='+**********', # 填写在active number处获得的号码 body=text, to=phone_number ) print("msg sent: {}".format(mes.sid))def renew(last_pool): data_list = get_page() # 接收data参数列表 pool = [] for data in data_list: with get(url=url, headers=headers, params=data) as resp: # 携带参数发送请求 text = resp.text # await 等待知道获取完整数据 text_dict = json.loads(text)['data']['cards'] parse_dict = {} for card in text_dict: if card['card_type'] == 9: scheme = card['scheme'] if card['mblog']['isLongText'] is False: text = card['mblog']['text'] text = re.sub(r'<.*?>|\n+', '', text) else: text = card['mblog']['longText']['longTextContent'] user = card['mblog']['user']['profile_url'] id = card['mblog']['id'] time = card['mblog']['created_at'] if not check_time(id, last_pool, time) or id in pool: text = "{} new wblog published recently\n".format(len(pool)) print(text) if len(pool) > 0: sent_message(text, "+86*********") return pool parse_dict['url'] = scheme parse_dict['text'] = text parse_dict['author'] = user parse_dict['time'] = time pool.append(id) # print('new weibo detected:\n' # 'author home page: {}\n' # 'content: {}\n' # 'time: {}\n'.format(parse_dict['author'], parse_dict['text'], parse_dict['time'])) text = "{} new wblog published recently\n".format(len(pool)) print(text) if len(pool) > 0: sent_message(text, "+86***********") return pooldef main(): """ 开启任务循环 :return: None """ last_pool = [[], []] while True: print("\nnew detect launched") new_pool = renew(last_pool) last_pool = last_pool[1:] last_pool.append(new_pool) time.sleep(30)if __name__ == '__main__': kw = input("Please input the keyword you wanna overwatch: ") headers = { 'user-agent': 'Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/534.57.2 (KHTML, like Gecko) Version/5.1.7 Safari/534.57.2' } url = 'https://m.weibo.cn/api/container/getIndex' main()。