diff --git a/.env.example b/.env.example index 40aca23..00a0845 100644 --- a/.env.example +++ b/.env.example @@ -30,4 +30,13 @@ SMTP= # --- Optional: Telegram Bot Notifications --- TG_BOT_TOKEN= -TG_CHAT_ID= \ No newline at end of file +TG_CHAT_ID= + +# --- Optional: Bark Notifications --- +# Find the device key in the Bark app's push URL: https://api.day.app/YOUR_DEVICE_KEY/... +BARK_DEVICE_KEY= +# Leave empty to use Bark's public server, or set your self-hosted Bark server URL. +BARK_SERVER= +# Optional notification group and sound name. +BARK_GROUP=CEACStatusBot +BARK_SOUND= diff --git a/.github/workflows/actions.yml b/.github/workflows/actions.yml index 8b155de..ec42a11 100644 --- a/.github/workflows/actions.yml +++ b/.github/workflows/actions.yml @@ -49,6 +49,10 @@ jobs: SMTP: ${{ secrets.SMTP }} TG_BOT_TOKEN: ${{ secrets.TG_BOT_TOKEN }} TG_CHAT_ID: ${{ secrets.TG_CHAT_ID }} + BARK_DEVICE_KEY: ${{ secrets.BARK_DEVICE_KEY }} + BARK_SERVER: ${{ secrets.BARK_SERVER }} + BARK_GROUP: ${{ secrets.BARK_GROUP }} + BARK_SOUND: ${{ secrets.BARK_SOUND }} GH_TOKEN: ${{ secrets.GH_TOKEN }} GITHUB_REPOSITORY: ${{ github.repository }} run: uv run trigger.py diff --git a/CEACStatusBot/notification/__init__.py b/CEACStatusBot/notification/__init__.py index 2ae1640..ad65ba6 100644 --- a/CEACStatusBot/notification/__init__.py +++ b/CEACStatusBot/notification/__init__.py @@ -1,4 +1,5 @@ from .handle import * from .manager import * from .email import * -from .telegram import * \ No newline at end of file +from .telegram import * +from .bark import * diff --git a/CEACStatusBot/notification/bark.py b/CEACStatusBot/notification/bark.py new file mode 100644 index 0000000..ab64d6d --- /dev/null +++ b/CEACStatusBot/notification/bark.py @@ -0,0 +1,59 @@ +import requests + +from .handle import NotificationHandle + +DEFAULT_BARK_SERVER = "https://api.day.app" +DEFAULT_BARK_GROUP = "CEACStatusBot" +CEAC_STATUS_URL = "https://ceac.state.gov/CEACStatTracker/Status.aspx?App=NIV" +BARK_SUCCESS_CODE = 200 + + +class BarkNotificationHandle(NotificationHandle): + def __init__( + self, + device_key: str, + server_url: str = DEFAULT_BARK_SERVER, + group: str = DEFAULT_BARK_GROUP, + sound: str | None = None, + ) -> None: + super().__init__() + if not device_key.strip(): + error_message = "Bark device key must not be empty" + raise ValueError(error_message) + + normalized_server_url = server_url.strip().rstrip("/") or DEFAULT_BARK_SERVER + self.__device_key = device_key.strip() + self.__api_url = f"{normalized_server_url}/push" + self.__group = group.strip() or DEFAULT_BARK_GROUP + self.__sound = sound.strip() if sound else None + + def send(self, result: dict) -> None: + title = f"[CEACStatusBot] {result['application_num_origin']}: {result['status']}" + body_parts = [ + f"Visa type: {result.get('visa_type', 'Unknown')}", + f"Case created: {result.get('case_created', 'Unknown')}", + f"Last updated: {result.get('case_last_updated', 'Unknown')}", + ] + if description := result.get("description"): + body_parts.extend(("", description)) + + payload = { + "device_key": self.__device_key, + "title": title, + "body": "\n".join(body_parts), + "group": self.__group, + "url": CEAC_STATUS_URL, + } + if self.__sound: + payload["sound"] = self.__sound + + response = requests.post(self.__api_url, json=payload, timeout=15) + response.raise_for_status() + + response_data = response.json() + if response_data.get("code") != BARK_SUCCESS_CODE: + message = response_data.get("message", "unknown Bark API error") + error_message = f"Failed to send Bark notification: {message}" + raise RuntimeError(error_message) + + print("Bark notification sent successfully") diff --git a/CEACStatusBot/request/query.py b/CEACStatusBot/request/query.py index 391b2c7..2b387a2 100644 --- a/CEACStatusBot/request/query.py +++ b/CEACStatusBot/request/query.py @@ -4,6 +4,11 @@ from CEACStatusBot.captcha import CaptchaHandle, OnnxCaptchaHandle + +def _normalize_application_num(application_num: str) -> str: + return "".join(character for character in application_num if character.isalnum()).casefold() + + def query_status(location, application_num, passport_number, surname, captchaHandle: CaptchaHandle = OnnxCaptchaHandle("captcha.onnx")): failCount = 0 result = { @@ -103,13 +108,20 @@ def update_from_current_page(cur_page, name, data): if not status_tag: continue - application_num_returned = soup.find("span", id="ctl00_ContentPlaceHolder1_ucApplicationStatusView_lblCaseNo").string - assert application_num_returned == application_num - status = status_tag.string - visa_type = soup.find("span", id="ctl00_ContentPlaceHolder1_ucApplicationStatusView_lblAppName").string - case_created = soup.find("span", id="ctl00_ContentPlaceHolder1_ucApplicationStatusView_lblSubmitDate").string - case_last_updated = soup.find("span", id="ctl00_ContentPlaceHolder1_ucApplicationStatusView_lblStatusDate").string - description = soup.find("span", id="ctl00_ContentPlaceHolder1_ucApplicationStatusView_lblMessage").string + application_num_tag = soup.find("span", id="ctl00_ContentPlaceHolder1_ucApplicationStatusView_lblCaseNo") + if not application_num_tag: + continue + + application_num_returned = application_num_tag.get_text(strip=True) + if _normalize_application_num(application_num_returned) != _normalize_application_num(application_num): + print("CEAC returned a different application number; retrying.") + continue + + status = status_tag.get_text(strip=True) + visa_type = soup.find("span", id="ctl00_ContentPlaceHolder1_ucApplicationStatusView_lblAppName").get_text(strip=True) + case_created = soup.find("span", id="ctl00_ContentPlaceHolder1_ucApplicationStatusView_lblSubmitDate").get_text(strip=True) + case_last_updated = soup.find("span", id="ctl00_ContentPlaceHolder1_ucApplicationStatusView_lblStatusDate").get_text(strip=True) + description = soup.find("span", id="ctl00_ContentPlaceHolder1_ucApplicationStatusView_lblMessage").get_text(strip=True) result.update({ "success": True, diff --git a/README.Chinese.md b/README.Chinese.md index ed02109..478e0eb 100644 --- a/README.Chinese.md +++ b/README.Chinese.md @@ -52,6 +52,18 @@ Telegram Bot [创建教程](https://www.cytron.io/tutorial/how-to-create-a-teleg - TG_CHAT_ID: 聊天 ID，获取方法见教程 +#### Bark 通知 + +安装并打开 [Bark](https://github.com/Finb/Bark)，从 App 显示的推送地址中复制设备密钥（例如 `https://api.day.app/YOUR_DEVICE_KEY/...` 中的 `YOUR_DEVICE_KEY` 部分）。 + +- BARK_DEVICE_KEY: Bark 设备密钥 + +- BARK_SERVER: 可选，自建 Bark 服务地址；默认使用 `https://api.day.app` + +- BARK_GROUP: 可选，通知分组；默认为 `CEACStatusBot` + +- BARK_SOUND: 可选，Bark 通知声音名称 + ### 在 Github Actions 的使用方法 @@ -62,7 +74,7 @@ Telegram Bot [创建教程](https://www.cytron.io/tutorial/how-to-create-a-teleg ![image](docs/github.new.secret.png) -3. 查看 `Github Actions` 中的 `workflows` 是否正常运行并检查邮箱是否收到邮件。 +3. 查看 `Github Actions` 中的 `workflows` 是否正常运行，并检查邮箱、Telegram 或 Bark 是否收到通知。 ### 本地使用 @@ -97,4 +109,4 @@ uv run trigger.py - [ceac_tracker](https://github.com/lixin-wei/ceac_tracker) -- [CEACStatTracker](https://github.com/yuzeming/CEACStatTracker) \ No newline at end of file +- [CEACStatTracker](https://github.com/yuzeming/CEACStatTracker) diff --git a/README.md b/README.md index b5cbeb4..04914f6 100644 --- a/README.md +++ b/README.md @@ -47,6 +47,18 @@ Create a Telegram bot and get the info below according to [this tutorial](https: - TG_CHAT_ID: the chat id you want to receive the notification +#### Notification by Bark + +Install and open [Bark](https://github.com/Finb/Bark), then copy the device key from the push URL shown in the app (for example, the `YOUR_DEVICE_KEY` part of `https://api.day.app/YOUR_DEVICE_KEY/...`). + +- BARK_DEVICE_KEY: your Bark device key + +- BARK_SERVER: optional, your self-hosted Bark server URL; defaults to `https://api.day.app` + +- BARK_GROUP: optional, notification group; defaults to `CEACStatusBot` + +- BARK_SOUND: optional, a Bark sound name + ### Github Actions 1. folk this repo @@ -54,7 +66,7 @@ Create a Telegram bot and get the info below according to [this tutorial](https: 2. set your Environment Variables in `Github -> Settings -> Secrets and variables -> Actions -> New repository secret` ![image](docs/github.new.secret.png) -3. check your workflow in Actions and your Mailbox / Telegram +3. check your workflow in Actions and your Mailbox / Telegram / Bark app ### Local Usage You can also run this bot locally. @@ -84,4 +96,4 @@ uv run trigger.py Part of the code in this repo refers to the following project. Thank you for your pretty work. - [ceac_tracker](https://github.com/lixin-wei/ceac_tracker) -- [CEACStatTracker](https://github.com/yuzeming/CEACStatTracker) \ No newline at end of file +- [CEACStatTracker](https://github.com/yuzeming/CEACStatTracker) diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_bark.py b/tests/test_bark.py new file mode 100644 index 0000000..e239d26 --- /dev/null +++ b/tests/test_bark.py @@ -0,0 +1,64 @@ +import unittest +from unittest.mock import Mock, patch + +from CEACStatusBot.notification.bark import CEAC_STATUS_URL, BarkNotificationHandle + + +class BarkNotificationHandleTest(unittest.TestCase): + def setUp(self) -> None: + self.result = { + "application_num_origin": "AA00123456", + "status": "Issued", + "visa_type": "NONIMMIGRANT VISA APPLICATION", + "case_created": "20-Aug-2026", + "case_last_updated": "21-Aug-2026", + "description": "Your visa is in final processing.", + } + + @patch("CEACStatusBot.notification.bark.requests.post") + def test_send_uses_v2_json_api(self, post: Mock) -> None: + response = post.return_value + response.json.return_value = {"code": 200, "message": "success"} + + notification = BarkNotificationHandle( + "device-key", + "https://bark.example.com/", + "Visa", + "minuet", + ) + notification.send(self.result) + + post.assert_called_once_with( + "https://bark.example.com/push", + json={ + "device_key": "device-key", + "title": "[CEACStatusBot] AA00123456: Issued", + "body": ( + "Visa type: NONIMMIGRANT VISA APPLICATION\n" + "Case created: 20-Aug-2026\n" + "Last updated: 21-Aug-2026\n\n" + "Your visa is in final processing." + ), + "group": "Visa", + "url": CEAC_STATUS_URL, + "sound": "minuet", + }, + timeout=15, + ) + response.raise_for_status.assert_called_once_with() + + @patch("CEACStatusBot.notification.bark.requests.post") + def test_send_raises_for_bark_api_error(self, post: Mock) -> None: + post.return_value.json.return_value = {"code": 400, "message": "invalid device key"} + notification = BarkNotificationHandle("device-key") + + with self.assertRaisesRegex(RuntimeError, "invalid device key"): + notification.send(self.result) + + def test_empty_device_key_is_rejected(self) -> None: + with self.assertRaisesRegex(ValueError, "must not be empty"): + BarkNotificationHandle(" ") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_query.py b/tests/test_query.py new file mode 100644 index 0000000..4407907 --- /dev/null +++ b/tests/test_query.py @@ -0,0 +1,21 @@ +import unittest + +from CEACStatusBot.request.query import _normalize_application_num + + +class NormalizeApplicationNumberTest(unittest.TestCase): + def test_ignores_spacing_hyphens_and_case(self) -> None: + self.assertEqual( + _normalize_application_num(" aa00-20 akax "), + _normalize_application_num("AA0020AKAX"), + ) + + def test_keeps_distinct_application_numbers_distinct(self) -> None: + self.assertNotEqual( + _normalize_application_num("AA0020AKAX"), + _normalize_application_num("AA0020AKAY"), + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/trigger.py b/trigger.py index 7bb144d..8bb7e13 100644 --- a/trigger.py +++ b/trigger.py @@ -5,6 +5,7 @@ from dotenv import load_dotenv from CEACStatusBot import ( + BarkNotificationHandle, EmailNotificationHandle, NotificationManager, TelegramNotificationHandle, @@ -78,5 +79,23 @@ def download_artifact(): print("Telegram bot notification config missing or incomplete") +# --- Optional: Bark notifications --- +BARK_DEVICE_KEY = os.getenv("BARK_DEVICE_KEY") +BARK_SERVER = os.getenv("BARK_SERVER") or "https://api.day.app" +BARK_GROUP = os.getenv("BARK_GROUP") or "CEACStatusBot" +BARK_SOUND = os.getenv("BARK_SOUND") + +if BARK_DEVICE_KEY: + bark_notification = BarkNotificationHandle( + BARK_DEVICE_KEY, + BARK_SERVER, + BARK_GROUP, + BARK_SOUND, + ) + notificationManager.addHandle(bark_notification) +else: + print("Bark notification config missing or incomplete") + + # --- Send notifications --- notificationManager.send()