全文共6162字，閱讀大約需要10分鐘

最近在群里看到一個(gè)好玩的消息推送場景，如下圖所示，原理是在微信或者企業(yè)微信通過調(diào)用官方的接口實(shí)現(xiàn)每日定時(shí)推送消息。今天就帶大家來研究下它是怎么實(shí)現(xiàn)的。

效果圖

整個(gè)代碼會(huì)分幾個(gè)部分來講解

• 日志：為了實(shí)時(shí)監(jiān)測(cè)程序的運(yùn)行狀態(tài)，及后期問題排查

• 天氣API詳解：會(huì)講述如何調(diào)用免費(fèi)的天氣API接口

• Python日期處理：Python中日期轉(zhuǎn)換及日期天數(shù)的計(jì)算

• 完整的消息推送

1.日志

Python日志記錄的代碼，可在任何場景下復(fù)用，它能夠?qū)崟r(shí)監(jiān)測(cè)程序的運(yùn)行狀態(tài)，輕松解決測(cè)試和問題排查的難題。

注意：log_home需要改為自己本地路徑

_tb_nm = '微信每日推送'
_tb_nm_cn = "微信每日推送"
_service_code = _tb_nm
# 日志目錄
log_home = '/home/xusl/log/wx'

# 日志level
log_level = logging.INFO

# 日志打印到控制臺(tái)
log_to_console = True

log_config = {
    'version': 1,
    'formatters': {
        'generic': {
            'format': '%(asctime)s %(levelname)-5.5s [%(name)s:%(lineno)s][%(threadName)s] %(message)s',
        },
        'simple': {
            'format': '%(asctime)s %(levelname)-5.5s %(message)s',
        },
    },
    'handlers': {
        'console': {
            'class': 'logging.StreamHandler',
            'formatter': 'generic',
        },
        'file': {
            'class': 'logging.FileHandler',
            'filename': os.path.join(log_home, 'excel_to_data.log'),
            'encoding': 'utf-8',
            'formatter': 'generic',

        },
    },
    'root': {
        'level': log_level,
        'handlers': ['console', 'file', ] if log_to_console else ['file', ],
    }
}
logging.config.dictConfig(log_config)
logger = logging.getLogger(__name__)

2.天氣API詳解

在這里提供一個(gè)網(wǎng)站 天氣API說明，感謝作者提供了8個(gè)天氣接口，響應(yīng)效率高，可以達(dá)到不限制次數(shù)。關(guān)鍵是免費(fèi)的，JSON兩種方式返回。

接口返回的天氣指數(shù)數(shù)據(jù)很全面，如：溫度、最高溫度、最低溫度、風(fēng)、天氣、空氣質(zhì)量指數(shù)。

參數(shù)只有一個(gè)，就是cityId。

比如上海市的cityId是101020100，獲取天氣的API接口就是http://t.weather.sojson.com/api/weather/city/101020100

訪問這個(gè)地址，返回的數(shù)據(jù)如下：

因?yàn)榉祷氐慕Y(jié)果有近15天的天氣，所以要對(duì)結(jié)果做一個(gè)遍歷，取出今天對(duì)應(yīng)的天氣信息。同時(shí)天氣更新時(shí)間為每天的：3點(diǎn),8點(diǎn),13點(diǎn),19點(diǎn)，所以建議不要凌晨去獲取，加上CDN有1個(gè)小時(shí)的緩存，建議4點(diǎn),9點(diǎn),14點(diǎn),20點(diǎn)后獲取。

注意：因?yàn)槲覀兊某绦蚴敲咳胀扑鸵淮?，所以沒有對(duì)天氣結(jié)果進(jìn)行緩存處理，但如果你的程序需要頻繁調(diào)用天氣接口，為了減少對(duì)方的CDN加速費(fèi)用，一定要在代碼里加入緩存，API接口是每8小時(shí)更新一次，機(jī)制是CDN緩存8小時(shí)更新一次。

