Compare commits

..
3 Commits
3 changed files with 90 additions and 45 deletions
+8 -6
View File
@@ -1,9 +1,8 @@
name: CET Monitor
on:
schedule:
- cron: "17 * * * *"
- cron: '*/30 6-18 * * *'
workflow_dispatch:
inputs:
@@ -18,18 +17,21 @@ jobs:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Checkout
uses: actions/checkout@v4
- uses: actions/setup-python@v5
- name: Setup Python
uses: actions/setup-python@v5
with:
python-version: "3.12"
- run: pip install -r requirements.txt
- name: Install
run: pip install -r requirements.txt
- name: Run monitor
env:
PUSH_TOKEN: ${{ secrets.PUSH_TOKEN }}
FORCE_PUSH: ${{ inputs.force_push }}
FORCE_PUSH: ${{ github.event.inputs.force_push }}
run: python main.py
- name: Commit state UTC
+1 -1
View File
@@ -101,5 +101,5 @@
"hash": "6084abd031ee4d800065b9648e665982dce30b0de38d553af6df7b3f828a0d3e"
}
],
"last_check": "2026-08-05T04:10:00.906883+00:00"
"last_check_utc": "2026-08-05T04:16:50.540062+00:00"
}
+81 -38
View File
@@ -7,44 +7,60 @@ 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():
r = requests.get(URL, headers={"User-Agent": "Mozilla/5.0"}, timeout=20)
r.encoding = "utf-8"
headers = {"User-Agent": "Mozilla/5.0"}
soup = BeautifulSoup(r.text, "lxml")
result = []
for retry in range(3):
try:
r = requests.get(URL, headers=headers, timeout=20)
r.encoding = "utf-8"
for a in soup.find_all("a"):
title = a.get_text(strip=True)
if title and any(k in title for k in ["四六级", "CET", "成绩", "报名", "考试", "准考证"]):
result.append({
"title": title,
"url": a.get("href")
})
soup = BeautifulSoup(r.text, "lxml")
result = []
return result[:20]
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 h(item):
def make_hash(item):
return hashlib.sha256(
(item["title"] + str(item["url"])).encode()
(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, encoding="utf-8") as f:
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,
@@ -54,49 +70,76 @@ def save_state(items):
def push(items):
token = os.getenv("PUSH_TOKEN")
if not token:
print("PUSH_TOKEN missing")
print("ERROR: PUSH_TOKEN 未配置")
return False
content = "📢 CET四六级新通知\n\n"
for x in items:
content += f"{x['title']}\n{x['url']}\n\n"
for item in items:
content += (
f"{item['title']}\n"
f"{item['url']}\n\n"
)
r = requests.post(
PUSH_URL,
data={
"token": token,
"title": "四六级新通知",
"content": content
},
timeout=10
)
# ShowDoc Push token 在 URL 路径中
url = f"https://push.showdoc.com.cn/server/api/push/{token}"
print(r.status_code, r.text)
return r.ok
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 = {x["hash"] for x in state.get("items", [])}
old_hash = {
x["hash"]
for x in state.get("items", [])
}
force = os.getenv("FORCE_PUSH", "false").lower() == "true"
new = []
new_items = []
for item in notices:
item["hash"] = h(item)
if force or item["hash"] not in old:
new.append(item)
item["hash"] = make_hash(item)
if new:
if push(new):
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("no update")
print("无新增")
if __name__ == "__main__":