diff --git a/.gitea/workflows/monitor.yml b/.gitea/workflows/monitor.yml index 1e56441..5f4a350 100644 --- a/.gitea/workflows/monitor.yml +++ b/.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 diff --git a/main.py b/main.py index 19512c4..14900f9 100644 --- a/main.py +++ b/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 = { - "User-Agent": "Mozilla/5.0" - } + r = requests.get(URL, headers={"User-Agent": "Mozilla/5.0"}, timeout=20) + r.encoding = "utf-8" - for i in range(3): - try: - r = requests.get(URL, headers=headers, timeout=20) - r.encoding = "utf-8" + soup = BeautifulSoup(r.text, "lxml") + result = [] - 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 = [] - - 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 + return result[:20] -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, "r", encoding="utf-8") as f: + with open(STATE_FILE, 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: - content += ( - f"{category(item['title'])}\n" - f"【{item['title']}】\n" - f"{item['url']}\n\n" - ) + for x in items: + content += f"【{x['title']}】\n{x['url']}\n\n" - try: - r = requests.post( - PUSH_URL, - data={ - "token": token, - "title": "四六级新通知", - "content": content - }, - timeout=10 - ) + 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 + print(r.status_code, r.text) + return r.ok def main(): - current = fetch_notices() - - if current is None: - print("获取公告失败,本轮跳过") - return + notices = fetch_notices() 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_items = [] + new = [] - for item in current: - item["hash"] = make_hash(item) + for item in notices: + item["hash"] = h(item) + if force or item["hash"] not in old: + new.append(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) + if new: + if push(new): + save_state(notices) else: - print("推送失败,不更新状态,等待下次重试") + print("no update") if __name__ == "__main__":