【原创】Python小实例(节假日微信推送)
本帖最后由 流月汐志 于 2020-5-8 18:26 编辑# 效果图
[!(https://s2.ax1x.com/2020/01/13/lHMq2t.md.jpg)](https://imgchr.com/i/lHMq2t)
# 整体思路
1. 从国务办网站爬取所有URL到本地
> chinese_holiday_urls.json
```json
{
"20191121": [
"国务院办公厅关于2020年部分节假日安排的通知",
"http://www.gov.cn/zhengce/content/2019-11/21/content_5454164.htm"
],
...
}
```
2. 读取选择本年度的发布URL
> 读取标题包含本年(2020)的URL
> 每年可能发布多次文件
3. 爬取发布关于本年的URL
> 正则取数据[节假日，多少天，调休日]
> chinese_holiday.json
```json
{
"holidays": {
"20200101": [
"元旦",
"星期三",
"1"
],
...
},
"workdays": {
"20200119": [
"调休日",
"星期天"
],
...
}
```
4. 节假日数据整合
> 把周末加入到以上的 `holidays` 字段
> 然后 `holidays` 字段所有数据按 key 重新排序
> 剔除掉加入的周末数据里的调休日
> 将最后的数据重新保存到 Json 文件
> chinese_holiday.json
```json
{
"holidays": {
"20200101": [
"元旦",
"星期三",
"1"
],
"20200104": [
"星期六"
],
...
},
"workdays": {
"20200119": [
"调休日",
"星期天"
],
...
}
```
5. 推送
> 使用 apscheduler 每天一次从 chinese_holiday.json 读取数据推送
> 使用 apscheduler 每月一次从网站爬取更新数据到两个 Json 文件
> 使用 wxpusher 推送消息
# 代码部分
## 导入库
```python
import os
import re
import sys
import json
import requests
import calendar
from collections import OrderedDict
from bs4 import BeautifulSoup
from datetime import datetime, timedelta
from wxpusher import WxPusher
from apscheduler.schedulers.blocking import BlockingScheduler
```
## 从国务办网站爬取所有URL到本地
> 知识点01： requests|BeautifulSoup 基础网页处理
> 知识点02： Json 文件的保存与读取
> + ensure_ascii=False 设置为 False, 会禁止中文转码为 Unicode 码
> + indent=4 设置 Json 每行缩进为 4 格
> + sort_keys=True 设置保存 json 文件时，按key 排序保存
```python
###############################################################################
# 保存所有节假日发布网址 url
def SaveUrl(url):
respose_html = requests.get(url, params=PARAMS, headers=HEADERS)
soup_html = BeautifulSoup(respose_html.content, "html.parser")
info_elems = soup_html.find_all("td", attrs={"class": "info"})
# today_str = datetime.today().__format__('%Y年%m月%d日')
for info_elem in info_elems:
push_date_str = info_elem.find_all("li")[-1].text
push_date_str = push_date_str.split('：')[-1].strip(' ')
push_date_str = ''.join(re.findall('+', push_date_str))
href_url = info_elem.find('a')['href']
title = info_elem.find('a').text
if os.path.exists(JSON_FILE) and os.path.getsize(JSON_FILE):
with open(JSON_FILE, 'r', encoding='utf-8') as ff:
json_obj = json.load(ff)
if push_date_str in json_obj.keys():
break
else:
json_obj =
with open(JSON_FILE, 'w', encoding='utf-8') as ff:
json.dump(json_obj, ff, indent=4, sort_keys=True, ensure_ascii=False)
else:
json_obj = {}
json_obj =
with open(JSON_FILE, 'w', encoding='utf-8') as ff:
json.dump(json_obj, ff, indent=4, sort_keys=True, ensure_ascii=False)
return JSON_FILE
```
## 读取并爬取需要的URL
> 知识点：正则，正则，还是正则！
> 官方文档：https://docs.python.org/zh-cn/3/library/re.html
```python
###############################################################################
# 爬取当年的节假日发布网址
# HOLIDAY_DIC = {"20200101":["元旦", "星期三"], }
# WORKDAY_DIC = {"20200119": ["调休","星期天"],}
def CrawPage(href):
global PARAMS, HEADERS, THIS_YEAR, HOLIDAY_DIC, WEEKDAYS
respose_html = requests.get(href, params=PARAMS, headers=HEADERS)
soup_html = BeautifulSoup(respose_html.content, "html.parser")
info_elems = soup_html.find_all("p")
for info in info_elems:
text = info.text
regx = '^.{2}(?P<holiday>.*)：(\d{4}年)?(?P<startday>\d+月\d+日)至?' \
'(\d+月)?(?P<endday>\d+日)?放假(调休)?，共(?P<offdays>+天)。' \
'(?P<ondays>(\d+月\d+日（星期.）、?)+上班。)?$'
re_obj = re.search(regx, text)
if re_obj:
# 春节
holiday = re_obj.group('holiday')
# 1月24日
startday = re_obj.group('startday')
startday = str(THIS_YEAR) + ''.join(format_date(startday))
# month = re.search('\d+月', startday).group(0)
# 1月30日
# endday = re_obj.group('endday')
# if endday is not None:
# endday = month + endday
# 休假 7 天
offdays = re_obj.group('offdays')
offdays = int(re.sub('\D', '', offdays))
start_dt = datetime.strptime(startday, "%Y%m%d")
# 放假的日期列表
offdates = list(gen_dates(start_dt, offdays))
for offday in offdates:
HOLIDAY_DIC = re.split('、', holiday) + , str(offdays)]
# 调休['1月19日','2月1日']
ondays = re_obj.group('ondays')
if ondays is not None:
ondays = re.findall('\d+月\d+日', ondays)
for onday in ondays:
onday = str(THIS_YEAR) + ''.join(format_date(onday))
WORKDAY_DIC = ["调休日", WEEKDAYS]
```
## 节假日数据整合
> 知识点： calendar库的使用
```python
###############################################################################
# 数据处理
# WEEKEND_DIC = {"20200104": "星期六", "20200104": "星期天"}
def All_WEEEK():
global WEEKEND_DIC, THIS_YEAR, THIS_MONTH
for month in range(THIS_MONTH, 13):
month_cal = calendar.monthcalendar(THIS_YEAR, month)
for week in month_cal:
sa = week[-2]
su = week[-1]
if sa != 0:
date_str = ''.join(format_date(str(month) + '-' + str(sa)))
date_str = str(THIS_YEAR) + date_str
WEEKEND_DIC = "星期六"
if su != 0:
date_str = ''.join(format_date(str(month) + '-' + str(su)))
date_str = str(THIS_YEAR) + date_str
WEEKEND_DIC = "星期天"
return WEEKEND_DIC
# MULTI_DIC = {}
def HolidayMain():
global HOLIDAY_DIC, WORKDAY_DIC
# 计算所有节假日和周末
WEEKEND_DIC = All_WEEEK()
for dd in WEEKEND_DIC.keys():
if dd not in HOLIDAY_DIC:
HOLIDAY_DIC = ]
# 节假日按时间key排序
TEMP_DIC = HOLIDAY_DIC
HOLIDAY_DIC = OrderedDict()
for key in sorted(TEMP_DIC.keys()):
HOLIDAY_DIC = TEMP_DIC
# 剔除调休日
for key in WORKDAY_DIC.keys():
if key in HOLIDAY_DIC.keys():
HOLIDAY_DIC.pop(key)
MULTI_DIC['holidays'] = HOLIDAY_DIC
MULTI_DIC['workdays'] = WORKDAY_DIC
# 保存到 json
with open(BASE_FILE, 'w', encoding='utf-8') as ff:
json.dump(MULTI_DIC, ff, indent=4, sort_keys=True, ensure_ascii=False)
```
## 微信推送
> 需要去 wxpusher 创建应用
> 使用 markdown 格式发送
> 主要需要从 json 文件中计算出需要的数据
```python
MESSAGE = """### 节假日推送
> 今天:
> + {today_key}
> + {today_value}
>
> 明天:
> + {tomorrow_key}
> + {tomorrow_value}
>
> 下一个节假日:
> + {next_key}
> + {next_value}
>
> 今年剩余节假日:
> + {last_holidays} 天
"""
def MsgPusher():
# 查询用户获取 UID
query_user_ret = WxPusher.query_user(1, 100, APP_TOKEN)
users_info = query_user_ret['data']['records']
for each_user in users_info:
UID_LIST.append(each_user['uid'])
with open(BASE_FILE, 'r', encoding='utf-8') as ff:
multi_json = json.load(ff)
today_key = DATE_NOW.strftime("%Y%m%d")
today_value = PushInfo(today_key, multi_json)
tomorrow_key = (DATE_NOW + timedelta(days=1)).strftime("%Y%m%d")
tomorrow_value = PushInfo(tomorrow_key, multi_json)
# 计算下个节假日
for key in multi_json['holidays'].keys():
value = multi_json['holidays']
if today_key in multi_json['holidays'].keys():
t_value = multi_json['holidays']
else:
t_value = None
if key > today_key and (t_value is None or value != t_value):
if len(value) > 2:
next_key = key
# 计算相差多少天
next_dt = datetime.strptime(next_key, "%Y%m%d")
today_dt = datetime.strptime(today_key, "%Y%m%d")
diff_days = (next_dt - today_dt).days
next_value = "|".join(value[:-1]) + \
", 再过 {} 天就可以玩 {} 天了哦！". \
format(diff_days, value[-1])
break
else:
next_key = None
next_value = None
else:
next_key = None
next_value = None
# 计算今年剩余节假日天数
temp_list = []
for key in multi_json['holidays'].keys():
value = multi_json['holidays']
if key > today_key:
if len(value) > 2:
# 以元组的形式保存到set
temp_list.append(value)
last_holidays = len(temp_list)
message = MESSAGE.format(
today_key=today_key, today_value=today_value,
tomorrow_key=tomorrow_key, tomorrow_value=tomorrow_value,
next_key=next_key, next_value=next_value,
last_holidays=str(last_holidays))
print(">>> 开始推送消息")
# print(message)
result = WxPusher.send_message(message, UID_LIST, APP_TOKEN, content_type=3)
print(result)
print("<<< 推送消息完成")
```
为什么最后推送消息的地方WxPusher.send_message(message, UID_LIST, APP_TOKEN, content_type=3)我按你的格式写总是提示有问题wxpusher.exceptions.WxPusherNoneTokenException
我的写法WxPusher.send_message("hello,world",uids='UidXXXXX',appToken='AT_XXXX',contentType=3)
请问哪里出问题了 不羁的阳光 发表于 2021-7-23 23:56
为什么最后推送消息的地方WxPusher.send_message(message, UID_LIST, APP_TOKEN, content_type=3)我按你的 ...
搞定了，看了GIHUB文档 使劲顶楼猪的屁屁一下下 这个可以啊 感谢分享 听书解锁版 学习借鉴一下，谢谢分享 有些意思，学习了 kun5815 发表于 2020-1-14 19:24
调试了下，忘了修改时间回来了 牛的一批
页:
[1]
2