再次修复

.gitea/workflows/monitor.yml

@@ -6,6 +6,11 @@ on:
     - cron: "17 * * * *"
 
   workflow_dispatch:
+    inputs:
+      force_push:
+        description: "强制推送当前公告"
+        required: false
+        default: "false"
 
 
 jobs:
@@ -13,26 +18,27 @@ jobs:
     runs-on: ubuntu-latest
 
     steps:
-      - name: Checkout
+      - uses: actions/checkout@v4
-        uses: actions/checkout@v4
 
-      - name: Setup Python
+      - uses: actions/setup-python@v5
-        uses: actions/setup-python@v5
         with:
           python-version: "3.12"
 
-      - name: Install
+      - run: pip install -r requirements.txt
-        run: pip install -r requirements.txt
 
       - name: Run monitor
         env:
           PUSH_TOKEN: ${{ secrets.PUSH_TOKEN }}
+          FORCE_PUSH: ${{ 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

main.py

@@ -7,154 +7,96 @@ 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 = {
+    r = requests.get(URL, headers={"User-Agent": "Mozilla/5.0"}, timeout=20)
-        "User-Agent": "Mozilla/5.0"
+    r.encoding = "utf-8"
-    }
 
-    for i in range(3):
+    soup = BeautifulSoup(r.text, "lxml")
-        try:
+    result = []
-            r = requests.get(URL, headers=headers, timeout=20)
-            r.encoding = "utf-8"
 
-            soup = BeautifulSoup(r.text, "lxml")
+    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")
+            })
 
-            notices = []
+    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", "成绩",
-                    "报名", "考试", "准考证"
-                ]):
-                    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):
+def h(item):
     return hashlib.sha256(
-        (item["title"] + str(item["url"])).encode("utf-8")
+        (item["title"] + str(item["url"])).encode()
     ).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,
-            "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")
-
     if not token:
-        print("ERROR: PUSH_TOKEN 未配置")
+        print("PUSH_TOKEN missing")
         return False
 
     content = "📢 CET四六级新通知\n\n"
 
-    for item in items:
+    for x in items:
-        content += (
+        content += f"【{x['title']}】\n{x['url']}\n\n"
-            f"{category(item['title'])}\n"
-            f"【{item['title']}】\n"
-            f"{item['url']}\n\n"
-        )
 
-    try:
+    r = requests.post(
-        r = requests.post(
+        PUSH_URL,
-            PUSH_URL,
+        data={
-            data={
+            "token": token,
-                "token": token,
+            "title": "四六级新通知",
-                "title": "四六级新通知",
+            "content": content
-                "content": content
+        },
-            },
+        timeout=10
-            timeout=10
+    )
-        )
 
-        print("Push status:", r.status_code)
+    print(r.status_code, r.text)
-        print("Push response:", r.text)
+    return r.ok
-
-        return r.ok
-
-    except Exception as e:
-        print("Push异常:", e)
-        return False
 
 
 def main():
-    current = fetch_notices()
+    notices = fetch_notices()
-
-    if current is None:
-        print("获取公告失败，本轮跳过")
-        return
 
     state = load_state()
+    old = {x["hash"] for x in state.get("items", [])}
 
-    old_hash = {
+    force = os.getenv("FORCE_PUSH", "false").lower() == "true"
-        x["hash"]
-        for x in state.get("items", [])
-    }
 
-    new_items = []
+    new = []
 
-    for item in current:
+    for item in notices:
-        item["hash"] = make_hash(item)
+        item["hash"] = h(item)
+        if force or item["hash"] not in old:
+            new.append(item)
 
-        if item["hash"] not in old_hash:
+    if new:
-            new_items.append(item)
+        if push(new):
-
+            save_state(notices)
-    if not new_items:
-        print("没有新公告")
-        save_state(current)
-        return
-
-    print("发现新公告:", len(new_items))
-
-    if push(new_items):
-        print("推送成功，保存状态")
-        save_state(current)
     else:
-        print("推送失败，不更新状态，等待下次重试")
+        print("no update")
 
 
 if __name__ == "__main__":