Telegram Notifications

Relevant source files

The following files were used as context for generating this wiki page:

• notification_service.py
• test_telegram.py
• config.example.json
• install_dependencies.bat
• run_service.bat
• build_notification_exe.bat
• README_Notifications.txt
• pos_app.py
• requirements.txt

The Telegram Notification Service is a standalone, multi-threaded Python application designed to monitor the Point of Sale (POS) system's database and send real-time alerts and daily reports via Telegram. It operates independently of the main pos_app.py GUI application, providing continuous background monitoring of inventory levels and sales performance.

The service is responsible for two primary functions: alerting when a product's stock reaches zero and sending a daily summary of the previous day's sales. It uses a separate state-tracking database to prevent duplicate notifications and ensure reports are sent only once per day. The entire system is configurable through a config.json file.

Sources: notification_service.py:8-13, README_Notifications.txt:1-5

Architecture

The service is built around a central NotificationService class that orchestrates several specialized components, each running in its own thread. This allows for concurrent monitoring of different conditions without blocking. The main components are the POSDatabaseReader for accessing inventory data, the StateTracker for managing notification history, and the TelegramNotifier for communicating with the Telegram API.

Sources: notification_service.py:532-549

The following diagram illustrates the high-level architecture of the notification service.

Core Components

The service is composed of four main classes that encapsulate specific responsibilities.

Configuration

The service's behavior is controlled by config.json. An example configuration is provided in config.example.json.

Section	Key	Description	Default Value
telegram	bot_token	The API token for the Telegram bot, obtained from @BotFather.	"123...-EXAMPLE"
telegram	chat_id	The unique identifier for the target chat where notifications will be sent.	"123456789"
polling	zero_stock_interval_seconds	The interval in seconds for checking for out-of-stock products.	300
polling	health_check_interval_seconds	The interval in seconds for the service to log a health check message.	60
daily_report	send_time	The time of day (HH:MM) to send the daily report, in the specified timezone.	"08:00"
daily_report	timezone	The timezone for the send_time.	"America/Mexico_City"
daily_report	top_products_count	The number of top-selling products to include in the daily report.	10
daily_report	metric	The metric to determine top products (quantity or revenue).	"quantity"
database	pos_db_path	The file path to the main POS inventory database.	"pos_inventory.db"
database	state_db_path	The file path for the service's state-tracking database.	"notification_state.db"
logging	log_level	The logging level for console output.	"INFO"
logging	log_file	Path to the main log file.	"logs/notification_service.log"
logging	error_log_file	Path to the error log file.	"logs/notification_errors.log"
logging	max_bytes	Maximum size of a log file in bytes before rotation.	10485760 (10 MB)
logging	backup_count	Number of old log files to keep.	5
notifications	out_of_stock	Contains settings for out-of-stock alerts, including an enabled flag and message_template.	{ "enabled": true, ... }
notifications	daily_report	Contains settings for daily reports, including an enabled flag and message_template.	{ "enabled": true, ... }

Sources: config.example.json

Notification Flows

Out-of-Stock Alerts

This flow runs continuously, polling the POS database at a configured interval to check for products that have a quantity of zero. To avoid spamming, it uses the StateTracker to ensure an alert is sent only once for each product until it is restocked.

The following sequence diagram shows the process for sending an out-of-stock alert.

Daily Sales Report

This flow is scheduled to run once per day at a configured time. It queries the database for the previous day's sales data, generates a summary report, and sends it to the configured Telegram chat. The StateTracker ensures the report is not sent more than once per day, even if the service restarts.

The following sequence diagram shows the process for sending the daily report.

State Management (notification_state.db)

The service uses a dedicated SQLite database, notification_state.db, to track which notifications have been sent. This prevents duplicate alerts and ensures scheduled tasks run only once. This database is managed by the StateTracker class.

Sources: notification_service.py:27-30, config.example.json:16

Database Schema

The schema for notification_state.db consists of three tables to track product alerts, daily reports, and general service state.

Table	Column	Type	Description
notified_products	product_id	INTEGER	The unique ID of the product from pos_inventory.db. Primary Key.
	barcode	TEXT	The product's barcode.
	product_name	TEXT	The name of the product.
	notified_at	TEXT	ISO format timestamp of when the out-of-stock notification was sent.
daily_report_log	id	INTEGER	Auto-incrementing primary key.
	report_date	TEXT	The date for which the report was generated (e.g., "2023-10-27").
	sent_at	TEXT	ISO format timestamp of when the report was sent.
	top_products_count	INTEGER	The number of top products included in the report.
	total_sales	REAL	The total sales revenue for the reported day.
service_state	key	TEXT	A key for storing arbitrary service metadata. Primary Key.
	value	TEXT	The value associated with the key.

Sources: notification_service.py:35-60

Setup and Execution

Dependencies

The service requires requests and pytz. These can be installed using the provided batch script or requirements.txt.

python -m pip install -r requirements.txt

Sources: install_dependencies.bat:23, requirements.txt:5-6

Configuration Test

Before running the service, the test_telegram.py script can be used to verify the bot token and chat ID in config.json. The script performs two checks:

1. Validates the bot token by calling the /getMe Telegram API endpoint.
2. Sends a formatted test message to the configured chat_id.

Sources: test_telegram.py:31-97

Running the Service

The service can be started by running the notification_service.py script directly or by using the run_service.bat launcher. The launcher script first verifies that config.json exists before executing the Python script.

python notification_service.py

Sources: run_service.bat:25

For deployment on Windows, build_notification_exe.bat uses PyInstaller to create a single TelegramNotifications.exe file, which can be run as a background process.

Sources: build_notification_exe.bat:16-19