城市數(shù)據(jù)請(qǐng)?jiān)诎俣染W(wǎng)盤下載：

鏈接: https://pan.baidu.com/s/1JFAwnH2MRLc5OD3hsJZwGQ 提取碼: u8sk

3.Python日期處理

考慮到程序中有日期轉(zhuǎn)字符串，字符串轉(zhuǎn)日期，日期相減，所以寫了幾個(gè)方法供大家參考，同時(shí)兼顧了國歷和農(nóng)歷生日信息的獲取，具體如下

import datetime

from time import localtime


def get_now_datetime():
    """
    獲取當(dāng)前日期
    :return: datetime now
    """
    return datetime.datetime.now()


def get_datetime_str(d_date=None, pattern='%Y-%m-%d'):
    """
    獲取指定日期 字符格式
    :param d_date:
    :param pattern:
    :return:
    """
    if not d_date:
        d_date = get_now_datetime()
    return datetime.datetime.strftime(d_date, pattern)


def parse_str2date(s_date, pattern='%Y-%m-%d'):
    """
    將字符串轉(zhuǎn)換為日期格式
    :param s_date:
    :param pattern:
    :return:
    """
    return datetime.datetime.strptime(s_date, pattern)


def get_birthday(config, year, today_dt):
    """
    獲取距離下次生日的時(shí)間
    :return:
    """
    logger.info('獲取距離下次生日的時(shí)間...................')
    birthday = config["birth_day"]  # 獲取生日日期
    birthday_year = birthday.split("-")[0]  # 2023 or r2023
    # 將str日期轉(zhuǎn)換為日期型
    # d_birthday = datetime.datetime.strptime(birthday, "%Y-%m-%d")

    # 判斷是否為農(nóng)歷生日
    if birthday_year[0] == "r":
        # 獲取農(nóng)歷生日的今年對(duì)應(yīng)的月和日
        try:
            r_mouth = int(birthday.split("-")[1])
            r_day = int(birthday.split("-")[2])
            nl_birthday = ZhDate(year, r_mouth, r_day).to_datetime().date()
        except TypeError:
            logger.error("請(qǐng)檢查生日的日子是否在今年存在")
            # 調(diào)用系統(tǒng)命令行執(zhí)行 pause 命令，目的是在控制臺(tái)窗口顯示 "請(qǐng)按任意鍵繼續(xù). . ." 的提示信息，并等待用戶按下任意鍵后繼續(xù)執(zhí)行程序
            # os.system("pause")
            sys.exit(1)     # 異常退出

        birthday_month = nl_birthday.month
        birthday_day = nl_birthday.day
        # 今年生日
        year_date = datetime.date(int(year), birthday_month, birthday_day)
    else:
        # 獲取國歷生日的今年對(duì)應(yīng)月和日
        birthday_month = int(birthday.split("-")[1])
        birthday_day = int(birthday.split("-")[2])
        # 獲取國歷生日的今年對(duì)應(yīng)月和日
        year_date = datetime.date(int(year), birthday_month, birthday_day)

    # 計(jì)算生日年份，如果還沒過，按當(dāng)年減，如果過了需要+1
    year_date = get_datetime_str(year_date)
    if today_dt > year_date:
        if birthday_year[0] == "r":
            r_mouth = int(birthday.split("-")[1])
            r_day = int(birthday.split("-")[2])
            # 獲取農(nóng)歷明年生日的月和日
            r_last_birthday = ZhDate((int(year) + 1), r_mouth, r_day).to_datetime().date()
            birth_date = datetime.date((int(year) + 1), r_last_birthday.month, r_last_birthday.day)
            print(type(birth_date))
        else:
            # 獲取國歷明年生日的月和日
            birth_date = datetime.date((int(year) + 1), birthday_month, birthday_day)

            str_birth_date = get_datetime_str(birth_date)
        birth_day = (datetime.datetime.strptime(str_birth_date, "%Y-%m-%d").date() - datetime.datetime.strptime(today_dt, "%Y-%m-%d").date()).days

    elif today_dt == year_date:
        birth_day = 0
    else:
        birth_day = (datetime.datetime.strptime(year_date, "%Y-%m-%d").date() - datetime.datetime.strptime(today_dt, "%Y-%m-%d").date()).days

    if birth_day == 0:
        birthday_data = "生日快樂，祝福你無事絆心弦，所念皆如愿。"
    else:
        birthday_data = "平安喜樂，得償所愿。"

    return birth_day, birthday_data

