从前慢现在也慢

这个屌丝很懒，什么也没留下！

热门标签

python实现微信公众号自动定时推送天气、以及公众号聊天机器人（给心爱的人一点惊喜）_python向微信公众号推送消息

作者：从前慢现在也慢 | 2024-02-08 20:35:30

踩

python向微信公众号推送消息

前言

一、项目展示

二、使用步骤

1.安装库

2.源代码

3.部署到服务器

总结

前言

配置需求:

1.一台服务器

2.一个域名（用于聊天机器人）

3.一个微信测试公众号

4.python环境（这里用到的版本是3.6.4）

5.爬虫所需的库:bs4、requests

一、项目展示

二、使用步骤

1.安装库

1.bs4、requests的安装

pip install -i http://pypi.douban.com/simple/ --trusted-host=pypi.douban.com/simple包名

2.注册微信公众测试号

微信公众平台 (qq.com)

3.获取测试号appid、appsecret、userid、templateId

2.源代码

1.config.json


{
  "appId": "你自己的",
  "appsecret": "你自己的",
  "userId": [
    "你自己的","xxxx"
  ],
  "templateId": "你自己的"
}

我这里使用的模板是:

{{date.DATA}} 现在位于：{{city.DATA}} 天气{{today_weather.DATA}}，{{today_weather_str.DATA}} 今天的最低温度为{{today_min.DATA}}，最高温度为{{today_max.DATA}}，风力为{{today_wind.DATA}} tips: {{today_str.DATA}} 顺便给小琳播报一下{{tomorrow.DATA}}的天气：{{tomorrow_weather.DATA}}，最低温度{{tomorrow_min.DATA}}，最高温度{{tomorrow_max.DATA}}，风力{{tomorrow_wind.DATA}} 每日一句情话：{{daily_love.DATA}} 每日一句英语：{{note_ch.DATA}}{{note_en.DATA}} 每天都要元气满满哦~

2.wechat_sendmsg.py

参考文章:

(97条消息) python实现微信公众号给你的重要用户推送消息_python向微信公众号推送消息_DreamAtDawn的博客-CSDN博客

（这里采用的是白天的天气晚上获取会出现问题）

代码如下（示例）：


from datetime import datetime
from requests import get, post
from time import localtime
import json
import os
import random
import sys
import bs4
import requests
 
 
def get_weather():
    city_name = '张家界'
 
    headers = {
        'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9',
        'Accept-Encoding': 'gzip, deflate',
        'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6',
        'Cache-Control': 'max-age=0',
        'Connection': 'keep-alive',
        'Cookie': 'HttpOnly; userNewsPort0=1; Hm_lvt_080dabacb001ad3dc8b9b9049b36d43b=1660634957; f_city=%E9%87%91%E5%8D%8E%7C101210901%7C; defaultCty=101210901; defaultCtyName=%u91D1%u534E; Hm_lpvt_080dabacb001ad3dc8b9b9049b36d43b=1660639816',
        'Host': 'www.weather.com.cn',
        'Referer': 'http://www.weather.com.cn/weather1d/101251101.shtml',
        'Upgrade-Insecure-Requests': '1',
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/104.0.5112.81 Safari/537.36 Edg/104.0.1293.54'
 
    }
    url = "http://www.weather.com.cn/weather/101251101.shtml"
    response = get(url=url, headers=headers)
 
    text = response.content.decode('utf-8')
    soup = bs4.BeautifulSoup(text, 'html.parser')
    # 存放日期
    list_day = []
    i = 0
    day_list = soup.find_all('h1')
    for each in day_list:
        if i <= 1:  # 今明两天的数据
            list_day.append(each.text.strip())
            i += 1
 
    # 天气情况
    list_weather = []
    weather_list = soup.find_all('p', class_='wea')  # 之前报错是因为写了class 改为class_就好了
    for i in weather_list:
        list_weather.append(i.text)
    list_weather = list_weather[0:2]  # 只取今明两天
 
    # 存放当前温度，和明天的最高温度和最低温度
    tem_list = soup.find_all('p', class_='tem')
 
    i = 0
    list_tem = []
    for each in tem_list:
        if i >= 0 and i < 2:
            list_tem.append([each.span.text, each.i.text])  # 记录明天的最高温度和最低温度
            i += 1
 
    # 风力
    list_wind = []
    wind_list = soup.find_all('p', class_='win')
    for each in wind_list:
        list_wind.append(each.i.text.strip())
    list_wind = list_wind[0:2]  # 同只取今明两天
 
    today_date = list_day[0]
    today_weather = list_weather[0]
    today_max = list_tem[0][0] + '℃'
    today_min = list_tem[0][1]
 
    today_wind = list_wind[0]
    tomorrow = list_day[1]
    tomorrow_weather = list_weather[1]
    tomorrow_max = list_tem[1][0] + '℃'
    tomorrow_min = list_tem[1][1]
    tomorrow_wind = list_wind[1]
 
    return city_name, today_date, today_weather, today_min, today_max, today_wind, tomorrow, tomorrow_weather, tomorrow_min, tomorrow_max, tomorrow_wind
 
 
