Python系列之——利用Python實現微博監控

01-30

0x00 前言:

前幾個星期在寫一個微博監控系統可謂是一波三折啊獲取到微博後因為一些字元編碼問題導致心態爆炸開發中斷但是就在昨天發現了另外一個微博的介面

一個手機微博的介面歡迎登錄 - 新浪微博經過一番測試後認為這個介面滿足我的要求就繼續完成未完成的使命吧

0x01 分析:

這個介面直接訪問的話會有一個302跳轉到一個登陸界面

也就是說這裡需要先模擬登陸一下才可以訪問到微博

抓個包分析了一下

發現只要用戶名和密碼正確既返回200且json部分的retcode會返回20000000

少了驗證碼這一大坑那模擬登陸就相當簡單啦

登陸完後訪問用戶主頁例如:https://m.weibo.cn/u/3023940914

可以在審查元素的Network模塊看到這裡用了兩個xhr來載入用戶信息及微博信息

分別是

https://m.weibo.cn/api/container/getIndex?type=uid&amp;value=3023940914&amp;containerid=1005053023940914

https://m.weibo.cn/api/container/getIndex?type=uid&value=3023940914&containerid=1076033023940914

經過測試這個介面直接加上type和value參數訪問就相當於第一個介面不必加上containerid參數

而第二個介面的containerid參數則是通過第一個介面獲取的

獲取到第二個containerid參數訪問第二個介面就可以獲取到這個uid發布的微博了

返回的是json格式的數據用戶的微博信息都在cards列表裡每條數據的mblog數組裡面包括微博正文、圖片、來源與時間等

其中card_type標識的是微博類型例如:文字微博圖片微博視頻微博轉發等經過測試文字微博和圖片微博的card_type標識都一樣為9

這裡初步只開發監控文字和圖片微博的功能<del>其實就是懶</del>

0x02 開發

首先需要模擬登陸後續的操作都需要基於登陸的格調來進行也是需要在同個會話進行可以使用requests.session() 方法來完成

代碼片段:

