From f7f0b039dd1c6907ae6d349de8933f379a5dc714 Mon Sep 17 00:00:00 2001 From: Zakhar Timoshenko Date: Sat, 3 Oct 2026 00:45:41 +0300 Subject: [PATCH] Add Telegram cancellation import functionality with tests and documentation - Implemented Telegram cancellation parsing logic in `telegramCancellationParser.js`. - Created unit tests for cancellation parsing in `telegramCancellationParser.test.js`. - Developed integration tests for Telegram import service in `telegramImport.integration.test.js`. - Added API tests for Telegram import endpoints in `telegramImportApi.test.js`. - Introduced service tests for Telegram import logic in `telegramImportService.test.js`. - Documented the Telegram cancellation import process in `telegram-cancellation-import.md`. - Built a React component for managing Telegram imports in `TelegramImportPanel.jsx`. --- .env.example | 14 ++ .gitignore | 3 + README.md | 1 + .../migrations/013_telegram_alert_import.sql | 45 ++++ backend/package-lock.json | 128 +++++++++++ backend/package.json | 2 + backend/scripts/telegram-login.js | 50 +++++ backend/src/config/telegram.js | 47 ++++ backend/src/index.js | 4 + backend/src/routes/alerts.js | 4 +- backend/src/routes/api.js | 1 + backend/src/routes/telegramImport.js | 33 +++ backend/src/services/alertCreationService.js | 2 +- .../src/services/alertNotificationService.js | 15 +- backend/src/services/alertSchedulerService.js | 4 +- .../services/telegramCancellationParser.js | 139 ++++++++++++ backend/src/services/telegramImportService.js | 206 ++++++++++++++++++ backend/src/services/telegramSource.js | 57 +++++ backend/test/alertNotificationService.test.js | 15 +- .../test/telegramCancellationParser.test.js | 89 ++++++++ .../test/telegramImport.integration.test.js | 130 +++++++++++ backend/test/telegramImportApi.test.js | 40 ++++ backend/test/telegramImportService.test.js | 82 +++++++ docker-compose.yml | 12 + docs/telegram-cancellation-import.md | 97 +++++++++ .../src/components/TelegramImportPanel.jsx | 135 ++++++++++++ frontend/src/pages/AlertsPage.jsx | 37 +++- frontend/src/services/api.js | 6 + 28 files changed, 1381 insertions(+), 17 deletions(-) create mode 100644 backend/migrations/013_telegram_alert_import.sql create mode 100644 backend/scripts/telegram-login.js create mode 100644 backend/src/config/telegram.js create mode 100644 backend/src/routes/telegramImport.js create mode 100644 backend/src/services/telegramCancellationParser.js create mode 100644 backend/src/services/telegramImportService.js create mode 100644 backend/src/services/telegramSource.js create mode 100644 backend/test/telegramCancellationParser.test.js create mode 100644 backend/test/telegramImport.integration.test.js create mode 100644 backend/test/telegramImportApi.test.js create mode 100644 backend/test/telegramImportService.test.js create mode 100644 docs/telegram-cancellation-import.md create mode 100644 frontend/src/components/TelegramImportPanel.jsx diff --git a/.env.example b/.env.example index 31cc869..0bb8140 100644 --- a/.env.example +++ b/.env.example @@ -28,6 +28,20 @@ FIREBASE_SECRETS_DIR=/opt/polotsk-transit/secrets # with the absolute host path to the JSON file. GOOGLE_APPLICATION_CREDENTIALS=/run/secrets/firebase-service-account.json +# Telegram cancellation import through a separate user account (MTProto). +# Get API credentials at my.telegram.org. Never put a session in the checkout. +TELEGRAM_IMPORT_ENABLED=false +TELEGRAM_AUTO_PUBLISH=true +TELEGRAM_API_ID= +TELEGRAM_API_HASH= +TELEGRAM_CHANNEL=avp2Polotsk +TELEGRAM_CITY_SLUG=polotsk +TELEGRAM_POLL_INTERVAL_MS=60000 +# Host directory, mounted read-only into the backend container. +TELEGRAM_SECRETS_DIR=/opt/polotsk-transit/telegram-secrets +# For direct Node.js launches, replace this with the absolute host path. +TELEGRAM_SESSION_FILE=/run/telegram-secrets/account.session + # Bootstrap admin BOOTSTRAP_ADMIN=true BOOTSTRAP_ADMIN_RESET_PASSWORD=false diff --git a/.gitignore b/.gitignore index 8ccbc34..61b6d42 100644 --- a/.gitignore +++ b/.gitignore @@ -39,6 +39,9 @@ nginx/ssl/ # Runtime secrets secrets/ +*.session +*.session-journal +telegram-session* # Temp files *.tmp diff --git a/README.md b/README.md index fa72588..df8b193 100644 --- a/README.md +++ b/README.md @@ -14,6 +14,7 @@ - ✅ Аутентификация по API ключам - ✅ Rate limiting - ✅ Серверные push-уведомления об алертах через Firebase Cloud Messaging +- ✅ Импорт отмен из Telegram через отдельный аккаунт, проверка в админке и срок действия до полуночи ([настройка](docs/telegram-cancellation-import.md)) ### Технологический стек diff --git a/backend/migrations/013_telegram_alert_import.sql b/backend/migrations/013_telegram_alert_import.sql new file mode 100644 index 0000000..57af009 --- /dev/null +++ b/backend/migrations/013_telegram_alert_import.sql @@ -0,0 +1,45 @@ +-- Admin alert forms already store UTC clock values. Give these columns an +-- explicit timezone so imports, API serialization and midnight expiry agree +-- regardless of the PostgreSQL or Node.js host timezone. Safe to rerun. +DO $$ +DECLARE col text; +BEGIN + FOREACH col IN ARRAY ARRAY['start_time', 'end_time', 'created_at'] LOOP + IF EXISTS (SELECT 1 FROM information_schema.columns + WHERE table_schema = current_schema() AND table_name = 'alerts' + AND column_name = col AND data_type = 'timestamp without time zone') THEN + EXECUTE format('ALTER TABLE alerts ALTER COLUMN %I TYPE timestamptz USING %I AT TIME ZONE ''UTC''', col, col); + END IF; + END LOOP; +END $$; + +ALTER TABLE alerts ADD COLUMN IF NOT EXISTS notification_pending BOOLEAN NOT NULL DEFAULT false; + +CREATE TABLE IF NOT EXISTS telegram_import_messages ( + id SERIAL PRIMARY KEY, + city_id INTEGER NOT NULL REFERENCES cities(id), + channel_id TEXT NOT NULL, + channel_username VARCHAR(32) NOT NULL, + message_id INTEGER NOT NULL CHECK (message_id > 0), + published_at TIMESTAMPTZ NOT NULL, + edited_at TIMESTAMPTZ, + content_hash CHAR(64) NOT NULL, + raw_text TEXT NOT NULL, + has_media BOOLEAN NOT NULL DEFAULT false, + status VARCHAR(20) NOT NULL CHECK (status IN ('review', 'published', 'dismissed', 'expired')), + reasons JSONB NOT NULL DEFAULT '[]', + parsed JSONB NOT NULL DEFAULT '{}', + reviewed_by INTEGER REFERENCES users(id) ON DELETE SET NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + UNIQUE (city_id, channel_id, message_id) +); + +CREATE TABLE IF NOT EXISTS telegram_import_alerts ( + message_id INTEGER NOT NULL REFERENCES telegram_import_messages(id) ON DELETE CASCADE, + route_id INTEGER NOT NULL REFERENCES routes(id) ON DELETE CASCADE, + alert_id INTEGER UNIQUE REFERENCES alerts(id) ON DELETE SET NULL, + PRIMARY KEY (message_id, route_id) +); + +CREATE INDEX IF NOT EXISTS idx_telegram_import_review ON telegram_import_messages(city_id, status, published_at DESC); diff --git a/backend/package-lock.json b/backend/package-lock.json index 9d02f61..48bd33d 100644 --- a/backend/package-lock.json +++ b/backend/package-lock.json @@ -25,6 +25,7 @@ "pg": "^8.11.3", "redis": "^4.6.10", "socket.io": "^4.7.2", + "teleproto": "1.229.1", "uuid": "^9.0.1", "winston": "^3.11.0", "ws": "^8.20.0" @@ -1087,6 +1088,15 @@ "integrity": "sha512-V/Hy/X9Vt7f3BbPJEi8BdVFMByHi+jNXrYkW3huaybV/kQ0KJg0Y6PkEMbn+zeT+i+SiKZ/HMqJGIIt4LZDqNQ==", "license": "MIT" }, + "node_modules/big-integer": { + "version": "1.6.52", + "resolved": "https://registry.npmjs.org/big-integer/-/big-integer-1.6.52.tgz", + "integrity": "sha512-QxD8cf2eVqJOOz63z6JIN9BzvVs/dlySa5HGSBH5xtR8dPteIRQnBxxKqkNTiT6jbDTF6jAfrd4oMcND9RGbQg==", + "license": "Unlicense", + "engines": { + "node": ">=0.6" + } + }, "node_modules/bignumber.js": { "version": "9.3.1", "resolved": "https://registry.npmjs.org/bignumber.js/-/bignumber.js-9.3.1.tgz", @@ -2732,6 +2742,12 @@ "url": "https://github.com/sponsors/ljharb" } }, + "node_modules/graceful-fs": { + "version": "4.2.11", + "resolved": "https://registry.npmjs.org/graceful-fs/-/graceful-fs-4.2.11.tgz", + "integrity": "sha512-RbJ5/jmFcNNCcDV5o9eTnBLJ/HszWV0P73bc+Ff4nS/rJj+YaS6IGyiOL0VoBYX+l1Wrl3k63h/KrH+nhJ0XvQ==", + "license": "ISC" + }, "node_modules/gtoken": { "version": "7.1.0", "resolved": "https://registry.npmjs.org/gtoken/-/gtoken-7.1.0.tgz", @@ -2987,12 +3003,30 @@ "dev": true, "license": "ISC" }, + "node_modules/imurmurhash": { + "version": "0.1.4", + "resolved": "https://registry.npmjs.org/imurmurhash/-/imurmurhash-0.1.4.tgz", + "integrity": "sha512-JmXMZ6wuvDmLiHEml9ykzqO6lwFbof0GG4IkcGaENdCRDDmMVnny7s5HsIgHCbaq0w2MyPhDqkhTUgS2LU2PHA==", + "license": "MIT", + "engines": { + "node": ">=0.8.19" + } + }, "node_modules/inherits": { "version": "2.0.4", "resolved": "https://registry.npmjs.org/inherits/-/inherits-2.0.4.tgz", "integrity": "sha512-k/vGaX4/Yla3WzyMCvTQOXYeIHvqOKtnqBduzTHpzpQZzAskKMhZ2K+EnBiSM9zGSoIFeMpXKxa4dYeZIQqewQ==", "license": "ISC" }, + "node_modules/ip-address": { + "version": "10.7.3", + "resolved": "https://registry.npmjs.org/ip-address/-/ip-address-10.7.3.tgz", + "integrity": "sha512-A1kdq/tSb5QjvKvAMgIoEvDBIgL7qaqVP/jkvSwYYRZ9iEzvPpopxp2wQfu3SuZRHtpHNxMn8Fs0bS+gf5Xmwg==", + "license": "MIT", + "engines": { + "node": ">= 12" + } + }, "node_modules/ipaddr.js": { "version": "1.9.1", "resolved": "https://registry.npmjs.org/ipaddr.js/-/ipaddr.js-1.9.1.tgz", @@ -3544,6 +3578,18 @@ "node-gyp-build-test": "build-test.js" } }, + "node_modules/node-localstorage": { + "version": "2.2.1", + "resolved": "https://registry.npmjs.org/node-localstorage/-/node-localstorage-2.2.1.tgz", + "integrity": "sha512-vv8fJuOUCCvSPjDjBLlMqYMHob4aGjkmrkaE42/mZr0VT+ZAU10jRF8oTnX9+pgU9/vYJ8P7YT3Vd6ajkmzSCw==", + "license": "MIT", + "dependencies": { + "write-file-atomic": "^1.1.4" + }, + "engines": { + "node": ">=0.12" + } + }, "node_modules/nodemon": { "version": "3.1.11", "resolved": "https://registry.npmjs.org/nodemon/-/nodemon-3.1.11.tgz", @@ -4403,6 +4449,25 @@ "node": ">=10" } }, + "node_modules/slide": { + "version": "1.1.6", + "resolved": "https://registry.npmjs.org/slide/-/slide-1.1.6.tgz", + "integrity": "sha512-NwrtjCg+lZoqhFU8fOwl4ay2ei8PaqCBOUV3/ektPY9trO1yQ1oXEfmHAhKArUVUr/hOHvy5f6AdP17dCM0zMw==", + "license": "ISC", + "engines": { + "node": "*" + } + }, + "node_modules/smart-buffer": { + "version": "4.2.0", + "resolved": "https://registry.npmjs.org/smart-buffer/-/smart-buffer-4.2.0.tgz", + "integrity": "sha512-94hK0Hh8rPqQl2xXc3HsaBoOXKV20MToPkcXvwbISWLEs+64sBq5kFgn2kJDHb1Pry9yrP0dxrCI9RRci7RXKg==", + "license": "MIT", + "engines": { + "node": ">= 6.0.0", + "npm": ">= 3.0.0" + } + }, "node_modules/socket.io": { "version": "4.8.1", "resolved": "https://registry.npmjs.org/socket.io/-/socket.io-4.8.1.tgz", @@ -4534,6 +4599,20 @@ "integrity": "sha512-6FlzubTLZG3J2a/NVCAleEhjzq5oxgHyaCU9yYXvcLsvoVaHJq/s5xXI6/XXP6tz7R9xAOtHnSO/tXtF3WRTlA==", "license": "MIT" }, + "node_modules/socks": { + "version": "2.8.10", + "resolved": "https://registry.npmjs.org/socks/-/socks-2.8.10.tgz", + "integrity": "sha512-e0VyvkVTwVYViNovRkZ9aodhxVlyoMn7eJhVUPxZ+eK9P/7CBkxvvsBOHqFPEH416726W8tLXXXjKwqgTErrCQ==", + "license": "MIT", + "dependencies": { + "ip-address": "^10.1.1", + "smart-buffer": "^4.2.0" + }, + "engines": { + "node": ">= 10.0.0", + "npm": ">= 3.0.0" + } + }, "node_modules/split2": { "version": "4.2.0", "resolved": "https://registry.npmjs.org/split2/-/split2-4.2.0.tgz", @@ -4561,6 +4640,12 @@ "node": ">= 0.8" } }, + "node_modules/store2": { + "version": "2.14.4", + "resolved": "https://registry.npmjs.org/store2/-/store2-2.14.4.tgz", + "integrity": "sha512-srTItn1GOvyvOycgxjAnPA63FZNwy0PTyUBFMHRM+hVFltAeoh0LmNBz9SZqUS9mMqGk8rfyWyXn3GH5ReJ8Zw==", + "license": "MIT" + }, "node_modules/stream-events": { "version": "1.0.5", "resolved": "https://registry.npmjs.org/stream-events/-/stream-events-1.0.5.tgz", @@ -4804,6 +4889,38 @@ "license": "MIT", "optional": true }, + "node_modules/teleproto": { + "version": "1.229.1", + "resolved": "https://registry.npmjs.org/teleproto/-/teleproto-1.229.1.tgz", + "integrity": "sha512-QZBSWF/OWeIEi2LRvAPgaga7Z7K2TM1f2Xl/IW7WHTRRHGKWQVRDxQHbSf9OPtyBeSYEh2KMSY6v+Ima+U+hIw==", + "license": "MIT", + "dependencies": { + "big-integer": "^1.6.48", + "mime": "^3.0.0", + "node-localstorage": "^2.2.1", + "socks": "^2.6.2", + "store2": "^2.13.0" + }, + "engines": { + "node": ">=18" + }, + "funding": { + "type": "github", + "url": "https://github.com/sponsors/sanyok12345" + } + }, + "node_modules/teleproto/node_modules/mime": { + "version": "3.0.0", + "resolved": "https://registry.npmjs.org/mime/-/mime-3.0.0.tgz", + "integrity": "sha512-jSCU7/VB1loIWBZe14aEYHU/+1UMEHoaO7qxCOVJOw9GgH72VAWppxNcjU+x9a2k3GSIBXNKxXQFqRvvZ7vr3A==", + "license": "MIT", + "bin": { + "mime": "cli.js" + }, + "engines": { + "node": ">=10.0.0" + } + }, "node_modules/text-hex": { "version": "1.0.0", "resolved": "https://registry.npmjs.org/text-hex/-/text-hex-1.0.0.tgz", @@ -5230,6 +5347,17 @@ "license": "ISC", "optional": true }, + "node_modules/write-file-atomic": { + "version": "1.3.4", + "resolved": "https://registry.npmjs.org/write-file-atomic/-/write-file-atomic-1.3.4.tgz", + "integrity": "sha512-SdrHoC/yVBPpV0Xq/mUZQIpW2sWXAShb/V4pomcJXh92RuaO+f3UTWItiR3Px+pLnV2PvC2/bfn5cwr5X6Vfxw==", + "license": "ISC", + "dependencies": { + "graceful-fs": "^4.1.11", + "imurmurhash": "^0.1.4", + "slide": "^1.1.5" + } + }, "node_modules/ws": { "version": "8.20.0", "resolved": "https://registry.npmjs.org/ws/-/ws-8.20.0.tgz", diff --git a/backend/package.json b/backend/package.json index e492145..c1bf2c5 100644 --- a/backend/package.json +++ b/backend/package.json @@ -9,6 +9,7 @@ "test": "node --test", "migrate": "node src/utils/migrate.js", "create-admin": "node scripts/create-admin.js", + "telegram:login": "node scripts/telegram-login.js", "seed": "node src/utils/seed.js" }, "keywords": [ @@ -35,6 +36,7 @@ "pg": "^8.11.3", "redis": "^4.6.10", "socket.io": "^4.7.2", + "teleproto": "1.229.1", "uuid": "^9.0.1", "winston": "^3.11.0", "ws": "^8.20.0" diff --git a/backend/scripts/telegram-login.js b/backend/scripts/telegram-login.js new file mode 100644 index 0000000..8f1c187 --- /dev/null +++ b/backend/scripts/telegram-login.js @@ -0,0 +1,50 @@ +require('dotenv').config(); +const fs = require('node:fs/promises'); +const readline = require('node:readline'); +const { getTelegramConfig, missingTelegramConfig, assertExternalSessionFile } = require('../src/config/telegram'); +const { publicError } = require('../src/services/telegramImportService'); + +function ask(prompt, hidden = false) { + return new Promise((resolve, reject) => { + const rl = readline.createInterface({ input: process.stdin, output: process.stdout, terminal: true }); + rl.question(prompt, answer => { rl.close(); process.stdout.write('\n'); resolve(answer.trim()); }); + if (hidden) rl._writeToOutput = () => {}; + rl.on('SIGINT', () => { rl.close(); reject(new Error('LOGIN_CANCELLED')); }); + }); +} + +async function main() { + const config = getTelegramConfig(); + if (!process.stdin.isTTY) throw new Error('LOGIN_REQUIRES_TERMINAL'); + if (missingTelegramConfig(config).length) throw new Error('LOGIN_CONFIG_REQUIRED'); + // The operator creates a private directory outside the checkout first. + const file = await assertExternalSessionFile(config.sessionFile); + let session = ''; + try { session = (await fs.readFile(file, 'utf8')).trim(); } + catch (error) { if (error.code !== 'ENOENT') throw error; } + const { TelegramClient } = require('teleproto'); + const { StringSession } = require('teleproto/sessions'); + const { Logger } = require('teleproto/extensions/Logger'); + const client = new TelegramClient(new StringSession(session), config.apiId, config.apiHash, { baseLogger: new Logger('none'), connectionRetries: 2 }); + try { + await client.start({ + phoneNumber: () => ask('Номер отдельного Telegram-аккаунта: '), + phoneCode: () => ask('Код из Telegram: ', true), + password: () => ask('Пароль двухэтапной проверки: ', true), + // Do not register a new account or accept registration terms from this utility. + firstAndLastNames: () => { throw new Error('USE_EXISTING_ACCOUNT'); }, + onError: () => true + }); + if (!await client.checkAuthorization() || (await client.getMe()).bot) throw new Error('USER_ACCOUNT_REQUIRED'); + const handle = await fs.open(file, session ? 'w' : 'wx', 0o600); + try { await handle.writeFile(client.session.save()); await handle.sync(); } + finally { await handle.close(); } + console.log('Авторизация сохранена в защищённый файл. Теперь можно включить TELEGRAM_IMPORT_ENABLED=true.'); + } finally { await client.destroy(); } +} + +main().catch(error => { + const local = ['LOGIN_CANCELLED', 'LOGIN_REQUIRES_TERMINAL', 'LOGIN_CONFIG_REQUIRED', 'USE_EXISTING_ACCOUNT']; + console.error(`Авторизация не завершена (${local.includes(error.message) ? error.message : publicError(error)}).`); + process.exitCode = 1; +}); diff --git a/backend/src/config/telegram.js b/backend/src/config/telegram.js new file mode 100644 index 0000000..cedf542 --- /dev/null +++ b/backend/src/config/telegram.js @@ -0,0 +1,47 @@ +const fs = require('node:fs/promises'); +const path = require('node:path'); + +function getTelegramConfig(env = process.env) { + return { + enabled: env.TELEGRAM_IMPORT_ENABLED === 'true', + autoPublish: env.TELEGRAM_AUTO_PUBLISH !== 'false', + apiId: Number(env.TELEGRAM_API_ID), + apiHash: env.TELEGRAM_API_HASH || '', + sessionFile: env.TELEGRAM_SESSION_FILE || '', + channel: (env.TELEGRAM_CHANNEL || 'avp2Polotsk').replace(/^@/, ''), + citySlug: env.TELEGRAM_CITY_SLUG || 'polotsk', + intervalMs: Math.max(30_000, Number(env.TELEGRAM_POLL_INTERVAL_MS) || 60_000) + }; +} + +function missingTelegramConfig(config) { + const missing = []; + if (!Number.isSafeInteger(config.apiId) || config.apiId <= 0) missing.push('TELEGRAM_API_ID'); + if (!/^[a-f0-9]{32}$/i.test(config.apiHash)) missing.push('TELEGRAM_API_HASH'); + if (!path.isAbsolute(config.sessionFile)) missing.push('TELEGRAM_SESSION_FILE'); + if (!/^[a-zA-Z][a-zA-Z0-9_]{4,31}$/.test(config.channel)) missing.push('TELEGRAM_CHANNEL'); + if (!/^[a-z0-9]+(?:-[a-z0-9]+)*$/.test(config.citySlug)) missing.push('TELEGRAM_CITY_SLUG'); + return missing; +} + +async function assertExternalSessionFile(file) { + if (!path.isAbsolute(file)) throw new Error('SESSION_FILE_INVALID'); + const root = await fs.realpath(path.resolve(__dirname, '../../..')); + // Resolve the parent as well, so a symlink cannot put the credential in the checkout. + const parent = await fs.realpath(path.dirname(file)); + const resolved = path.join(parent, path.basename(file)); + if (resolved === root || resolved.startsWith(`${root}${path.sep}`)) { + throw new Error('SESSION_FILE_IN_CHECKOUT'); + } + try { + const stat = await fs.lstat(resolved); + if (!stat.isFile() || stat.isSymbolicLink() || (stat.mode & 0o077) !== 0) { + throw new Error('SESSION_FILE_PERMISSIONS'); + } + } catch (error) { + if (error.code !== 'ENOENT') throw error; + } + return resolved; +} + +module.exports = { getTelegramConfig, missingTelegramConfig, assertExternalSessionFile }; diff --git a/backend/src/index.js b/backend/src/index.js index d7c05ff..3d88415 100644 --- a/backend/src/index.js +++ b/backend/src/index.js @@ -13,11 +13,13 @@ const { ensureRealtimeSchema } = require('./services/realtimeSchema'); const { attachRealtimeServer, closeRealtimeServer } = require('./services/realtimeService'); const { ensureBootstrapAdmin } = require('./utils/bootstrapAdmin'); const { startAlertScheduler } = require('./services/alertSchedulerService'); +const { startTelegramImporter } = require('./services/telegramImportService'); const app = express(); const PORT = process.env.PORT || 3000; let server; let stopAlertScheduler = () => {}; +let stopTelegramImporter = () => Promise.resolve(); // Security middleware app.use(helmet({ @@ -128,6 +130,7 @@ function shutdown(signal) { closeRealtimeServer(); stopAlertScheduler(); + stopTelegramImporter().catch(() => {}); if (!server) { process.exit(0); @@ -168,6 +171,7 @@ async function startServer() { await ensureRealtimeSchema(); await ensureBootstrapAdmin(); stopAlertScheduler = startAlertScheduler(); + stopTelegramImporter = startTelegramImporter(); server = app.listen(PORT, () => { logger.info(`Server is running on port ${PORT}`); diff --git a/backend/src/routes/alerts.js b/backend/src/routes/alerts.js index 70c78f5..c049c5f 100644 --- a/backend/src/routes/alerts.js +++ b/backend/src/routes/alerts.js @@ -70,7 +70,7 @@ router.get('/active', async (req, res) => { WHERE a.city_id = $1 AND a.is_active = true AND a.start_time <= NOW() - AND (a.end_time IS NULL OR a.end_time >= NOW()) + AND (a.end_time IS NULL OR a.end_time > NOW()) `; const params = [req.city.id]; @@ -241,7 +241,7 @@ router.post('/cleanup', async (req, res) => { SET is_active = false WHERE city_id = $1 AND is_active = true AND end_time IS NOT NULL - AND end_time < NOW() + AND end_time <= NOW() RETURNING id` , [req.city.id]); diff --git a/backend/src/routes/api.js b/backend/src/routes/api.js index b464435..78e30fa 100644 --- a/backend/src/routes/api.js +++ b/backend/src/routes/api.js @@ -24,6 +24,7 @@ router.use('/cities', require('./cities')); router.use(resolveCity); router.use('/schedule-import', require('./scheduleImport')); +router.use('/telegram-import', require('./telegramImport')); router.get('/quality', authenticateToken, requireAdmin, qualityController.getReport.bind(qualityController)); router.get('/schedule-center', authenticateToken, requireAdmin, scheduleCenterController.getOverview.bind(scheduleCenterController)); diff --git a/backend/src/routes/telegramImport.js b/backend/src/routes/telegramImport.js new file mode 100644 index 0000000..4eb3238 --- /dev/null +++ b/backend/src/routes/telegramImport.js @@ -0,0 +1,33 @@ +const express = require('express'); +const { authenticateToken, requireAdmin } = require('../middleware/auth'); +const { getTelegramImportService } = require('../services/telegramImportService'); + +function createTelegramImportRouter(service = getTelegramImportService()) { + const router = express.Router(); + router.use(authenticateToken, requireAdmin); + const handle = action => async (req, res) => { + try { res.json(await action(req)); } + catch (error) { res.status(error.status || 500).json({ error: error.status ? error.message : 'Не удалось обработать импорт Telegram. Проверьте миграцию 013 и подключение базы.' }); } + }; + router.get('/', handle(async req => ({ status: service.getStatus(), messages: await service.list(req.city.id) }))); + router.post('/sync', handle(req => { + if (req.city.slug !== service.config.citySlug) { + const error = new Error('Источник Telegram настроен для другого города.'); + error.status = 409; + throw error; + } + return service.sync(); + })); + router.post('/:id/review', handle(req => { + if (!/^\d+$/.test(req.params.id)) { + const error = new Error('Некорректный идентификатор сообщения.'); + error.status = 400; + throw error; + } + return service.review(Number(req.params.id), req.city.id, req.body, req.user.id); + })); + return router; +} + +module.exports = createTelegramImportRouter(); +module.exports.createTelegramImportRouter = createTelegramImportRouter; diff --git a/backend/src/services/alertCreationService.js b/backend/src/services/alertCreationService.js index d50e922..bdd387f 100644 --- a/backend/src/services/alertCreationService.js +++ b/backend/src/services/alertCreationService.js @@ -113,7 +113,7 @@ function isAlertDue(alert, now = new Date()) { } const endTime = new Date(alert.end_time); - return !Number.isNaN(endTime.getTime()) && endTime >= now; + return !Number.isNaN(endTime.getTime()) && endTime > now; } class AlertCreationService { diff --git a/backend/src/services/alertNotificationService.js b/backend/src/services/alertNotificationService.js index b482904..d68e4e9 100644 --- a/backend/src/services/alertNotificationService.js +++ b/backend/src/services/alertNotificationService.js @@ -90,6 +90,8 @@ function isAlertActive(alert = {}) { function buildAlertCreatedMessage(alert, options = {}) { const alertId = getAlertId(alert); const topic = options.topic || DEFAULT_FCM_TOPIC; + const endTime = toIsoString(firstDefined(alert.end_time, alert.endTime)); + const now = options.now || new Date(); return { topic, @@ -100,12 +102,13 @@ function buildAlertCreatedMessage(alert, options = {}) { routeId: getRouteId(alert), routeNumber: getRouteNumber(alert), startTime: toIsoString(firstDefined(alert.start_time, alert.startTime)), - endTime: toIsoString(firstDefined(alert.end_time, alert.endTime)), + endTime, title: getNotificationTitle(alert), body: getNotificationBody(alert) }, android: { - priority: 'high' + priority: 'high', + ...(endTime ? { ttl: Math.max(0, Math.min(28 * 86_400_000, Date.parse(endTime) - new Date(now).getTime())) } : {}) } }; } @@ -124,6 +127,7 @@ class AlertNotificationService { this.enabled = options.enabled === undefined ? firebaseConfig.enabled : options.enabled; this.topic = options.topic || firebaseConfig.topic; this.logger = options.logger || getDefaultLogger(); + this.now = options.now || (() => new Date()); this.getMessaging = options.getMessaging || options.messagingFactory || (() => ( options.messaging || getFirebaseMessaging({ ...firebaseConfig, @@ -144,13 +148,18 @@ class AlertNotificationService { return { sent: false, skipped: true, reason: 'inactive' }; } + const endTime = toIsoString(firstDefined(alert.end_time, alert.endTime)); + if (endTime && Date.parse(endTime) <= this.now().getTime()) { + return { sent: false, skipped: true, reason: 'expired' }; + } + try { const messaging = await this.getMessaging(); if (!messaging || typeof messaging.send !== 'function') { throw new Error('Firebase Messaging is unavailable'); } - const message = buildAlertCreatedMessage(alert, { topic: this.topic }); + const message = buildAlertCreatedMessage(alert, { topic: this.topic, now: this.now() }); const messageId = await messaging.send(message); const normalizedMessageId = toDataString(messageId); diff --git a/backend/src/services/alertSchedulerService.js b/backend/src/services/alertSchedulerService.js index faaf30f..ea566a2 100644 --- a/backend/src/services/alertSchedulerService.js +++ b/backend/src/services/alertSchedulerService.js @@ -3,9 +3,9 @@ const CLAIM_DUE_ALERTS_SQL = ` SELECT a.id FROM alerts a WHERE a.is_active = true - AND a.start_time > a.created_at + AND (a.start_time > a.created_at OR a.notification_pending = true) AND a.start_time <= CURRENT_TIMESTAMP - AND (a.end_time IS NULL OR a.end_time >= CURRENT_TIMESTAMP) + AND (a.end_time IS NULL OR a.end_time > CURRENT_TIMESTAMP) AND a.notification_sent_at IS NULL AND a.notification_attempts < 5 AND ( diff --git a/backend/src/services/telegramCancellationParser.js b/backend/src/services/telegramCancellationParser.js new file mode 100644 index 0000000..10794e6 --- /dev/null +++ b/backend/src/services/telegramCancellationParser.js @@ -0,0 +1,139 @@ +const { getLocalDateTime } = require('../utils/etaTime'); + +const MONTHS = ['января', 'февраля', 'марта', 'апреля', 'мая', 'июня', 'июля', 'августа', 'сентября', 'октября', 'ноября', 'декабря']; +const ROUTE = /маршрут(?:у|а|е|ом)?\s*(?:№|номер|no\.?)?\s*(\d{1,3}[а-яa-z]?)(?![\dа-яa-z])/giu; +const TIME = /(? `${y}-${String(month).padStart(2, '0')}-${String(day).padStart(2, '0')}`; + if (year) return format(Number(year)); + const base = Number(publicationDate.slice(0, 4)); + return [base - 1, base, base + 1].map(format).filter(validDate) + .sort((a, b) => Math.abs(Date.parse(a) - Date.parse(publicationDate)) - Math.abs(Date.parse(b) - Date.parse(publicationDate)))[0] || format(base); +} + +function parseDate(text, publishedAt) { + const publicationDate = getLocalDateTime(new Date(publishedAt)).date; + const dates = []; + let dateText = text; + const patterns = [ + /(? { + const [full, d, m] = args; + const y = i === 1 ? undefined : args[3]; + const offset = args.at(-2); + // In carrier posts “с остановки ... на 09.10” is a departure, not a date. + if (i === 1 && /с\s+остановк[иа]\s+[^\n]*$/i.test(dateText.slice(0, offset))) return full; + dates.push(inferYear(d, i === 2 ? MONTHS.indexOf(m.toLowerCase()) + 1 : m, y, publicationDate)); + return ' '; + }); + } + if (/сегодня/i.test(text)) dates.push(publicationDate); + if (/завтра/i.test(text)) dates.push(addDays(publicationDate, 1)); + if (/послезавтра/i.test(text)) return { reason: 'Дата «послезавтра» требует проверки.' }; + if (dates.some(d => !validDate(d))) return { reason: 'Некорректная дата отмены.' }; + if (new Set(dates).size > 1) return { reason: 'Указаны разные даты отмены.' }; + return { date: dates[0] || publicationDate, dateText, inferred: dates.length === 0 }; +} + +function normalizeRoute(value) { return String(value).trim().toLowerCase(); } +function normalizeStop(value) { + const key = String(value).toLowerCase().replace(/ё/g, 'е').replace(/[^а-яa-z0-9]/g, ''); + return key === 'аввокзал' ? 'автовокзал' : key; +} +function canonicalStop(value) { return normalizeStop(value) === 'автовокзал' ? 'Автовокзал' : value; } + +function parseCancellation({ text = '', publishedAt, hasMedia = false }) { + const normalized = text.replace(/\u00a0/g, ' ').replace(/\r/g, '').trim(); + let serviceDate; + const review = reason => ({ status: 'review', reasons: [reason], groups: [], ...(serviceDate ? { serviceDate } : {}) }); + if (!normalized) return hasMedia ? review('Публикация содержит вложение без текста. Нужна проверка.') : { status: 'ignored', reasons: [], groups: [] }; + if (!/(?:не\s+буд(?:ут|ет)\s+выполняться|не\s+выполняются|отмен(?:ены|яется|яются|а\s+рейс)|рейс[^\n]*отмен)/i.test(normalized)) { + return { status: 'ignored', reasons: [], groups: [] }; + } + if (hasMedia) return review('В публикации есть вложение. Проверьте все отмены по оригиналу.'); + if (normalized.length > 20_000) return review('Слишком длинное сообщение. Нужна проверка.'); + if (/возобнов|восстанов|кроме|только|ежеднев|еженедель|понедельник|вторник|сред[ау]|четверг|пятниц|суббот|воскресен|будн|выходн|период|дальнейш|неопредел|(?:с\s+\d[^\n]+\s+по\s+\d)|до\s+\d/i.test(normalized)) { + return review('Есть условия, период или восстановление движения. Нужна проверка.'); + } + if (/буд(?:ут|ет)\s+выполняться/i.test(normalized.replace(/не\s+буд(?:ут|ет)\s+выполняться/gi, '')) || /отмена\s+отменена/i.test(normalized)) { + return review('В сообщении есть сведения о выполнении рейсов. Нужна проверка.'); + } + if (Number.isNaN(new Date(publishedAt).getTime())) return review('Неизвестно время публикации.'); + const parsedDate = parseDate(normalized, publishedAt); + if (parsedDate.reason) return review(parsedDate.reason); + serviceDate = parsedDate.date; + const routeMatches = [...parsedDate.dateText.matchAll(ROUTE)]; + if (!routeMatches.length) return review('Не удалось определить номер маршрута.'); + const groups = []; + for (const [i, route] of routeMatches.entries()) { + const section = parsedDate.dateText.slice(route.index + route[0].length, routeMatches[i + 1]?.index); + const stopPattern = /с\s+остановк[иа]\s+(.+?)\s+(?:на|в)\s+((?:\d{1,2}[:.]\d{2}(?:\s*[,;]\s*|\s+и\s+|\s+(?=\d))?)+)/giu; + const departures = []; + for (const stop of section.matchAll(stopPattern)) { + const times = [...stop[2].matchAll(TIME)].map(t => `${t[1].padStart(2, '0')}:${t[2]}`); + if (times.some(t => Number(t.slice(0, 2)) > 23 || Number(t.slice(3)) > 59)) return review('Некорректное время отправления.'); + const stopName = stop[1].replace(/^[\s«"“]+|[\s»"”.,:;]+$/g, '').trim(); + if (!stopName || !times.length) return review('Не удалось определить остановку и время.'); + departures.push({ stop: canonicalStop(stopName), times: [...new Set(times)] }); + } + if (!departures.length || [...section.matchAll(TIME)].length !== departures.reduce((n, d) => n + d.times.length, 0)) { + return review('Не удалось однозначно связать все времена с остановками.'); + } + groups.push({ routeNumber: normalizeRoute(route[1]), departures }); + } + return { status: 'ready', reasons: [], serviceDate: parsedDate.date, inferredDate: parsedDate.inferred, groups }; +} + +function matchCancellationRoutes(parsed, routes) { + const plans = new Map(); + for (const group of parsed.groups) { + const candidates = routes.filter(r => normalizeRoute(r.route_number) === group.routeNumber); + for (const departure of group.departures) { + const matches = candidates.length === 1 ? candidates : candidates.filter(r => normalizeStop(r.origin_stop || '') === normalizeStop(departure.stop)); + if (matches.length !== 1) return { reasons: [`Маршрут №${group.routeNumber}, остановка «${departure.stop}»: направление не определено однозначно.`], plans: [] }; + const route = matches[0]; + if (!plans.has(route.id)) plans.set(route.id, { routeId: route.id, routeNumber: route.route_number, departures: [] }); + plans.get(route.id).departures.push(departure); + } + } + return { reasons: [], plans: [...plans.values()] }; +} + +function buildCancellationAlert(plan, serviceDate, sourceUrl, publishedAt, cityId) { + const bounds = dayBounds(serviceDate); + const dateLabel = serviceDate.split('-').reverse().join('.'); + const start = new Date(Math.max(Date.parse(bounds.start), Date.parse(publishedAt))).toISOString(); + return { + city_id: cityId, route_id: plan.routeId, alert_type: 'cancellation', + title: `Отмена отдельных рейсов маршрута №${plan.routeNumber}`, + message: `${dateLabel}: не выполняются отправления ${plan.departures.map(d => `с остановки «${d.stop}» в ${d.times.join(', ')}`).join('; ')}.\nИсточник: ${sourceUrl}`, + start_time: start, end_time: bounds.end, is_active: true + }; +} + +module.exports = { parseCancellation, matchCancellationRoutes, buildCancellationAlert, dayBounds, validDate, addDays }; diff --git a/backend/src/services/telegramImportService.js b/backend/src/services/telegramImportService.js new file mode 100644 index 0000000..3f896aa --- /dev/null +++ b/backend/src/services/telegramImportService.js @@ -0,0 +1,206 @@ +const crypto = require('node:crypto'); +const { getTelegramConfig, missingTelegramConfig } = require('../config/telegram'); +const { getLocalDateTime } = require('../utils/etaTime'); +const { TelegramSource } = require('./telegramSource'); +const { parseCancellation, matchCancellationRoutes, buildCancellationAlert, dayBounds, validDate, addDays } = require('./telegramCancellationParser'); +const { AlertCreationService } = require('./alertCreationService'); + +function httpError(status, message) { return Object.assign(new Error(message), { status }); } +function sourceUrl(row) { return `https://t.me/${row.channel_username}/${row.message_id}`; } +function contentHash(message) { + return crypto.createHash('sha256').update(JSON.stringify([message.text, message.hasMedia, message.publishedAt])).digest('hex'); +} + +function publicError(error) { + const known = ['AUTH_REQUIRED', 'USER_ACCOUNT_REQUIRED', 'SESSION_MISSING', 'SESSION_FILE_INVALID', 'SESSION_FILE_IN_CHECKOUT', 'SESSION_FILE_PERMISSIONS', 'CHANNEL_INVALID', 'CHANNEL_PRIVATE', 'USERNAME_NOT_OCCUPIED', 'HISTORY_LIMIT', 'AUTH_KEY_UNREGISTERED', 'SESSION_REVOKED']; + const value = error.errorMessage || error.message; + if (known.includes(value)) return value; + if (error.code === 'ENOENT') return 'SESSION_MISSING'; + if (error.seconds) return 'FLOOD_WAIT'; + return 'IMPORT_FAILED'; +} + +class TelegramImportService { + constructor({ pool, config = getTelegramConfig(), source, logger, now = () => new Date() } = {}) { + this.pool = pool || require('../config/database'); + this.config = config; + this.source = source || new TelegramSource(config); + this.logger = logger || require('../config/logger'); + this.now = now; + this.running = false; + this.stopped = false; + this.state = { connection: config.enabled ? 'waiting' : 'disabled', lastSyncAt: null, lastError: null, retryAt: null }; + this.alerts = new AlertCreationService({ pool: this.pool, logger: this.logger }); + } + + getStatus() { + const missing = missingTelegramConfig(this.config); + return { + ...this.state, enabled: this.config.enabled, autoPublish: this.config.autoPublish, + channel: this.config.channel, citySlug: this.config.citySlug, + connection: !this.config.enabled ? 'disabled' : missing.length ? 'configuration_required' : this.state.connection, + missing, syncing: this.running + }; + } + + async sync() { + if (!this.config.enabled) throw httpError(409, 'Импорт Telegram выключен в настройках сервера.'); + if (missingTelegramConfig(this.config).length) throw httpError(409, 'Укажите Telegram API ID, API hash и файл авторизации на сервере.'); + if (this.running || this.stopped) throw httpError(409, 'Импорт уже выполняется или остановлен.'); + if (this.state.retryAt && Date.parse(this.state.retryAt) > this.now().getTime()) throw httpError(429, 'Telegram ограничил частоту запросов. Дождитесь следующей попытки.'); + this.running = true; + this.state.connection = 'connecting'; + try { + const city = (await this.pool.query('SELECT id FROM cities WHERE slug = $1 AND is_active = true', [this.config.citySlug])).rows[0]; + if (!city) throw httpError(409, 'Город источника отсутствует или отключён.'); + const today = getLocalDateTime(this.now()).date; + const messages = await this.source.fetchMessages(new Date(dayBounds(addDays(today, -2)).start)); + const counts = { published: 0, review: 0, unchanged: 0, ignored: 0, expired: 0 }; + for (const message of messages) { + if (this.stopped) break; + const result = await this.processMessage(message, city.id); + counts[result.status] = (counts[result.status] || 0) + 1; + } + this.state = { connection: 'ready', lastSyncAt: this.now().toISOString(), lastError: null, retryAt: null }; + return counts; + } catch (error) { + const code = publicError(error); + const delay = error.seconds ? Math.min(86_400, Math.max(30, Number(error.seconds) || 60)) : 60; + this.state.connection = 'error'; + this.state.lastError = code; + this.state.retryAt = new Date(this.now().getTime() + delay * 1000).toISOString(); + this.logger.warn('Telegram channel import failed', { code }); + await this.source.close().catch(() => {}); + if (error.status) throw error; + throw httpError(503, `Не удалось прочитать канал Telegram (${code}). Проверьте подключение и авторизацию.`); + } finally { this.running = false; } + } + + async getRoutes(db, cityId) { + return (await db.query(`SELECT r.id, r.route_number, r.name, origin.name AS origin_stop + FROM routes r LEFT JOIN LATERAL ( + SELECT s.name FROM route_stops rs JOIN stops s ON s.id = rs.stop_id + WHERE rs.route_id = r.id ORDER BY rs.sequence LIMIT 1 + ) origin ON true WHERE r.city_id = $1 AND r.is_active = true`, [cityId])).rows; + } + + async processMessage(message, cityId) { + if (!Number.isSafeInteger(message.messageId) || message.messageId <= 0 || !/^\d+$/.test(message.channelId) + || message.channelUsername.toLowerCase() !== this.config.channel.toLowerCase()) throw new Error('CHANNEL_INVALID'); + const parsed = parseCancellation(message); + const hash = contentHash(message); + const db = await this.pool.connect(); + try { + await db.query('BEGIN'); + // The unique key and row lock serialize duplicate readers and concurrent reviews. + if (parsed.status !== 'ignored') { + await db.query(`INSERT INTO telegram_import_messages + (city_id, channel_id, channel_username, message_id, published_at, edited_at, content_hash, raw_text, has_media, status) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,'review') ON CONFLICT DO NOTHING`, + [cityId, message.channelId, message.channelUsername, message.messageId, message.publishedAt, message.editedAt, hash, message.text.slice(0, 20_000), message.hasMedia]); + } + const row = (await db.query(`SELECT * FROM telegram_import_messages + WHERE city_id = $1 AND channel_id = $2 AND message_id = $3 FOR UPDATE`, [cityId, message.channelId, message.messageId])).rows[0]; + if (!row) { await db.query('COMMIT'); return { status: 'ignored' }; } + const initialized = row.parsed && Object.keys(row.parsed).length > 0; + if ((initialized && row.content_hash === hash) || (row.edited_at && (!message.editedAt || Date.parse(message.editedAt) < new Date(row.edited_at).getTime()))) { + await db.query('COMMIT'); return { status: 'unchanged' }; + } + if (initialized) await this.deactivate(db, row.id); + let status = row.status === 'dismissed' ? 'dismissed' : 'review'; + let reasons = parsed.reasons; + let plans = []; + if (parsed.status === 'ignored') reasons = ['Текст исходной публикации изменился: отмена больше не распознаётся.']; + if (parsed.status === 'ready') { + const matched = matchCancellationRoutes(parsed, await this.getRoutes(db, cityId)); + plans = matched.plans; + reasons = matched.reasons; + if (Date.parse(dayBounds(parsed.serviceDate).end) <= this.now().getTime()) status = status === 'dismissed' ? status : 'expired'; + else if (!reasons.length && this.config.autoPublish && status !== 'dismissed') status = 'published'; + else if (!reasons.length) reasons = ['Автоматическая публикация выключена. Подтвердите уведомление.']; + } + if (status === 'published') { + for (const plan of plans) await this.saveAlert(db, row.id, buildCancellationAlert(plan, parsed.serviceDate, sourceUrl(row), message.publishedAt, cityId)); + } + await db.query(`UPDATE telegram_import_messages SET content_hash=$2, raw_text=$3, edited_at=$4, + has_media=$5, status=$6, reasons=$7::jsonb, parsed=$8::jsonb, reviewed_by=NULL, updated_at=now() WHERE id=$1`, + [row.id, hash, message.text.slice(0, 20_000), message.editedAt, message.hasMedia, status, JSON.stringify(reasons), JSON.stringify({ ...parsed, plans })]); + await db.query('COMMIT'); + return { status, id: row.id }; + } catch (error) { await db.query('ROLLBACK'); throw error; } + finally { db.release(); } + } + + async deactivate(db, messageId) { + await db.query(`UPDATE alerts SET is_active=false, notification_pending=false + WHERE id IN (SELECT alert_id FROM telegram_import_alerts WHERE message_id=$1)`, [messageId]); + } + + async saveAlert(db, messageId, payload) { + const existing = (await db.query('SELECT alert_id FROM telegram_import_alerts WHERE message_id=$1 AND route_id=$2', [messageId, payload.route_id])).rows[0]; + if (existing) { + // A deleted alert stays deleted, and editing a publication doesn't resend a delivered push. + if (existing.alert_id) await db.query(`UPDATE alerts SET title=$2, message=$3, start_time=$4, end_time=$5, + is_active=true, notification_pending=true, + notification_attempts=CASE WHEN notification_sent_at IS NULL THEN 0 ELSE notification_attempts END + WHERE id=$1`, [existing.alert_id, payload.title, payload.message, payload.start_time, payload.end_time]); + return; + } + const result = await this.alerts.insertAlert(db, payload); + await db.query('UPDATE alerts SET notification_pending=true WHERE id=$1', [result.alert.id]); + await db.query('INSERT INTO telegram_import_alerts(message_id,route_id,alert_id) VALUES ($1,$2,$3)', [messageId, payload.route_id, result.alert.id]); + } + + async list(cityId) { + return (await this.pool.query(`SELECT m.*, COALESCE((SELECT json_agg(t.alert_id) FROM telegram_import_alerts t + WHERE t.message_id=m.id AND t.alert_id IS NOT NULL), '[]') AS alert_ids + FROM telegram_import_messages m WHERE m.city_id=$1 ORDER BY (m.status='review') DESC,m.published_at DESC,m.id DESC LIMIT 100`, [cityId])).rows.map(row => ({ + ...row, source_url: sourceUrl(row), + status: row.status === 'published' && validDate(row.parsed.serviceDate) && Date.parse(dayBounds(row.parsed.serviceDate).end) <= this.now().getTime() ? 'expired' : row.status + })); + } + + async review(id, cityId, input, userId) { + const db = await this.pool.connect(); + try { + await db.query('BEGIN'); + const row = (await db.query('SELECT * FROM telegram_import_messages WHERE id=$1 AND city_id=$2 FOR UPDATE', [id, cityId])).rows[0]; + if (!row) throw httpError(404, 'Сообщение не найдено.'); + if (input.content_hash !== row.content_hash) throw httpError(409, 'Публикация изменилась. Обновите список и проверьте новый текст.'); + if (!['publish', 'dismiss'].includes(input.action)) throw httpError(400, 'Неизвестное действие.'); + if (input.action === 'publish') { + if (!validDate(input.service_date)) throw httpError(400, 'Укажите корректную дату отмены.'); + if (Date.parse(dayBounds(input.service_date).end) <= this.now().getTime()) throw httpError(410, 'День отмены уже закончился.'); + const route = (await this.getRoutes(db, cityId)).find(r => r.id === Number(input.route_id)); + if (!route) throw httpError(400, 'Выберите действующий маршрут выбранного города.'); + const title = String(input.title || '').trim(); + const message = String(input.message || '').trim(); + if (!title || title.length > 200 || !message || message.length > 5000) throw httpError(400, 'Укажите заголовок (до 200 символов) и текст (до 5000 символов).'); + await this.deactivate(db, row.id); + const bounds = dayBounds(input.service_date); + await this.saveAlert(db, row.id, { city_id: cityId, route_id: route.id, alert_type: 'cancellation', title, + message: `${message}\nИсточник: ${sourceUrl(row)}`, start_time: new Date(Math.max(Date.parse(bounds.start), this.now().getTime())).toISOString(), end_time: bounds.end, is_active: true }); + } else await this.deactivate(db, row.id); + const parsed = input.action === 'publish' ? { ...row.parsed, serviceDate: input.service_date, manual: true } : row.parsed; + await db.query(`UPDATE telegram_import_messages SET status=$2, reviewed_by=$3, parsed=$4::jsonb, updated_at=now() WHERE id=$1`, [row.id, input.action === 'publish' ? 'published' : 'dismissed', userId, JSON.stringify(parsed)]); + await db.query('COMMIT'); + return { id: row.id, status: input.action === 'publish' ? 'published' : 'dismissed' }; + } catch (error) { await db.query('ROLLBACK'); throw error; } + finally { db.release(); } + } + + async stop() { this.stopped = true; await this.source.close(); } +} + +let singleton; +function getTelegramImportService() { return singleton || (singleton = new TelegramImportService()); } +function startTelegramImporter(service = getTelegramImportService()) { + if (!service.config.enabled) return () => service.stop(); + const run = () => service.sync().catch(() => {}); // sync records a sanitized error for the admin status. + run(); + const timer = setInterval(run, service.config.intervalMs); + timer.unref?.(); + return () => { clearInterval(timer); return service.stop(); }; +} + +module.exports = { TelegramImportService, getTelegramImportService, startTelegramImporter, contentHash, publicError }; diff --git a/backend/src/services/telegramSource.js b/backend/src/services/telegramSource.js new file mode 100644 index 0000000..034c6c1 --- /dev/null +++ b/backend/src/services/telegramSource.js @@ -0,0 +1,57 @@ +const fs = require('node:fs/promises'); +const { assertExternalSessionFile } = require('../config/telegram'); + +class TelegramSource { + constructor(config) { this.config = config; this.client = null; this.channel = null; } + + async connect() { + if (this.client?.connected && this.channel) return; + if (this.client) await this.close(); + const file = await assertExternalSessionFile(this.config.sessionFile); + const session = (await fs.readFile(file, 'utf8')).trim(); + if (!session) throw new Error('SESSION_MISSING'); + const { TelegramClient } = require('teleproto'); + const { StringSession } = require('teleproto/sessions'); + const { Logger } = require('teleproto/extensions/Logger'); + const client = new TelegramClient(new StringSession(session), this.config.apiId, this.config.apiHash, { + connectionRetries: 2, requestRetries: 2, floodSleepThreshold: 0, baseLogger: new Logger('none') + }); + this.client = client; + await client.connect(); + if (!await client.checkAuthorization()) throw new Error('AUTH_REQUIRED'); + if ((await client.getMe()).bot) throw new Error('USER_ACCOUNT_REQUIRED'); + const channel = await client.getEntity(this.config.channel); + if (channel.className !== 'Channel' || !channel.broadcast || channel.username?.toLowerCase() !== this.config.channel.toLowerCase()) { + throw new Error('CHANNEL_INVALID'); + } + this.channel = channel; + } + + async fetchMessages(since) { + await this.connect(); + const messages = []; + // Pagination revisits recent messages so edits are detected after restarts. + // The bound prevents an accidental full-history import of a busy channel. + for await (const message of this.client.iterMessages(this.channel, { limit: 1000, waitTime: 1 })) { + if (message.date && message.date * 1000 < since.getTime()) return messages.reverse(); + if (message.className !== 'Message' || !message.date) continue; + messages.push({ + channelId: this.channel.id.toString(), channelUsername: this.channel.username, + messageId: message.id, text: message.message || '', hasMedia: Boolean(message.media && message.media.className !== 'MessageMediaWebPage'), + publishedAt: new Date(message.date * 1000).toISOString(), + editedAt: message.editDate ? new Date(message.editDate * 1000).toISOString() : null + }); + } + if (messages.length >= 1000) throw new Error('HISTORY_LIMIT'); + return messages.reverse(); + } + + async close() { + const client = this.client; + this.client = null; + this.channel = null; + if (client) await client.destroy(); + } +} + +module.exports = { TelegramSource }; diff --git a/backend/test/alertNotificationService.test.js b/backend/test/alertNotificationService.test.js index 30bca0c..d268173 100644 --- a/backend/test/alertNotificationService.test.js +++ b/backend/test/alertNotificationService.test.js @@ -8,6 +8,17 @@ const { } = require('../src/services/alertNotificationService'); const { createFirebaseMessaging, getFirebaseConfig } = require('../src/config/firebase'); +test('expires a cancellation at Minsk midnight and does not send a stale push', async () => { + const now = new Date('2026-10-02T20:59:00Z'); + const alert = { id: 1, alert_type: 'cancellation', end_time: '2026-10-02T21:00:00Z' }; + assert.equal(buildAlertCreatedMessage(alert, { now }).android.ttl, 60_000); + let sends = 0; + const service = new AlertNotificationService({ enabled: true, now: () => new Date('2026-10-02T21:00:00Z'), + getMessaging: async () => { sends++; }, logger: createLogger() }); + assert.deepEqual(await service.sendAlertCreated(alert), { sent: false, skipped: true, reason: 'expired' }); + assert.equal(sends, 0); +}); + function createLogger() { const errors = []; return { @@ -23,6 +34,7 @@ test('sends an exact data-only alert-created message with string-only data', asy const service = new AlertNotificationService({ enabled: true, topic: 'route_alerts', + now: () => new Date('2026-07-11T08:30:00Z'), androidChannelId: 'route_alerts', messaging: { async send(message) { @@ -64,7 +76,8 @@ test('sends an exact data-only alert-created message with string-only data', asy body: 'Задержка около 15 минут' }, android: { - priority: 'high' + priority: 'high', + ttl: 3_600_000 } }); assert.equal('notification' in messages[0], false); diff --git a/backend/test/telegramCancellationParser.test.js b/backend/test/telegramCancellationParser.test.js new file mode 100644 index 0000000..f8fc81b --- /dev/null +++ b/backend/test/telegramCancellationParser.test.js @@ -0,0 +1,89 @@ +const test = require('node:test'); +const assert = require('node:assert/strict'); +const { parseCancellation, matchCancellationRoutes, buildCancellationAlert, dayBounds } = require('../src/services/telegramCancellationParser'); + +const text = `Уважаемые пассажиры! +Не будут выполняться рейсы:  +ПО МАРШРУТУ № 27 С ОСТАНОВКИ АВ/ВОКЗАЛ НА 15.43,16.43. +                С ОСТАНОВКИ ЗОДЧЕГО НА 15.58,16.58. +С УВАЖЕНИЕМ +Филиал "Автобусный парк №2 г.Полоцка" ОАО "Витебскоблавтотранс"`; +const parse = (value = text, publishedAt = '2026-10-02T12:52:00Z') => parseCancellation({ text: value, publishedAt }); + +test('carrier post preserves four individual departures and canonicalizes АВ/ВОКЗАЛ', () => { + const result = parse(); + assert.equal(result.status, 'ready'); + assert.equal(result.serviceDate, '2026-10-02'); + assert.deepEqual(result.groups, [{ routeNumber: '27', departures: [ + { stop: 'Автовокзал', times: ['15:43', '16:43'] }, { stop: 'ЗОДЧЕГО', times: ['15:58', '16:58'] } + ] }]); +}); + +test('implicit date comes from publication in Minsk, never from import day or edit date', () => { + assert.equal(parse(text, '2026-10-02T22:30:00Z').serviceDate, '2026-10-03'); +}); + +test('tomorrow, numeric and Russian dates take precedence over publication day', () => { + for (const prefix of ['Завтра\n', '03.10.2026\n', 'На 03.10\n', '3 октября 2026 года\n']) { + assert.equal(parse(prefix + text).serviceDate, '2026-10-03', prefix); + } + assert.equal(parse('Завтра\n' + text, '2026-12-31T17:00:00Z').serviceDate, '2027-01-01'); + assert.equal(parse('На 01.01\n' + text, '2026-12-31T17:00:00Z').serviceDate, '2027-01-01'); +}); + +test('09.10 in a departure list remains a time', () => { + assert.deepEqual(parse(text.replace('15.43', '09.10')).groups[0].departures[0].times, ['09:10', '16:43']); +}); + +test('ambiguous dates, conditions, invalid times and unspecified departures require review', () => { + for (const value of [ + 'Сегодня и завтра\n' + text, 'На 31.02.2026\n' + text, + 'С 02.10 по 05.10\n' + text, 'По пятницам\n' + text, + text.replace('15.43', '25.43'), text.replace('15.43', '15.99'), + 'Не будут выполняться рейсы по маршруту №27.', text + '\nКроме отправления на 18:00.', + 'Послезавтра\n' + text + ]) assert.equal(parse(value).status, 'review', value); +}); + +test('news is ignored; a picture without text stays on review', () => { + assert.equal(parse('С праздником, уважаемые пассажиры!').status, 'ignored'); + assert.equal(parseCancellation({ text: '', hasMedia: true, publishedAt: '2026-10-02T10:00:00Z' }).status, 'review'); +}); + +test('review retains a recognized future date; attached media cannot be silently auto-published', () => { + const result = parse('Завтра\nНе будут выполняться рейсы по маршруту №27.'); + assert.equal(result.status, 'review'); + assert.equal(result.serviceDate, '2026-10-03'); + assert.equal(parseCancellation({ text, hasMedia: true, publishedAt: '2026-10-02T12:00:00Z' }).status, 'review'); +}); + +test('different route numbers in one post are separate groups', () => { + const result = parse(text + '\nПО МАРШРУТУ № 5А С ОСТАНОВКИ ЦЕНТР НА 18:00.'); + assert.equal(result.status, 'ready'); + assert.deepEqual(result.groups.map(g => g.routeNumber), ['27', '5а']); +}); + +test('route number with two directions is matched by origin stop, with no guess on missing origin', () => { + const routes = [ + { id: 1, route_number: '27', origin_stop: 'Автовокзал' }, + { id: 2, route_number: '27', origin_stop: 'Зодчего' }, + { id: 3, route_number: '5', origin_stop: 'Центр' } + ]; + const matched = matchCancellationRoutes(parse(), routes); + assert.deepEqual(matched.plans.map(p => p.routeId), [1, 2]); + assert.equal(matched.reasons.length, 0); + assert.equal(matchCancellationRoutes(parse(), routes.map(r => ({ ...r, origin_stop: '' }))).plans.length, 0); + assert.equal(matchCancellationRoutes(parse(), []).reasons.length, 1); +}); + +test('midnight expiry is an explicit UTC instant, including year boundary', () => { + assert.deepEqual(dayBounds('2026-10-02'), { start: '2026-10-01T21:00:00.000Z', end: '2026-10-02T21:00:00.000Z' }); + assert.equal(dayBounds('2026-12-31').end, '2026-12-31T21:00:00.000Z'); + const plan = matchCancellationRoutes(parse(), [{ id: 1, route_number: '27' }]).plans[0]; + const alert = buildCancellationAlert(plan, '2026-10-02', 'https://t.me/avp2Polotsk/1166', '2026-10-02T12:52:00Z', 1); + assert.equal(alert.end_time, '2026-10-02T21:00:00.000Z'); + assert.match(alert.message, /Автовокзал.*15:43, 16:43/); + assert.match(alert.title, /отдельных рейсов/); +}); + +module.exports = { carrierText: text }; diff --git a/backend/test/telegramImport.integration.test.js b/backend/test/telegramImport.integration.test.js new file mode 100644 index 0000000..0713d93 --- /dev/null +++ b/backend/test/telegramImport.integration.test.js @@ -0,0 +1,130 @@ +const test = require('node:test'); +const assert = require('node:assert/strict'); +const fs = require('node:fs/promises'); +const path = require('node:path'); +const { TelegramImportService } = require('../src/services/telegramImportService'); +const { CLAIM_DUE_ALERTS_SQL } = require('../src/services/alertSchedulerService'); + +// Run against an isolated in-memory PostgreSQL engine, never the working database. +// TELEGRAM_TEST_PGLITE_MODULE can point to a temporary npm installation. +const modulePath = process.env.TELEGRAM_TEST_PGLITE_MODULE; +const carrierText = 'Не будут выполняться рейсы:\nПО МАРШРУТУ № 27 С ОСТАНОВКИ АВ/ВОКЗАЛ НА 15.43,16.43.\nС ОСТАНОВКИ ЗОДЧЕГО НА 15.58,16.58.'; +const config = { enabled: true, autoPublish: true, apiId: 123, apiHash: 'a'.repeat(32), sessionFile: '/tmp/private/account.session', channel: 'avp2Polotsk', citySlug: 'polotsk', intervalMs: 60_000 }; +const message = (id = 1166, text = carrierText) => ({ channelId: '1234567', channelUsername: 'avp2Polotsk', messageId: id, publishedAt: '2026-10-02T12:52:00Z', editedAt: null, hasMedia: false, text }); + +async function setup() { + const { PGlite } = require(modulePath); + const db = new PGlite(); + await db.exec(`CREATE TABLE cities(id serial PRIMARY KEY,slug text,is_active boolean DEFAULT true); + CREATE TABLE users(id serial PRIMARY KEY); + CREATE TABLE routes(id serial PRIMARY KEY,route_number text,name text,city_id int REFERENCES cities(id),is_active boolean DEFAULT true); + CREATE TABLE stops(id serial PRIMARY KEY,name text); + CREATE TABLE route_stops(route_id int,stop_id int,sequence int); + CREATE TABLE alerts(id serial PRIMARY KEY,route_id int REFERENCES routes(id),alert_type text,title text,message text, + city_id int,start_time timestamp,end_time timestamp,is_active boolean DEFAULT true,created_at timestamp DEFAULT now(), + notification_sent_at timestamptz,notification_claimed_at timestamptz,notification_attempts int DEFAULT 0); + INSERT INTO cities(slug) VALUES ('polotsk'),('novopolotsk'); INSERT INTO users DEFAULT VALUES; + INSERT INTO routes(route_number,name,city_id) VALUES ('27','Автовокзал → Зодчего',1),('27','Зодчего → Автовокзал',1),('27','Другой город',2); + INSERT INTO stops(name) VALUES ('Автовокзал'),('Зодчего'); + INSERT INTO route_stops VALUES (1,1,1),(2,2,1); + INSERT INTO alerts(route_id,city_id,start_time,end_time,title,message,alert_type) VALUES (1,1,'2026-10-02 12:00','2026-10-02 21:00','Old','Old','info'); + SET TIME ZONE 'Europe/Minsk';`); + const migration = await fs.readFile(path.join(__dirname, '../migrations/013_telegram_alert_import.sql'), 'utf8'); + await db.exec(migration); + await db.exec(migration); // idempotent upgrades + // A session is leased for the whole transaction, just like a pg Pool client. + let tail = Promise.resolve(); + async function lease() { + const previous = tail; + let release; + tail = new Promise(resolve => { release = resolve; }); + await previous; + return { query: (sql, args) => db.query(sql, args), release }; + } + const pool = { connect: lease, query: async (sql, args) => { + const client = await lease(); + try { return await client.query(sql, args); } finally { client.release(); } + } }; + let now = new Date('2026-10-02T13:00:00Z'); + const service = new TelegramImportService({ pool, config, source: { close: async () => {} }, logger: { warn() {}, error() {} }, now: () => now }); + return { db, pool, service, clock: value => { now = new Date(value); } }; +} + +test('Telegram SQL lifecycle: atomic import, duplicates, edits, expiry, deletion and city isolation', { skip: !modulePath }, async () => { + const { db, pool, service, clock } = await setup(); + try { + assert.equal((await pool.query('SELECT end_time FROM alerts WHERE id=1')).rows[0].end_time.toISOString(), '2026-10-02T21:00:00.000Z'); + const results = await Promise.all([service.processMessage(message(), 1), service.processMessage(message(), 1)]); + assert.deepEqual(results.map(r => r.status), ['published', 'unchanged']); + let alerts = (await pool.query('SELECT * FROM alerts WHERE notification_pending=true ORDER BY route_id')).rows; + assert.equal(alerts.length, 2); + assert.deepEqual(alerts.map(a => a.route_id), [1, 2]); + assert.equal(alerts[0].end_time.toISOString(), '2026-10-02T21:00:00.000Z'); + assert.match(alerts[0].message, /Автовокзал.*15:43, 16:43/); + assert.ok(alerts.every(a => a.notification_sent_at === null)); + + // Delivered pushes aren't repeated on text edits, while the stored alert changes. + await pool.query('UPDATE alerts SET notification_sent_at=now() WHERE notification_pending=true'); + const edited = { ...message(), text: carrierText.replace('15.43', '15.45'), editedAt: '2026-10-02T13:01:00Z' }; + assert.equal((await service.processMessage(edited, 1)).status, 'published'); + alerts = (await pool.query('SELECT * FROM alerts WHERE notification_pending=true ORDER BY route_id')).rows; + assert.equal(alerts.length, 2); + assert.ok(alerts.every(a => a.notification_sent_at !== null)); + assert.match(alerts[0].message, /15:45/); + assert.equal((await service.processMessage(message(), 1)).status, 'unchanged', 'stale Telegram snapshot cannot undo an edit'); + + // Missing text after an edit withdraws the old alerts and requires a review. + assert.equal((await service.processMessage({ ...edited, text: 'Движение восстановлено.', editedAt: '2026-10-02T13:02:00Z' }, 1)).status, 'review'); + assert.equal((await pool.query('SELECT count(*)::int AS n FROM alerts WHERE notification_pending=true')).rows[0].n, 0); + const row = (await service.list(1))[0]; + const input = { action: 'publish', content_hash: row.content_hash, service_date: '2026-10-02', route_id: 1, title: 'Отмена рейса', message: 'Автовокзал, 15:45' }; + await assert.rejects(service.review(row.id, 2, input, 1), /не найдено/); + await assert.rejects(service.review(row.id, 1, { ...input, route_id: 3 }, 1), /выбранного города/); + await assert.rejects(service.review(row.id, 1, { ...input, content_hash: 'stale' }, 1), /изменилась/); + await service.review(row.id, 1, input, 1); + await service.review(row.id, 1, { action: 'dismiss', content_hash: row.content_hash }, 1); + assert.equal((await service.processMessage({ ...edited, editedAt: '2026-10-02T14:00:00Z' }, 1)).status, 'dismissed'); + assert.equal((await pool.query('SELECT count(*)::int AS n FROM alerts WHERE notification_pending=true')).rows[0].n, 0); + + // Administrator deletion is respected on a subsequent source edit. + await service.processMessage(message(1167), 1); + const deletedId = (await pool.query('SELECT alert_id FROM telegram_import_alerts WHERE message_id=(SELECT id FROM telegram_import_messages WHERE message_id=1167) AND route_id=1')).rows[0].alert_id; + await pool.query('DELETE FROM alerts WHERE id=$1', [deletedId]); + await service.processMessage({ ...message(1167), text: carrierText.replace('15.43', '15.46'), editedAt: '2026-10-02T14:01:00Z' }, 1); + assert.equal((await pool.query('SELECT alert_id FROM telegram_import_alerts WHERE message_id=(SELECT id FROM telegram_import_messages WHERE message_id=1167) AND route_id=1')).rows[0].alert_id, null); + + // At midnight, late imports and manual publication cannot reactivate yesterday's cancellation. + clock('2026-10-02T21:00:00Z'); + assert.equal((await service.processMessage(message(1168), 1)).status, 'expired'); + const expired = (await service.list(1)).find(m => m.message_id === 1168); + await assert.rejects(service.review(expired.id, 1, { ...input, content_hash: expired.content_hash }, 1), /уже закончился/); + assert.equal((await service.list(2)).length, 0); + + // Pending imported alerts are claimed even when start_time is before created_at. + await pool.query(`UPDATE alerts SET is_active=true, notification_pending=true, start_time=now()-interval '1 minute', + end_time=now()+interval '1 hour', notification_sent_at=NULL WHERE id<>1`); + const claimed = await pool.query(CLAIM_DUE_ALERTS_SQL, [50]); + assert.ok(claimed.rows.length > 0); + assert.ok(claimed.rows.every(a => a.notification_attempts === 1)); + assert.equal((await pool.query(CLAIM_DUE_ALERTS_SQL, [50])).rows.length, 0); + } finally { await db.close(); } +}); + +test('Telegram SQL import rolls back all alerts if a later route insert fails', { skip: !modulePath }, async () => { + const { db, pool, service } = await setup(); + try { + const connect = pool.connect; + pool.connect = async () => { + const client = await connect(); + const query = client.query; + client.query = (sql, args) => { + if (/INSERT INTO alerts/.test(sql) && args[0] === 2) throw new Error('Injected failure'); + return query(sql, args); + }; + return client; + }; + await assert.rejects(service.processMessage(message(), 1), /Injected failure/); + assert.equal((await pool.query('SELECT count(*)::int AS n FROM telegram_import_messages')).rows[0].n, 0); + assert.equal((await pool.query('SELECT count(*)::int AS n FROM alerts')).rows[0].n, 1); + } finally { await db.close(); } +}); diff --git a/backend/test/telegramImportApi.test.js b/backend/test/telegramImportApi.test.js new file mode 100644 index 0000000..5623a30 --- /dev/null +++ b/backend/test/telegramImportApi.test.js @@ -0,0 +1,40 @@ +const test = require('node:test'); +const assert = require('node:assert/strict'); +process.env.JWT_SECRET = 'telegram-api-tests-only-secret'; +process.env.NODE_ENV = 'production'; +// This router test never connects to a real database. +require.cache[require.resolve('../src/config/database')] = { exports: { query: async () => ({ rows: [] }) } }; +const express = require('express'); +const jwt = require('jsonwebtoken'); +const { createTelegramImportRouter } = require('../src/routes/telegramImport'); + +test('Telegram admin API rejects anonymous/non-admin access and preserves selected-city scope', async () => { + const calls = []; + const service = { + config: { citySlug: 'polotsk' }, + getStatus: () => ({ enabled: false }), + list: async cityId => { calls.push(['list', cityId]); return []; }, + sync: async () => { calls.push(['sync']); return {}; }, + review: async (...args) => { calls.push(['review', ...args]); return { status: 'dismissed' }; } + }; + const app = express(); + app.use(express.json()); + app.use((req, res, next) => { req.city = { id: Number(req.get('X-City-Id') || 1), slug: req.get('X-City-Id') === '2' ? 'novopolotsk' : 'polotsk' }; next(); }); + app.use('/telegram-import', createTelegramImportRouter(service)); + const server = await new Promise(resolve => { const s = app.listen(0, '127.0.0.1', () => resolve(s)); }); + const url = `http://127.0.0.1:${server.address().port}/telegram-import`; + const token = role => jwt.sign({ id: 7, role }, process.env.JWT_SECRET); + try { + assert.equal((await fetch(url)).status, 401); + assert.equal((await fetch(url, { headers: { Authorization: `Bearer ${token('user')}` } })).status, 403); + assert.equal(calls.length, 0); + const headers = { Authorization: `Bearer ${token('admin')}`, 'X-City-Id': '2', 'Content-Type': 'application/json' }; + assert.equal((await fetch(url, { headers })).status, 200); + assert.deepEqual(calls[0], ['list', 2]); + assert.equal((await fetch(`${url}/sync`, { method: 'POST', headers })).status, 409); + assert.equal(calls.length, 1); + assert.equal((await fetch(`${url}/4/review`, { method: 'POST', headers, body: JSON.stringify({ action: 'dismiss', content_hash: 'hash' }) })).status, 200); + assert.deepEqual(calls[1], ['review', 4, 2, { action: 'dismiss', content_hash: 'hash' }, 7]); + assert.equal((await fetch(`${url}/not-an-id/review`, { method: 'POST', headers, body: '{}' })).status, 400); + } finally { await new Promise(resolve => server.close(resolve)); } +}); diff --git a/backend/test/telegramImportService.test.js b/backend/test/telegramImportService.test.js new file mode 100644 index 0000000..54d4bcf --- /dev/null +++ b/backend/test/telegramImportService.test.js @@ -0,0 +1,82 @@ +const test = require('node:test'); +const assert = require('node:assert/strict'); +const { TelegramImportService, publicError } = require('../src/services/telegramImportService'); +const { TelegramSource } = require('../src/services/telegramSource'); +const { getTelegramConfig } = require('../src/config/telegram'); + +const config = { enabled: true, autoPublish: true, apiId: 123, apiHash: 'a'.repeat(32), sessionFile: '/tmp/private/account.session', channel: 'avp2Polotsk', citySlug: 'polotsk', intervalMs: 60_000 }; +function setup(source, overrides = {}) { + return new TelegramImportService({ config: { ...config, ...overrides }, source, + pool: { query: async () => ({ rows: [{ id: 1 }] }) }, logger: { warn() {} }, now: () => new Date('2026-10-02T12:00:00Z') }); +} + +test('disabled or incomplete setup never connects to Telegram and never reveals credentials in status', async () => { + let calls = 0; + const source = { fetchMessages: async () => { calls++; } }; + const disabled = setup(source, { enabled: false }); + await assert.rejects(disabled.sync(), /выключен/); + const missing = setup(source, { apiHash: '' }); + assert.equal(missing.getStatus().connection, 'configuration_required'); + await assert.rejects(missing.sync(), /Укажите/); + assert.equal(calls, 0); + assert.equal(JSON.stringify(disabled.getStatus()).includes(config.apiHash), false); + assert.equal(JSON.stringify(disabled.getStatus()).includes(config.sessionFile), false); + assert.equal(getTelegramConfig({}).enabled, false); +}); + +test('worker polls a bounded three-day window, processes sequentially, and reports success', async () => { + const processed = []; + const source = { async fetchMessages(since) { + assert.equal(since.toISOString(), '2026-09-29T21:00:00.000Z'); + return [{ id: 1 }, { id: 2 }]; + } }; + const service = setup(source); + service.processMessage = async (message, cityId) => { processed.push([message.id, cityId]); return { status: 'published' }; }; + const counts = await service.sync(); + assert.equal(counts.published, 2); + assert.deepEqual(processed, [[1, 1], [2, 1]]); + assert.equal(service.getStatus().connection, 'ready'); +}); + +test('overlapping worker requests do not start two channel readers', async () => { + let finish; + const service = setup({ fetchMessages: () => new Promise(resolve => { finish = resolve; }) }); + const running = service.sync(); + await new Promise(resolve => setImmediate(resolve)); + await assert.rejects(service.sync(), /уже выполняется/); + finish([]); + await running; +}); + +test('FloodWait backs off and only a sanitized error code is exposed or logged', async () => { + const logs = []; + let attempts = 0; + const service = setup({ async fetchMessages() { attempts++; throw Object.assign(new Error('private-session-secret'), { seconds: 120 }); }, async close() {} }); + service.logger = { warn: (...args) => logs.push(args) }; + await assert.rejects(service.sync(), /FLOOD_WAIT/); + assert.equal(service.getStatus().retryAt, '2026-10-02T12:02:00.000Z'); + await assert.rejects(service.sync(), /частоту запросов/); + assert.equal(attempts, 1); + assert.equal(JSON.stringify([logs, service.getStatus()]).includes('private-session-secret'), false); + assert.equal(publicError(new Error('some sensitive content')), 'IMPORT_FAILED'); +}); + +test('source stops reading old history and keeps original publication and edit timestamps', async () => { + const source = new TelegramSource(config); + source.channel = { id: { toString: () => '1234567' }, username: config.channel }; + source.connect = async () => {}; + let iterations = 0; + source.client = { async *iterMessages() { + for (const entry of [ + { className: 'Message', id: 7, date: Date.parse('2026-10-02T10:00Z') / 1000, editDate: Date.parse('2026-10-02T11:00Z') / 1000, message: 'Отмена', media: null }, + { className: 'MessageService', id: 6, date: Date.parse('2026-10-02T09:00Z') / 1000 }, + { className: 'Message', id: 5, date: Date.parse('2026-09-01T10:00Z') / 1000 }, + { className: 'Message', id: 4, date: Date.parse('2026-08-01T10:00Z') / 1000 } + ]) { iterations++; yield entry; } + } }; + const messages = await source.fetchMessages(new Date('2026-10-01T21:00Z')); + assert.equal(iterations, 3); + assert.equal(messages.length, 1); + assert.equal(messages[0].publishedAt, '2026-10-02T10:00:00.000Z'); + assert.equal(messages[0].editedAt, '2026-10-02T11:00:00.000Z'); +}); diff --git a/docker-compose.yml b/docker-compose.yml index e4efeed..6049fc3 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -63,6 +63,14 @@ services: FCM_ANDROID_CHANNEL_ID: "${FCM_ANDROID_CHANNEL_ID:-route_alerts}" FIREBASE_PROJECT_ID: "${FIREBASE_PROJECT_ID:-}" GOOGLE_APPLICATION_CREDENTIALS: "${GOOGLE_APPLICATION_CREDENTIALS:-/run/secrets/firebase-service-account.json}" + TELEGRAM_IMPORT_ENABLED: "${TELEGRAM_IMPORT_ENABLED:-false}" + TELEGRAM_AUTO_PUBLISH: "${TELEGRAM_AUTO_PUBLISH:-true}" + TELEGRAM_API_ID: "${TELEGRAM_API_ID:-}" + TELEGRAM_API_HASH: "${TELEGRAM_API_HASH:-}" + TELEGRAM_CHANNEL: "${TELEGRAM_CHANNEL:-avp2Polotsk}" + TELEGRAM_CITY_SLUG: "${TELEGRAM_CITY_SLUG:-polotsk}" + TELEGRAM_POLL_INTERVAL_MS: "${TELEGRAM_POLL_INTERVAL_MS:-60000}" + TELEGRAM_SESSION_FILE: "${TELEGRAM_SESSION_FILE:-/run/telegram-secrets/account.session}" ports: - "3000:3000" depends_on: @@ -78,6 +86,10 @@ services: - avatars_data:/app/public/uploads/avatars - city_images_data:/app/public/uploads/cities - ${FIREBASE_SECRETS_DIR:-./secrets}:/run/secrets:ro + - type: bind + source: ${TELEGRAM_SECRETS_DIR:-./secrets} + target: /run/telegram-secrets + read_only: true frontend: build: diff --git a/docs/telegram-cancellation-import.md b/docs/telegram-cancellation-import.md new file mode 100644 index 0000000..7a2f549 --- /dev/null +++ b/docs/telegram-cancellation-import.md @@ -0,0 +1,97 @@ +# Отмены рейсов из @avp2Polotsk + +Первый этап читает публичный канал через **отдельный существующий Telegram-аккаунт** и MTProto-клиент `teleproto`. Права управления каналом и Bot API не нужны. Получение выполняется раз в минуту; просматриваются публикации за текущий и два предыдущих дня, включая правки. Вложения не скачиваются. Расписания рейсов не изменяются: импорт создаёт предупреждения существующего типа `cancellation`. + +## Распознавание и срок действия + +Формат «Не будут выполняться рейсы: ПО МАРШРУТУ № 27 С ОСТАНОВКИ АВ/ВОКЗАЛ НА 15.43,16.43. С ОСТАНОВКИ ЗОДЧЕГО НА 15.58,16.58.» разбирается в четыре конкретных отправления. `АВ/ВОКЗАЛ` означает **Автовокзал**, точки в часах заменяются двоеточием. Отмена всего маршрута из такого объявления не выводится. + +Если у номера несколько направлений, они сопоставляются по начальной остановке в базе. Неоднозначный номер, неизвестное направление, разные даты, условия, периоды и изображения без текста требуют проверки администратором. Обычные новости не импортируются. Пока нет OCR, распознавания произвольных формулировок и автоматического редактирования расписаний. + +Явная дата, «сегодня» или «завтра» определяют день отмены. Если дата отсутствует, используется **день оригинальной публикации по Europe/Minsk**, а не день импорта или правки. Дата без года выбирается ближайшей к публикации. Конец действия — **00:00 следующего дня по Минску**, граница исключительная. Например, для 02.10.2026 это `2026-10-02T21:00:00Z`. Будущие отмены активируются в начале указанного дня и получают push тогда же. Просроченные объявления сохраняются как завершённые без push. + +Новые однозначные отмены публикуются автоматически. `TELEGRAM_AUTO_PUBLISH=false` переводит все распознанные отмены на проверку. В админке на странице **«Уведомления» → «Отмены из Telegram»** доступны состояние подключения, ручная проверка канала, исходные сообщения, ссылки на оригинал и история. При ручной публикации оператор проверяет маршрут, текст и день отмены; срок до полуночи назначается сервером. + +## Установка + +Для существующей базы применить миграцию **013** один раз, до запуска новой версии backend. Она также переводит UTC-значения времени существующих алертов из `timestamp` в `timestamptz`. Это соответствует принятому формату хранения UTC в админке. При наличии вручную внесённых локальных времен сначала проверить их и сделать резервную копию базы. + +```bash +docker compose exec -T postgres psql -U transit_admin -d polotsk_transit \ + -f /docker-entrypoint-initdb.d/013_telegram_alert_import.sql +``` + +На чистой базе Docker применяет миграцию автоматически. **Не запускайте `npm run migrate` на существующей базе**: старые init-миграции проекта не поддерживают повторный запуск. + +### Авторизация аккаунта + +1. Создать API-приложение на [my.telegram.org](https://my.telegram.org) и получить `api_id` и `api_hash`. +2. На компьютере/сервере оператора создать закрытый каталог **вне checkout**. Пример для локального запуска: + + ```bash + mkdir -p "$HOME/.config/polotsk-transit-telegram" + chmod 700 "$HOME/.config/polotsk-transit-telegram" + ``` + +3. В `backend/.env` указать: + + ```dotenv + TELEGRAM_IMPORT_ENABLED=false + TELEGRAM_API_ID=your_api_id + TELEGRAM_API_HASH=your_api_hash + TELEGRAM_SESSION_FILE=/absolute/private/directory/account.session + TELEGRAM_CHANNEL=avp2Polotsk + TELEGRAM_CITY_SLUG=polotsk + TELEGRAM_AUTO_PUBLISH=true + TELEGRAM_POLL_INTERVAL_MS=60000 + ``` + +4. Из каталога `backend` выполнить `npm ci`, затем `npm run telegram:login`. Номер, код и пароль вводятся оператором локально. Код и пароль скрыты; session string не печатается. Утилита сохраняет файл с правами `600` и работает только с существующим пользовательским аккаунтом. Если Telegram требует регистрацию/дополнительный неподдерживаемый шаг, завершить его в официальном клиенте и повторить авторизацию. +5. Включить `TELEGRAM_IMPORT_ENABLED=true` и перезапустить backend. Для push также должны быть настроены Firebase и `FCM_ENABLED=true`. + +Файл сессии даёт доступ к аккаунту. Его нельзя коммитить, передавать frontend или присылать в чат. Backend не выполняет интерактивную авторизацию; отсутствующая/отозванная сессия отображается как ошибка подключения. В статус и логи не попадают API hash, сессия, телефон, коды или сырые ошибки клиента. + +### Docker / Proxmox + +Авторизацию выполнить на доверенном хосте. Поместить созданный `account.session` в закрытый каталог на Docker-хосте, например `/opt/polotsk-transit/telegram-secrets`, с правами `600`. В корневом `.env` настроить: + +```dotenv +TELEGRAM_SECRETS_DIR=/opt/polotsk-transit/telegram-secrets +TELEGRAM_SESSION_FILE=/run/telegram-secrets/account.session +TELEGRAM_API_ID=your_api_id +TELEGRAM_API_HASH=your_api_hash +TELEGRAM_IMPORT_ENABLED=true +TELEGRAM_AUTO_PUBLISH=true +``` + +Compose монтирует каталог только для чтения, отдельно от Firebase. После миграции выполнить `docker compose up -d --build backend frontend`. + +## Повторы, правки и доставка + +Идентичность публикации: город + числовой ID канала + ID сообщения. Снимок текста, результат распознавания и связи с алертами сохраняются одной транзакцией; повторный импорт и одновременные читатели не создают второй алерт. Более старая правка не заменяет свежую. Администратор публикует только проверенную версию по `content_hash`; изменившийся текст даёт `409` и требует нового просмотра. + +Правка публикации обновляет связанные алерты. Если новая версия неоднозначна или больше не распознаётся как отмена, прежние алерты выключаются и сообщение возвращается на проверку. Отклонение сохраняется при последующих правках. Удалённый администратором алерт не воссоздаётся при правке исходного сообщения. Автоудаление сообщения в Telegram не отменяет предупреждение раньше полуночи; удаление не трактуется как восстановление движения. + +Импорт помечает алерты как `notification_pending` в той же транзакции. Существующий scheduler отправляет их после начала действия, использует блокировки и до пяти попыток; Firebase отключён — ожидание сохраняется. Уже доставленный push при правке того же алерта не отправляется повторно. Остаётся обычное окно между приёмом FCM и записью результата в БД: при аварии в этот момент возможен повтор доставки. `alertId` позволяет клиенту показывать один экземпляр уведомления. + +FCM TTL ограничен остатком срока действия алерта; истёкший алерт не отправляется. Backend передаёт абсолютный `endTime`. **Удаление уже показанного уведомления из системной шторки требует поддержки Android-клиента**: проверить `endTime` при получении, не показывать просроченный push, поставить `setTimeoutAfter(endTime-now)` и исключать истёкшие алерты из локального списка. Изменения Android-клиента не входят в этот этап backend/админки. Правки и отключение алерта не отзывают уже показанный push. + +При FloodWait worker выдерживает паузу, не запускает параллельные циклы и отображает время следующей попытки. За один цикл лимит составляет 1000 публикаций; превышение отображается как ошибка, а не как успешное чтение всей истории. Сообщения, исчезнувшие из Telegram во время длительного простоя сервера, восстановить нельзя. + +## API и проверка + +Все endpoint требуют JWT администратора и учитывают `X-City-Id`: + +- `GET /api/v1/telegram-import` — безопасный статус и до 100 сообщений, сначала требующие проверки. +- `POST /api/v1/telegram-import/sync` — чтение настроенного канала; запрещено в другом городе. +- `POST /api/v1/telegram-import/:id/review` — `content_hash`, `action: publish|dismiss`; для публикации также `service_date`, `route_id`, `title`, `message`. + +Обычные unit-тесты: `cd backend && npm test`. SQL-тесты используют временный PostgreSQL runtime PGlite и не обращаются к рабочей базе: + +```bash +npm install --prefix /tmp/polotsk-telegram-test-runtime @electric-sql/pglite --ignore-scripts +TELEGRAM_TEST_PGLITE_MODULE=/tmp/polotsk-telegram-test-runtime/node_modules/@electric-sql/pglite \ + node --test test/telegramImport.integration.test.js +``` + +Реальное чтение Telegram и доставка на устройство проверяются после локальной авторизации аккаунта, применения миграции и включения Firebase. diff --git a/frontend/src/components/TelegramImportPanel.jsx b/frontend/src/components/TelegramImportPanel.jsx new file mode 100644 index 0000000..521c057 --- /dev/null +++ b/frontend/src/components/TelegramImportPanel.jsx @@ -0,0 +1,135 @@ +import { useEffect, useState } from 'react'; +import { ExternalLink, RefreshCw, Send, X } from 'lucide-react'; +import { telegramImportApi } from '../services/api'; +import { useCity } from '../contexts/CityContext'; + +const connectionLabels = { + disabled: 'Импорт выключен', configuration_required: 'Нужна настройка аккаунта', + waiting: 'Ожидание подключения', connecting: 'Подключение к Telegram', ready: 'Канал подключён', error: 'Ошибка подключения' +}; +const errorLabels = { + SESSION_MISSING: 'Не найден файл авторизации. Выполните авторизацию аккаунта на сервере.', + AUTH_REQUIRED: 'Требуется повторная авторизация Telegram-аккаунта.', + AUTH_KEY_UNREGISTERED: 'Авторизация отозвана. Подключите аккаунт повторно.', + SESSION_REVOKED: 'Авторизация отозвана. Подключите аккаунт повторно.', + SESSION_FILE_PERMISSIONS: 'Файл авторизации должен быть доступен только владельцу.', + SESSION_FILE_IN_CHECKOUT: 'Перенесите файл авторизации за пределы проекта.', + USER_ACCOUNT_REQUIRED: 'Нужен отдельный пользовательский аккаунт Telegram.', + CHANNEL_PRIVATE: 'Канал недоступен этому аккаунту.', CHANNEL_INVALID: 'Проверьте имя канала в настройках сервера.', + USERNAME_NOT_OCCUPIED: 'Канал с таким именем не найден.', + FLOOD_WAIT: 'Telegram ограничил запросы. Импорт продолжится после паузы.', + HISTORY_LIMIT: 'В канале слишком много сообщений за последние три дня. Нужна проверка импорта.', + IMPORT_FAILED: 'Проверьте соединение с Telegram и миграцию базы данных.' +}; +const formatTime = value => value ? new Date(value).toLocaleString('ru-RU', { timeZone: 'Europe/Minsk' }) : 'ещё не выполнялось'; + +export default function TelegramImportPanel({ routes, onChanged }) { + const { activeCity } = useCity(); + const [data, setData] = useState(null); + const [error, setError] = useState(''); + const [busy, setBusy] = useState(false); + const [editing, setEditing] = useState(null); + const [showHistory, setShowHistory] = useState(false); + + const load = async () => { + try { setData((await telegramImportApi.getAll()).data); setError(''); } + catch (err) { setError(err.response?.data?.error || 'Не удалось загрузить импорт Telegram.'); } + }; + useEffect(() => { + load(); + const timer = setInterval(load, 30_000); + return () => clearInterval(timer); + }, []); + + const sync = async () => { + setBusy(true); + try { await telegramImportApi.sync(); await load(); onChanged(); } + catch (err) { setError(err.response?.data?.error || 'Не удалось прочитать канал.'); } + finally { setBusy(false); } + }; + const review = async (row, values) => { + setBusy(true); + try { + await telegramImportApi.review(row.id, { ...values, content_hash: row.content_hash }); + setEditing(null); await load(); onChanged(); + } catch (err) { setError(err.response?.data?.error || 'Не удалось сохранить решение.'); } + finally { setBusy(false); } + }; + + const status = data?.status; + const configuredCity = status?.citySlug === activeCity?.slug; + const pending = data?.messages.filter(m => m.status === 'review') || []; + const rows = showHistory ? data?.messages || [] : pending; + return ( +
+
+
+

