背景
容易忘记女朋友的纪念日（现在是老婆），为了避免挨批，楼主在几年前写了这个程序，最近又把它拿出来翻修了一下，更新了天气接口。
代码打包分享：见最后面的代码下载
目录结构：
.
├── LICENSE
├── README.md
├── config.dev.yaml
├── config.py
├── requirements.txt
└── send_love_msg.py
使用
1.申请wxpusher（免费,可以跳过，可以选择使用企业微信机器人）
2.高德天气API（免费）
3.创建企业微信机器人（免费）
4.复制配置 config.dev.yaml, 创建配置 config.yaml，将相关的参数填入其中。
5.将代码上传至 Github
6.每天9-10点之间等待它发送消息。
（为什么不设置成0点？GithubAction 定时任务执行延迟问题：GitHub 的官方文档明确写了是无法保证准时执行的。因为 Github 是国际网站，默认时区是 UTC 时间（协调世界时间），大致时间是冰岛和英国的时区。UTC 比北京时间慢 8 小时，用目标时间减 8 就可以了。这也就是为什么我设置的是0点执行，但是最终执行时间是北京时间9点到10点左右 ）
最终效果图
企业微信机器人推送：
企业微信机器人推送
wxpusher 推送：
wxpusher-推送1
wxpusher-推送2
核心代码
import json
from datetime import datetime
import pytz
import requests
from config import loadConfig
# Load configuration
config = loadConfig("config.yaml")
qywxWebhookKey = config.weChatWork.webhookKey
wxpushAppToken = config.wxPusher.appToken
city = config.weather.city
monthOfBirthday = config.lover.monthOfBirthday
dayOfBirthday = config.lover.dayOfBirthday
expressLoveTimestamp = config.lover.expressLoveTimestamp
meetingTimestamp = config.lover.meetingTimestamp
weatherApiKey = config.weather.apiKey
def getMsgHeader():
tz = pytz.timezone("Asia/Shanghai")
dt = datetime.now(tz)
h = '今天是 <font color="info">{}</font>'.format(dt.strftime("%Y-%m-%d %A"))
return h
def getMsgHeaderToWechat():
tz = pytz.timezone("Asia/Shanghai")
dt = datetime.now(tz)
h = '今天是 <font color="#87CEEB">{}</font>'.format(dt.strftime("%Y-%m-%d %A"))
return h
class Weather:
def __init__(self):
self.city = ""
self.adcode = ""
self.province = ""
self.reporttime = ""
self.date = ""
self.week = ""
self.dayweather = ""
self.nightweather = ""
self.daytemp = ""
self.nighttemp = ""
self.daywind = ""
self.nightwind = ""
self.daypower = ""
self.nightpower = ""
def isValide(self) -> bool:
return self.city != ""
def jsonDecode(self, jsonTex):
self.city = jsonTex["city"]
self.adcode = jsonTex["adcode"]
self.province = jsonTex["province"]
self.reporttime = jsonTex["reporttime"]
casts = jsonTex["casts"][0]
self.date = casts["date"]
self.week = casts["week"]
self.dayweather = casts["dayweather"]
self.nightweather = casts["nightweather"]
self.daytemp = casts["daytemp"]
self.nighttemp = casts["nighttemp"]
self.daywind = casts["daywind"]
self.nightwind = casts["nightwind"]
self.daypower = casts["daypower"]
self.nightpower = casts["nightpower"]
def getWeatherTextToWechatWork(self):
tex = '武汉天气\n > <font color="info">{}</font>, 白天温度: <font color="info">{}</font> ~ 晚上温度: <font color="info">{}</font>\n白天风力:{}-{}，晚上风力:{}-{}。'.format(
self.dayweather,
self.daytemp,
self.nighttemp,
self.daypower,
self.daywind,
self.nightpower,
self.nightwind,
)
return tex
def getWeatherTextToWechat(self):
tex = '<hr>武汉天气 <br> <font color="green">{}</font>, 白天温度: <font color="green">{}</font> ~ 晚上温度: <font color="green">{}</font>， 白天风力:{}-{}，晚上风力:{}-{}。'.format(
self.dayweather,
self.daytemp,
self.nighttemp,
self.daypower,
self.daywind,
self.nightpower,
self.nightwind,
)
return tex
def getWeather() -> Weather:
url = "https://restapi.amap.com/v3/weather/weatherInfo"
params = {
"city": city,
"extensions": "all",
"key": weatherApiKey,
}
try:
response = requests.get(url, params=params)
response.raise_for_status() # 检查请求是否成功
data = response.json()
if data.get("status") != "1":
raise ValueError(f"API Error: {data.get('info')}")
forecasts_data = data.get("forecasts", [])
if not forecasts_data:
raise ValueError("No forecasts data available.")
forecast = forecasts_data[0]
weather = Weather()
weather.jsonDecode(forecast)
return weather
except requests.RequestException as e:
print(f"Request error: {e}")
except ValueError as e:
print(f"Value error: {e}")
except Exception as e:
print(f"Unexpected error: {e}")
return Weather() # 返回一个无效的 Weather 对象
def getMeetingDay():
tz = pytz.timezone("Asia/Shanghai")
now = datetime.now(tz)
day = int((now.timestamp() - meetingTimestamp) / (24 * 60 * 60))
print(day)
print("相遇的:", day)
return day
def getBirthDayOfLover():
tz = pytz.timezone("Asia/Shanghai")
yearNow = datetime.now(tz)
dt = datetime(yearNow.year, yearNow.month, yearNow.day)
# 判断今年的生日是否已经过去
birthday = datetime(yearNow.year, monthOfBirthday, dayOfBirthday)
if birthday.timestamp() < yearNow.timestamp():
# 下一年的生日
birthday = datetime(birthday.year + 1, monthOfBirthday, dayOfBirthday)
day = int((birthday.timestamp() - dt.timestamp()) / (24 * 60 * 60))
print("生日:", day)
return day
def getExpressLoveDay():
# unixTimeStamp = 1599148800
tz = pytz.timezone("Asia/Shanghai")
now = datetime.now(tz)
day = int((now.timestamp() - expressLoveTimestamp) / (24 * 60 * 60))
print(day)
print("相爱的天:", day)
return day
class DailyWord:
def __init__(self):
self.sid = ""
self.note = ""
self.content = ""
self.pic = ""
def isValide(self) -> bool:
return self.sid != ""
def getDailyWordHtml(self) -> str:
return '<br>每日一句<br>{}<br>{}<br><img src="{}" align="center">'.format(
self.content, self.note, self.pic
)
def getDailyWord() -> DailyWord:
url = "http://open.iciba.com/dsapi"
r = requests.get(url)
r.encoding = "utf-8"
result = r.json()
dw = DailyWord()
if result.get("sid"):
dw.sid = result["sid"]
dw.note = result["note"]
dw.content = result["content"]
dw.pic = result["fenxiang_img"]
return dw
def sendDailyWordToWechatWork(dw: DailyWord):
if dw.isValide():
webhook = (
f"https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key={qywxWebhookKey}"
)
header = {"Content-Type": "application/json", "Charset": "UTF-8"}
message = {
"msgtype": "news",
"news": {
"articles": [
{
"title": "每日一句",
"description": dw.content,
"url": dw.pic,
"picurl": dw.pic,
}
]
},
}
message_json = json.dumps(message)
requests.post(url=webhook, data=message_json, headers=header)
return
def sendAlarmMsg(mdTex):
wechatwork(mdTex)
def wechatwork(tex):
webhook = f"https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key={qywxWebhookKey}"
header = {"Content-Type": "application/json", "Charset": "UTF-8"}
message = {"msgtype": "markdown", "markdown": {"content": tex}}
print(f"wechat send msg, key:{qywxWebhookKey}")
print(message)
message_json = json.dumps(message)
try:
requests.post(url=webhook, data=message_json, headers=header)
except requests.exceptions.RequestException as e:
print("unable to connect to wechat server, err:", e)
except Exception as e2:
print("send message to wechat server, err:", e2)
sendAlarmMsg(str(e2))
def wxPusher(tex):
url = "http://wxpusher.zjiecode.com/api/send/message"
header = {"Content-Type": "application/json", "Charset": "UTF-8"}
message = {
"appToken": wxpushAppToken,
"content": tex,
"summary": "相爱一生",
"contentType": 2,
"topicIds": [6931],
"url": "http://wxpusher.zjiecode.com",
}
message_json = json.dumps(message)
try:
info = requests.post(url=url, data=message_json, headers=header)
print(info.text)
except requests.exceptions.RequestException as e:
print("unable to connect to wx, err:", e)
sendAlarmMsg(str(e))
except Exception as e:
print("send message to wx, err:", e)
sendAlarmMsg(str(e))
if __name__ == "__main__":
h = getMsgHeader()
w = getWeather()
bd = getBirthDayOfLover()
md = getMeetingDay()
ed = getExpressLoveDay()
dw = getDailyWord()
# 企业微信
w1 = w.getWeatherTextToWechatWork()
# tex1 = '{}\n> 今天是我们相爱的<font color="warning"> {} </font>天\n我们已经相遇<font color="warning"> {}
# </font>天({})\n距离你的生日还有<font color="warning"> {} </font>天\n\n{}'.format(
# h, ed, md,datetime.utcfromtimestamp(meetingTimestamp).strftime('%Y-%m-%d %H:%M:%S') , bd, w1
# )
# 一行代码完成转换和格式化，并插入到原始字符串中
tex1 = (
'{}\n> 今天是我们相爱的<font color="warning"> {} </font>天（{}）\n'
'我们已经相遇<font color="warning"> {} </font>天（{}）\n'
'距离你的生日还有<font color="warning"> {} </font>天'
).format(
h,
ed,
datetime.fromtimestamp(expressLoveTimestamp, tz=pytz.utc)
.astimezone(pytz.timezone("Asia/Shanghai"))
.strftime("%Y-%m-%d"),
md,
datetime.fromtimestamp(meetingTimestamp, tz=pytz.utc)
.astimezone(pytz.timezone("Asia/Shanghai"))
.strftime("%Y-%m-%d"),
bd,
)
wechatwork(tex1)
sendDailyWordToWechatWork(dw)
# wxpusher
h2 = getMsgHeaderToWechat()
w2 = w.getWeatherTextToWechat()
dw2 = dw.getDailyWordHtml()
tex2 = (
'{}<br> 今天是我们相爱的<font color="green"> {} </font>天（{}）<br>'
'我们已经相遇<font color="green">{}</font>天（{}）<br>'
'距离你的生日还有<font color="green"> {} </font>天<br><br>{}<br>{}'
).format(
h2,
ed,
datetime.fromtimestamp(expressLoveTimestamp, tz=pytz.utc)
.astimezone(pytz.timezone("Asia/Shanghai"))
.strftime("%Y-%m-%d"),
md,
datetime.fromtimestamp(meetingTimestamp, tz=pytz.utc)
.astimezone(pytz.timezone("Asia/Shanghai"))
.strftime("%Y-%m-%d"),
bd,
w2,
dw2,
)
配置文件
wxPusher:
appToken: "xxxxxxxxxxxxxxxxxxxxxx" # wxPusher的appToken
topicIds:
- xxxx # wxPusher的topicIds，支持多个ID
wechatWork:
webhookKey: "xxxxxx" # 企业微信的webhook key
weather:
apiKey: "xxxxxxxxxxxxx" # 高德天气接口的key
city: 420100 # https://a.amap.com/lbs/static/code_resource/AMap_adcode_citycode.zip
lover:
expressLoveTimestamp: 1136185445 # 表白的日子
monthOfBirthday: 1 # 出生日期：月
dayOfBirthday: 1 #出生日期：日
meetingTimestamp: 1136185445 # 相遇的日子
代码下载
v1.0.0 初版
通过网盘分享的文件：FunnyCode-1.0.0.zip
链接: https://pan.baidu.com/s/1M8acZNCMHkr7B8Sj3YDScw?pwd=xw6g 提取码: xw6g
v1.0.1
改动：
- 修复 wxpusher topicIds，支持配置
通过网盘分享的文件：FunnyCode-1.0.1.zip
链接: https://pan.baidu.com/s/1O2P4qwpILh-SnYO6PsegOg?pwd=1itx 提取码: 1itx
发表于 2024-8-3 01:34
|
发表于 2024-8-3 01:37
