147 lines
3.1 KiB
Python
147 lines
3.1 KiB
Python
|
|
import os
|
|
import json
|
|
import hashlib
|
|
from datetime import datetime, timezone
|
|
|
|
import requests
|
|
from bs4 import BeautifulSoup
|
|
|
|
|
|
URL = "https://cet.neea.edu.cn/"
|
|
STATE_FILE = "data/state.json"
|
|
|
|
|
|
def fetch_notices():
|
|
headers = {"User-Agent": "Mozilla/5.0"}
|
|
|
|
for retry in range(3):
|
|
try:
|
|
r = requests.get(URL, headers=headers, timeout=20)
|
|
r.encoding = "utf-8"
|
|
|
|
soup = BeautifulSoup(r.text, "lxml")
|
|
result = []
|
|
|
|
for a in soup.find_all("a"):
|
|
title = a.get_text(strip=True)
|
|
href = a.get("href")
|
|
|
|
if title and any(k in title for k in [
|
|
"四六级", "CET", "成绩",
|
|
"报名", "考试", "准考证"
|
|
]):
|
|
result.append({
|
|
"title": title,
|
|
"url": href
|
|
})
|
|
|
|
return result[:20]
|
|
|
|
except Exception as e:
|
|
print(f"获取失败 {retry + 1}/3:", e)
|
|
|
|
return None
|
|
|
|
|
|
def make_hash(item):
|
|
return hashlib.sha256(
|
|
(item["title"] + str(item["url"])).encode("utf-8")
|
|
).hexdigest()
|
|
|
|
|
|
def load_state():
|
|
if not os.path.exists(STATE_FILE):
|
|
return {"items": []}
|
|
|
|
with open(STATE_FILE, "r", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
|
|
|
|
def save_state(items):
|
|
os.makedirs("data", exist_ok=True)
|
|
|
|
with open(STATE_FILE, "w", encoding="utf-8") as f:
|
|
json.dump({
|
|
"items": items,
|
|
"last_check_utc": datetime.now(timezone.utc).isoformat()
|
|
}, f, ensure_ascii=False, indent=2)
|
|
|
|
|
|
def push(items):
|
|
token = os.getenv("PUSH_TOKEN")
|
|
|
|
if not token:
|
|
print("ERROR: PUSH_TOKEN 未配置")
|
|
return False
|
|
|
|
content = "📢 CET四六级新通知\n\n"
|
|
|
|
for item in items:
|
|
content += (
|
|
f"【{item['title']}】\n"
|
|
f"{item['url']}\n\n"
|
|
)
|
|
|
|
# ShowDoc Push token 在 URL 路径中
|
|
url = f"https://push.showdoc.com.cn/server/api/push/{token}"
|
|
|
|
try:
|
|
r = requests.post(
|
|
url,
|
|
data={
|
|
"title": "四六级新通知",
|
|
"content": content
|
|
},
|
|
timeout=10
|
|
)
|
|
|
|
print("Push status:", r.status_code)
|
|
print("Push response:", r.text)
|
|
|
|
return r.status_code == 200
|
|
|
|
except Exception as e:
|
|
print("Push异常:", e)
|
|
return False
|
|
|
|
|
|
def main():
|
|
notices = fetch_notices()
|
|
|
|
if notices is None:
|
|
print("获取公告失败")
|
|
return
|
|
|
|
state = load_state()
|
|
|
|
old_hash = {
|
|
x["hash"]
|
|
for x in state.get("items", [])
|
|
}
|
|
|
|
force = os.getenv("FORCE_PUSH", "false").lower() == "true"
|
|
|
|
new_items = []
|
|
|
|
for item in notices:
|
|
item["hash"] = make_hash(item)
|
|
|
|
if force or item["hash"] not in old_hash:
|
|
new_items.append(item)
|
|
|
|
if new_items:
|
|
print("发现通知:", len(new_items))
|
|
|
|
if push(new_items):
|
|
print("推送成功")
|
|
save_state(notices)
|
|
else:
|
|
print("推送失败，不更新状态")
|
|
else:
|
|
print("无新增")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|