Отмены из Telegram

+

+ {status ? configuredCity ? connectionLabels[status.connection] : 'Для выбранного города источник не настроен' : 'Загрузка источника…'} + {status && <> · @{status.channel}} +

+ {status?.connection === 'disabled' &&

Подключите отдельный аккаунт на сервере. Отмены будут поступать автоматически.

} + {status?.enabled &&

Последняя проверка: {formatTime(status.lastSyncAt)} · {status.autoPublish ? 'Однозначные отмены публикуются автоматически' : 'Публикация после проверки'}

} + {status?.lastError &&

{errorLabels[status.lastError] || 'Не удалось подключиться к Telegram.'}

} + {status?.retryAt &&

Следующая попытка не раньше: {formatTime(status.retryAt)}

} +

Срок действия — до полуночи дня отмены по Минску. АВ/ВОКЗАЛ распознаётся как Автовокзал.

+
+ +
+ {error &&

{error}

} + {data && ( +
+
+

На проверке: {pending.length}

+ +
+ {rows.map(row => ( +
+
+ {formatTime(row.published_at)} · {{ review: 'На проверке', published: 'Опубликовано', dismissed: 'Отклонено', expired: 'День отмены завершён' }[row.status]} + Оригинал +
+

{row.raw_text || 'Вложение без текста — откройте оригинал.'}

+ {row.status === 'review' && <> + {row.reasons.map(reason =>

{reason}

)} +
+ + +
+ } +
+ ))} + {!rows.length &&

{showHistory ? 'Сообщений об отменах пока нет.' : 'Нет сообщений, требующих проверки.'}

} +
+ )} + {editing && setEditing(null)} onSubmit={values => review(editing, values)} />} +
+ ); +} + +function TelegramReviewForm({ row, routes, busy, onSubmit, onCancel }) { + const publicationDate = new Intl.DateTimeFormat('en-CA', { + timeZone: 'Europe/Minsk', year: 'numeric', month: '2-digit', day: '2-digit' + }).format(new Date(row.published_at)); + const [form, setForm] = useState({ + action: 'publish', service_date: row.parsed?.serviceDate || publicationDate, route_id: row.parsed?.plans?.length === 1 ? row.parsed.plans[0].routeId : '', + title: 'Отмена отдельных рейсов', message: row.raw_text.slice(0, 5000) + }); + const change = e => setForm({ ...form, [e.target.name]: e.target.value }); + return ( +
{ e.preventDefault(); onSubmit(form); }}> +

Проверка отмены

+
+ + +
+ +