reqHeaders = {n User-Agent: Mozilla/5.0 (Windows NT 6.1; WOW64; rv:54.0) Gecko/20100101 Firefox/54.0,n Content-Type: application/x-www-form-urlencoded,n Referer: https://passport.weibo.cn/signin/login,n Connection: close,n Accept-Language: zh-CN,zh;q=0.8,en-US;q=0.5,en;q=0.3n}nloginApi = https://passport.weibo.cn/sso/loginnloginPostData = {n username:userName,n password:passWord,n savestate:1,n r:,n ec:0,n pagerefer:,n entry:mweibo,n wentry:,n loginfrom:,n client_id:,n code:,n qq:,n mainpageflag:1,n hff:,n hfp:n}n#get user sessionnsession = requests.session()ntry:n r = session.post(loginApi,data=loginPostData,headers=reqHeaders)n if r.status_code == 200 and json.loads(r.text)[retcode] == 20000000:n #successfuln #do sometingn else:n #Logon failure!nexcept Exception as e:n #Logon failure!n

登陸完成後就可以拼接用戶id訪問前面說的第一個介面了

訪問完後再拼接containerid參數獲取微博信息的json數據

代碼片段:

#get user weibo containeridnuserInfo https://m.weibo.cn/api/container/getIndex?uid=%s&type=uid&value=%s%(wbUserId,wbUserId)ntry:n r = session.get(userInfo,headers=reqHeaders)n for i in r.json()[tabsInfo][tabs]:n if i[tab_type] == weibo:n conId = i[containerid]nexcept Exception as e:n #failure!n#get user weibo indexnweiboInfo = https://m.weibo.cn/api/container/getIndex?uid=%s&type=uid&value=%s&containerid=%s%(wbUserId,wbUserId,conId)ntry:n r = session.get(weiboInfo,headers=reqHeaders)n itemIds = [] #WBQueuen for i in r.json()[cards]:n if i[card_type] == 9:n itemIds.append(i[mblog][id])nexcept Exception as e:n #failure!n

這裡把所有獲取到的微博的id存起來後面繼續訪問是發現有新的微博id不在這個列表裡就證明是新發布的微博

代碼片段:

def startMonitor():n returnDict = {}n try:n r = session.get(weiboInfo,headers=reqHeaders)n for i in r.json()[cards]:n if i[card_type] == 9:n if str(i[mblog][id]) not in itemIds:n itemIds.append(i[mblog][id])n #Got a new weibon #@ return returnDict dictn returnDict[created_at] = i[mblog][created_at]n returnDict[text] = i[mblog][text]n returnDict[source] = i[mblog][source]n returnDict[nickName] = i[mblog][user][screen_name]n #if has photosn if i[mblog].has_key(pics):n returnDict[picUrls] = []n for j in i[mblog][pics]:n returnDict[picUrls].append(j[url])n return returnDictn except Exception as e:n #failure!n

將這些方法封裝成了一個類完整代碼如下

#!/usr/bin/env pythonn# -*- coding: utf-8 -*-n# Author : 奶權n# Action : 微博監控n# Desc : 微博監控主模塊nnimport requests,json,sysnfrom lxml import etreennclass weiboMonitor():n """n @ Class self :n """n def __init__(self, ):n self.session = requests.session()n self.reqHeaders = {n User-Agent: Mozilla/5.0 (Windows NT 6.1; WOW64; rv:54.0) Gecko/20100101 Firefox/54.0,n Content-Type: application/x-www-form-urlencoded,n Referer: https://passport.weibo.cn/signin/login,n Connection: close,n Accept-Language: zh-CN,zh;q=0.8,en-US;q=0.5,en;q=0.3n }nn """n @ Class self :n @ String userName : The username of weibo.cnn @ String passWord : The password of weibo.cnn """n def login(self, userName, passWord):n loginApi = https://passport.weibo.cn/sso/loginn loginPostData = {n username:userName,n password:passWord,n savestate:1,n r:,n ec:0,n pagerefer:,n entry:mweibo,n wentry:,n loginfrom:,n client_id:,n code:,n qq:,n mainpageflag:1,n hff:,n hfp:n }n #get user sessionn try:n r = self.session.post(loginApi,data=loginPostData,headers=self.reqHeaders)n if r.status_code == 200 and json.loads(r.text)[retcode] == 20000000:n self.echoMsg(Info,Login successful! UserId:+json.loads(r.text)[data][uid])n else:n self.echoMsg(Error,Logon failure!)n sys.exit()n except Exception as e:n self.echoMsg(Error,e)n sys.exit()nn """n @ Class self :n @ String wbUserId : The user you want to monitoredn """n def getWBQueue(self, wbUserId):n #get user weibo containeridn userInfo = https://m.weibo.cn/api/container/getIndex?uid=%s&type=uid&value=%s%(wbUserId,wbUserId)n try:n r = self.session.get(userInfo,headers=self.reqHeaders)n for i in r.json()[tabsInfo][tabs]:n if i[tab_type] == weibo:n conId = i[containerid]n except Exception as e:n self.echoMsg(Error,e)n sys.exit()n #get user weibo indexn self.weiboInfo = https://m.weibo.cn/api/container/getIndex?uid=%s&type=uid&value=%s&containerid=%s%(wbUserId,wbUserId,conId)n try:n r = self.session.get(self.weiboInfo,headers=self.reqHeaders)n self.itemIds = [] #WBQueuen for i in r.json()[cards]:n if i[card_type] == 9:n self.itemIds.append(i[mblog][id])n self.echoMsg(Info,Got weibos)n self.echoMsg(Info,Has %d id(s)%len(self.itemIds))n except Exception as e:n self.echoMsg(Error,e)n sys.exit()n """n @ Class self :n """n def startMonitor(self, ):n returnDict = {}n try:n r = self.session.get(self.weiboInfo,headers=self.reqHeaders)n for i in r.json()[cards]:n if i[card_type] == 9:n if str(i[mblog][id]) not in self.itemIds:n self.itemIds.append(i[mblog][id])n self.echoMsg(Info,Got a new weibo)n #@ return returnDict dictn returnDict[created_at] = i[mblog][created_at]n returnDict[text] = i[mblog][text]n returnDict[source] = i[mblog][source]n returnDict[nickName] = i[mblog][user][screen_name]n #if has photosn if i[mblog].has_key(pics):n returnDict[picUrls] = []n for j in i[mblog][pics]:n returnDict[picUrls].append(j[url])n return returnDictn self.echoMsg(Info,Has %d id(s)%len(self.itemIds))n except Exception as e:n self.echoMsg(Error,e)n sys.exit()nn """n @ String level : Info/Errorn @ String msg : The message you want to shown """n def echoMsg(self, level, msg):n if level == Info:n print [Info] %s%msgn elif level == Error:n print [Error] %s%msgn

寫了個一發現有新微博就發郵件提醒的功能完整代碼見Github地址 naiquann/WBMonitor

0x03 測試

運行代碼

填寫完相關的登陸信息及要監控的用戶的id後

這裡寫了一個心跳包每三秒訪問一次看看有沒有新微博發布

測試的時候這樣比較方便要是拿來用的話可以酌情增加間隔時間

當有微博發布的時候

大功告成啦監控小姐姐的微博去嘍~

本文作者:奶權#米斯特安全團隊轉載請註明出處