From bcea6361faedecdc4fd00a10a980560834ebdd3b Mon Sep 17 00:00:00 2001 From: biss Date: Wed, 5 Aug 2026 12:07:02 +0800 Subject: [PATCH] =?UTF-8?q?=E5=B0=9D=E8=AF=95=E4=BF=AE=E5=A4=8Dpush?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitea/workflows/monitor.yml | 22 +++- main.py | 198 ++++++++++++++++++++++++++--------- 2 files changed, 166 insertions(+), 54 deletions(-) diff --git a/.gitea/workflows/monitor.yml b/.gitea/workflows/monitor.yml index 8df5c88..1e56441 100644 --- a/.gitea/workflows/monitor.yml +++ b/.gitea/workflows/monitor.yml @@ -1,23 +1,35 @@ + name: CET Monitor + on: schedule: - cron: "17 * * * *" + workflow_dispatch: + jobs: monitor: runs-on: ubuntu-latest + steps: - - uses: actions/checkout@v4 - - uses: actions/setup-python@v5 + - name: Checkout + uses: actions/checkout@v4 + + - name: Setup Python + uses: actions/setup-python@v5 with: python-version: "3.12" - - run: pip install -r requirements.txt - - name: Run + + - name: Install + run: pip install -r requirements.txt + + - name: Run monitor env: PUSH_TOKEN: ${{ secrets.PUSH_TOKEN }} run: python main.py - - name: Save state + + - name: Commit state run: | git config user.name "gitea-actions" git config user.email "actions@localhost" diff --git a/main.py b/main.py index 449d93e..19512c4 100644 --- a/main.py +++ b/main.py @@ -1,61 +1,161 @@ -import os,json,hashlib -from datetime import datetime,timezone + +import os +import json +import hashlib +from datetime import datetime, timezone + import requests from bs4 import BeautifulSoup -URL="https://cet.neea.edu.cn/" -STATE="data/state.json" -def category(t): - if "报名" in t:return "📝 报名" - if "成绩" in t:return "📊 成绩" - if "准考证" in t:return "🎫 准考证" - if "考试" in t:return "📝 考试安排" +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 fetch(): - r=requests.get(URL,headers={"User-Agent":"Mozilla/5.0"},timeout=20) - r.encoding="utf-8" - soup=BeautifulSoup(r.text,"lxml") - out=[] - for a in soup.find_all("a"): - t=a.get_text(strip=True) - if t and any(k in t for k in ["四六级","CET","成绩","报名","考试","准考证"]): - out.append({"title":t,"url":a.get("href"),"category":category(t)}) - return out[:20] - -def load(): - if not os.path.exists(STATE): return [] - return json.load(open(STATE,encoding="utf-8")).get("items",[]) - -def save(items): - os.makedirs("data",exist_ok=True) - json.dump({"items":items,"last_check":datetime.now(timezone.utc).isoformat()},open(STATE,"w",encoding="utf-8"),ensure_ascii=False,indent=2) - -def h(x): - return hashlib.sha256((x["title"]+str(x["url"])).encode()).hexdigest() def push(items): - token=os.getenv("PUSH_TOKEN") - if not token:return - msg="📢 CET四六级新通知\n\n" - for x in items: - msg+=f"{x['category']}\n{x['title']}\n{x['url']}\n\n" - requests.post("https://push.showdoc.com.cn/server/api/push", - data={"title":"四六级新通知","content":msg,"token":token},timeout=10) + 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() - old={x.get("hash") for x in load()} - fresh=[] - saved=[] - for x in current: - x["hash"]=h(x) - saved.append(x) - if x["hash"] not in old: - fresh.append(x) - if fresh: push(fresh) - save(saved) + current = fetch_notices() -if __name__=="__main__": + 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()