def get_daily_love():
    # 每日一句情话
    url = "https://api.lovelive.tools/api/SweetNothings/Serialization/Json"
    r = requests.get(url)
    all_dict = json.loads(r.text)
    sentence = all_dict['returnObj'][0]
    daily_love = sentence
    return daily_love
 
 
def get_ciba():
    # 每日一句英语（来自爱词霸）
    url = "http://open.iciba.com/dsapi/"
    r = get(url)
    note_en = r.json()["content"]
    note_ch = r.json()["note"]
    return note_ch, note_en
 
 
def get_color():
    # 往list中填喜欢的颜色即可
    color_list = ['#6495ED', '#3CB371']
    return random.choice(color_list)
 
 
def get_config():
    with open("config.json", "r") as f:
        config = json.load(f)
        return config
 
 
def get_token(config):
    url = "https://api.weixin.qq.com/cgi-bin/token"
    param = {
        "grant_type": "client_credential",
        "appid": config['appId'],
        "secret": config['appsecret']
    }
    response = requests.get(url=url, params=param)
    result = response.json().get("access_token")
    return result
 
 
def send_message(to_user, access_token, city_name, today_date, today_weather, today_max, today_min, today_wind,
                 tomorrow, tomorrow_weather, tomorrow_max, tomorrow_min, tomorrow_wind, daily_love, note_ch, note_en,
                 today_weather_str, today_str):
    url = "https://api.weixin.qq.com/cgi-bin/message/template/send?access_token={}".format(access_token)
    week_list = ["星期日", "星期一", "星期二", "星期三", "星期四", "星期五", "星期六"]
    year = localtime().tm_year
    month = localtime().tm_mon
    day = localtime().tm_mday
    today = datetime.date(datetime(year=year, month=month, day=day))
    week = week_list[today.isoweekday() % 7]
    print(today_max, today_min)
 
    for i in to_user:
        data = {
            "touser": i,
            "template_id": config["templateId"],
            "url": "http://www.baidu.com",
            "topcolor": "#FF0000",
            "data": {
                "date": {
                    "value": "{} {}".format(today, week),
                    "color": get_color()
                },
                "city": {
                    "value": city_name,
                    "color": get_color()
                },
                "today": {
                    "value": today_date,
                    "color": get_color()
                },
                "today_weather": {
                    "value": today_weather,
                    "color": get_color()
                },
                "today_max": {
                    "value": today_max,
                    "color": get_color()
                },
                "today_min": {
                    "value": today_min,
                    "color": get_color()
                },
                "today_wind": {
                    "value": today_wind,
                    "color": get_color()
                },
                "tomorrow": {
                    "value": tomorrow,
                    "color": get_color()
                },
                "tomorrow_weather": {
                    "value": tomorrow_weather,
                    "color": get_color()
                },
                "tomorrow_max": {
                    "value": tomorrow_max,
                    "color": get_color()
                },
                "tomorrow_min": {
                    "value": tomorrow_min,
                    "color": get_color()
                },
                "tomorrow_wind": {
                    "value": tomorrow_wind,
                    "color": get_color()
                },
                "daily_love": {
                    "value": daily_love,
                    "color": get_color()
                },
 
                "note_en": {
                    "value": note_en,
                    "color": get_color()
                },
                "note_ch": {
                    "value": note_ch,
                    "color": get_color()
                },
                "today_weather_str": {
                    "value": today_weather_str,
                    "color": get_color()
                },
                "today_str": {
                    "value": today_str,
                    "color": get_color()
                }
            }
        }
        headers = {
            'Content-Type': 'application/json',
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) '
                          'AppleWebKit/537.36 (KHTML, like Gecko) Chrome/103.0.0.0 Safari/537.36'
        }
        response = post(url, headers=headers, json=data).json()
        if response["errcode"] == 40037:
            print("推送消息失败，请检查模板id是否正确")
        elif response["errcode"] == 40036:
            print("推送消息失败，请检查模板id是否为空")
        elif response["errcode"] == 40003:
            print("推送消息失败，请检查微信号是否正确")
        elif response["errcode"] == 0:
            print("推送消息成功")
        else:
            print(response)
 
 
