前言
哔咘：“你看下明天天气呢？我好准备明天穿的衣服鞋子。”
叭卟没做回应，心想怎么又问啊，自己不知道看吗。
哔咘一个温柔的眼神，叭卟立刻回答道：“要得，天气预报马上看。”
......
于是乎，叭卟产生了一个大胆的想法——
整个自动发送天气预报的东东。哔咘一高兴，然后就可以...嘻嘻嘻嘻嘻嘻......
不用帮看天气预报了
故事讲完了，来，我们搞起！
1. 准备道具 1.1 天气预报和笑话key的获取
聚合数据注册免费的API：https://www.juhe.cn/docs/api/id/73
注册申请成功后，在【个人中心】查看自己的key（key_joke、key_weather）
1.2 天气灾害预警城市id和key的获取https://id.qweather.com/和风天气注册免费账号：https://id.qweather.com/
登录后进入【应用管理】，新建应用 -添加Key（key_alarm）
预警城市id获取：https://dev.qweather.com/docs/api/geo/city-lookup/
1.3 企业微信相关id、Secret获取及设置
注册企业微信：https://work.weixin.qq.com/
进入【我的企业】，查看企业id（corpid）
【应用管理】新建应用，查看应用AgentId和Secret（agentid、appsecret）
查看应用Secret需要下载企业微信app，被邀请人不需要下载
【通讯录】中邀请微信好友，也可以通过企业微信app邀请
查看成员的账号（userid_list）
企业微信app【新消息通知】中关闭“仅在企业微信中接收消息”
不关闭的话微信端收不到消息哦
2. 搞起搞起 2.1 Python代码示例 2.1.1 配置文件（Config.py）
# -*- coding:utf-8 -*- # 天气预报及笑话参数 city = '重庆' # 要查询的城市——重庆 key_weather = '*************************' # 聚合数据的天气预报key url_weather = 'http://apis.juhe.cn/simpleWeather/query' # 聚合数据的天气预报URL key_joke = '*************************' # 聚合数据的笑话key url_joke = 'http://v.juhe.cn/joke/randJoke.php' # 聚合数据的笑话URL url_oneword = 'http://open.iciba.com/dsapi?' # 金山词霸每日一句URL # 天气灾害预警参数 cityid_CQ = 101040100 # 要查询的城市id,和风天气城市列表API文档：https://dev.qweather.com/docs/api/warning/weather-warning-city-list/ key_alarm = '*************************' # 和风天气的Key url_alarm = 'https://devapi.qweather.com/v7/warning/now' # 和风天气灾害预警URL url_alarm_text = 'https://weather.cma.cn/web/channel-380.html' # 中国气象局每日天气提示URL # 企业微信参数 corpid = '***************' # 企业的id agentid = ******** # 应用的id appsecret = '*************************' # 应用的Secret userid_list = ['********', '********'] # 企业成员的账号，即要发送给的好友列表 userid_str = '|'.join(userid_list) # 企业微信格式要求 url_token = 'https://qyapi.weixin.qq.com/cgi-bin/gettoken' # 企业微信token获取URL url_send = 'https://qyapi.weixin.qq.com/cgi-bin/message/send' # 企业微信发送消息URL2.1.2 获取现在的时间和明天是星期几（GetTime.py）
# -*- coding:utf-8 -*-
import datetime
# 获取现在的时间和明天是星期几
def get_time():
time_now = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
time_tomorrow = (datetime.timedelta(days=1) + datetime.datetime.now()).strftime('%Y-%m-%d %H:%M:%S')
week_num = datetime.datetime.strptime(time_tomorrow, '%Y-%m-%d %H:%M:%S').weekday()
week_list = ['一', '二', '三', '四', '五', '六', '日']
week_tomorrow = '星期' + week_list[week_num]
return time_now, week_tomorrow
2.1.3 从企业微信获取token（GetToken.py）
# -*- coding:utf-8 -*-
import requests, json, Config
# 参数-从Config文件引入
corpid = Config.corpid
appsecret = Config.appsecret
url_token = Config.url_token
# 获取企业微信的token，API文档：https://open.work.weixin.qq.com/api/doc/90000/90135/91039
def get_token():
response = requests.get(f'{url_token}?corpid={corpid}&corpsecret={appsecret}')
data = json.loads(response.text)
token = data['access_token']
return token
2.1.4 获取天气预报并发送（SendWeather.py）
# -*- coding:utf-8 -*-
import requests, json, traceback, Config
from GetToken import get_token
from GetTime import get_time
# 参数-从Config文件引入
url_weather = Config.url_weather
key = Config.key_weather
city = Config.city
url_oneword = Config.url_oneword
corpid = Config.corpid
appsecret = Config.appsecret
userid = Config.userid_str
agentid = Config.agentid
url_send = Config.url_send
# 聚合数据获取天气数据，API文档：https://www.juhe.cn/docs/api/id/73
def get_weather():
params_weather = {'key': key, 'city': city} # 天气参数体
response_weather = requests.get(url_weather,
params=params_weather)
r_weather = json.loads(response_weather.text) # 将返回的数据构造为json格式
city_r = r_weather['result']['city'] # 城市
date = r_weather['result']['future'][1]['date'] # 明天的日期
tem = r_weather['result']['future'][1]['temperature'] # 明天的温度
wea = r_weather['result']['future'][1]['weather'] # 明天的天气
# 构造要发送消息的内容和格式
title = '☀天气预报' + 'n'
weather = title + ('%s %sn%s: %s %s' % (date, get_time()[1], city_r, tem, wea))
+ get_one_word() # 添加了金山词霸每日一句
return weather
# 金山词霸获取每日一句英文和中文内容，金山词霸开放平台API文档：http://open.iciba.com/index.php?c=wiki
def get_one_word():
r_oneword = requests.get(url_oneword).json()
content = r_oneword.get('content') + 'n' # 英文内容，这里添加了换行符
note = r_oneword.get('note') # 中文内容
oneword = 'n' + content + note # 每日一句的内容，添加了换行符
return oneword
# 通过企业微信发送天气预报，发送应用消息API文档：https://open.work.weixin.qq.com/api/doc/90000/90135/90236
def send_weather():
msg_content = get_weather()
json_data = {
'touser': userid,
'msgtype': 'text', # 消息类型：文本消息
'agentid': agentid,
'text': {'content': msg_content}
} # 构造请求参数
str_f = '发送天气预报失败!!'
str_s = '发送天气预报成功!!'
try:
response_send = requests.post(
f'{url_send}?access_token={get_token()}', json=json_data)
r_wea = json.loads(response_send.text)
if r_wea['errmsg'] == 'ok': # 判断消息是否发送成功
print('%s %s' % (str_s, get_time()[0]))
print('=' * 56)
else:
print('%s %s' % (str_f, get_time()[0]))
print(r_wea)
except Exception as err:
print('%s %s' % (str_f, get_time()[0]))
print(err.args)
print('-' * 56)
print(traceback.format_exc())
if __name__ == '__main__':
send_weather()
2.1.5 获取笑话并发送（SendJoke.py）
# -*- coding:utf-8 -*-
import requests, json, traceback, Config
from GetToken import get_token
from GetTime import get_time
# 参数-从Config文件引入
key = Config.key_joke
url = Config.url_joke
userid = Config.userid_str
agentid = Config.agentid
url_send = Config.url_send
# 聚合数据获取笑话数据，API文档：https://www.juhe.cn/docs/api/id/95
def get_joke():
params_joke = {'key': key}
response_joke = requests.get(url, params=params_joke)
r_joke = json.loads(response_joke.text)
title_one = 'U0001F604一份开心：' + 'n'
title_two = 'U0001F604两份快乐：' + 'n'
# U0001F604表情unicode编码详见：https://apps.timwhitlock.info/emoji/tables/unicode#block-6c-other-additional-symbols
joke_one = r_joke['result'][0]['content'] + 'nn' # 第一条笑话内容
joke_two = r_joke['result'][1]['content'] # 第二条笑话内容
# 要发送的笑话内容
jokes = title_one + joke_one + title_two + joke_two
return jokes
# 通过企业微信发送笑话，企业微信发送应用消息API文档：https://open.work.weixin.qq.com/api/doc/90000/90135/90236
def send_joke():
msg_content = get_joke()
json_data = {
'touser': userid,
'msgtype': "text", # 消息类型：文本消息
'agentid': agentid,
'text': {'content': msg_content}
} # 构造请求参数
str_f = '发送笑话失败!!'
str_s = '发送笑话成功!!'
try:
response_joke = requests.post(
f"{url_send}?access_token={get_token()}", json=json_data)
r_joke = json.loads(response_joke.text)
if r_joke['errmsg'] == 'ok': # 判断消息是否发送成功
print('%s %s' % (str_s, get_time()[0]))
print('=' * 56)
else:
print('%s %s' % (str_f, get_time()[0]))
print(r_joke)
except Exception as err:
print('%s %s' % (str_f, get_time()[0]))
print(err.args)
print('-' * 56)
print(traceback.format_exc())
if __name__ == '__main__':
send_joke()
2.1.6 获取天气灾害预警并发送（SendAlarm.py）
# -*- coding:utf-8 -*-
import requests, json, traceback, Config
from GetToken import get_token
from GetTime import get_time
# 参数-从Config文件引入
cityid = Config.cityid_CQ
url = Config.url_alarm
key = Config.key_alarm
url_alarm_text = Config.url_alarm_text
userid = Config.userid_str
agentid = Config.agentid
url_send = Config.url_send
# 和风天气获取天气灾害预警数据，API文档：https://dev.qweather.com/docs/api/warning/weather-warning/
def get_alarm():
try:
alarm_content = 0 # 预警内容初始值：0
response_alarm = requests.get(f'{url}?location={cityid}&key={key}') # 获取对应城市（cityid）的灾害预警
r_alarm = json.loads(response_alarm.text)
if r_alarm['warning'] != []: # 判断是否有天气灾害预警内容
alarm_title = r_alarm['warning'][0]['title'] # 天气灾害预警的标题
list = alarm_title.split('[') # 这里对返回的标题做了拆分处理
title_head = list[0] # 灾害预警的名称
title_tail = list[1].split(']')[0] # 灾害预警的等级
alarm_type = r_alarm['warning'][0]['typeName'] # 灾害类型
alarm_level = r_alarm['warning'][0]['level'] # 灾害颜色级别
alarm_text = r_alarm['warning'][0]['text'] # 灾害详情内容
# 构造要发送的天气灾害预警的内容和格式
alarm_content = {'content': f'★{title_head}n★ {title_tail}n类型：{alarm_type}n'
f'级别：{alarm_level}n详情：{alarm_text}n更多：{url_alarm_text}'} # 添加了中国气象局每日天气提示的URL
else:
print(f'目前没有天气灾害预警消息!! {get_time()[0]}')
return alarm_content
except Exception as err:
print(err.args)
print('-' * 56)
print(traceback.format_exc())
# 通过企业微信发送天气灾害预警，发送应用消息API文档：https://open.work.weixin.qq.com/api/doc/90000/90135/90236
def send_alarm():
content = get_alarm()
json_data = {
'touser': userid,
'msgtype': 'text', # 消息类型：文本消息
'agentid': agentid,
'text': content
} # 构造请求参数
str_f = '发送天气灾害预警失败!!'
str_s = '发送天气灾害预警成功!!'
try:
if content != 0: # 判断是否有天气灾害预警
response = requests.post(f'{url_send}?access_token={get_token()}', json=json_data)
r_alarm = json.loads(response.text)
if r_alarm['errmsg'] == 'ok': # 判断消息是否发送成功
print('%s %s' % (str_s, get_time()[0]))
print('=' * 56)
else:
print('%s %s' % (str_f, get_time()[0]))
print(r_alarm)
except Exception as err:
print('%s %s' % (str_f, get_time()[0]))
print(err.args)
print('-' * 56)
print(traceback.format_exc())
if __name__ == '__main__':
send_alarm()
2.2 部署代码到Linux服务器
2.2.1 WinSCP工具将以上6个Python文件拷贝到Linux服务器中的同一个文件夹下
2.2.2 Linux 定时任务配置
crontab -e命令写入以下定时任务
crontab命令详解见参考资料2
Linux命令详解见参考资料3
3. 搞完收工# 每天6:26后台执行SendJoke.py并输出日志保存到SendJoke.log中
26 6 * * * nohup python3 -u /usr/local/pythonfiles/JokeWeather/SendJoke.py >> /usr/local/pythonfiles/JokeWeather/SendJoke.log 2>&1 &# 每天18:18后台执行SendAlarm.py并输出日志保存到SendAlarm.log中
18 18 * * * nohup python3 -u /usr/local/pythonfiles/JokeWeather/SendAlarm.py >> /usr/local/pythonfiles/JokeWeather/SendAlarm.log 2>&1 &# 每天21:18后台执行SendWeather.py并输出日志保存到SendWeather.log中
18 21 * * * nohup python3 -u /usr/local/pythonfiles/JokeWeather/SendWeather.py >> /usr/local/pythonfiles/JokeWeather/SendWeather.log 2>&1 &
效果如下
参考资料
1. 利用Python实现企业微信发送文件消息_Lin_Xiao_Dai的博客-CSDN博客_python 企业微信发送文件
2. Linux定时运行Python脚本_孤影惆怅的博客-CSDN博客_linux定时运行python脚本
3. 如何让你的 Python 代码在腾讯云 Linux 中一直运行着_mukes的博客-CSDN博客
欢迎分享，转载请注明来源：内存溢出
微信扫一扫
支付宝扫一扫
评论列表（0条）