bump: release version 0.1.1

.env.example

@@ -1 +1,10 @@
-VIRTFUSION_TOKEN=your_token_here
+# Токен доступа к API VirtFusion (передается в заголовке Authorization: Bearer <TOKEN>)
+VIRTFUSION_TOKEN=your_virtfusion_token_here
+
+# Токен Telegram-бота, полученный от @BotFather
+TELEGRAM_BOT_TOKEN=123456789:ABCdefGhIJKlmNoPQRsTUVwxyZ
+
+# ID чата или канала для отправки уведомлений
+# Для персонального чата: 123456789
+# Для публичного/приватного канала: -1001234567890 (обязательно с префиксом -100)
+TELEGRAM_CHAT_ID=-1001234567890

.gitignore

@@ -222,3 +222,4 @@ __marimo__/
 
 # This
 config.json
+state.json

Makefile

@@ -0,0 +1,11 @@
+.PHONY: install-deps build clean
+
+install-deps:
+	echo "Проверка и установка зависимостей для сборки deb-пакета..."
+	sudo apt update && sudo apt install -y debhelper dh-python pybuild-plugin-pyproject python3-build python3-all python3-setuptools build-essential devscripts
+
+build: install-deps
+	dpkg-buildpackage -us -uc -b
+
+clean:
+	debian/rules clean

config.json.example

@@ -1,14 +1,21 @@
 {
+  "_comment": "Конфигурационный файл локально хранится в корне проекта (./config.json), а на сервере — в /etc/traffic-checker/config.json",
+  "thresholds": {
+    "warning_percents": [
+      80.0,
+      90.0
+    ]
+  },
   "servers": [
     {
-            "uuid": "00000000-0000-0000-0000-000000000000",
-            "target": "Target name from xray-manager",
-            "pretty_name": "Pretty name from xray-manager"
+      "uuid": "11111111-2222-4333-8444-555555555555",
+      "target": "nl1",
+      "pretty_name": "🇳🇱 Netherlands #1"
     },
     {
-            "uuid": "11111111-1111-1111-1111-111111111111",
-            "target": "Target name from xray-manager",
-            "pretty_name": "Pretty name from xray-manager"
+      "uuid": "a1b2c3d4-e5f6-4a5b-8c9d-0e1f2a3b4c5d",
+      "target": "de1",
+      "pretty_name": "🇩🇪 Germany #1"
     }
   ]
 }

debian/changelog

@@ -0,0 +1,12 @@
+traffic-checker (0.1.1-1) unstable; urgency=medium
+
+  * Add logging function for traffic check status in stdout.
+  * Defer state save until successful telegram delivery.
+
+ -- Vladimir Khvan <thisisnotablownfuse@gmail.com>  Sun, 02 Aug 2026 02:54:53 +0300
+
+traffic-checker (0.1-1) unstable; urgency=medium
+
+  * Initial release.
+
+ -- Vladimir Khvan <thisisnotablownfuse@gmail.com>  Sat, 01 Aug 2026 04:00:00 +0300

debian/control

@@ -0,0 +1,14 @@
+Source: traffic-checker
+Section: python
+Priority: optional
+Maintainer: Your Name <you@example.com>
+Build-Depends: debhelper-compat (=13),
+               dh-python,
+               python3-all,
+               python3-setuptools
+Standards-Version: 4.6.2
+
+Package: traffic-checker
+Architecture: all
+Depends: ${misc:Depends}, ${python3:Depends}
+Description: Traffic consumption monitoring CLI tool for remote servers with automated alerting

debian/install

@@ -0,0 +1,3 @@
+config.json.example /etc/traffic-checker/
+.env.example /etc/traffic-checker/
+state.json.example /var/lib/traffic-checker/

debian/rules

@@ -0,0 +1,9 @@
+#!/usr/bin/make -f
+
+export DH_VERBOSE = 1
+
+%:
+	dh $@ --with python3 --buildsystem=pybuild
+
+override_dh_installsystemd:
+	dh_installsystemd --no-enable

debian/traffic-checker.service

@@ -0,0 +1,11 @@
+[Unit]
+Description=Traffic Checker Monitoring Service
+After=network-online.target
+Wants=network-online.target
+
+[Service]
+Type=oneshot
+User=traffic-checker
+Group=traffic-checker
+ExecStart=/usr/bin/traffic-checker
+EnvironmentFile=-/etc/traffic-checker/environment