if __name__ == "__main__":
    try:
        config = get_config()
    except FileNotFoundError:
        print("推送消息失败，请检查config.txt文件是否与程序位于同一路径")
        os.system("pause")
        sys.exit(1)
    except SyntaxError:
        print("推送消息失败，请检查配置文件格式是否正确")
        os.system("pause")
        sys.exit(1)
    # 获取accessToken
    accessToken = get_token(config)
    # 接收的用户
    users = config["userId"]
    # 传入省份和市获取天气信息
    city_name, today_date, today_weather, today_min, today_max, today_wind, tomorrow, tomorrow_weather, tomorrow_min, tomorrow_max, tomorrow_wind = get_weather()
    # 判断今天天气并返回对应响应
    # print(today_weather)
    today_weather_str = ""
    str_weather = ['雨', '云', '晴', '阴']
    # 提示可以多想几条 放入列表 random.choice([])
    today_weather_str_rain = ['今日可能有雨 记得带伞哦', '下雨天和巧克力更配哦~', '天街小雨润如酥 草色遥看近却无']
    today_weather_str_cloud = ['等风起，漫随天外云卷云舒', '今日有云 气候宜人', '应是天仙狂醉，乱把白云揉碎']
    today_weather_str_sun = ['快醒醒 太阳已经晒屁gu啦~', '今日天气晴 不止天气', '今天天气不错 多出门走走']
    today_weather_str_overcast = ['今日天气较阴 适宜懒人瘫在家里~', '阴天雨天暴雨天 爱你爱到发晒癫']
    for i in str_weather:
        if i in today_weather and i == '雨':
            today_weather_str = random.choice(today_weather_str_rain)
            break
        elif i in today_weather and i == '云':
            today_weather_str = random.choice(today_weather_str_cloud)
            break
        elif i in today_weather and i == '晴':
            today_weather_str = random.choice(today_weather_str_sun)
            break
        elif i in today_weather and i == '阴':
            today_weather_str = random.choice(today_weather_str_overcast)
            break
        else:
            today_weather_str = '啊哦 今天公众号有点不太聪明~ '
    # print(today_weather_str)
    # 判断今日温度并返回对应响应
    today_min_judge = today_min.split('℃')
    today_max_judge = today_max.split('℃')
    today_min_str = ''
    today_max_str = ''
    weather_min_str = ['今日温度较低 天冷记得加衣哦', '今日温度较低 要注意保暖哦']
    weather_max_str = ['今日温度较高 奖励自己一个冰淇淋降降温吧~', '今天好像比较热 记得防晒和消暑降温哦']
    weather_center_str = ['今日温度适宜 适合运动跳芭蕾~(doge)', '今日温度适宜 适合出去happy~']
    if int(today_min_judge[0]) < 10:
        today_str = random.choice(weather_min_str)
    elif int(today_max[0]) > 30:
        today_str = random.choice(weather_max_str)
    else:
        today_str = random.choice(weather_center_str)
    # print(today_str)
    # 获取每日情话
    daily_love = get_daily_love()
    # 获取每日一句英语
    note_ch, note_en = get_ciba()
    # 公众号推送消息
    # 传入提醒信息
    send_message(users, accessToken, city_name, today_date, today_weather, today_max, today_min, today_wind,
                 tomorrow, tomorrow_weather, tomorrow_max, tomorrow_min, tomorrow_wind, daily_love, note_ch,
                 note_en, today_weather_str, today_str)
    # 这里系统暂停乱码 需要将pycharm设置成与cmd相同的GBK编码 在setting file encode中
    os.system("pause")

