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

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

效果圖

整個代碼會分幾個部分來講解

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

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

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

• 完整的消息推送

1.日志

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

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

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

# 日志level
log_level = logging.INFO

# 日志打印到控制臺
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詳解

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

接口返回的天氣指數(shù)數(shù)據(jù)很全面瞧挤，如：溫度、最高溫度儡湾、最低溫度特恬、風(fēng)、天氣徐钠、空氣質(zhì)量指數(shù)癌刽。

參數(shù)只有一個，就是cityId尝丐。

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

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

因為返回的結(jié)果有近15天的天氣爹袁，所以要對結(jié)果做一個遍歷远荠，取出今天對應(yīng)的天氣信息。同時天氣更新時間為每天的：3點,8點,13點,19點失息，所以建議不要凌晨去獲取譬淳，加上CDN有1個小時的緩存，建議4點,9點,14點,20點后獲取盹兢。

注意：因為我們的程序是每日推送一次邻梆，所以沒有對天氣結(jié)果進行緩存處理，但如果你的程序需要頻繁調(diào)用天氣接口蛤迎，為了減少對方的CDN加速費用确虱，一定要在代碼里加入緩存，API接口是每8小時更新一次替裆，機制是CDN緩存8小時更新一次校辩。

城市數(shù)據(jù)請在百度網(wǎng)盤下載：

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

3.Python日期處理

考慮到程序中有日期轉(zhuǎn)字符串窘问，字符串轉(zhuǎn)日期，日期相減宜咒，所以寫了幾個方法供大家參考惠赫，同時兼顧了國歷和農(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):
    """
    獲取距離下次生日的時間
    :return:
    """
    logger.info('獲取距離下次生日的時間...................')
    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)歷生日的今年對應(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("請檢查生日的日子是否在今年存在")
            # 調(diào)用系統(tǒng)命令行執(zhí)行 pause 命令故黑，目的是在控制臺窗口顯示 "請按任意鍵繼續(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:
        # 獲取國歷生日的今年對應(yīng)月和日
        birthday_month = int(birthday.split("-")[1])
        birthday_day = int(birthday.split("-")[2])
        # 獲取國歷生日的今年對應(yīng)月和日
        year_date = datetime.date(int(year), birthday_month, birthday_day)

    # 計算生日年份，如果還沒過场晶，按當(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.完整的消息推送腳本

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

• 申請相關(guān)的信息app_id半醉、app_secret、template_id劝术、user 去微信申請缩多，微信公眾平臺接口測試賬號申請：https://mp.weixin.qq.com/debug/cgi-bin/sandbox?t=sandbox/login，
  

• 如下圖所示养晋，登錄后會有對應(yīng)的app_id瞧壮、app_secret
  

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

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

?美好的一天開始啦(′⊙ω⊙`) 
?今天是：{{date.DATA}} 
?下面開始為你播報{{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

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

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

wx_message.py

# Created on 2024/1/19
# @title: '微信公眾號發(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

# 日志打印到控制臺
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 = [
    "會好蛾娶，遲早。",
    "生命幾許潜秋，遵從自己蛔琅，別趕路，感受路峻呛。",
    "去愛具體的生活罗售。",
    "拐個彎辜窑，與生活和解，得失都隨意寨躁。",
    "不要預(yù)知明天的煩惱穆碎。",
    "后來重聞往事如耳旁過風(fēng)，不慌不忙职恳。",
    "勇敢的人先享受世界所禀。",
    "玫瑰不用長高，晚霞自會俯腰放钦，愛意隨風(fēng)奔跑色徘，溫柔漫過山腰。",
    "春風(fēng)得意馬蹄疾操禀，一日看盡長安花褂策。",
    "你若決定燦爛，山無遮床蜘，海無攔辙培。",
    "中途下車的人很多，你不必耿耿于懷邢锯。",
    "內(nèi)心豐盈者扬蕊，獨行也如眾。",
    "你記得花丹擎，花就不怕枯萎尾抑。",
    "春日不遲，相逢終有時蒂培。",
    "日升月落總有黎明再愈。",
    "有人等煙雨，有人怪雨急护戳。",
    "等風(fēng)來翎冲，不如追風(fēng)去。",
    "真誠永遠可貴媳荒。",
    "喜樂有分享抗悍，共度日月長。",
    "在過程中追逐意義钳枕。"
]


def get_color():
    # 獲取隨機顏色
    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失敗缴渊，請檢查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("天氣信息獲取失敗，請檢查天氣API是否正常")
        return False, response["status"]


def get_birthday(config, year, today_dt):
    """
    獲取距離下次生日的時間
    :return:
    """
    logger.info('獲取距離下次生日的時間...................')
    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)歷生日的今年對應(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("請檢查生日的日子是否在今年存在")
            # 調(diào)用系統(tǒng)命令行執(zhí)行 pause 命令鱼炒，目的是在控制臺窗口顯示 "請按任意鍵繼續(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:
        # 獲取國歷生日的今年對應(yīng)月和日
        birthday_month = int(birthday.split("-")[1])
        birthday_day = int(birthday.split("-")[2])
        # 獲取國歷生日的今年對應(yīng)月和日
        year_date = datetime.date(int(year), birthday_month, birthday_day)

    # 計算生日年份，如果還沒過，按當(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("推送消息失敗糟把，請檢查模板id是否正確")
    elif response["errcode"] == 40036:
        logger.error("推送消息失敗绢涡，請檢查模板id是否為空")
    elif response["errcode"] == 40003:
        logger.error("推送消息失敗，請檢查微信號是否正確")
    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("推送消息失敗遣疯，請檢查config.txt文件是否與程序位于同一路徑")

5.結(jié)尾

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

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

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

  
  
  
  

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

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

最后的最后抗蠢，希望單身的朋友有雙向暗戀，早日追到心選思劳，早日心動。希望不單身的朋友彼此珍惜妨猩，和對象長久潜叛。希望所有人都能擁有愛，所有人都能被愛擁有。