debian/traffic-checker.timer

@@ -0,0 +1,9 @@
+[Unit]
+Description=Run Traffic Checker every 5 minutes
+
+[Timer]
+OnCalendar=*:0/5
+Persistent=true
+
+[Install]
+WantedBy=timers.target

pyproject.toml

@@ -0,0 +1,26 @@
+[build-system]
+requires = ["setuptools>=61.0"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "traffic-checker"
+version = "0.1.1"
+description = "Traffic consumption monitoring CLI tool for remote servers with automated alerting"
+requires-python = ">=3.11"
+authors = [
+    { name = "Vladimir Khvan", email = "thisisnotablownfuse@gmail.com" }
+]
+classifiers = [
+    "Programming Language :: Python :: 3",
+    "Programming Language :: Python :: 3.11",
+    "Operating System :: POSIX :: Linux",
+]
+dependencies = [
+    "requests>=2.32.3",
+]
+
+[project.scripts]
+traffic-checker = "traffic_checker.main:main"
+
+[tool.setuptools.packages.find]
+where = ["src"]

src/traffic_checker.py

@@ -1,51 +0,0 @@
-import json
-import os
-from pathlib import Path
-
-import requests
-
-
-def load_config() -> dict:
-    dev_config = Path(__file__).parent.parent / "config.json"
-    sys_config = Path("/etc/traffic-checker/config.json")
-
-    if dev_config.exists():
-        config_path = dev_config
-    elif sys_config.exists():
-        config_path = sys_config
-    else:
-        raise FileNotFoundError(
-            "Configuration file config.json not found in dev path or /etc/traffic-checker/"
-        )
-
-    with open(config_path, "r", encoding="utf-8") as f:
-        return json.load(f)
-
-
-def fetch_server_data(server_uuid: str) -> dict:
-    token = os.getenv("VIRTFUSION_TOKEN")
-
-    if not token:
-        raise RuntimeError("VIRTFUSION_TOKEN environment variable is not set.")
-
-    url = f"https://virtfusion-nat.gullo.me/api/server/{server_uuid}"
-
-    headers = {
-        "Accept": "application/json",
-        "Authorization": f"Bearer {token}",
-    }
-
-    params = {"state": "true"}
-
-    response = requests.get(url, headers=headers, params=params, timeout=10)
-    response.raise_for_status()
-
-    return response.json()
-
-
-if __name__ == "__main__":
-    config = load_config()
-
-    for server in config["servers"]:
-        data = fetch_server_data(server["uuid"])
-        print(data)

src/traffic_checker/main.py

@@ -0,0 +1,303 @@
+import json
+import os
+from datetime import datetime, timedelta
+from pathlib import Path
+from zoneinfo import ZoneInfo
+
+import requests
+
+
+def load_config() -> dict:
+    dev_config = Path(__file__).parent.parent.parent / "config.json"
+    sys_config = Path("/etc/traffic-checker/config.json")
+
+    if dev_config.exists():
+        config_path = dev_config
+    elif sys_config.exists():
+        config_path = sys_config
+    else:
+        raise FileNotFoundError(
+            "Configuration file config.json not found in dev path or /etc/traffic-checker/"
+        )
+
+    with open(config_path, "r", encoding="utf-8") as f:
+        return json.load(f)
+
+
+def get_state_file_path() -> Path:
+    dev_state = Path(__file__).parent.parent.parent / "state.json"
+    sys_state = Path("/var/lib/traffic-checker/state.json")
+
+    if dev_state.exists() or not sys_state.parent.exists():
+        return dev_state
+
+    return sys_state
+
+
+def load_state(path: Path) -> dict:
+    if not path.exists():
+        return {}
+
+    try:
+        with open(path, "r", encoding="utf-8") as f:
+            return json.load(f)
+    except json.JSONDecodeError:
+        return {}
+
+
+def save_state(path: Path, state: dict) -> None:
+    with open(path, "w", encoding="utf-8") as f:
+        json.dump(state, f, indent=2, ensure_ascii=False)
+
+
+def fetch_server_data(server_uuid: str) -> dict:
+    token = os.getenv("VIRTFUSION_TOKEN")
+
+    if not token:
+        raise RuntimeError("VIRTFUSION_TOKEN environment variable is not set.")
+
+    url = f"https://virtfusion-nat.gullo.me/api/server/{server_uuid}"
+
+    headers = {
+        "Accept": "application/json",
+        "Authorization": f"Bearer {token}",
+    }
+
+    params = {"state": "true"}
+
+    response = requests.get(url, headers=headers, params=params, timeout=10)
+    response.raise_for_status()
+
+    return response.json()
+
+
+def process_traffic_stats(server_data: dict) -> dict:
+    traffic_limit_str = str(server_data["data"]["network"]["primary"]["limit"])
+    traffic_limit_gbyte = float(traffic_limit_str.removesuffix("GB").strip())
+
+    traffic_usage_byte = int(
+        server_data["data"]["state"]["network"]["primary"]["traffic"]["total"]
+    )
+
+    traffic_usage_gbyte = traffic_usage_byte / 1024**3
+    traffic_usage_percent = traffic_usage_gbyte / traffic_limit_gbyte * 100
+
+    return {
+        "limit": traffic_limit_gbyte,
+        "used": round(traffic_usage_gbyte, 2),
+        "percentage": round(traffic_usage_percent, 2),
+    }
+
+
+def process_monthly_period(
+    server_data: dict, timezone: str = "Europe/Moscow"
+) -> dict:
+    period_start_utc = datetime.fromisoformat(
+        server_data["data"]["currentMonthlyPeriod"]["start"]
+    )
+
+    period_end_utc = datetime.fromisoformat(
+        server_data["data"]["currentMonthlyPeriod"]["end"]
+    )
+
+    period_reset_utc = period_end_utc + timedelta(microseconds=1)
+
+    return {
+        "start": period_start_utc.astimezone(ZoneInfo(timezone)),
+        "end": period_end_utc.astimezone(ZoneInfo(timezone)),
+        "reset": period_reset_utc.astimezone(ZoneInfo(timezone)),
+    }
+
+
+def process_traffic_event(
+    server_uuid: str,
+    warning_thresholds: list[float],
+    traffic_percentage: float,
+    period_reset_iso: str,
+    state: dict,
+) -> dict:
+    server_state = state.get(
+        server_uuid,
+        {
+            "last_threshold": 0.0,
+            "is_critical": False,
+            "period_reset": period_reset_iso,
+        },
+    )
+
+    if server_state.get("period_reset") != period_reset_iso:
+        new_server_state = {
+            "last_threshold": 0.0,
+            "is_critical": False,
+            "period_reset": period_reset_iso,
+        }
+
+        was_critical = server_state.get("is_critical", False)
+
+        return {
+            "event": "reset" if was_critical else "none",
+            "server_state": new_server_state,
+        }
+
+    last_threshold = server_state.get("last_threshold", 0)
+
+    if traffic_percentage >= 100.0:
+        if last_threshold < 100.0:
+            new_server_state = {
+                "last_threshold": 100.0,
+                "is_critical": True,
+                "period_reset": period_reset_iso,
+            }
+
+            return {
+                "event": "critical",
+                "server_state": new_server_state,
+            }
+
+    passed_thresholds = [
+        t for t in warning_thresholds if traffic_percentage >= t
+    ]
+
+    if passed_thresholds:
+        max_passed = max(passed_thresholds)
+        if max_passed > last_threshold:
+            new_server_state = {
+                "last_threshold": max_passed,
+                "is_critical": False,
+                "period_reset": period_reset_iso,
+            }
+
+            return {"event": "warning", "server_state": new_server_state}
+
+    return {"event": "none", "server_state": server_state}
+
+
+def format_message(
+    event_type: str,
+    server_target: str,
+    server_pretty_name: str,
+    traffic_percentage: float,
+    period_reset: datetime,
+) -> str:
+
+    if event_type == "warning":
+        return f"⚠️ На сервере {server_target} ({server_pretty_name}) израсходовано {traffic_percentage}% трафика."
+
+    MONTHS_RU = {
+        1: "января",
+        2: "февраля",
+        3: "марта",
+        4: "апреля",
+        5: "мая",
+        6: "июня",
+        7: "июля",
+        8: "августа",
+        9: "сентября",
+        10: "октября",
+        11: "ноября",
+        12: "декабря",
+    }
+
+    period_reset_str = f"{period_reset.day} {MONTHS_RU[period_reset.month]} в {period_reset.strftime("%H:%M")}"
+
+    if event_type == "critical":
+        return (
+            f"⛔️ На сервере {server_target} ({server_pretty_name}) введено ограничение скорости в 625 килобит/секунду из-за исчерпания объема трафика.\n\n"
+            f"Снятие ограничений скорости состоится {period_reset_str}."
+        )
+
+    if event_type == "reset":
+        return f"🔄 На сервере {server_target} ({server_pretty_name}) снято ограничение скорости."
+
+    raise ValueError(
+        f"""Unknown event type: "{event_type}". """
+        f"""Expected one of: "warning", "critical", "reset"."""
+    )
+
+
+def send_telegram_message(
+    message: str, chat_id: int | str | None = None
+) -> None:
+    token = os.getenv("TELEGRAM_BOT_TOKEN")
+    if not token:
+        raise RuntimeError(
+            "TELEGRAM_BOT_TOKEN environment variable is not set."
+        )
+
+    target_chat_id = chat_id or os.getenv("TELEGRAM_CHAT_ID")
+    if not target_chat_id:
+        raise RuntimeError("TELEGRAM_CHAT_ID is not provided.")
+
+    url = f"https://api.telegram.org/bot{token}/sendMessage"
+
+    payload = {
+        "chat_id": target_chat_id,
+        "text": message,
+    }
+
+    response = requests.post(url, json=payload, timeout=10)
+    response.raise_for_status()
+
+
+def to_log(
+    server_target: str,
+    event_type: str,
+    traffic_stats: dict,
+    monthly_period: dict,
+):
+    period_start_str = monthly_period["start"].strftime("%Y-%m-%d %H:%M:%S")
+    period_reset_str = monthly_period["reset"].strftime("%Y-%m-%d %H:%M:%S")
+
+    print(f"Traffic check for server: {server_target}")
+    print(f"Event: {event_type}")
+    print(
+        f"Traffic stats: {traffic_stats["percentage"]}%, "
+        f"{traffic_stats["used"]} GB / {traffic_stats["limit"]} GB"
+    )
+    print(f"Monthly period: {period_start_str} -> {period_reset_str}")
+
+
+def main():
+    config = load_config()
+
+    state_file_path = get_state_file_path()
+    state = load_state(state_file_path)
+
+    for server in config["servers"]:
+        try:
+            data = fetch_server_data(server["uuid"])
+            traffic_stats = process_traffic_stats(data)
+            monthly_period = process_monthly_period(data)
+
+            event = process_traffic_event(
+                server["uuid"],
+                config["thresholds"]["warning_percents"],
+                traffic_stats["percentage"],
+                monthly_period["reset"].isoformat(),
+                state,
+            )
+
+            if event["event"] != "none":
+                message = format_message(
+                    event["event"],
+                    server["target"],
+                    server["pretty_name"],
+                    traffic_stats["percentage"],
+                    monthly_period["reset"],
+                )
+                send_telegram_message(message)
+
+            to_log(
+                server["target"], event["event"], traffic_stats, monthly_period
+            )
+
+            state[server["uuid"]] = event["server_state"]
+
+            save_state(state_file_path, state)
+
+        except Exception as e:
+            print(f"""Error processing server {server["target"]}: {e}""")
+
+
+if __name__ == "__main__":
+    main()

state.json.example

@@ -0,0 +1,13 @@
+{
+  "_comment": "Файл состояния локально хранится в корне проекта (./state.json), а на сервере — в /var/lib/traffic-checker/state.json",
+  "11111111-2222-4333-8444-555555555555": {
+    "last_threshold": 80.0,
+    "is_critical": false,
+    "period_reset": "2026-08-26T03:00:00+03:00"
+  },
+  "a1b2c3d4-e5f6-4a5b-8c9d-0e1f2a3b4c5d": {
+    "last_threshold": 100.0,
+    "is_critical": true,
+    "period_reset": "2026-08-03T03:00:00+03:00"
+  }
+}