3.部署到服务器

本地测试无误后相同的操作部署到服务器

1.服务器这里用到系统是centos7 依据自己的情况

安装python 安装所需库然后将代码上传到服务器（finalshell直接复制粘贴）

2.服务器运行代码测试 python /x/x.py

3.linux定时任务配置/etc/crontab

这里是每天早上九点定时推送根据个人情况修改

到这里公众号推送就已经完成了

配置对话机器人

这里就会稍微有些麻烦我就言简意赅一点

1.进入微信测试号绑定接口配置url

a.这里我使用的是django来处理微信回复消息

b.服务器安装django django-admin startproject xx 生成django项目

c.绑定服务器必须是域名形式不能是公网ip地址（绑定时微信会通过get请求发送四个参数）参考文章：接入概述 | 微信开放文档 (qq.com)

我这里是http://www.xxx.xxx/wechat_test/(你的django url地址)

d.django中view.py的代码def wechat_test():(这里获取四个参数并指定加密方式来绑定微信url)

import hashlib
import json
import requests
import time
from django.shortcuts import render,HttpResponse
from xml.etree import ElementTree


def wechat_test(request):
	if request.method == 'GET':
		token = 'misimnya'
		timestamp = request.GET.get('timestamp')
		nonce = request.GET.get('nonce')
		echostr = request.GET.get('echostr')
		signature = request.GET.get('signature')
		# 进行字典排序
		sort_list = [token, timestamp, nonce]
		sort_list.sort()
		
		result = hashlib.sha1((sort_list[0] + sort_list[1] + sort_list[2]).encode('utf-8')).hexdigest()
		# 判断参数是否相同
		if result == signature:
			return HttpResponse(echostr)
		else:
			return HttpResponse('error!')
	elif	request.method == 'POST':
	 	return HttpResponse('no get')

e.启动服务 python manage.py runserver 去微信测试号绑定提交

token自定义和代码参数一致就行

f.绑定成功后你已经完成一大半了接下来就是对收到的消息进行处理（收到的是xml格式的消息）采用什么机器人回复（这里是用的是免费的思知机器人）

进去官网注册：思知机器人 | 对话机器人 | 聊天机器人 | 知识图谱 - OwnThink

你也可以用官方的我这里用的是自己的机器人

机器人管理登陆后创建自己的机器人然后拿到机器人的appid

views源代码:


import hashlib
import json
import requests
import time
from django.shortcuts import render,HttpResponse
from xml.etree import ElementTree
 
 
 
# Create your views here.
class ParseXmlMsg(object):
    def __init__(self, xmlData):
        self.ToUserName = xmlData.find('ToUserName').text
        self.FromUserName = xmlData.find('FromUserName').text
        self.CreateTime = xmlData.find('CreateTime').text
        self.MsgType = xmlData.find('MsgType').text
        self.MsgId = xmlData.find('MsgId').text
 
        if self.MsgType == 'text':
            self.Content = xmlData.find('Content').text.encode('utf-8')
        elif self.MsgType == 'image':
            self.PicUrl = xmlData.find('PicUrl').text
            self.MediaId = xmlData.find('MediaId').text
 
 
