效果图

整体思路

从国务办网站爬取所有URL到本地

chinese_holIDAy_urls.json

{

"20191121": [

"国务院办公厅关于2020年部分节假日安排的通知",

"http://www.gov.cn/zhengce/content/2019-11/21/content_5454164.htm"

],

...

}

读取选择本年度的发布URL

读取标题包含本年(2020)的URL

每年可能发布多次文件

爬取发布关于本年的URL

正则取数据[节假日，多少天，调休日]

chinese_holiday.json

{

"holidays": {

"20200101": [

"元旦",

"星期三",

"1"

],

...

},

"workdays": {

"20200119": [

"调休日",

"星期天"

],

...

}

节假日数据整合

把周末加入到以上的 holidays 字段

然后 holidays 字段所有数据按 key 重新排序

剔除掉加入的周末数据里的调休日

将最后的数据重新保存到 Json 文件

chinese_holiday.json

{

"holidays": {

"20200101": [

"元旦",

"星期三",

"1"

],

"20200104": [

"星期六"

],

...

},

"workdays": {

"20200119": [

"调休日",

"星期天"

],

...

}

推送

使用 apscheduler 每天一次从 chinese_holiday.json 读取数据推送

使用 apscheduler 每月一次从网站爬取更新数据到两个 Json 文件

使用 wxpusher 推送消息

代码部分

导入库

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 排序保存

###############################################################################

# 保存所有节假日发布网址 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('[0-9]+', 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[push_date_str] = [title, href_url]

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[push_date_str] = [title, href_url]

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

###############################################################################

# 爬取当年的节假日发布网址

# 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.*)：(\d{4}年)?(?P\d+月\d+日)至?' \

'(\d+月)?(?P\d+日)?放假(调休)?，共(?P[1-9]+天)。' \

'(?P(\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[offday] = re.split('、', holiday) + [WEEKDAYS[get_week(offday)], 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[onday] = ["调休日", WEEKDAYS[get_week(onday)]]

节假日数据整合

知识点： calendar库的使用

###############################################################################

# 数据处理

# 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[date_str] = "星期六"

if su != 0:

date_str = ''.join(format_date(str(month) + '-' + str(su)))

date_str = str(THIS_YEAR) + date_str

WEEKEND_DIC[date_str] = "星期天"

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[dd] = [WEEKEND_DIC[dd]]

# 节假日按时间key排序

TEMP_DIC = HOLIDAY_DIC

HOLIDAY_DIC = OrderedDict()

for key in sorted(TEMP_DIC.keys()):

HOLIDAY_DIC[key] = TEMP_DIC[key]

# 剔除调休日

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 文件中计算出需要的数据

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'][key]

if today_key in multi_json['holidays'].keys():

t_value = multi_json['holidays'][today_key]

else:

t_value = None

if key > today_key and (t_value is None or value[0] != t_value[0]):

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'][key]

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("<<< 推送消息完成")