162 lines
3.4 KiB
Python
162 lines
3.4 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"
|
|
PUSH_URL = "https://push.showdoc.com.cn/server/api/push"
|
|
|
|
|
|
def fetch_notices():
|
|
headers = {
|
|
"User-Agent": "Mozilla/5.0"
|
|
}
|
|
|
|
for i in range(3):
|
|
try:
|
|
r = requests.get(URL, headers=headers, timeout=20)
|
|
r.encoding = "utf-8"
|
|
|
|
soup = BeautifulSoup(r.text, "lxml")
|
|
|
|
notices = []
|
|
|
|
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", "成绩",
|
|
"报名", "考试", "准考证"
|
|
]):
|
|
notices.append({
|
|
"title": title,
|
|
"url": href
|
|
})
|
|
|
|
return notices[:20]
|
|
|
|
except Exception as e:
|
|
print(f"请求失败 {i+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": datetime.now(timezone.utc).isoformat()
|
|
}, f, ensure_ascii=False, indent=2)
|
|
|
|
|
|
def category(title):
|
|
if "报名" in title:
|
|
return "📝 报名"
|
|
if "成绩" in title:
|
|
return "📊 成绩"
|
|
if "准考证" in title:
|
|
return "🎫 准考证"
|
|
if "考试" in title:
|
|
return "📅 考试安排"
|
|
return "📢 其他"
|
|
|
|
|
|
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"{category(item['title'])}\n"
|
|
f"【{item['title']}】\n"
|
|
f"{item['url']}\n\n"
|
|
)
|
|
|
|
try:
|
|
r = requests.post(
|
|
PUSH_URL,
|
|
data={
|
|
"token": token,
|
|
"title": "四六级新通知",
|
|
"content": content
|
|
},
|
|
timeout=10
|
|
)
|
|
|
|
print("Push status:", r.status_code)
|
|
print("Push response:", r.text)
|
|
|
|
return r.ok
|
|
|
|
except Exception as e:
|
|
print("Push异常:", e)
|
|
return False
|
|
|
|
|
|
def main():
|
|
current = fetch_notices()
|
|
|
|
if current is None:
|
|
print("获取公告失败，本轮跳过")
|
|
return
|
|
|
|
state = load_state()
|
|
|
|
old_hash = {
|
|
x["hash"]
|
|
for x in state.get("items", [])
|
|
}
|
|
|
|
new_items = []
|
|
|
|
for item in current:
|
|
item["hash"] = make_hash(item)
|
|
|
|
if item["hash"] not in old_hash:
|
|
new_items.append(item)
|
|
|
|
if not new_items:
|
|
print("没有新公告")
|
|
save_state(current)
|
|
return
|
|
|
|
print("发现新公告:", len(new_items))
|
|
|
|
if push(new_items):
|
|
print("推送成功，保存状态")
|
|
save_state(current)
|
|
else:
|
|
print("推送失败，不更新状态，等待下次重试")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|