class TextMsg(object):
    def __init__(self, toUserName, fromUserName, content):
        # 私有对象，禁止外部访问
        self.__dict = dict()
        self.__dict['ToUserName'] = toUserName
        self.__dict['FromUserName'] = fromUserName
        self.__dict['CreateTime'] = int(time.time())
        self.__dict['Content'] = content
 
    def send(self):
        XmlForm = """
            <xml>
                <ToUserName><![CDATA[{ToUserName}]]></ToUserName>
                <FromUserName><![CDATA[{FromUserName}]]></FromUserName>
                <CreateTime>{CreateTime}</CreateTime>
                <MsgType><![CDATA[text]]></MsgType>
                <Content><![CDATA[{Content}]]></Content>
            </xml>
            """
        return XmlForm.format(**self.__dict)
 
 
class ImageMsg(object):
    def __init__(self, toUserName, fromUserName, mediaId):
        self.__dict = dict()
        self.__dict['ToUserName'] = toUserName
        self.__dict['FromUserName'] = fromUserName
        self.__dict['CreateTime'] = int(time.time())
        self.__dict['MediaId'] = mediaId
 
    def send(self):
        XmlForm = """
            <xml>
                <ToUserName><![CDATA[{ToUserName}]]></ToUserName>
                <FromUserName><![CDATA[{FromUserName}]]></FromUserName>
                <CreateTime>{CreateTime}</CreateTime>
                <MsgType><![CDATA[image]]></MsgType>
                <Image>
                <MediaId><![CDATA[{MediaId}]]></MediaId>
                </Image>
            </xml>
            """
        return XmlForm.format(**self.__dict)
def wechat_test(request):
	if request.method == 'GET':
		token = 'misimnya'
		timestamp = request.GET.get('timestamp')
		nonce = request.GET.get('nonce')
		echostr = request.GET.get('echostr')
		signature = request.GET.get('signature')
		# 进行字典排序
		sort_list = [token, timestamp, nonce]
		sort_list.sort()
		
		result = hashlib.sha1((sort_list[0] + sort_list[1] + sort_list[2]).encode('utf-8')).hexdigest()
		# 判断参数是否相同
		if result == signature:
			return HttpResponse(echostr)
		else:
			return HttpResponse('error!')
	elif	request.method == 'POST':
	 	# 解析发送过来的body
	 	webData = request.body
	 	xmlData = ElementTree.fromstring(webData)
	 	recMsg = ParseXmlMsg(xmlData)
	 	if recMsg.MsgType == 'text':
	 		toUser = recMsg.FromUserName
	 		fromUser = recMsg.ToUserName
	 		# 获取用户发送过来的消息文本
	 		usercontent = recMsg.Content
	 		# 一定要传解码的消息文本 这里获取到的是 /x84形式的
	 		# print(usercontent.decode('utf-8'))
	 		# ownthink 返回json格式数据的对象
            # 自己的机器人id userid
	 		ownthink_reply = requests.get('https://api.ownthink.com/bot?appid=自己的机器人id&userid=自己的userid&spoken={}'.format(usercontent.decode('utf-8')))
	 		# text方法获取返回json数据
	 		content_json = ownthink_reply.text
	 		answer = json.loads(content_json) 
	 		# print(answer)
	 		content = answer['data']['info']['text']
	 		# content = 'test'
	 		replyMsg = TextMsg(toUser, fromUser, content)
	 		return HttpResponse(content=replyMsg.send())
	 	elif recMsg.MsgType == 'image':
        		print('暂时不做处理')
        		return HttpResponse(content='success')
 
def index(request):
	return render(request, "love_story.html")

g.更改setting.py将app加入将csrf这里注释（post请求会拦截报错）用户发送的信息就是post请求过来

h.总的一句话就是配置url 指向views里面的处理函数也可以不加入app

不懂的话可以搜索看看django教程

最后就是关于django在服务器的后台挂起

参考文章:(97条消息) Centos下django项目的启动和停止（后台运行）_停止django项目_noob_python的博客-CSDN博客

总结

由于微信对公众号的整改 5.4日起微信公众号的颜色无了不过问题不大哈哈

如果这篇文章对你有帮助的话就点个赞谢谢大家~

大佬轻喷~

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/从前慢现在也慢/article/detail/70719