更新 .gitea/workflows/monitor.yml

运行时间修改

Signed-off-by: biss <1+biss@noreply.localhost>

.gitea/workflows/monitor.yml

@@ -1,11 +1,15 @@
-
 name: CET Monitor
 
 on:
   schedule:
-    - cron: "17 * * * *"
+    - cron: '*/30 6-18 * * *'
 
   workflow_dispatch:
+    inputs:
+      force_push:
+        description: "强制推送当前公告"
+        required: false
+        default: "false"
 
 
 jobs:
@@ -27,12 +31,16 @@ jobs:
       - name: Run monitor
         env:
           PUSH_TOKEN: ${{ secrets.PUSH_TOKEN }}
+          FORCE_PUSH: ${{ github.event.inputs.force_push }}
         run: python main.py
 
-      - name: Commit state
+      - name: Commit state UTC
         run: |
           git config user.name "gitea-actions"
           git config user.email "actions@localhost"
+
           git add data/state.json
-          git diff --cached --quiet || git commit -m "update cet state"
+
+          git diff --cached --quiet || git commit -m "update cet state $(date -u '+%Y-%m-%dT%H:%M:%SZ')"
+
           git push

data/state.json

@@ -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"
 }

main.py

@@ -10,22 +10,18 @@ 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 = {
+    headers = {"User-Agent": "Mozilla/5.0"}
-        "User-Agent": "Mozilla/5.0"
-    }
 
-    for i in range(3):
+    for retry in range(3):
         try:
             r = requests.get(URL, headers=headers, timeout=20)
             r.encoding = "utf-8"
 
             soup = BeautifulSoup(r.text, "lxml")
-
+            result = []
-            notices = []
 
             for a in soup.find_all("a"):
                 title = a.get_text(strip=True)
@@ -35,15 +31,15 @@ def fetch_notices():
                     "四六级", "CET", "成绩",
                     "报名", "考试", "准考证"
                 ]):
-                    notices.append({
+                    result.append({
                         "title": title,
                         "url": href
                     })
 
-            return notices[:20]
+            return result[:20]
 
         except Exception as e:
-            print(f"请求失败 {i+1}/3:", e)
+            print(f"获取失败 {retry + 1}/3:", e)
 
     return None
 
@@ -68,22 +64,10 @@ def save_state(items):
     with open(STATE_FILE, "w", encoding="utf-8") as f:
         json.dump({
             "items": items,
-            "last_check": datetime.now(timezone.utc).isoformat()
+            "last_check_utc": 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")
 
@@ -95,16 +79,17 @@ def push(items):
 
     for item in items:
         content += (
-            f"{category(item['title'])}\n"
             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(
-            PUSH_URL,
+            url,
             data={
-                "token": token,
                 "title": "四六级新通知",
                 "content": content
             },
@@ -114,7 +99,7 @@ def push(items):
         print("Push status:", r.status_code)
         print("Push response:", r.text)
 
-        return r.ok
+        return r.status_code == 200
 
     except Exception as e:
         print("Push异常:", e)
@@ -122,10 +107,10 @@ def push(items):
 
 
 def main():
-    current = fetch_notices()
+    notices = fetch_notices()
 
-    if current is None:
+    if notices is None:
-        print("获取公告失败，本轮跳过")
+        print("获取公告失败")
         return
 
     state = load_state()
@@ -135,26 +120,26 @@ def main():
         for x in state.get("items", [])
     }
 
+    force = os.getenv("FORCE_PUSH", "false").lower() == "true"
+
     new_items = []
 
-    for item in current:
+    for item in notices:
         item["hash"] = make_hash(item)
 
-        if item["hash"] not in old_hash:
+        if force or item["hash"] not in old_hash:
             new_items.append(item)
 
-    if not new_items:
+    if new_items:
-        print("没有新公告")
+        print("发现通知:", len(new_items))
-        save_state(current)
-        return
 
-    print("发现新公告:", len(new_items))
+        if push(new_items):
-
+            print("推送成功")
-    if push(new_items):
+            save_state(notices)
-        print("推送成功，保存状态")
+        else:
-        save_state(current)
+            print("推送失败，不更新状态")
     else:
-        print("推送失败，不更新状态，等待下次重试")
+        print("无新增")
 
 
 if __name__ == "__main__":