Python系列之——利用Python实现微博监控

0x00 前言:

前几个星期在写一个微博监控系统 可谓是一波三折啊 获取到微博后因为一些字符编码问题 导致心态爆炸开发中断 但是就在昨天发现了另外一个微博的接口

一个手机微博的接口https://m.weibo.cn/经过一番测试后认为这个接口满足我的要求 就继续完成未完成的使命吧

0x01 分析:

这个接口直接访问的话会有一个302跳转到一个登陆界面


 也就是说这里需要先模拟登陆一下才可以访问到微博
抓个包分析了一下

发现只要用户名和密码正确既返回200且json部分的retcode会返回20000000
少了验证码这一大坑 那模拟登陆就相当简单啦

登陆完后访问用户主页 例如:https://m.weibo.cn/u/3023940914
可以在审查元素的Network模块看到 这里用了两个xhr来加载用户信息及微博信息

分别是
https://m.weibo.cn/api/container/getIndex?type=uid&value=3023940914&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 = {
    ‘User-Agent‘: ‘Mozilla/5.0 (Windows NT 6.1; WOW64; rv:54.0) Gecko/20100101 Firefox/54.0‘,
    ‘Content-Type‘: ‘application/x-www-form-urlencoded‘,
    ‘Referer‘: ‘https://passport.weibo.cn/signin/login‘,
    ‘Connection‘: ‘close‘,
    ‘Accept-Language‘: ‘zh-CN,zh;q=0.8,en-US;q=0.5,en;q=0.3‘
}
loginApi = ‘https://passport.weibo.cn/sso/login‘
loginPostData = {
    ‘username‘:userName,
    ‘password‘:passWord,
    ‘savestate‘:1,
    ‘r‘:‘‘,
    ‘ec‘:‘0‘,
    ‘pagerefer‘:‘‘,
    ‘entry‘:‘mweibo‘,
    ‘wentry‘:‘‘,
    ‘loginfrom‘:‘‘,
    ‘client_id‘:‘‘,
    ‘code‘:‘‘,
    ‘qq‘:‘‘,
    ‘mainpageflag‘:1,
    ‘hff‘:‘‘,
    ‘hfp‘:‘‘
}
#get user session
session = requests.session()
try:
    r = session.post(loginApi,data=loginPostData,headers=reqHeaders)
    if r.status_code == 200 and json.loads(r.text)[‘retcode‘] == 20000000:
        #successful
        #do someting
    else:
        #Logon failure!
except Exception as e:
    #Logon failure!

登陆完成后就可以拼接用户id访问前面说的第一个接口了
访问完后再拼接containerid参数获取微博信息的json数据
代码片段:

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

这里把所有获取到的微博的id存起来 后面继续访问是发现有新的微博id不在这个列表里就证明是新发布的微博
代码片段:

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

将这些方法封装成了一个类 完整代码如下

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Author    : 奶权
# Action    : 微博监控
# Desc      : 微博监控主模块

import requests,json,sys
from lxml import etree

class weiboMonitor():
    """
        @   Class self  :
    """
    def __init__(self, ):
        self.session = requests.session()
        self.reqHeaders = {
            ‘User-Agent‘: ‘Mozilla/5.0 (Windows NT 6.1; WOW64; rv:54.0) Gecko/20100101 Firefox/54.0‘,
            ‘Content-Type‘: ‘application/x-www-form-urlencoded‘,
            ‘Referer‘: ‘https://passport.weibo.cn/signin/login‘,
            ‘Connection‘: ‘close‘,
            ‘Accept-Language‘: ‘zh-CN,zh;q=0.8,en-US;q=0.5,en;q=0.3‘
        }

"""
        @   Class self  :
        @   String userName  : The username of weibo.cn
        @   String passWord  : The password of weibo.cn
    """
    def login(self, userName, passWord):
        loginApi = ‘https://passport.weibo.cn/sso/login‘
        loginPostData = {
            ‘username‘:userName,
            ‘password‘:passWord,
            ‘savestate‘:1,
            ‘r‘:‘‘,
            ‘ec‘:‘0‘,
            ‘pagerefer‘:‘‘,
            ‘entry‘:‘mweibo‘,
            ‘wentry‘:‘‘,
            ‘loginfrom‘:‘‘,
            ‘client_id‘:‘‘,
            ‘code‘:‘‘,
            ‘qq‘:‘‘,
            ‘mainpageflag‘:1,
            ‘hff‘:‘‘,
            ‘hfp‘:‘‘
        }
        #get user session
        try:
            r = self.session.post(loginApi,data=loginPostData,headers=self.reqHeaders)
            if r.status_code == 200 and json.loads(r.text)[‘retcode‘] == 20000000:
                self.echoMsg(‘Info‘,‘Login successful! UserId:‘+json.loads(r.text)[‘data‘][‘uid‘])
            else:
                self.echoMsg(‘Error‘,‘Logon failure!‘)
                sys.exit()
        except Exception as e:
            self.echoMsg(‘Error‘,e)
            sys.exit()

