From c4af6642612ddcf8cedfa7e1397f09bc00118e17 Mon Sep 17 00:00:00 2001 From: biss Date: Wed, 5 Aug 2026 12:15:35 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E6=AD=A3=20ShowDoc=20Push=20=E5=9C=B0?= =?UTF-8?q?=E5=9D=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitea/workflows/monitor.yml | 11 ++-- main.py | 119 ++++++++++++++++++++++++----------- 2 files changed, 88 insertions(+), 42 deletions(-) diff --git a/.gitea/workflows/monitor.yml b/.gitea/workflows/monitor.yml index 5f4a350..2515383 100644 --- a/.gitea/workflows/monitor.yml +++ b/.gitea/workflows/monitor.yml @@ -18,18 +18,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 diff --git a/main.py b/main.py index 14900f9..4f2d411 100644 --- a/main.py +++ b/main.py @@ -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__":