Python爬虫抓取中国天气并发送到微信
Python爬取中国天气
随便点进一个城市的详细天气预报,这里以北京为例。
按照惯例开下F12。这种实时更新的界面一般是通过AJAX传入JSON文件实现的,打开network选项卡刷新验证一下。
url = "https://d1.weather.com.cn/sk_2d/101010100.html?_=1618886817920" requests_url = requests.get(url)
这里的url是刚才那个文件的url,在Network选项卡中右键复制即可获得。
可以给requests.get加一个请求头,放一个假UA,防止被反爬。
message = json.loads(requests_url.text.encode("latin1").decode("utf8").replace("var dataSK = ", ""))
获取文件信息
这里我先用latin1 编码,再用utf-8解码,发现可以提取出正常的文本信息,这时通过str文本的replace方法把里面的代码部分var dataSK =去除,再用json库的loads方法将其转化成Python字典。
到这一步就很简单了,通过字典的键获取对应的值。
cityname = message['cityname'] aqi = int(message['aqi']) sd = message['sd'] wd = message['WD'] ws = message['WS'] temp = message['temp'] weather = message['weather']
最后按照想要的格式输出就可以了,为了方便整个程序的操作,我把这一段代码封装成了函数。
def get_weather(): url = "https://d1.weather.com.cn/sk_2d/101010100.html?_=1618886817920" requests_url = requests.get(url) message = json.loads(requests_url.text.encode("latin1").decode("utf8").replace("var dataSK = ", "")) cityname = message['cityname'] aqi = int(message['aqi']) sd = message['sd'] wd = message['WD'] ws = message['WS'] temp = message['temp'] weather = message['weather'] if aqi <= 50: airQuality = "优" elif aqi <= 100: airQuality = "良" elif aqi <= 150: airQuality = "轻度污染" elif aqi <= 200: airQuality = "中度污染" elif aqi <= 300: airQuality = "重度污染" else: airQuality = "严重污染" return cityname + " " + '今日天气:' + weather + ' 温度:' + temp + ' 摄氏度 ' + wd + ws + ' 相对湿度:' + sd + ' 空气质量:' + str(aqi) + "(" + airQuality + ")"
爬取百度热搜
不多说,直接打开F12,Network选项卡中却没有我们想要的,看来这次只能直接爬取网页了。这里要使用BeautifulSoup库。
requests_page = requests.get('http://top.baidu.com/buzz?b=1&c=513&fr=topbuzz_b42_c513') soup = BeautifulSoup(requests_page.text, "lxml")
这里解析出所有的html代码,再F12一下看看我们需要的内容在哪个标签下。定位一下主页里的标题,啊,我一看,原来是个a标签class='list-title',这好办。
soup_text = soup.find_all("a", class_='list-title')
然后我们把它输出出来。
for text in soup_text: print(text.string)
然后发现,还他喵的是乱码,我们再试试.encode("latin1").decode("GBK")
for text in soup_text: print(text.string.encode("latin1").decode("GBK"))
还真成了,我吐了,敢情你俩是一家的。。。
这里改动一下,我们把他们封装进列表里面,方便整理。
def get_top_list(): requests_page = requests.get('http://top.baidu.com/buzz?b=1&c=513&fr=topbuzz_b42_c513') soup = BeautifulSoup(requests_page.text, "lxml") soup_text = soup.find_all("a", class_='list-title') top_list = [] for text in soup_text: top_list.append(text.string.encode("latin1").decode("GBK")) return top_list
爬取金山词霸每日一句
我直接找到他的每日一句文件。
直接开始写代码,轻车熟路。这里很简单,不解释。
def get_daily_sentence(): url = "http://open.iciba.com/dsapi/" r = requests.get(url) r = json.loads(r.text) content = r["content"] note = r["note"] daily_sentence = content + "n" + note return daily_sentence
整理信息
简单的调用一下我们写的函数,将返回信息整理到一个字符串内即可,方便我们下一步的发送。具体代码跟随后面整体代码展示一遍展示。
通过企业微信机器人发送
首先将机器人添加到群聊,具体步骤不演示,不会自行百度或查阅官方文档。
然后获取你的机器人的webhook链接。(不要把这个链接散播出去,要不然谁都可以调用你的机器人发送信息,造成垃圾信息污染)
我们直接向这个链接发送post请求就可以完成机器人发送信息了,十分的简单。
url = #这里填写你的机器人的webhook链接 headers = {"Content-Type": "text/plain"} data = { "msgtype": "text", "text": { "content": #这里填写要发送的内容,这里选择了纯文本模式 } } requests_url = requests.post(url, headers=headers, data=json.dumps(data))
完成。
完整代码
import simplejson as json import requests import datetime import fake_useragent # 这个库可以不用 from bs4 import BeautifulSoup import time def get_fake_ua(): #这个函数是用来获取随机UA的,可以不用 location = '/fake_useragent_0.1.11.json' #这里是我导入的fakeuseragent库文件,可以不用 ua = fake_useragent.UserAgent(path=location) headers = { 'user-agent': ua.random } return headers def get_week_day(date): week_day_dict = { 0: '星期一', 1: '星期二', 2: '星期三', 3: '星期四', 4: '星期五', 5: '星期六', 6: '星期天', } day = date.weekday() return "今天日期为:" + str(datetime.date.today()) + ' ' + week_day_dict[day] def get_weather(): url = "https://d1.weather.com.cn/sk_2d/101010100.html?_=1618886817920" r_url = requests.get(url, headers=get_fake_ua()) message = json.loads(r_url.text.encode("latin1").decode("utf8").replace("var dataSK = ", "")) cityname = message['cityname'] aqi = int(message['aqi']) sd = message['sd'] wd = message['WD'] ws = message['WS'] temp = message['temp'] weather = message['weather'] if aqi <= 50: airQuality = "优" elif aqi <= 100: airQuality = "良" elif aqi <= 150: airQuality = "轻度污染" elif aqi <= 200: airQuality = "中度污染" elif aqi <= 300: airQuality = "重度污染" else: airQuality = "严重污染" return cityname + " " + '今日天气:' + weather + ' 温度:' + temp + ' 摄氏度 ' + wd + ws + ' 相对湿度:' + sd + ' 空气质量:' + str(aqi) + "(" + airQuality + ")" def get_top_list(): requests_page = requests.get('http://top.baidu.com/buzz?b=1&c=513&fr=topbuzz_b42_c513') soup = BeautifulSoup(requests_page.text, "lxml") soup_text = soup.find_all("a", class_='list-title') i = 0 top_list = [] for text in soup_text: i += 1 top_list.append(text.string.encode("latin1").decode("GBK")) if i == 10: break return top_list def get_daily_sentence(): url = "http://open.iciba.com/dsapi/" r = requests.get(url, headers=get_fake_ua()) r = json.loads(r.text) content = r["content"] note = r["note"] daily_sentence = content + "n" + note return daily_sentence def get_sendContent(): sendContent = get_week_day(datetime.date.today()) + "nn" + get_weather() + "nn" + str(get_top_list()).replace( "', '", 'n').replace("['", "").replace("']", "") + "nn" + get_daily_sentence() return sendContent def send(content): url = # 填写你的webhook链接 headers = {"Content-Type": "text/plain"} data = { "msgtype": "text", "text": { "content": content, } } requests_url = requests.post(url, headers=headers, data=json.dumps(data)) if requests_url.text == '{"errcode":0,"errmsg":"ok"}': return "发送成功" else: return "发送失败" + requests_url.text print(send(get_sendContent()))
本文结束