"""
        @   Class self  :
        @   String wbUserId  : The user you want to monitored
    """
    def getWBQueue(self, wbUserId):
        #get user weibo containerid
        userInfo = ‘https://m.weibo.cn/api/container/getIndex?uid=%s&type=uid&value=%s‘%(wbUserId,wbUserId)
        try:
            r = self.session.get(userInfo,headers=self.reqHeaders)
            for i in r.json()[‘tabsInfo‘][‘tabs‘]:
                if i[‘tab_type‘] == ‘weibo‘:
                    conId = i[‘containerid‘]
        except Exception as e:
            self.echoMsg(‘Error‘,e)
            sys.exit()
        #get user weibo index
        self.weiboInfo = ‘https://m.weibo.cn/api/container/getIndex?uid=%s&type=uid&value=%s&containerid=%s‘%(wbUserId,wbUserId,conId)
        try:
            r = self.session.get(self.weiboInfo,headers=self.reqHeaders)
            self.itemIds = []   #WBQueue
            for i in r.json()[‘cards‘]:
                if i[‘card_type‘] == 9:
                    self.itemIds.append(i[‘mblog‘][‘id‘])
            self.echoMsg(‘Info‘,‘Got weibos‘)
            self.echoMsg(‘Info‘,‘Has %d id(s)‘%len(self.itemIds))
        except Exception as e:
            self.echoMsg(‘Error‘,e)
            sys.exit()
    """
        @   Class self  :
    """
    def startMonitor(self, ):
        returnDict = {}
        try:
            r = self.session.get(self.weiboInfo,headers=self.reqHeaders)
            for i in r.json()[‘cards‘]:
                if i[‘card_type‘] == 9:
                    if str(i[‘mblog‘][‘id‘]) not in self.itemIds:
                        self.itemIds.append(i[‘mblog‘][‘id‘])
                        self.echoMsg(‘Info‘,‘Got a new weibo‘)
                        #@ return returnDict dict
                        returnDict[‘created_at‘] = i[‘mblog‘][‘created_at‘]
                        returnDict[‘text‘] = i[‘mblog‘][‘text‘]
                        returnDict[‘source‘] = i[‘mblog‘][‘source‘]
                        returnDict[‘nickName‘] = i[‘mblog‘][‘user‘][‘screen_name‘]
                        #if has photos
                        if i[‘mblog‘].has_key(‘pics‘):
                            returnDict[‘picUrls‘] = []
                            for j in i[‘mblog‘][‘pics‘]:
                                returnDict[‘picUrls‘].append(j[‘url‘])
                        return returnDict
            self.echoMsg(‘Info‘,‘Has %d id(s)‘%len(self.itemIds))
        except Exception as e:
            self.echoMsg(‘Error‘,e)
            sys.exit()

"""
        @   String level   : Info/Error
        @   String msg     : The message you want to show
    """
    def echoMsg(self, level, msg):
        if level == ‘Info‘:
            print ‘[Info] %s‘%msg
        elif level == ‘Error‘:
            print ‘[Error] %s‘%msg

写了个一发现有新微博就发邮件提醒的功能 完整代码见Github地址 https://github.com/naiquann/WBMonitor
#### 0x03 测试
运行代码


 填写完相关的登陆信息及要监控的用户的id后

这里写了一个心跳包 每三秒访问一次看看有没有新微博发布
测试的时候这样比较方便 要是拿来用的话可以酌情增加间隔时间

当有微博发布的时候

时间： 2024-10-13 11:08:07

Python系列之——利用Python实现微博监控的相关文章

【k哥学Python系列】一Python入门(一)

前言 各位看博客的园友们,大家好,我就是那个风流倜傥的KK,还记得我那篇2019年的年中总结博客吗?我想有许多看博客的园友是没有读过我那篇文章的,KK很生气,后果很严重(开个玩笑了,怎么可能).给大家一个传送门2019年终总结,这篇博客总结了KK2019年求职的一些门槛和自己2019年的收获,希望对正在求职的你或者马上需要求职的园友们一些宝贵的建议(虽说是水文,但还是有点作用的),文章结尾的时候我给自己2020年定下了几个目标. 计划二的时候说要学一门后台开发语言,正在看标题的你已经知道是什么开