*注：生日當(dāng)天的文案可根據(jù)自己的風(fēng)格修改??

4.完整的消息推送腳本

整個(gè)消息推送腳本分兩個(gè)文件，分別是配置信息和python腳本，開始之前我們要準(zhǔn)備一下配置信息。

• 申請(qǐng)相關(guān)的信息app_id、app_secret、template_id、user 去微信申請(qǐng)，微信公眾平臺(tái)接口測(cè)試賬號(hào)申請(qǐng)：https://mp.weixin.qq.com/debug/cgi-bin/sandbox?t=sandbox/login，
  

• 如下圖所示，登錄后會(huì)有對(duì)應(yīng)的app_id、app_secret
  

• 然后掃碼關(guān)注，就能看到對(duì)應(yīng)的微信號(hào)信息，這里的微信號(hào)是放在配置文件的user的
  

• 接著是最重要的一步，寫入下方的模板信息，模板ID即為我們template_id
  

?美好的一天開始啦(′⊙ω⊙`) 
?今天是：{{date.DATA}} 
?下面開始為你播報(bào){{city_nm.DATA}}的天氣

?今天的天氣：{{weather.DATA}}
?最高：{{max_temperature.DATA}}
?最低：{{min_temperature.DATA}}

?????：{{glowing_terms.DATA}}

?距離寶貝的生日：{{birth_day.DATA}} 天
???：{{birthday_data.DATA}}

• 最后修改我們配置信息config.txt和wx_message.py。

config.txt

{
# 微信公眾號(hào)配置
"app_id": "xxx",
"app_secret": "xxx",
"template_id": "xxx",
"user": ["xxx"], # 接收消息的微信號(hào)，多個(gè)微信用英文逗號(hào)間隔，例如["wx1", "wx2"]

# 信息配置
"city_id": "101020100",
"city_nm": "上海市",
"birth_day":"2023-08-22",   # 生日若為農(nóng)歷在最前面加上r即可
}

wx_message.py

# Created on 2024/1/19
# @title: '微信公眾號(hào)發(fā)送消息'
# @author: Xusl

import logging.config
import random
import datetime
import sys, json
import os
import requests

from time import localtime
from requests import get, post
from zhdate import ZhDate


_tb_nm = '微信每日推送'
_tb_nm_cn = "微信每日推送"
_service_code = _tb_nm
# 日志目錄
log_home = '/home/xusl/log/wx'

# 日志level
log_level = logging.INFO

# 日志打印到控制臺(tái)
log_to_console = True


log_config = {
    'version': 1,
    'formatters': {
        'generic': {
            'format': '%(asctime)s %(levelname)-5.5s [%(name)s:%(lineno)s][%(threadName)s] %(message)s',
        },
        'simple': {
            'format': '%(asctime)s %(levelname)-5.5s %(message)s',
        },
    },
    'handlers': {
        'console': {
            'class': 'logging.StreamHandler',
            'formatter': 'generic',
        },
        'file': {
            'class': 'logging.FileHandler',
            'filename': os.path.join(log_home, _tb_nm + '.log'),
            'encoding': 'utf-8',
            'formatter': 'generic',

        },
    },
    'root': {
        'level': log_level,
        'handlers': ['console', 'file', ] if log_to_console else ['file', ],
    }
}
logging.config.dictConfig(log_config)
logger = logging.getLogger(_tb_nm)

# 每日一言
lines = [
    "會(huì)好，遲早。",
    "生命幾許，遵從自己，別趕路，感受路。",
    "去愛具體的生活。",
    "拐個(gè)彎，與生活和解，得失都隨意。",
    "不要預(yù)知明天的煩惱。",
    "后來重聞往事如耳旁過風(fēng)，不慌不忙。",
    "勇敢的人先享受世界。",
    "玫瑰不用長高，晚霞自會(huì)俯腰，愛意隨風(fēng)奔跑，溫柔漫過山腰。",
    "春風(fēng)得意馬蹄疾，一日看盡長安花。",
    "你若決定燦爛，山無遮，海無攔。",
    "中途下車的人很多，你不必耿耿于懷。",
    "內(nèi)心豐盈者，獨(dú)行也如眾。",
    "你記得花，花就不怕枯萎。",
    "春日不遲，相逢終有時(shí)。",
    "日升月落總有黎明。",
    "有人等煙雨，有人怪雨急。",
    "等風(fēng)來，不如追風(fēng)去。",
    "真誠永遠(yuǎn)可貴。",
    "喜樂有分享，共度日月長。",
    "在過程中追逐意義。"
]


def get_color():
    # 獲取隨機(jī)顏色
    get_colors = lambda n: list(map(lambda i: "#" + "%06x" % random.randint(0, 0xFFFFFF), range(n)))
    color_list = get_colors(100)
    return random.choice(color_list)


def get_access_token(config):
    logger.info('獲取access_token...................')
    # appId
    app_id = config["app_id"]
    # appSecret
    app_secret = config["app_secret"]
    post_url = ("https://api.weixin.qq.com/cgi-bin/token?grant_type=client_credential&appid={}&secret={}"
                .format(app_id, app_secret))
    try:
        access_token = get(post_url).json()['access_token']
    except KeyError:
        logging.error("獲取access_token失敗，請(qǐng)檢查app_id和app_secret是否正確")
        sys.exit(1)
    return access_token


def get_now_datetime():
    """
    獲取當(dāng)前日期
    :return: datetime now
    """
    return datetime.datetime.now()


def get_datetime_str(d_date=None, pattern='%Y-%m-%d'):
    """
    獲取指定日期 字符格式
    :param d_date:
    :param pattern:
    :return:
    """
    if not d_date:
        d_date = get_now_datetime()
    return datetime.datetime.strftime(d_date, pattern)


def parse_str2date(s_date, pattern='%Y-%m-%d'):
    """
    將字符串轉(zhuǎn)換為日期格式
    :param s_date:
    :param pattern:
    :return:
    """
    return datetime.datetime.strptime(s_date, pattern)


def get_weather_info(config, today_dt):
    """
    獲取城市當(dāng)日天氣
    :return:
    """
    logger.info('獲取天氣...................')
    headers = {
        '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'
    }
    city_id = config["city_id"]
    region_url = "http://t.weather.sojson.com/api/weather/city/{}".format(city_id)
    response = get(region_url, headers=headers).json()

    if response["status"] == 200:
        forecast = response["data"]["forecast"]
        for item in forecast:
            if item["ymd"] == today_dt:
                return True, item
    else:
        logging.error("天氣信息獲取失敗，請(qǐng)檢查天氣API是否正常")
        return False, response["status"]


def get_birthday(config, year, today_dt):
    """
    獲取距離下次生日的時(shí)間
    :return:
    """
    logger.info('獲取距離下次生日的時(shí)間...................')
    birthday = config["birth_day"]  # 獲取生日日期
    birthday_year = birthday.split("-")[0]  # 2023 or r2023
    # 將str日期轉(zhuǎn)換為日期型
    # d_birthday = datetime.datetime.strptime(birthday, "%Y-%m-%d")

    # 判斷是否為農(nóng)歷生日
    if birthday_year[0] == "r":
        # 獲取農(nóng)歷生日的今年對(duì)應(yīng)的月和日
        try:
            r_mouth = int(birthday.split("-")[1])
            r_day = int(birthday.split("-")[2])
            nl_birthday = ZhDate(year, r_mouth, r_day).to_datetime().date()
        except TypeError:
            logger.error("請(qǐng)檢查生日的日子是否在今年存在")
            # 調(diào)用系統(tǒng)命令行執(zhí)行 pause 命令，目的是在控制臺(tái)窗口顯示 "請(qǐng)按任意鍵繼續(xù). . ." 的提示信息，并等待用戶按下任意鍵后繼續(xù)執(zhí)行程序
            # os.system("pause")
            sys.exit(1)     # 異常退出

        birthday_month = nl_birthday.month
        birthday_day = nl_birthday.day
        # 今年生日
        year_date = datetime.date(int(year), birthday_month, birthday_day)
    else:
        # 獲取國歷生日的今年對(duì)應(yīng)月和日
        birthday_month = int(birthday.split("-")[1])
        birthday_day = int(birthday.split("-")[2])
        # 獲取國歷生日的今年對(duì)應(yīng)月和日
        year_date = datetime.date(int(year), birthday_month, birthday_day)

    # 計(jì)算生日年份，如果還沒過，按當(dāng)年減，如果過了需要+1
    year_date = get_datetime_str(year_date)
    if today_dt > year_date:
        if birthday_year[0] == "r":
            r_mouth = int(birthday.split("-")[1])
            r_day = int(birthday.split("-")[2])
            # 獲取農(nóng)歷明年生日的月和日
            r_last_birthday = ZhDate((int(year) + 1), r_mouth, r_day).to_datetime().date()
            birth_date = datetime.date((int(year) + 1), r_last_birthday.month, r_last_birthday.day)
            print(type(birth_date))
        else:
            # 獲取國歷明年生日的月和日
            birth_date = datetime.date((int(year) + 1), birthday_month, birthday_day)

            str_birth_date = get_datetime_str(birth_date)
        birth_day = (datetime.datetime.strptime(str_birth_date, "%Y-%m-%d").date() - datetime.datetime.strptime(today_dt, "%Y-%m-%d").date()).days

    elif today_dt == year_date:
        birth_day = 0
    else:
        birth_day = (datetime.datetime.strptime(year_date, "%Y-%m-%d").date() - datetime.datetime.strptime(today_dt, "%Y-%m-%d").date()).days

    if birth_day == 0:
        birthday_data = "生日快樂，祝福你無事絆心弦，所念皆如愿。"
    else:
        birthday_data = "平安喜樂，得償所愿。"

    return birth_day, birthday_data


def get_image_url():
    url = "https://cn.bing.com/HPImageArchive.aspx?format=js&idx=0&n=1"
    headers = {
        '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',
        'Content-type': 'application/x-www-form-urlencoded'

    }
    r = requests.get(url, headers=headers, verify=False)
    r.encoding = 'UTF-8-sig'
    image_url = "https://cn.bing.com" + json.loads(r.text)["images"][0]["url"]

    return image_url


def send_message(to_user, access_token, template_id, result, city_nm, birth_day, birthday_data):
    """
    發(fā)送微信通知
    :param to_user:
    :param access_token:
    :param template_id:
    :param result:
    :param city_nm:
    :param birth_day:
    :param birthday_data:
    :return:
    """
    logger.info('發(fā)送微信通知...................')
    weather = result["type"]  # 天氣
    max_temperature = result["high"]  # 高溫
    min_temperature = result["low"]  # 低溫
    glowing_terms = random.choice(lines)  # 每日一言

    url = "https://api.weixin.qq.com/cgi-bin/message/template/send?access_token={}".format(access_token)
    week_list = ["星期日", "星期一", "星期二", "星期三", "星期四", "星期五", "星期六"]
    year = localtime().tm_year  # 年 2024
    month = localtime().tm_mon  # 月 1
    day = localtime().tm_mday  # 日 19

    today = datetime.date(year=year, month=month, day=day)
    logging.info('today：%s ' % today)
    week = week_list[today.isoweekday() % 7]
    logging.info('week：%s ' % week)

    logging.info('城市：{}，天氣：{}，高溫：{}，低溫：{}'.format(city_nm, weather, max_temperature, min_temperature))
    data = {
        "touser": to_user,
        "template_id": template_id,
        "url": "http://weixin.qq.com/download",
        "topcolor": "#FF0000",
        "data": {
            "date": {
                "value": "{} {}".format(today, week),
                "color": get_color()
            },
            "city_nm": {
                "value": city_nm,
                "color": get_color()
            },
            "weather": {
                "value": weather,
                "color": get_color()
            },
            "max_temperature": {
                "value": max_temperature,
                "color": get_color()
            },
            "min_temperature": {
                "value": min_temperature,
                "color": get_color()
            },
            "glowing_terms": {
                "value": glowing_terms,
                "color": get_color()
            },
            "birth_day": {
                "value": birth_day,
                "color": get_color()
            },
            "birthday_data": {
                "value": birthday_data,
                "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:
        logger.error("推送消息失敗，請(qǐng)檢查模板id是否正確")
    elif response["errcode"] == 40036:
        logger.error("推送消息失敗，請(qǐng)檢查模板id是否為空")
    elif response["errcode"] == 40003:
        logger.error("推送消息失敗，請(qǐng)檢查微信號(hào)是否正確")
    elif response["errcode"] == 0:
        logger.info("推送消息成功")
    else:
        logger.info(response)


if __name__ == '__main__':
    today_dt = get_datetime_str()       # 獲取當(dāng)日日期
    t_year = today_dt.split("-")[0]       # 當(dāng)年

    try:
        with open("config.txt", encoding="utf-8") as f:
            config = eval(f.read())
            access_token = get_access_token(config)
            birth_day, birthday_data = get_birthday(config, t_year, today_dt)    # 生日祝福語
            city_nm = config["city_nm"]                 # 城市名
            # 獲取城市當(dāng)日天氣
            flag, result = get_weather_info(config, today_dt)
            template_id = config["template_id"]         # 模板ID
            # 接收的用戶
            to_user = config["user"]                    # 用戶里誒奧

            if flag is True:
                logging.info(f'天氣獲取成功 {result}: {str(result)}')
                for user in to_user:
                    send_message(user, access_token, template_id, result, city_nm, birth_day, birthday_data)
            else:
                logging.error(f'異常 {result}: {str(result)}')
    except FileNotFoundError:
        logging.error("推送消息失敗，請(qǐng)檢查config.txt文件是否與程序位于同一路徑")

5.結(jié)尾

整個(gè)程序盡可能在代碼里體現(xiàn)了備注信息，仍然有幾點(diǎn)美中不足

• 每日一言部分太少，解決方案可以優(yōu)化成獲取相關(guān)網(wǎng)站的數(shù)據(jù)保證每天不是重復(fù)的，或者可以直接擴(kuò)大詞庫lines。

• 抬頭部分不能自定義修改，最早的想法是改成自己的公眾號(hào)，每日定時(shí)推送，研究發(fā)現(xiàn)公眾號(hào)不能自定義模板，只能從官方的模板里挑選，局限性就太大了。

  
  
  
  

  解決方法可以把代碼移植至企業(yè)微信，這樣抬頭支持自定義，換湯不換藥，唯一需要更改的就是申請(qǐng)注冊(cè)企業(yè)微信，同時(shí)更換為企業(yè)微信相關(guān)配置信息，如果時(shí)間允許，我盡量再出一版企業(yè)微信的教程。(? ??_??)?

• 最后的定時(shí)任務(wù)就不再過多詳解了，直接使用服務(wù)器的crontab即可

最后的最后，希望單身的朋友有雙向暗戀，早日追到心選，早日心動(dòng)。希望不單身的朋友彼此珍惜，和對(duì)象長久。希望所有人都能擁有愛，所有人都能被愛擁有。