| layout | title | tags | lang |
|---|
article | 每天推送鸡汤 | | zh-Hans |
前几日在掘金上看到一篇文章&＃xff1a;100行代码&＃xff0c;每天早上喝一碗鸡汤
大意是通过高德api获取一个固定城市的天气&＃xff0c;通过金山词霸的api获取每日推送的鸡汤文&＃xff0c;看着挺不错的样子&＃xff0c;我就打算研究一下&＃xff0c;原文中有的地方我就不再重复了&＃xff0c;记录下copy的过程。
api准备
首先&＃xff0c;作者在文章下面贴上了完整代码&＃xff0c;完整代码中的Server酱的推送地址和高德api url中的key是需要我们自己在Server酱和高德开放平台上面自行申请替换&＃xff0c;都是免费的配额足够自用了。
高德查询天气api还需要城市编码&＃xff0c;可以通过高德开放平台上面发布的城市编码表查询所需要查询天气的城市编码。
定时任务框架APScheduler
定时发送使用的APScheduler&＃xff0c;APScheduler支持多种方式存储任务队列&＃xff0c;默认使用的是在内存中存放&＃xff0c;看起来挺好玩的&＃xff0c;我使用的是sqlite方式&＃xff0c;运行后可以打开数据库查看任务队列&＃xff1b;
APScheduler框架也挺有意思的&＃xff0c;比如说它的工作流程&＃xff1a;
参考教程&＃xff1a;花10分钟让你彻底学会Python定时任务框架apscheduler
中间碰到比较坑的是服务器时间&＃xff0c;我是在下午调试的&＃xff0c;还以为服务器时间使用的是不是24小时制&＃xff0c;部署了很多次都未按照预期运行&＃xff0c;直到我灵光一闪&＃xff0c;想到了时区问题。。。
纽约和北京的时差相差12个小时
知道真相的我眼泪掉下来。。。
翻翻APScheduler的用户指南&＃xff0c;找到一个叫timezone的构造参数&＃xff0c;尝试使用下&＃xff1a;
tz
&＃61;
pytz
timezone
&＃39;Asia/Shanghai&＃39;
scheduler
&＃61;
BlockingScheduler
timezone
&＃61;
tz
嗯&＃xff0c;完美运行&＃xff0c;又涨姿势了
部署
nohup命令&＃xff0c;后台运行进程&＃xff0c;并输出日志到nohup.out文件中&＃xff1a;
查看当前后台运行进程&＃xff1a;
根据PID关闭相应进程&＃xff1a;
推送
推送使用的是Server酱的微信群组推送&＃xff0c;和单独推送使用起来大差不差&＃xff1b;
又向里面加入了钉钉群机器人的推送。
效果图
钉钉&＃xff1a;
微信&＃xff1a;
总结
总的来说&＃xff0c;还是挺好玩的一个东西&＃xff1b;
美中不足的是因为推送时候使用的是Markdown&＃xff0c;获取天气只能把地址位置固定&＃xff0c;不能根据用户位置进行调整&＃xff0c;有待优化&＃xff1b;
微信Server酱推送以后消息只能在服务器端保存3天&＃xff0c;3天前的消息都不能查看了&＃xff0c;有待优化。
具体代码
#!/usr/bin/python3
# -*- coding: utf-8 -*-
import
requests
json
sys
os
time
pytz
from
apscheduler
schedulers
blocking
import
BlockingScheduler
soup_url
&＃61;
"http://open.iciba.com/dsapi/"
weather_url
&＃61;
"http://restapi.amap.com/v3/weather/weatherInfo?city&＃61;城市编码&key&＃61;高德地图key&extensions&＃61;all"
dd_urls
&＃61;
"https://oapi.dingtalk.com/robot/send?access_token&＃61;机器人token"
# Server酱群组推送
group_push_key
&＃61;
"key"
group_push_url
&＃61;
"https://pushbear.ftqq.com/sub"
tz
&＃61;
pytz
timezone
&＃39;Asia/Shanghai&＃39;
scheduler
&＃61;
BlockingScheduler
timezone
&＃61;
tz
def
get_time
"""获取当前时间"""
return
time
strftime
"%Y-%m-%d %H:%M:%S"
time
localtime
def
get_soup
"""获取鸡汤"""
response
&＃61;
requests
get
soup_url
json_data
&＃61;
response
json
date
&＃61;
json_data
&＃39;dateline&＃39;
content
&＃61;
json_data
&＃39;content&＃39;
note
&＃61;
json_data
&＃39;note&＃39;
picture
&＃61;
json_data
&＃39;picture&＃39;
picture2
&＃61;
json_data
&＃39;picture2&＃39;
translation
&＃61;
json_data
&＃39;translation&＃39;
return
date
content
note
picture
translation
picture2
def
get_weather
"""获取天气"""
response
&＃61;
requests
get
weather_url
json_data
&＃61;
response
json
if
json_data
&＃39;status&＃39;
&＃61;&＃61;
&＃39;1&＃39;
return
json_data
&＃39;forecasts&＃39;
0
&＃39;casts&＃39;
0
else
print
get_time
&＃43;
" 天气获取失败&＃xff1a;"
&＃43;
json_data
&＃39;info&＃39;
return
None
def
make_dd_soup
soup
weather
"""制作鸡汤 钉钉"""
if
weather
is
None
weather
&＃61;
get_weather
msg
&＃61;
"msgtype"
"actionCard"
"actionCard"
"title"
"[每日鸡汤]"
"text"
" \n#### {date}\n\n#### 天气 {weather}&＃xff0c;温度{nighttemp}℃ ~ {daytemp}℃。\n\n*{content}*\n\n#### {note}\n\n{translation}\n\n###### \n"
format
date
&＃61;
soup
0
weather
&＃61;
weather
&＃39;dayweather&＃39;
if
weather
&＃39;dayweather&＃39;
&＃61;&＃61;
weather
&＃39;nightweather&＃39;
else
weather
&＃39;dayweather&＃39;
&＃43;
"转"
&＃43;
weather
&＃39;nightweather&＃39;
nighttemp
&＃61;
weather
&＃39;nighttemp&＃39;
daytemp
&＃61;
weather
&＃39;daytemp&＃39;
content
&＃61;
soup
1
note
&＃61;
soup
2
picture
&＃61;
soup
5
translation
&＃61;
soup
4
"hideAvatar"
"0"
"btnOrientation"
"0"
"singleTitle"
""
"singleURL"
""
return
msg
def
make_wx_soup
soup
weather
"""制作鸡汤 微信"""
if
weather
is
None
weather
&＃61;
get_weather
title
&＃61;
"早上好&＃xff01;"
desp
&＃61;
"#### {date}\n\n天气&＃xff1a;{weather}&＃xff0c;温度{nighttemp}℃ ~ {daytemp}℃。\n\n*{content}*\n\n{note}\n\n\n\n{translation}"
format
date
&＃61;
soup
0
weather
&＃61;
weather
&＃39;dayweather&＃39;
if
weather
&＃39;dayweather&＃39;
&＃61;&＃61;
weather
&＃39;nightweather&＃39;
else
weather
&＃39;dayweather&＃39;
&＃43;
"转"
&＃43;
weather
&＃39;nightweather&＃39;
nighttemp
&＃61;
weather
&＃39;nighttemp&＃39;
daytemp
&＃61;
weather
&＃39;daytemp&＃39;
content
&＃61;
soup
1
note
&＃61;
soup
2
picture
&＃61;
soup
5
translation
&＃61;
soup
4
return
title
desp
def
push_wx
text
&＃61;
None
desp
&＃61;
""
"""推送消息到微信"""
params
&＃61;
"text"
text
"desp"
desp
response
&＃61;
requests
get
server_url
params
&＃61;
params
json_data
&＃61;
response
json
if
json_data
&＃39;errno&＃39;
&＃61;&＃61;
0
print
get_time
&＃43;
" 推送成功。"
else
print
"{0} 推送失败&＃xff1a;{1} \n {2}"
format
get_time
json_data
&＃39;errno&＃39;
json_data
&＃39;errmsg&＃39;
def
push_wx_group
text
desp
"""推送消息到微信群组"""
params
&＃61;
&＃39;sendkey&＃39;
group_push_key
&＃39;text&＃39;
text
&＃39;desp&＃39;
desp
response
&＃61;
requests
post
group_push_url
params
&＃61;
params
json_data
&＃61;
response
json
if
json_data
&＃39;code&＃39;
&＃61;&＃61;
0
print
get_time
&＃43;
&＃39; 微群 &＃39;
&＃43;
json_data
&＃39;data&＃39;
else
print
"{0} 微群 推送失败&＃xff1a;{1} \n {2}"
format
get_time
&＃43;
json_data
&＃39;message&＃39;
json_data
&＃39;data&＃39;
def
push_dd
dd_url
msg
"""推送消息到钉钉"""
headers
&＃61;
&＃39;Content-Type&＃39;
&＃39;application/json;charset&＃61;utf-8&＃39;
response
&＃61;
requests
post
dd_url
data
&＃61;
json
dumps
msg
headers
&＃61;
headers
json_data
&＃61;
response
json
if
json_data
&＃39;errcode&＃39;
&＃61;&＃61;
0
print
get_time
&＃43;
" 钉钉 推送成功。"
else
print
"{0} 钉钉 推送失败&＃xff1a;{1} \n {2}\n{3}"
format
get_time
json_data
&＃39;errcode&＃39;
json_data
&＃39;errmsg&＃39;
dd_url
def
work
# 获取数据
soup
&＃61;
get_soup
weather
&＃61;
get_weather
# 生成
dd_msg
&＃61;
make_dd_soup
soup
weather
wx_msg
&＃61;
make_wx_soup
soup
weather
# 推送 钉钉
for
url
in
dd_urls
push_dd
url
dd_msg
# 推送 微信群组
push_wx_group
wx_msg
0
wx_msg
1
if
__name__
&＃61;&＃61;
&＃39;__main__&＃39;
url
&＃61;
sys
argv
1
if
len
sys
argv
>
1
else
&＃39;sqlite:///jobs.sqlite&＃39;
scheduler
add_jobstore
&＃39;sqlalchemy&＃39;
url
&＃61;
url
# 使用sqlite存储jobs table默认名为apscheduler_jobs
print
get_time
&＃43;
" 开始执行任务"
scheduler
add_job
work
&＃39;cron&＃39;
day_of_week
&＃61;
&＃39;0-6&＃39;
hour
&＃61;
8
minute
&＃61;
00
second
&＃61;
00
id
&＃61;
&＃39;data_id&＃39;
scheduler
start
京公网安备 11010802041100号