Python 学习记录----利用Python绘制奥运五环

1 import turtle #导入turtle模块 2 turtle.color("blue") #定义颜色 3 turtle.penup() #penup和pendown()设置画笔抬起或放下时是否绘制直线 4 turtle.goto(-110,-25) #初始位置以中心坐标为(0,0) 5 turtle.pendown() 6 turtle.circle(45) #绘制圆的半径 7 8 turtle.color("black") 9 turtle.penup

利用 Python yield 创建协程将异步编程同步化

在 Lua 和 Python 等脚本语言中,经常提到一个概念: 协程.也经常会有同学对协程的概念及其作用比较疑惑,本文今天就来探讨下协程的前世今生. 首先回答一个大家最关心的问题:协程的好处是什么? 通俗易懂的回答: 让原来要使用 异步 + 回调 方式写的非人类代码,可以用看似同步的方式写出来. 1.回顾同步与异步编程 同步编程即线性化编程,代码按照既定顺序执行,上一条语句执行完才会执行下一条,否则就一直等在那里. 但是许多实际操作都是CPU 密集型任务和 IO 密集型任务,比如网络请求,此时不

用 Python 脚本实现对 Linux 服务器的监控

hon 分享到:8 原文出处: 曹江华 目前 Linux 下有一些使用 Python 语言编写的 Linux 系统监控工具 比如 inotify-sync(文件系统安全监控软件).glances(资源监控工具)在实际工作中,Linux 系统管理员可以根据自己使用的服务器的具体情况编写一下简单实用的脚本实现对 Linux 服务器的监控. 本文介绍一下使用 Python 脚本实现对 Linux 服务器 CPU 内存 网络的监控脚本的编写. Python 版本说明 Python 是由 Guido va

利用python实现网卡流量图表绘制！！！

项目背景: 利用python实现一个自动化的网卡流量图表绘制,这对于我们实现自动化运维平台有更深入的理解, 也会让我们对于现有的一些监控软件的一些实现都有很大的帮助. 实现环境: 虚拟机VMware Workstation 12 player 服务器:centos6.5的系统  ip:192.168.0.25 python2.6.6 rrdtool模块.time模块.psutil模块. SecureCRT ssh远程连接软件 实验过程: 思路其实很清醒:创建rrd数据库---->数据写入rrd数

利用python yielding创建协程将异步编程同步化

转自:http://www.jackyshen.com/2015/05/21/async-operations-in-form-of-sync-programming-with-python-yielding/ 目录 回顾同步与异步编程 回顾多线程编程 yield与协程 异步编程同步化 回顾同步与异步编程 同步编程即线性化编程,代码按照既定顺序执行,上一条语句执行完才会执行下一条,否则就一直等在那里.但是许多实际操作都是CPU 密集型任务和 IO 密集型任务,比如网络请求,此时不能让这些任务阻塞

利用Python进行数据分析（一）简单介绍

一.处理数据的基本内容 数据分析 是指对数据进行控制.处理.整理.分析的过程. 在这里,“数据”是指结构化的数据,例如:记录.多维数组.Excel 里的数据.关系型数据库中的数据.数据表等. 二.说说 Python 这门语言 Python 是现在最受欢迎的动态编程语言之一(还有 Perl.Ruby 等).近些年非常流行用 Python 建站,比如流行的 Python Web 框架 Django. Python 这类语言被称为脚本语言,因为它们可以编写简短粗糙的小程序,即脚本.不过这好像在说 Py

利用python httplib模块 发送Post请求测试web服务是否正常起来！

最近在学习python,恰好老大最近让我搞个基于post请求测试web服务是否正常启用的小监控,上网查了下资料,发现强大的Python恰好能够用上,所以自己现学现卖,顺便锻炼下自己. 由于本人也刚接触这块不久属于菜鸟级别,所以在任务之前,只能上网把基于post请求的web监控了解清楚,这些资料网上很多,因为本人认为完成这类任务最重要的就是要将实现的原理研究清楚,写程序只是实现的工具,如果大的逻辑不正确,后面的都是白忙活. 了解post发送请求的原理后,利用Python的httplib模块进行逻辑

利用Python进行数据分析_Pandas_汇总和计算描述统计

申明:本系列文章是自己在学习<利用Python进行数据分析>这本书的过程中,为了方便后期自己巩固知识而整理. 原文地址:https://www.cnblogs.com/zhouwp/p/8485462.html