From 732a06f65dda5a87513908a491c4e38c6922401d Mon Sep 17 00:00:00 2001 From: gosangam Date: Sun, 18 Oct 2026 23:48:22 +0530 Subject: [PATCH] first commit --- .env.example | 119 ++++ .gitignore | 27 + .python-version | 1 + ENV_SETUP.md | 388 +++++++++++ Procfile | 1 + README.md | 510 ++++++++++++++ app.json | 189 +++++ bot_handlers.py | 542 +++++++++++++++ caption_parser.py | 108 +++ config.py | 438 ++++++++++++ database.py | 752 ++++++++++++++++++++ downloader.py | 246 +++++++ drive_client.py | 620 +++++++++++++++++ errors.py | 69 ++ link_parser.py | 112 +++ logging_config.py | 48 ++ main.py | 328 +++++++++ models.py | 258 +++++++ naming.py | 182 +++++ progress.py | 259 +++++++ pytest.ini | 4 + queue_manager.py | 1003 +++++++++++++++++++++++++++ requirements-dev.txt | 5 + requirements.txt | 10 + start.sh | 16 + telegram_client.py | 345 +++++++++ tests/conftest.py | 373 ++++++++++ tests/test_bot_handlers.py | 69 ++ tests/test_caption_parser.py | 102 +++ tests/test_config.py | 140 ++++ tests/test_database.py | 155 +++++ tests/test_deployment.py | 77 ++ tests/test_drive_client.py | 305 ++++++++ tests/test_link_parser.py | 66 ++ tests/test_naming.py | 84 +++ tests/test_progress_and_telegram.py | 107 +++ tests/test_queue_manager.py | 438 ++++++++++++ tests/test_upload_tasks.py | 447 ++++++++++++ 38 files changed, 8943 insertions(+) create mode 100644 .env.example create mode 100644 .gitignore create mode 100644 .python-version create mode 100644 ENV_SETUP.md create mode 100644 Procfile create mode 100644 README.md create mode 100644 app.json create mode 100644 bot_handlers.py create mode 100644 caption_parser.py create mode 100644 config.py create mode 100644 database.py create mode 100644 downloader.py create mode 100644 drive_client.py create mode 100644 errors.py create mode 100644 link_parser.py create mode 100644 logging_config.py create mode 100644 main.py create mode 100644 models.py create mode 100644 naming.py create mode 100644 progress.py create mode 100644 pytest.ini create mode 100644 queue_manager.py create mode 100644 requirements-dev.txt create mode 100644 requirements.txt create mode 100755 start.sh create mode 100644 telegram_client.py create mode 100644 tests/conftest.py create mode 100644 tests/test_bot_handlers.py create mode 100644 tests/test_caption_parser.py create mode 100644 tests/test_config.py create mode 100644 tests/test_database.py create mode 100644 tests/test_deployment.py create mode 100644 tests/test_drive_client.py create mode 100644 tests/test_link_parser.py create mode 100644 tests/test_naming.py create mode 100644 tests/test_progress_and_telegram.py create mode 100644 tests/test_queue_manager.py create mode 100644 tests/test_upload_tasks.py diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..eabf504 --- /dev/null +++ b/.env.example @@ -0,0 +1,119 @@ +# Copy to .env for local runs. On Heroku set these as Config Vars instead. +# Placeholder values only - never commit real secrets. + +# ---------------------------------------------------------------- Telegram +# user = your own account (full history, needs login); bot = bot admin of the channel +TELEGRAM_MODE=user +# From https://my.telegram.org -> API development tools (needed in both modes) +TELEGRAM_API_ID=1234567 +TELEGRAM_API_HASH=your_api_hash_here +# From @BotFather: admin command interface (required in bot mode, recommended in user mode) +TELEGRAM_BOT_TOKEN=123456789:replace-with-your-bot-token +# User mode on Heroku: output of `python main.py login --session-string`. +# Locally you can leave it empty and use `python main.py login` (session file). +TELEGRAM_SESSION_STRING= +# Numeric Telegram user IDs allowed to use the commands (comma-separated) +ADMIN_USER_IDS=111111111 +# Chats that receive progress/notifications (default: ADMIN_USER_IDS) +NOTIFY_CHAT_IDS= +# @username, https://t.me/... link, or -100... numeric ID +SOURCE_CHANNEL_ID=@your_channel +# existing | new | both +PROCESS_MODE=both +START_MESSAGE_ID=1 +# Optional ISO date: only messages on/after this date (e.g. 2024-01-31) +START_DATE= +INCLUDE_DOCUMENT_VIDEOS=true +# Optional comma-separated filters (case-insensitive) +FILTER_TOPICS= +FILTER_BATCHES= +# Bot mode history probing (bots cannot list history) +BOT_SCAN_MAX_EMPTY=200 +BOT_SCAN_END_ID= +# Telegram flood waits shorter than this (seconds) are slept through automatically +FLOOD_SLEEP_THRESHOLD=60 +SESSION_DIR=data/sessions + +# ---------------------------------------------------------------- Google Drive +# oauth (personal Drive) | service_account (Shared Drive) +GOOGLE_AUTH_MODE=oauth +# drive.file (least privilege; bot only sees folders it created) | drive (full access) +GOOGLE_DRIVE_SCOPE=drive.file +# Local OAuth files (used by `python main.py drive-auth`) +GOOGLE_CLIENT_SECRETS_FILE=credentials/client_secret.json +GOOGLE_TOKEN_FILE=credentials/token.json +# Heroku / file-less alternatives (raw JSON or base64 JSON): +GOOGLE_OAUTH_TOKEN_JSON= +GOOGLE_SERVICE_ACCOUNT_JSON= +GOOGLE_SERVICE_ACCOUNT_FILE= +# Folder that will contain the topic folders (empty = top level of My Drive) +GOOGLE_DRIVE_ROOT_FOLDER_ID= +DRIVE_SHARED_DRIVE_ID= +# none (private, default) | anyone_with_link +DRIVE_SHARE_MODE=none +# Multiple of 256KB; one chunk is held in memory at a time +UPLOAD_CHUNK_SIZE=16MB + +# ---------------------------------------------------------------- Organisation & naming +BATCH_SUBFOLDERS_ENABLED=false +FALLBACK_TOPIC_FOLDER=Uncategorized +FALLBACK_BATCH_FOLDER=Unknown Batch +# suffix_number | suffix_message_id | skip | fail (never overwrites) +DUPLICATE_POLICY=suffix_number +# Reuse an existing same-name file in the folder when size and MD5 match +REUSE_IDENTICAL_FILES=true +# skip | upload : the same Telegram media posted again under a new message ID +DUPLICATE_MEDIA_POLICY=skip +# fallback (Telegram file name / telegram_) | fail +MISSING_TITLE_POLICY=fallback +# keep | append : when the Title extension disagrees with the real container +EXTENSION_MISMATCH_POLICY=keep +# Replace characters Windows cannot store (<>:"|?*) +STRICT_FILENAMES=false +# Optional persistent sequence number prefix, e.g. "001 - Class 01.mp4" +SEQUENCE_PREFIX_ENABLED=false +SEQUENCE_PREFIX_FORMAT="{seq:03d} - " + +# ---------------------------------------------------------------- Storage +# PostgreSQL (required on Heroku; Heroku Postgres sets it automatically) +DATABASE_URL= +# SQLite file used when DATABASE_URL is empty (local development) +DATABASE_PATH=data/bot.db +# Only set to true if you accept losing all progress on every Heroku restart +ALLOW_EPHEMERAL_DATABASE=false +DATA_DIR=data +DOWNLOAD_DIR=downloads +LOG_DIR=logs +# 0 = unlimited +MAX_TEMP_STORAGE=0 +MIN_FREE_DISK_SPACE=1GB +KEEP_LOCAL_FILES=false +# Keep the local file of a video that FAILED (default: delete it; it can be downloaded again) +KEEP_FAILED_DOWNLOADS=false +# Seconds between sweeps that delete abandoned temp files (0 = only at startup / when disk is low) +TEMP_CLEANUP_INTERVAL=3600 +COMPUTE_SHA256=true + +# ---------------------------------------------------------------- Processing +MAX_RETRIES=4 +RETRY_DELAY=30 +RETRY_MAX_DELAY=1800 +# Retries of a single HTTP request inside one attempt +HTTP_RETRIES=5 +PROCESSING_DELAY=2 +# true = halt the whole queue when any video fails permanently +STRICT_STOP=false +# true = start processing at every boot (otherwise /run; then remembered across restarts) +AUTO_START=false +# Seconds to wait for another worker to release the queue lock (default 0 locally, 90 on Heroku) +INSTANCE_LOCK_WAIT= +PROGRESS_INTERVAL=10 + +# ---------------------------------------------------------------- /upload tasks +# Minutes before an unfinished /upload setup (waiting for links or /confirm) expires +UPLOAD_SETUP_TIMEOUT=30 +# Largest allowed range (end - start + 1) for one task +UPLOAD_MAX_RANGE=5000 +# true = links must point to SOURCE_CHANNEL_ID; false = any channel the Telegram account/bot can access +UPLOAD_RESTRICT_TO_SOURCE_CHANNEL=true +LOG_LEVEL=info diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..bb3bb44 --- /dev/null +++ b/.gitignore @@ -0,0 +1,27 @@ +# Secrets and credentials - never commit +.env +.env.* +!.env.example +credentials/ +client_secret*.json +token.json +*service-account*.json +*.session +*.session-journal +session_string.txt + +# Local state (SQLite, sessions, downloads, logs) +data/ +downloads/ +logs/ +*.db +*.db-wal +*.db-shm +*.lock + +# Python +.venv/ +venv/ +__pycache__/ +*.pyc +.pytest_cache/ diff --git a/.python-version b/.python-version new file mode 100644 index 0000000..e4fba21 --- /dev/null +++ b/.python-version @@ -0,0 +1 @@ +3.12 diff --git a/ENV_SETUP.md b/ENV_SETUP.md new file mode 100644 index 0000000..eabfe00 --- /dev/null +++ b/ENV_SETUP.md @@ -0,0 +1,388 @@ +# How to Obtain Every Environment Variable + +This guide explains where each configuration value comes from and how to get it. Work through the +sections in order: some steps (the Telegram session string and the Google token) need values from +earlier steps. + +- **Local runs:** put the values in a `.env` file (`cp .env.example .env`). +- **Heroku:** set them as Config Vars, either in the dashboard (*App → Settings → Reveal Config Vars*) + or with `heroku config:set NAME=value --app YOUR_APP_NAME`. + +> All values below are placeholders. Never commit real values or paste them into chats, issues or +> screenshots. `.env`, `credentials/` and `data/` are already in `.gitignore`. + +--- + +## Quick checklist + +| # | Variable | Required? | Where it comes from | +|---|---|---|---| +| 1 | `TELEGRAM_API_ID` | Always | my.telegram.org | +| 2 | `TELEGRAM_API_HASH` | Always | my.telegram.org | +| 3 | `TELEGRAM_BOT_TOKEN` | Always on Heroku (control commands); required in bot mode | @BotFather | +| 4 | `ADMIN_USER_IDS` | When a bot token is set | @userinfobot | +| 5 | `SOURCE_CHANNEL_ID` | Always | Channel username, link, or numeric ID | +| 6 | `TELEGRAM_MODE` | Optional (default `user`) | Your choice: `user` or `bot` | +| 7 | `TELEGRAM_SESSION_STRING` | `TELEGRAM_MODE=user` on Heroku | `python main.py login --session-string` | +| 8 | `GOOGLE_AUTH_MODE` | Optional (default `oauth`) | Your choice: `oauth` or `service_account` | +| 9 | `GOOGLE_OAUTH_TOKEN_JSON` | `oauth` on Heroku | `python main.py drive-auth` | +| 10 | `GOOGLE_SERVICE_ACCOUNT_JSON` | `service_account` on Heroku | Google Cloud Console key | +| 11 | `GOOGLE_DRIVE_ROOT_FOLDER_ID` | Recommended | Drive folder URL, or `python main.py drive-init` | +| 12 | `GOOGLE_DRIVE_SCOPE` | Optional (default `drive.file`) | Your choice | +| 13 | `DRIVE_SHARED_DRIVE_ID` | Shared Drive setups only | Shared Drive URL | +| 14 | `DATABASE_URL` | Required on Heroku | Heroku Postgres (set automatically) | + +Everything else has a sensible default (see [Optional settings](#optional-settings)). + +--- + +## Part 1 — Telegram + +### 1–2. `TELEGRAM_API_ID` and `TELEGRAM_API_HASH` + +These identify your Telegram API application. They are needed in **both** modes, because the bot talks +to Telegram through Telethon (MTProto). + +1. Open and log in with your phone number. Telegram sends a code to your + Telegram app, not by SMS. +2. Click **API development tools**. +3. If no app exists yet, fill in the form: + - *App title*: anything, for example `Drive Organizer`; + - *Short name*: 5–32 letters or digits, for example `driveorganizer`; + - *Platform*: `Desktop`; + - URL and description can stay empty. +4. Click **Create application**. +5. Copy the two values: + - **App api_id** (a number) → `TELEGRAM_API_ID` + - **App api_hash** (32 hex characters) → `TELEGRAM_API_HASH` + +```env +TELEGRAM_API_ID=1234567 +TELEGRAM_API_HASH=0123456789abcdef0123456789abcdef +``` + +> The API hash is a secret. If my.telegram.org shows an "ERROR" when you create the app, try another +> browser, disable ad blockers or VPNs, and use a short alphanumeric short name. + +### 3. `TELEGRAM_BOT_TOKEN` + +The bot provides the admin commands (`/run`, `/status`, …) and progress messages. In bot mode it also +reads the channel. + +1. In Telegram, open [@BotFather](https://t.me/BotFather) and send `/newbot`. +2. Enter a display name, then a username ending in `bot` (for example `my_drive_organizer_bot`). +3. BotFather replies with a token like `123456789:AAH...`. Copy it → `TELEGRAM_BOT_TOKEN`. +4. **Open your new bot and press Start.** Bots cannot message you until you do, so without this you + get no progress messages. + +```env +TELEGRAM_BOT_TOKEN=123456789:replace-with-your-bot-token +``` + +If the token leaks, send `/revoke` to @BotFather to get a new one. + +### 4. `ADMIN_USER_IDS` (and optional `NOTIFY_CHAT_IDS`) + +Only these numeric Telegram user IDs may control the bot. + +1. Open [@userinfobot](https://t.me/userinfobot) and press Start. +2. It replies with your **Id** (a number such as `111111111`). Copy it. +3. For several admins, ask each person to do the same and separate the IDs with commas. + +```env +ADMIN_USER_IDS=111111111,222222222 +``` + +`NOTIFY_CHAT_IDS` is optional. It lists the chats that receive progress and error messages, and +defaults to `ADMIN_USER_IDS`. Each person listed must have pressed Start in the bot. + +### 5. `SOURCE_CHANNEL_ID` + +Any of these formats is accepted: + +| Channel type | How to get it | Example value | +|---|---|---| +| Public channel | The `@username` shown in the channel info | `@my_channel` or `https://t.me/my_channel` | +| Private channel: post link | Right-click (or long-press) any post → **Copy Post Link** | `https://t.me/c/1234567890/55` | +| Private channel: numeric ID | Prefix the number from the post link with `-100` | `-1001234567890` | +| Private channel: Telegram Web | Open the channel at ; the URL ends with `#-100…` | `-1001234567890` | + +Another way: forward a channel post to [@JsonDumpBot](https://t.me/JsonDumpBot) and read +`forward_from_chat.id`. + +```env +SOURCE_CHANNEL_ID=-1001234567890 +``` + +> Bot mode with a private channel: the bot must be a channel **administrator**. If the numeric ID +> can't be resolved at the first start, post one message in the channel. The bot then learns the +> channel and remembers it in the database. + +### 6. `TELEGRAM_MODE` + +This is a choice, not a credential. + +| Value | Use when | Needs | +|---|---|---| +| `user` (default) | You need the channel's **existing history**, or you are a member but not an admin | `TELEGRAM_SESSION_STRING` (Heroku) or a local login | +| `bot` | You only need **new posts**, and you can make the bot a channel admin | Bot added as channel administrator | + +To add the bot as admin: open the channel → **Manage channel → Administrators → Add Admin** → search +for your bot's username → save. No extra rights are needed beyond the default. + +### 7. `TELEGRAM_SESSION_STRING` (user mode only) + +This lets the bot act as your Telegram account without a session file. Heroku's disk is wiped on +every restart, so a file wouldn't survive. You create it **once, on your own computer**. + +1. Do the [local setup](README.md#19-local-development): create the virtualenv and install + requirements. +2. Create `.env` containing at least: + ```env + TELEGRAM_API_ID=1234567 + TELEGRAM_API_HASH=0123456789abcdef0123456789abcdef + ``` +3. Run: + ```bash + python main.py login --session-string + ``` +4. Enter your phone number in international format (`+15551234567`), then the login code Telegram + sends to your app, then your 2FA password if you have one. **None of these are stored.** +5. The string is written to `data/sessions/session_string.txt` with owner-only permissions. +6. Put it on Heroku without printing it: + ```bash + heroku config:set TELEGRAM_SESSION_STRING="$(cat data/sessions/session_string.txt)" --app YOUR_APP_NAME + ``` + For a local `.env`, paste the file's contents after `TELEGRAM_SESSION_STRING=`. +7. Delete the file: `rm data/sessions/session_string.txt`. + +> The session string grants **full access to your Telegram account**. Treat it like a password. To +> revoke it: Telegram → *Settings → Devices* → terminate the session. Then generate a new string. + +For local-only use you can skip the string and run `python main.py login`, which creates +`data/sessions/user.session` instead. + +--- + +## Part 2 — Google Drive + +Pick **one** authentication method: + +| Your Drive | Use | Variable to obtain | +|---|---|---| +| Personal Google account (`@gmail.com`) or your own Workspace Drive | **OAuth** (`GOOGLE_AUTH_MODE=oauth`) | `GOOGLE_OAUTH_TOKEN_JSON` | +| Google Workspace **Shared Drive** | **Service account** (`GOOGLE_AUTH_MODE=service_account`) | `GOOGLE_SERVICE_ACCOUNT_JSON` | + +> A service account **cannot** upload into a personal My Drive: it has no storage quota there. Use +> OAuth for personal Drives. + +### Common first step: Google Cloud project and Drive API + +1. Open and sign in with the Google account that owns the + destination Drive. +2. Use the project selector at the top → **New Project** → give it a name → **Create**, then select + it. +3. Go to **APIs & Services → Library**, search for **Google Drive API**, and click **Enable**. + +### 9. `GOOGLE_OAUTH_TOKEN_JSON` (OAuth) + +#### a) Configure the consent screen + +1. Go to **APIs & Services → OAuth consent screen**. In the newer console this is **Google Auth + Platform → Branding / Audience**. +2. User type: **External** (or *Internal* if you use Workspace and only your organisation needs + access). +3. Fill in the app name, your support email and developer contact email, then save. +4. Under **Audience** (or *Test users*), add your own Google account. +5. **Publish the app** (*Publishing status → In production*). + > If the app stays in *Testing*, Google expires its refresh tokens after **7 days** and the bot + > will lose Drive access every week. With the default `drive.file` scope no Google verification + > is needed. With the full `drive` scope you'll see an "unverified app" warning during your own + > login: click *Advanced → Go to (app name)*. + +#### b) Create the OAuth client + +1. Go to **APIs & Services → Credentials → Create Credentials → OAuth client ID**. +2. Application type: **Desktop app**. Give it any name → **Create**. +3. Click **Download JSON** and save it in the project as `credentials/client_secret.json`. Create the + `credentials/` folder if needed; it is gitignored. + +#### c) Choose the scope *before* authorizing + +| `GOOGLE_DRIVE_SCOPE` | Meaning | Choose it if | +|---|---|---| +| `drive.file` (default, least privilege) | The bot only sees files and folders **it created** | You let the bot create its own root folder (step 11, option B) | +| `drive` | Full access to your Drive | You want uploads to go into a folder **you** created | + +Set it in `.env` (for example `GOOGLE_DRIVE_SCOPE=drive.file`). Use the **same value on Heroku**. + +#### d) Authorize and get the token + +```bash +python main.py drive-auth # opens a browser +python main.py drive-auth --no-browser # prints a URL to open yourself +``` + +Sign in, approve access, and the token is saved to `credentials/token.json`. It contains the refresh +token, client ID and client secret. Then set it on Heroku: + +```bash +heroku config:set GOOGLE_OAUTH_TOKEN_JSON="$(cat credentials/token.json)" --app YOUR_APP_NAME +``` + +If you prefer pasting into the Heroku dashboard, a single-line base64 version also works: + +```bash +base64 < credentials/token.json | tr -d '\n' +``` + +```env +GOOGLE_AUTH_MODE=oauth +GOOGLE_OAUTH_TOKEN_JSON={"token": "...", "refresh_token": "...", "client_id": "...", "client_secret": "...", ...} +``` + +> If you change `GOOGLE_DRIVE_SCOPE` later, delete `credentials/token.json`, run `drive-auth` again, +> and update the Config Var. To revoke access: . + +### 10. `GOOGLE_SERVICE_ACCOUNT_JSON` (Shared Drive) + +1. In the Cloud Console: **IAM & Admin → Service Accounts → Create service account**. Give it a name; + no project roles are needed. Click **Done**. +2. Open the service account → **Keys → Add key → Create new key → JSON**. The file downloads once. + Store it safely. +3. Copy the `client_email` value from the JSON (for example + `uploader@your-project.iam.gserviceaccount.com`). +4. In Google Drive, open the **Shared Drive** → **Manage members** → add that email as + **Content manager**. +5. Set the variables: + ```bash + heroku config:set GOOGLE_AUTH_MODE=service_account \ + GOOGLE_DRIVE_SCOPE=drive \ + GOOGLE_SERVICE_ACCOUNT_JSON="$(cat path/to/key.json)" --app YOUR_APP_NAME + ``` + Use `GOOGLE_DRIVE_SCOPE=drive`, because the service account must reach a Shared Drive folder it + did not create. + Locally you can instead set `GOOGLE_SERVICE_ACCOUNT_FILE=path/to/key.json`. + +> Your organisation may block service-account key creation (policy +> `iam.disableServiceAccountKeyCreation`). In that case use OAuth. + +### 11. `GOOGLE_DRIVE_ROOT_FOLDER_ID` + +This is the folder that will contain all topic folders. If you leave it empty, topic folders are +created at the top level of My Drive. + +**Option A: a folder you created** (needs `GOOGLE_DRIVE_SCOPE=drive`, or a Shared Drive folder) +1. Open the folder in Google Drive in a browser. +2. The URL looks like `https://drive.google.com/drive/folders/1AbCdEfGhIjKlMnOpQrStUvWxYz`. +3. The part after `/folders/` is the ID → `GOOGLE_DRIVE_ROOT_FOLDER_ID`. + +**Option B: let the bot create it** (works with the default `drive.file` scope) +1. Run this after `drive-auth`: + ```bash + python main.py drive-init --name "Telegram Videos" + ``` +2. It prints `Set GOOGLE_DRIVE_ROOT_FOLDER_ID=...`. Copy that ID. + +```env +GOOGLE_DRIVE_ROOT_FOLDER_ID=1AbCdEfGhIjKlMnOpQrStUvWxYz +``` + +At startup the bot checks that it can see this folder. If it can't, it stops and explains how to fix +the problem. + +### 13. `DRIVE_SHARED_DRIVE_ID` (Shared Drive only) + +Open the Shared Drive itself (its top level). The URL is +`https://drive.google.com/drive/folders/`, and that `` is the Shared Drive ID. + +```env +DRIVE_SHARED_DRIVE_ID=0AbCdEfGhIjKlUk9PVA +``` + +### `DRIVE_SHARE_MODE` + +Leave it as `none` so uploads stay private. Set `anyone_with_link` only if every uploaded video +should be viewable by anyone who has its link. + +--- + +## Part 3 — Database + +### 14. `DATABASE_URL` + +- **Heroku Deploy button:** the Heroku Postgres add-on is provisioned and `DATABASE_URL` is set + automatically. Nothing to do. +- **Manual Heroku setup:** + ```bash + heroku addons:create heroku-postgresql:essential-0 --app YOUR_APP_NAME + heroku config:get DATABASE_URL --app YOUR_APP_NAME # confirm it is set + ``` + This is a paid plan; check Heroku's current pricing. Don't copy the URL anywhere else, because + Heroku may rotate it. +- **Other PostgreSQL providers:** use their connection string + (`postgresql://USER:PASSWORD@HOST:5432/DBNAME`). TLS is added automatically for remote hosts. +- **Local development:** leave it empty to use SQLite at `data/bot.db`. + +On Heroku the bot refuses to start without `DATABASE_URL`, unless you set +`ALLOW_EPHEMERAL_DATABASE=true`. With that set, all progress is lost on every restart. + +--- + +## Optional settings + +These are preferences, not credentials. Defaults are shown; full descriptions are in +[`.env.example`](.env.example). + +| Variable | Default | Purpose | +|---|---|---| +| `PROCESS_MODE` | `both` | `existing` (history), `new` (new posts), or `both` | +| `START_MESSAGE_ID` | `1` | First message ID to scan | +| `START_DATE` | empty | Only messages on or after this ISO date, for example `2024-01-31` | +| `INCLUDE_DOCUMENT_VIDEOS` | `true` | Also process videos sent as files | +| `FILTER_TOPICS` / `FILTER_BATCHES` | empty | Comma-separated allow-lists | +| `BATCH_SUBFOLDERS_ENABLED` | `false` | `Root/Batch/Topic` folder layout | +| `DUPLICATE_POLICY` | `suffix_number` | `suffix_number`, `suffix_message_id`, `skip`, `fail` | +| `MAX_RETRIES` / `RETRY_DELAY` / `RETRY_MAX_DELAY` | `4` / `30` / `1800` | Retry behaviour (seconds) | +| `PROGRESS_INTERVAL` | `10` | Seconds between progress-message edits | +| `DOWNLOAD_DIR` | `downloads` (`/tmp/telegram_drive_bot` on Heroku) | Temporary video storage | +| `UPLOAD_CHUNK_SIZE` | `16MB` (`8MB` on Heroku) | Upload chunk size (multiple of 256KB) | +| `MIN_FREE_DISK_SPACE` | `1GB` (`512MB` on Heroku) | Free-disk reserve | +| `AUTO_START` | `false` | Start processing at every boot | +| `LOG_LEVEL` | `info` | `debug`, `info`, `warning`, `error` | +| `BOT_SCAN_MAX_EMPTY` / `BOT_SCAN_END_ID` | `200` / empty | Bot-mode history probing limits | + +--- + +## Verify your configuration + +Locally, with a filled `.env`: + +```bash +python main.py check-config +``` + +This prints `Configuration OK.` or lists every problem with a fix. On Heroku, run +`heroku logs --tail --app YOUR_APP_NAME` after deploying. A healthy start logs: +- `Database: PostgreSQL` +- `Google Drive credentials validated` +- `Source channel resolved` + +The bot then sends you *Bot online* in Telegram. + +## Minimal example (Heroku, user mode, OAuth) + +```env +TELEGRAM_MODE=user +TELEGRAM_API_ID=1234567 +TELEGRAM_API_HASH=0123456789abcdef0123456789abcdef +TELEGRAM_BOT_TOKEN=123456789:replace-with-your-bot-token +TELEGRAM_SESSION_STRING=1BVtsOK...replace-with-your-session-string +ADMIN_USER_IDS=111111111 +SOURCE_CHANNEL_ID=-1001234567890 +GOOGLE_AUTH_MODE=oauth +GOOGLE_DRIVE_SCOPE=drive.file +GOOGLE_OAUTH_TOKEN_JSON={"refresh_token": "...", "client_id": "...", "client_secret": "...", "token_uri": "https://oauth2.googleapis.com/token"} +GOOGLE_DRIVE_ROOT_FOLDER_ID=1AbCdEfGhIjKlMnOpQrStUvWxYz +# DATABASE_URL is set automatically by Heroku Postgres +``` diff --git a/Procfile b/Procfile new file mode 100644 index 0000000..e045cb2 --- /dev/null +++ b/Procfile @@ -0,0 +1 @@ +worker: python main.py run diff --git a/README.md b/README.md new file mode 100644 index 0000000..a147b17 --- /dev/null +++ b/README.md @@ -0,0 +1,510 @@ +# Telegram Channel → Google Drive Organizer + +[![Deploy to Heroku](https://www.herokucdn.com/deploy/button.svg)](https://heroku.com/deploy?template=https://github.com/YOUR_GITHUB_USERNAME/YOUR_REPOSITORY_NAME) + +> **Before the button works:** replace `https://github.com/YOUR_GITHUB_USERNAME/YOUR_REPOSITORY_NAME` +> in the line above with the URL of the **public** GitHub repository that contains this project +> (see [Publishing to GitHub](#publishing-to-github)). That URL is the only value to change. +> The button flow has not been tested against a live repository yet. Test it once after publishing. + +A long-running worker that reads video posts from a Telegram channel, downloads each video, and +uploads it to Google Drive. The caption's **Title** becomes the file name and its **Topic** picks the +Drive folder. Videos are processed one at a time. All progress is stored in a database, so the bot +resumes after crashes, restarts and redeploys without uploading anything twice. + +## Contents +1. [Overview](#1-overview) · 2. [Features](#2-features) · 3. [Caption format](#3-caption-format-and-metadata-extraction) · +4. [Drive organization](#4-google-drive-topic-folder-organization) · 5. [Prerequisites](#5-prerequisites) · +6. [One-click Heroku deployment](#6-one-click-heroku-deployment) · 7. [Telegram setup](#7-telegram-bot-setup) · +8. [Historical access](#8-historical-channel-access-requirements) · 9. [Google Drive auth](#9-google-drive-authentication) · +10. [Config Vars](#10-required-heroku-config-vars) · 11. [Database](#11-database-provisioning-and-configuration) · +12. [First deployment](#12-first-deployment-and-startup) · 13. [Commands](#13-telegram-commands) · 14. [Logs](#14-checking-logs) · +15. [Restart / redeploy](#15-restarting-and-redeploying) · 16. [Troubleshooting](#16-troubleshooting) · +17. [Security](#17-security-and-credential-handling) · 18. [Limitations](#18-known-limitations-and-heroku-resource-constraints) · +19. [Local development](#19-local-development) + +--- + +## 1. Overview + +``` +Telegram channel ──(Telethon / MTProto)──► queue worker (one video at a time) ──► Google Drive + │ (resumable upload, +Admin chat ◄── bot commands & progress ─────────┤ verified by size + MD5) + ▼ + PostgreSQL (Heroku) / SQLite (local) +``` + +For each video: **discover → parse caption → resolve the topic folder → download to a temp disk → +upload (resumable) → verify size and MD5 in Drive → save the Drive file ID → mark COMPLETED → delete +the temp file → next video.** + +## 2. Features + +- Caption parsing for `Index` (one or many), `Title`, `Topic`, `Batch` and `Extracted By`. +- Topic folders are created once and reused. Optional `Batch/Topic` layout. +- Files are never overwritten. Name collisions get `(2)`, `(3)`, … or `[message id]`, or are skipped or failed, per your policy. +- No duplicate uploads, even if the process dies just after Drive accepted the file. Every upload carries a hidden `appProperties` tag (`channel:message`) that is checked before uploading. +- Resumable uploads that survive restarts. The upload session URI is stored in the database. +- Resumable downloads, continuing from the last 1 MiB boundary of a partial file. +- Explicit state machine: `PENDING → DOWNLOADING → DOWNLOADED → UPLOADING → COMPLETED`, plus `RETRY_WAIT`, `FAILED` and `SKIPPED`. Every transition is validated and logged in the `transitions` table. +- Bounded retries with exponential backoff, honouring Telegram flood waits and Drive rate limits. Permission errors are not retried. +- Disk-space checks before each download. A full disk, Drive quota or expired auth halts the queue cleanly instead of burning retries. +- Admin-only Telegram commands, plus one progress message that is edited in place rather than spammed. +- Graceful shutdown on `SIGTERM` (Heroku restarts) at a resumable checkpoint. +- A single-worker guard (PostgreSQL advisory lock) prevents two dynos from processing the same queue. + +## 3. Caption format and metadata extraction + +``` +Index: 001 + +Index: 002 + +Title: Class 01 Aman sir.mp4 + +Topic: Misc• Grammar Practice + +Batch: English ( SSC MAINS ) + +Extracted By: @PUTINxINDIA +``` + +| Field | Use | +|---|---| +| `Title` | Drive file name (`Class 01 Aman sir.mp4`) | +| `Topic` | Drive folder (`Misc• Grammar Practice`) | +| `Index` | All values kept in order (`["001", "002"]`); never used as the file name. The Telegram message ID is the unique identity. | +| `Batch` | Stored; optional `Batch/Topic` folders | +| `Extracted By` | Stored in the record | + +Parsing rules: +- Fields can appear in any order. +- Labels are case-insensitive. +- Blank lines, emoji or bullet prefixes, extra spaces and full-width colons are allowed. +- Unicode text is preserved. +- Parsing is line-based, so one field never swallows the next. +- A flattened single-line caption is split at each known label. + +Missing data never makes a video silently disappear: + +| Case | Behaviour | +|---|---| +| No Topic | Folder `Uncategorized` (`FALLBACK_TOPIC_FOLDER`) | +| No Title | Telegram file name, else `telegram_.`. A warning is recorded. `MISSING_TITLE_POLICY=fail` marks it FAILED instead. | +| Title without extension | Extension taken from the real file's magic bytes, else from the MIME type | +| Title extension ≠ real container | Kept as written, with a warning. `EXTENSION_MISMATCH_POLICY=append` appends the real one. | +| `/` `\` or control characters | Replaced (`STRICT_FILENAMES=true` also replaces `<>:"\|?*`) | + +## 4. Google Drive topic-folder organization + +``` +/ +├── Misc• Grammar Practice/ +│ ├── Class 01 Aman sir.mp4 +│ └── Class 02 Aman sir.mp4 +├── Error Detection/ +│ └── Class 01 Aman sir.mp4 +└── Uncategorized/ +``` + +Folder lookup goes database cache → Drive search → create. If duplicate folders already exist, the +oldest is reused. If a folder-create response is lost, the bot searches again before retrying, so +it never creates a second copy. Set `BATCH_SUBFOLDERS_ENABLED=true` for `Root/Batch/Topic/`. + +**Duplicate policy (`DUPLICATE_POLICY`)** applies when a *different* file already has the target +name in the folder: + +| Value | Result | +|---|---| +| `suffix_number` (default) | `Class 01 Aman sir (2).mp4` | +| `suffix_message_id` | `Class 01 Aman sir [1234].mp4` | +| `skip` | Item marked SKIPPED with the reason | +| `fail` | Item marked FAILED | + +An existing file is reused, not re-uploaded, only when its identity is verified. That means either +its hidden tag matches this Telegram message, or it has the same size and MD5 +(`REUSE_IDENTICAL_FILES=true`). The same Telegram media re-posted under a new message ID is SKIPPED +by default (`DUPLICATE_MEDIA_POLICY`). + +## 5. Prerequisites + +- A Telegram account, plus API credentials from . +- A Telegram bot from [@BotFather](https://t.me/BotFather). +- A Google account with a Google Cloud project (free) with the Drive API enabled. +- **A local computer with Python 3.12** for two one-time interactive logins. Telegram and Google logins need a human, so they can't run on Heroku. +- A Heroku account with billing enabled. The worker dyno and Heroku Postgres are paid resources; check current pricing on Heroku. +- Optional: the [Heroku CLI](https://devcenter.heroku.com/articles/heroku-cli) and the [GitHub CLI](https://cli.github.com/). + +## 6. One-click Heroku deployment + +The button (top of this page) opens Heroku's deploy form. It: +- creates the app from this repository's `app.json`; +- builds it with the official Python buildpack; +- provisions **Heroku Postgres (`essential-0`)**, which sets `DATABASE_URL`; +- starts **one `basic` worker dyno** (`worker: python main.py run`); there is no web process. + +**It is not fully one-click.** Two secrets must be created on your computer first, because they need +interactive logins: + +1. `TELEGRAM_SESSION_STRING`, only for `TELEGRAM_MODE=user` (see [§7](#7-telegram-bot-setup)). +2. `GOOGLE_OAUTH_TOKEN_JSON` or `GOOGLE_SERVICE_ACCOUNT_JSON` (see [§9](#9-google-drive-authentication)). + +Prepare those, click the button, fill the form, then press **Deploy app**. If add-on provisioning +fails (for example because billing isn't verified or the plan isn't available), create the app +anyway and follow [§11](#11-database-provisioning-and-configuration) manually. The worker refuses +to start on Heroku without a database rather than silently losing progress. + +Manual alternative with the CLI (from the repository root): + +```bash +heroku create YOUR_APP_NAME --stack heroku-24 +heroku addons:create heroku-postgresql:essential-0 --app YOUR_APP_NAME +heroku config:set TELEGRAM_API_ID=1234567 TELEGRAM_API_HASH=your_api_hash \ + TELEGRAM_BOT_TOKEN=123456:your-bot-token ADMIN_USER_IDS=111111111 \ + SOURCE_CHANNEL_ID=@your_channel --app YOUR_APP_NAME +heroku config:set TELEGRAM_SESSION_STRING="$(cat data/sessions/session_string.txt)" --app YOUR_APP_NAME +heroku config:set GOOGLE_OAUTH_TOKEN_JSON="$(cat credentials/token.json)" --app YOUR_APP_NAME +git push heroku main +heroku ps:scale worker=1 --app YOUR_APP_NAME +``` + +### Publishing to GitHub + +This directory (the one containing `Procfile` and `app.json`) must be the **repository root**. + +```bash +cd telegram_drive_bot +git init -b main +git add . +git status # confirm: no .env, credentials/, data/, *.session, downloads/ +git commit -m "Telegram to Google Drive organizer" +gh repo create YOUR_REPOSITORY_NAME --public --source=. --push +# or create the repo on github.com, then: +# git remote add origin https://github.com/YOUR_GITHUB_USERNAME/YOUR_REPOSITORY_NAME.git && git push -u origin main +``` + +Then put that repository URL into the button link at the top of this README, commit, and push. +The Heroku button only works with **public** repositories. + +## 7. Telegram bot setup + +1. **API ID / hash:** log in at → *API development tools* → create an app. + Copy `api_id` → `TELEGRAM_API_ID` and `api_hash` → `TELEGRAM_API_HASH`. + These are needed in both modes, because the bot talks MTProto through Telethon. +2. **Bot:** talk to @BotFather → `/newbot` → copy the token → `TELEGRAM_BOT_TOKEN`. + **Open your bot and press Start.** A bot can't message you until you do, so you won't get progress messages otherwise. +3. **Admin IDs:** message [@userinfobot](https://t.me/userinfobot) to get your numeric ID → `ADMIN_USER_IDS` + (comma-separate several). Only these users can run commands. +4. **Source channel (`SOURCE_CHANNEL_ID`):** any of these forms work: + - a public channel: `@channelname` or `https://t.me/channelname`; + - a private channel: the numeric ID `-100…`, or a post link `https://t.me/c/1234567890/55` (use *Copy Post Link* in Telegram); + - forwarding a post to @userinfobot also shows the channel ID. +5. **Choose a mode (`TELEGRAM_MODE`):** + - `user` (default): reads the channel as *your* account, which must be a member. Create the session string locally: + ```bash + python main.py login --session-string + ``` + Telethon asks for your phone number, the login code and, only if enabled, your 2FA password. + None of these are stored. The resulting session string is written to + `data/sessions/session_string.txt` (mode 600). Copy it into `TELEGRAM_SESSION_STRING`, then delete the file. + - `bot`: add the bot to the channel as an **administrator**. No session string is needed. + +## 8. Historical channel access requirements + +| Capability | `user` mode | `bot` mode | +|---|---|---| +| New posts (live) | ✅ | ✅ (bot must be a channel admin) | +| Existing history | ✅ full history the account can see | ⚠️ best effort: bots can't list history, so message IDs are probed in batches until `BOT_SCAN_MAX_EMPTY` consecutive gaps (or up to `BOT_SCAN_END_ID`) | +| Large files | ✅ up to Telegram's file limit | ✅ uses MTProto, so the HTTP Bot API's 20 MB download limit does **not** apply | +| Private channel by numeric ID | ✅ | ⚠️ the bot learns the channel's access key from its first live post; the key is then stored in the database | +| Extra secret needed | `TELEGRAM_SESSION_STRING` | none | + +`PROCESS_MODE` controls what is queued: +- `existing`: history only (from `START_MESSAGE_ID` / `START_DATE`). +- `new`: only posts after the first start. +- `both` (default). + +The bot never deletes or edits Telegram messages and doesn't bypass channel restrictions. A video +in a channel with content protection (no forwarding/saving) may not be downloadable. + +## 9. Google Drive authentication + +Pick **one**. + +### Option A — OAuth (personal Google Drive, recommended) + +1. In the [Google Cloud Console](https://console.cloud.google.com/): create a project, then enable the **Google Drive API**. +2. Configure the **OAuth consent screen**: *External*, add yourself as a test user. + Then set the publishing status to **In production**. In *Testing* status, Google expires refresh + tokens after 7 days and the bot would lose Drive access weekly. With the default `drive.file` + scope no verification is required. With the full `drive` scope Google shows an "unverified app" + screen during your own login (*Advanced → continue*). +3. Under *Credentials*, create an **OAuth client ID** of type **Desktop app**. Download the JSON to + `credentials/client_secret.json` in this project. +4. Locally: `python main.py drive-auth`. This opens a browser; use `--no-browser` to get a URL to open yourself. + It writes `credentials/token.json` (mode 600). +5. Heroku: `heroku config:set GOOGLE_OAUTH_TOKEN_JSON="$(cat credentials/token.json)" --app YOUR_APP_NAME`. + Multi-line JSON works through the CLI. In the dashboard you can paste raw JSON, or a single-line + base64 version: `base64 < credentials/token.json | tr -d '\n'`. + +**Scope and root folder.** `GOOGLE_DRIVE_SCOPE=drive.file` (least privilege) lets the bot see only +files and folders *it created*. So either: +- let the bot create the root folder: `python main.py drive-init --name "Telegram Videos"` prints the ID for `GOOGLE_DRIVE_ROOT_FOLDER_ID`; or +- use a folder you created yourself: set `GOOGLE_DRIVE_SCOPE=drive` **before** running `drive-auth`. + +The folder ID is the part after `/folders/` in its URL. The bot checks at startup that it can see +the folder and stops with an explanation if it can't. + +### Option B — Service account (Google Workspace Shared Drive) + +A service account has **no storage quota in a personal My Drive**. Uploads into a folder that a +normal Gmail user merely shared with it fail with `storageQuotaExceeded`. Use it only with a +**Shared Drive**: + +1. Create a service account, then create a JSON key for it. +2. In the Shared Drive, *Manage members* → add the service account's `client_email` as **Content manager**. +3. Set: + - `GOOGLE_AUTH_MODE=service_account` + - `GOOGLE_SERVICE_ACCOUNT_JSON="$(cat key.json)"` + - `DRIVE_SHARED_DRIVE_ID=` + - `GOOGLE_DRIVE_ROOT_FOLDER_ID=` + - `GOOGLE_DRIVE_SCOPE=drive` (needed to use a folder the service account did not create). + +Credentials from environment variables are parsed in memory. Nothing is written to the dyno's disk, +and they are validated by an authenticated Drive call at startup. Uploaded files are **private** +unless you set `DRIVE_SHARE_MODE=anyone_with_link`. + +## 10. Required Heroku Config Vars + +Set these in the dashboard (*Settings → Reveal Config Vars*) or with +`heroku config:set NAME=value --app YOUR_APP_NAME`. + +| Variable | Required | Notes | +|---|---|---| +| `TELEGRAM_API_ID`, `TELEGRAM_API_HASH` | yes | my.telegram.org | +| `TELEGRAM_BOT_TOKEN` | yes | @BotFather; command interface (and channel access in bot mode) | +| `ADMIN_USER_IDS` | yes | numeric IDs, comma-separated | +| `SOURCE_CHANNEL_ID` | yes | `@name`, link, or `-100…` | +| `TELEGRAM_MODE` | yes (default `user`) | `user` or `bot` | +| `TELEGRAM_SESSION_STRING` | **when `TELEGRAM_MODE=user`** | from `main.py login --session-string` | +| `GOOGLE_AUTH_MODE` | yes (default `oauth`) | `oauth` or `service_account` | +| `GOOGLE_OAUTH_TOKEN_JSON` | **when `oauth`** | contents of `credentials/token.json` | +| `GOOGLE_SERVICE_ACCOUNT_JSON` | **when `service_account`** | key JSON | +| `GOOGLE_DRIVE_ROOT_FOLDER_ID` | recommended | empty = top level of My Drive | +| `GOOGLE_DRIVE_SCOPE` | no (`drive.file`) | must match the scope used in `drive-auth` | +| `DATABASE_URL` | **yes on Heroku** | set automatically by Heroku Postgres | +| `PROCESS_MODE` | no (`both`) | `existing` / `new` / `both` | +| `DUPLICATE_POLICY` | no (`suffix_number`) | see §4 | +| `BATCH_SUBFOLDERS_ENABLED` | no (`false`) | `Root/Batch/Topic` | +| `MAX_RETRIES` / `RETRY_DELAY` | no (`4` / `30`) | automatic retries / base delay in seconds | +| `PROGRESS_INTERVAL` | no (`10`) | seconds between progress-message edits | +| `DOWNLOAD_DIR` | no (`/tmp/telegram_drive_bot` in app.json) | temporary files | +| `UPLOAD_CHUNK_SIZE` | no (`8MB` in app.json) | multiple of 256KB; memory per chunk | +| `AUTO_START` | no (`false`) | start processing on every boot | + +Step-by-step instructions for obtaining every value are in **[ENV_SETUP.md](ENV_SETUP.md)**. Every other supported variable is listed with an explanation in [`.env.example`](.env.example). + +## 11. Database provisioning and configuration + +Production state lives in **PostgreSQL**: +- video records with all caption metadata; +- states, attempts, errors, timestamps and Drive file IDs; +- upload session URIs; +- the folder map; +- discovery and completion cursors; +- duplicate-detection data; +- the transition history; +- the pause/run/stop choice. + +- **Via the button:** `app.json` requests `heroku-postgresql:essential-0`, a paid plan. +- **Manually:** `heroku addons:create heroku-postgresql:essential-0 --app YOUR_APP_NAME` + (or any plan, or any PostgreSQL reachable through `DATABASE_URL`). +- The schema is created and migrated automatically at startup. The version is tracked in the `kv` table. +- TLS (`sslmode=require`) is added automatically for remote hosts. +- **Without `DATABASE_URL` on Heroku the worker refuses to start.** SQLite on a dyno would be wiped + at every restart, and the bot would re-upload everything. To accept that risk explicitly, set + `ALLOW_EPHEMERAL_DATABASE=true`; a warning is logged at every start. +- Locally, without `DATABASE_URL`, SQLite at `data/bot.db` is used. This is for development only. + +## 12. First deployment and startup + +1. Deploy (button or CLI). Watch the logs: `heroku logs --tail --app YOUR_APP_NAME`. + You should see the database backend, `Google Drive credentials validated`, and `Source channel resolved`. +2. The bot messages each admin: *Bot online (PostgreSQL); send /sync then /run to start.* +3. Send `/sync` to discover channel videos, then `/run`. Watch `/progress`. +4. After `/run`, the choice is stored. After any dyno restart the bot syncs and resumes on its own. + `/pause` and `/stop` are also remembered. + +## 13. Telegram commands + +Send these to the bot in a **private chat**. Users not in `ADMIN_USER_IDS` are refused. + +| Command | Effect | +|---|---| +| `/start`, `/help` | Instructions | +| `/status` | Queue state, the current video, live progress | +| `/progress` | Processed / pending / retrying / failed / skipped / remaining counts | +| `/sync` | Discover unseen messages and queue them (idempotent). `/sync all` rescans from `START_MESSAGE_ID`. | +| `/run` | Start processing | +| `/pause` | Stop the current transfer at the next chunk boundary (resumable) and wait | +| `/resume` | Continue | +| `/stop` | Checkpoint and end the worker until `/run` | +| `/retry` | Re-queue FAILED items with a fresh retry budget | +| `/failed` | Failed items with short error summaries | +| `/folders` | Folder count and uploads per topic | +| `/check ID` | Full record for one Telegram message ID | + +Progress is a single message, edited every `PROGRESS_INTERVAL` seconds: + +``` +Processing video 12 of 150 +Topic: Misc• Grammar Practice +Title: Class 01 Aman sir.mp4 +Status: Uploading +Download: 100% (1.2 GB / 1.2 GB) +Upload: 45% (540.0 MB / 1.2 GB) · 18.3 MB/s · ETA 36s +``` + +## 14. Checking logs + +```bash +heroku logs --tail --app YOUR_APP_NAME # live +heroku logs -n 1500 --app YOUR_APP_NAME # recent history +heroku ps --app YOUR_APP_NAME # dyno status +``` + +On Heroku, logs go to stdout only. Locally they also go to `logs/bot.log`, rotated at 10 MB. +Tokens, hashes, session strings, private keys and the database URL are redacted from log output. + +## 15. Restarting and redeploying + +```bash +heroku ps:restart worker --app YOUR_APP_NAME +git push heroku main # or push to GitHub with automatic deploys enabled +heroku ps:scale worker=1 --app YOUR_APP_NAME +``` + +On `SIGTERM`, Heroku allows 30 seconds before it kills the process. The bot: +1. stops at the next download or upload chunk boundary; +2. saves state; +3. disconnects. + +On the next start it: +1. returns interrupted downloads to PENDING; +2. checks Drive for any upload that may have finished, using the hidden tag; +3. re-downloads files lost with the old dyno's disk; +4. **continues the saved upload session**, when the re-downloaded file is byte-identical (MD5); +5. resumes processing if it was running. + +**Scaling:** run **exactly one** worker. A second worker is blocked by a PostgreSQL advisory lock. +It waits `INSTANCE_LOCK_WAIT` seconds (90 on Heroku, which covers deploy overlap), then exits with +code 3. Scaling above 1 therefore only produces crash-looping dynos, not parallel processing. The +pipeline is intentionally sequential. + +## 16. Troubleshooting + +| Symptom | Fix | +|---|---| +| `DATABASE_URL is not set` | Attach Heroku Postgres (§11) | +| `TELEGRAM_SESSION_STRING is required…` | Run `python main.py login --session-string` locally (§7), or use `TELEGRAM_MODE=bot` | +| `Telegram user session is not authorized` | The session was revoked (e.g. *Terminate all sessions*); create a new session string | +| `Cannot resolve SOURCE_CHANNEL_ID` | Use `@username` for public channels; make sure the account/bot is a member/admin; in bot mode post once in the channel | +| `DRIVE_ROOT_FOLDER_ID is not visible…` | `drive.file` scope can't see folders you created: use `drive-init`, or switch to `GOOGLE_DRIVE_SCOPE=drive` and redo `drive-auth` | +| `Google authorization expired or was revoked` | Redo `drive-auth` and update `GOOGLE_OAUTH_TOKEN_JSON`; publish the consent screen to *In production* (§9) | +| `invalid_scope` on refresh | The token was created with another scope; delete `credentials/token.json` and run `drive-auth` again | +| `storageQuotaExceeded` | Drive is full, or you used a service account without a Shared Drive | +| `Not enough free disk space` | A video is larger than the dyno's free disk; lower `MIN_FREE_DISK_SPACE` with care, or process that video locally | +| No progress messages | Press *Start* in your bot chat; check `ADMIN_USER_IDS` / `NOTIFY_CHAT_IDS` | +| `Another worker is already processing` | `heroku ps:scale worker=1`; locally, stop the other process | +| Item stuck in FAILED | `/check ID` shows the error; fix the cause, then `/retry` | +| Long pause after `FloodWait` | Telegram rate limit; the bot waits the time Telegram asks for | + +## 17. Security and credential handling + +- All secrets come from environment variables / Config Vars; nothing is hard-coded. +- `.gitignore` excludes `.env`, `credentials/`, `token.json`, `*.session`, `session_string.txt`, `data/` and `downloads/`. +- The session string and OAuth token are full-access credentials. If one leaks: + - Telegram: *Settings → Devices → terminate the session*; + - Google: . +- Uploaded files stay private by default. Drive scope defaults to least privilege (`drive.file`). +- Commands are admin-only and private-chat only. Replies never include credentials, local paths or stack traces. +- Telegram login codes and 2FA passwords are entered interactively, locally, and never stored. + +## 18. Known limitations and Heroku resource constraints + +- **Ephemeral disk:** each video is staged on the dyno's temporary filesystem. Heroku doesn't + guarantee a specific capacity, and the disk is wiped on every restart (at least daily). + - The bot checks free space before each download and halts with a message if a video won't fit. + - Multi-GB videos may not fit on a dyno; run the bot locally for those. +- **Memory:** one upload chunk (`UPLOAD_CHUNK_SIZE`) is held in RAM at a time. `8MB` is comfortable on 512 MB dynos. +- **Daily restarts:** in-flight transfers resume from their checkpoint. A partially downloaded file + on the old dyno is lost and is downloaded again, but the Drive upload session is reused. +- **Bot mode:** history is best effort (ID probing), and private channels need one live post before the numeric ID resolves. +- **Telegram:** flood limits apply. Downloads go through Telegram's servers at whatever speed they allow. +- **Drive:** resumable sessions expire after about a week. Older sessions are discarded and the + upload restarts after a Drive check for an already-finished copy. +- **Captions edited after discovery** are not re-read. Use `/check ID` to see what was stored. +- **Cost:** the worker dyno and Postgres add-on are paid Heroku resources. + +### What has and has not been verified + +- **Automated tests (171):** run against SQLite **and** a real local PostgreSQL 16, with mocked Telegram and Google Drive. +- **Not tested:** real Telegram logins and downloads, real Google Drive uploads, and an actual Heroku deployment. Follow the manual end-to-end procedure below before relying on it. + +## 19. Local development + +```bash +python3.12 -m venv .venv && source .venv/bin/activate +pip install -r requirements-dev.txt +cp .env.example .env # fill in values +python main.py login # user mode: creates data/sessions/user.session +python main.py drive-auth # creates credentials/token.json +python main.py check-config +python main.py run # or ./start.sh +``` + +Other subcommands: +- `python main.py sync [--rescan]` +- `python main.py process` (headless: sync, process everything, exit) +- `python main.py status` +- `python main.py drive-init --name "Telegram Videos"` + +### Tests + +```bash +pytest +``` + +`pgserver` from `requirements-dev.txt` starts a throwaway PostgreSQL so every database test runs on +both backends. Without it, only SQLite is tested. The suite covers: +- caption parsing (the supplied example, multiple Index lines, missing fields, Unicode, flattened captions); +- naming and collisions; +- Drive folder reuse and dedupe; +- the resumable-upload protocol; +- state transitions and duplicate messages; +- download and upload failure recovery; +- restart during upload, and restart with a wiped disk; +- a successful upload followed by a database-write failure; +- sequential processing, pause/resume and graceful shutdown; +- admin-only commands; +- configuration and Heroku validation, env-var credentials, log redaction; +- `app.json` / Procfile consistency. + +### Manual end-to-end test (real integrations) + +1. Create a private test channel. Post four videos: + - two with `Topic: Test A`; + - one with `Topic: Test B`; + - one with no caption. + Give two of them the same `Title`. +2. Point the bot at a fresh Drive folder. Then `/sync`, `/run`, and watch `/progress`. +3. Check Drive: + - `Test A/` holds two files, one with a `(2)` suffix if the titles matched; + - `Test B/` holds one; + - `Uncategorized/` holds the caption-less video; + - file sizes match Telegram's. +4. Kill the process (`kill -9`, or `heroku ps:restart`) **during an upload** of a large video. + Restart it. Confirm the upload continues and Drive ends up with exactly one copy. +5. `/stop`, post a new video, `/run`: only the new video is processed. +6. Revoke Drive access temporarily and confirm the queue halts with a clear message. Restore access, then `/resume`. diff --git a/app.json b/app.json new file mode 100644 index 0000000..b724921 --- /dev/null +++ b/app.json @@ -0,0 +1,189 @@ +{ + "name": "Telegram Drive Organizer", + "description": "Telegram bot that downloads channel videos (whole channel or an /upload message range) into Google Drive topic folders.", + "keywords": [ + "telegram", + "google-drive", + "telethon", + "video", + "worker" + ], + "stack": "heroku-24", + "buildpacks": [ + { + "url": "heroku/python" + } + ], + "formation": { + "worker": { + "quantity": 1, + "size": "basic" + } + }, + "addons": [ + { + "plan": "heroku-postgresql:essential-0", + "as": "DATABASE" + } + ], + "env": { + "TELEGRAM_MODE": { + "description": "How the bot reads the channel: 'user' (your Telegram account; full history; needs TELEGRAM_SESSION_STRING) or 'bot' (bot admin of the channel; new posts, best-effort history).", + "value": "user", + "required": true + }, + "TELEGRAM_API_ID": { + "description": "Numeric API ID from https://my.telegram.org > API development tools. Needed in both modes.", + "required": true + }, + "TELEGRAM_API_HASH": { + "description": "API hash from https://my.telegram.org > API development tools. Keep secret.", + "required": true + }, + "TELEGRAM_BOT_TOKEN": { + "description": "Bot token from @BotFather. Provides the admin command interface (/run, /status, ...); in bot mode it also reads the channel.", + "required": true + }, + "TELEGRAM_SESSION_STRING": { + "description": "Only for TELEGRAM_MODE=user: string session created locally with `python main.py login --session-string`. Grants access to your account - keep secret.", + "required": false + }, + "ADMIN_USER_IDS": { + "description": "Comma-separated numeric Telegram user IDs allowed to run commands (get yours from @userinfobot).", + "required": true + }, + "SOURCE_CHANNEL_ID": { + "description": "Source channel: @username, https://t.me/... link, or numeric -100... ID.", + "required": true + }, + "GOOGLE_AUTH_MODE": { + "description": "'oauth' for a personal Google Drive (uses GOOGLE_OAUTH_TOKEN_JSON) or 'service_account' for a Shared Drive (uses GOOGLE_SERVICE_ACCOUNT_JSON).", + "value": "oauth", + "required": true + }, + "GOOGLE_OAUTH_TOKEN_JSON": { + "description": "Required when GOOGLE_AUTH_MODE=oauth: contents of credentials/token.json produced locally by `python main.py drive-auth` (raw JSON or base64).", + "required": false + }, + "GOOGLE_SERVICE_ACCOUNT_JSON": { + "description": "Required when GOOGLE_AUTH_MODE=service_account: the service-account key JSON (raw or base64). Service accounts only work with a Shared Drive they are a member of.", + "required": false + }, + "GOOGLE_DRIVE_SCOPE": { + "description": "'drive.file' (least privilege; the bot only sees folders it created) or 'drive' (needed to use a folder you created yourself). Must match the scope used during drive-auth.", + "value": "drive.file", + "required": false + }, + "GOOGLE_DRIVE_ROOT_FOLDER_ID": { + "description": "ID of the Drive folder that will contain the topic folders (the part after /folders/ in its URL). Empty = top level of My Drive.", + "required": false + }, + "DRIVE_SHARED_DRIVE_ID": { + "description": "Shared Drive ID when the root folder lives in a Shared Drive (service-account setups).", + "required": false + }, + "DRIVE_SHARE_MODE": { + "description": "'none' keeps uploads private (default). 'anyone_with_link' makes every uploaded file readable by anyone with the link.", + "value": "none", + "required": false + }, + "PROCESS_MODE": { + "description": "'existing' (channel history only), 'new' (new posts only) or 'both'.", + "value": "both", + "required": false + }, + "START_MESSAGE_ID": { + "description": "First channel message ID to consider when scanning history.", + "value": "1", + "required": false + }, + "INCLUDE_DOCUMENT_VIDEOS": { + "description": "Also process videos that were sent as files/documents.", + "value": "true", + "required": false + }, + "FILTER_TOPICS": { + "description": "Optional comma-separated list of Topic values to process; others are recorded as SKIPPED.", + "required": false + }, + "FILTER_BATCHES": { + "description": "Optional comma-separated list of Batch values to process.", + "required": false + }, + "BATCH_SUBFOLDERS_ENABLED": { + "description": "true = Root/Batch/Topic folders; false = Root/Topic (default).", + "value": "false", + "required": false + }, + "DUPLICATE_POLICY": { + "description": "When a different file already has the same name in the topic folder: suffix_number ('Name (2).mp4'), suffix_message_id ('Name [123].mp4'), skip, or fail. Files are never overwritten.", + "value": "suffix_number", + "required": false + }, + "MAX_RETRIES": { + "description": "Automatic retries for a failing video before it is marked FAILED.", + "value": "4", + "required": false + }, + "RETRY_DELAY": { + "description": "Base retry delay in seconds (doubles on each retry, capped by RETRY_MAX_DELAY).", + "value": "30", + "required": false + }, + "PROGRESS_INTERVAL": { + "description": "Seconds between edits of the Telegram progress message.", + "value": "10", + "required": false + }, + "DOWNLOAD_DIR": { + "description": "Temporary download directory on the dyno's ephemeral disk.", + "value": "/tmp/telegram_drive_bot", + "required": false + }, + "UPLOAD_CHUNK_SIZE": { + "description": "Resumable upload chunk size (multiple of 256KB). Held in memory per chunk; 8MB suits 512MB dynos.", + "value": "8MB", + "required": false + }, + "MIN_FREE_DISK_SPACE": { + "description": "Free disk space to keep in reserve; a download is not started if it would go below this.", + "value": "512MB", + "required": false + }, + "AUTO_START": { + "description": "true = start processing at every boot. Otherwise processing starts with /run and then resumes automatically after restarts.", + "value": "false", + "required": false + }, + "LOG_LEVEL": { + "description": "debug, info, warning or error.", + "value": "info", + "required": false + }, + "UPLOAD_RESTRICT_TO_SOURCE_CHANNEL": { + "description": "true = /upload links must point to SOURCE_CHANNEL_ID; false = any channel the Telegram account or bot can access.", + "value": "true", + "required": false + }, + "UPLOAD_SETUP_TIMEOUT": { + "description": "Minutes before an unfinished /upload setup (waiting for links or /confirm) expires.", + "value": "30", + "required": false + }, + "UPLOAD_MAX_RANGE": { + "description": "Largest number of message IDs one /upload task may cover.", + "value": "5000", + "required": false + }, + "KEEP_FAILED_DOWNLOADS": { + "description": "Keep local files of failed videos on the dyno disk (default false: delete; they are downloaded again on retry).", + "value": "false", + "required": false + }, + "TEMP_CLEANUP_INTERVAL": { + "description": "Seconds between sweeps that delete abandoned temporary files.", + "value": "3600", + "required": false + } + } +} diff --git a/bot_handlers.py b/bot_handlers.py new file mode 100644 index 0000000..23ebf12 --- /dev/null +++ b/bot_handlers.py @@ -0,0 +1,542 @@ +"""Telegram control commands (admin only), the /upload conversation, and the notifier. + +The /upload conversation is persisted in the ``upload_tasks`` table (statuses +AWAITING_START -> AWAITING_END -> AWAITING_CONFIRMATION), keyed by the admin's +Telegram user ID, so every administrator has an independent setup and a restart +does not lose it. +""" +from __future__ import annotations + +import logging +import re +from datetime import datetime, timedelta + +from database import iso, utcnow +from errors import BotError +from link_parser import EXAMPLE, LinkError, parse_message_link +from models import ACTIVE_TASK_STATUSES, SETUP_TASK_STATUSES, State, TaskStatus, UploadTask, VideoRecord +from progress import HtmlReply, esc +from queue_manager import QueueManager, public_error + +log = logging.getLogger(__name__) + +HELP = """Commands (administrators only): +/upload - process a range of channel messages (asks for start and end links) +/confirm - start the prepared upload task +/cancel - cancel your upload setup or your active task (/cancel ID for a specific task) +/taskstatus - status of upload tasks (/taskstatus ID for one task) +/uploadhistory - recent upload tasks and their results +/status - current processing state +/progress - processed / pending / failed counts +/run - process every discovered channel video +/pause - pause at the next safe checkpoint +/resume - continue after a pause or /stop +/stop - checkpoint the current video and stop +/sync - discover new channel videos and queue them +/retry - re-queue failed items +/failed - list failed items with error summaries +/folders - topic folders and uploaded files +/check ID - details for one Telegram message ID +/help - this message""" + +START_PROMPT = ( + "📥 Start Upload Task\n\n" + "Please send the Telegram link of the first message you want to process.\n\n" + "Example:\nhttps://t.me/c/3909337424/14" +) +END_PROMPT = ( + "📌 Enter the End Link\n\n" + "Now send the Telegram link of the last message you want to process.\n\n" + "Example:\nhttps://t.me/c/3909337424/20" +) +_STEP = { + TaskStatus.AWAITING_START: "waiting for the start link", + TaskStatus.AWAITING_END: "waiting for the end link", + TaskStatus.AWAITING_CONFIRMATION: "waiting for /confirm", +} + +_COMMAND_RE = re.compile(r"^/(?P[A-Za-z_]+)(?:@\w+)?(?:\s+(?P.*))?$", re.DOTALL) +_LINK_HINT = re.compile(r"(?:t\.me|telegram\.me|telegram\.dog)/", re.IGNORECASE) + + +def _short(text: str | None, limit: int = 160) -> str: + if not text: + return "-" + text = " ".join(text.split()) + # never echo filesystem paths back to chat + text = re.sub(r"(?:/[^\s/]+)+/([^\s/]+)", r"…/\1", text) + return text if len(text) <= limit else text[: limit - 1] + "…" + + +class CommandRouter: + def __init__(self, config, db, queue: QueueManager): + self.config = config + self.db = db + self.queue = queue + + def is_admin(self, user_id: int | None) -> bool: + return user_id is not None and user_id in self.config.admin_ids + + async def handle(self, user_id: int | None, text: str) -> str | None: + stripped = (text or "").strip() + match = _COMMAND_RE.match(stripped) + if not match: + if not self.is_admin(user_id): + return None # ignore chatter from non-admins + try: + return await self.on_text(user_id, stripped) + except Exception as exc: + log.exception("Handling a message from %s failed", user_id) + return f"Could not process that message: {public_error(exc)}" + if not self.is_admin(user_id): + log.warning("Rejected command from non-admin user %s", user_id) + return "You are not authorized to control this bot." + command = match.group("cmd").lower() + handler = getattr(self, f"cmd_{command}", None) + if handler is None: + return "Unknown command. Send /help for the list of commands." + try: + return await handler((match.group("arg") or "").strip(), user_id) + except Exception as exc: + log.exception("Command /%s failed", command) + return f"Command failed: {public_error(exc)}" + + # -- /upload conversation -------------------------------------------------------------- + + def _active_setup(self, user_id: int) -> tuple[UploadTask | None, bool]: + """This admin's unfinished setup, expiring it if it is older than the timeout.""" + setups = self.db.tasks_with_status(*SETUP_TASK_STATUSES, admin_user_id=user_id) + if not setups: + return None, False + setup = setups[-1] + for stale in setups[:-1]: # defensive: only one setup per admin may exist + self.db.update_task(stale.id, status=TaskStatus.CANCELLED, last_error="superseded", completed_at=iso(utcnow())) + age = utcnow() - datetime.fromisoformat(setup.updated_at) + if age > timedelta(minutes=self.config.upload_setup_timeout): + self.db.update_task( + setup.id, status=TaskStatus.CANCELLED, last_error="setup expired", completed_at=iso(utcnow()) + ) + log.info("Upload setup #%s of admin %s expired", setup.id, user_id) + return None, True + return setup, False + + async def on_text(self, user_id: int, text: str) -> str | None: + setup, expired = self._active_setup(user_id) + if setup is None: + if _LINK_HINT.search(text): + note = "Your previous upload setup expired. " if expired else "" + return f"{note}To process a message range, send /upload first." + return None + if setup.status == TaskStatus.AWAITING_START: + return await self._receive_start(setup, text) + if setup.status == TaskStatus.AWAITING_END: + return await self._receive_end(setup, text) + return f"Upload Task #{setup.id} is ready. Send /confirm to begin or /cancel to discard it." + + async def _check_link(self, text: str, *, expected_channel_id: int | None = None): + """Return (link, channel_id, title) or an error message.""" + try: + link = parse_message_link(text) + except LinkError as exc: + return str(exc) + source = self.queue.source + if source is None: + return "Telegram access is not available right now." + try: + channel_id, title = await source.resolve_link(link) + except BotError as exc: + return exc.public + if expected_channel_id is not None and channel_id != expected_channel_id: + return "The end link belongs to a different channel than the start link. Both links must come from the same channel." + if ( + expected_channel_id is None + and self.config.upload_restrict_to_source_channel + and channel_id != source.channel_id + ): + return ( + "This link is not from the configured source channel " + f"({source.channel_title or self.config.source_channel}). " + "Set UPLOAD_RESTRICT_TO_SOURCE_CHANNEL=false to allow other channels the bot can access." + ) + try: + accessible = await source.message_accessible(channel_id, link.message_id) + except BotError as exc: + return exc.public + if not accessible: + return f"Message {link.message_id} was not found in that channel (deleted, or not visible to the bot's account)." + return link, channel_id, title + + async def _receive_start(self, setup: UploadTask, text: str) -> str: + result = await self._check_link(text) + if isinstance(result, str): + return HtmlReply(f"❌ {esc(result)}\n\nPlease send a valid start link, for example:\n{EXAMPLE}") + link, channel_id, title = result + self.db.update_task( + setup.id, + status=TaskStatus.AWAITING_END, + channel_ref=link.channel_ref, + channel_id=channel_id, + channel_title=title, + start_message_id=link.message_id, + ) + return HtmlReply(f"Start message: {link.message_id} ✔\n\n{END_PROMPT}") + + async def _receive_end(self, setup: UploadTask, text: str) -> str: + result = await self._check_link(text, expected_channel_id=setup.channel_id) + if isinstance(result, str): + return HtmlReply( + f"❌ {esc(result)}\n\nThe start link (message {setup.start_message_id}) is kept. " + "Please send a valid end link." + ) + link, _, _ = result + if link.message_id < setup.start_message_id: + return HtmlReply( + f"❌ The end message ID ({link.message_id}) must be greater than or equal to the start message ID " + f"({setup.start_message_id}). Please send a valid end link." + ) + size = link.message_id - setup.start_message_id + 1 + if size > self.config.upload_max_range: + return HtmlReply( + f"❌ That range covers {size} messages; the maximum is {self.config.upload_max_range} " + "(UPLOAD_MAX_RANGE). Please send a closer end link." + ) + task = self.db.update_task(setup.id, status=TaskStatus.AWAITING_CONFIRMATION, end_message_id=link.message_id) + return self._confirmation(task) + + def _channel_label(self, task: UploadTask) -> str: + source = self.queue.source + name = task.channel_title or str(task.channel_id) + if source is not None and task.channel_id == source.channel_id: + return f"{name} (configured source channel)" + return name + + def _confirmation(self, task: UploadTask) -> HtmlReply: + return HtmlReply( + "✅ Upload Task Ready\n\n" + f"Start Message ID: {task.start_message_id}\n" + f"End Message ID: {task.end_message_id}\n" + f"Channel: {esc(self._channel_label(task))}\n\n" + "The bot will process eligible videos within this message range, organize them into Google Drive " + "topic folders, and clean up local temporary files after successful uploads.\n\n" + "Send /confirm to begin or /cancel to cancel this task." + ) + + # -- task commands ---------------------------------------------------------------------- + + async def cmd_upload(self, arg: str, user_id: int) -> str: + setup, expired = self._active_setup(user_id) + if setup is not None: + hint = { + TaskStatus.AWAITING_START: "Send the start link to continue.", + TaskStatus.AWAITING_END: f"Start message {setup.start_message_id} is set; send the end link to continue.", + TaskStatus.AWAITING_CONFIRMATION: "Send /confirm to begin.", + }[setup.status] + return ( + f"⚠️ You already have an upload setup in progress (Task #{setup.id}, {_STEP[setup.status]}).\n" + f"{hint}\nOr send /cancel to discard it and start again." + ) + self.db.create_task(user_id) + prefix = "Your previous unfinished setup expired.\n\n" if expired else "" + return HtmlReply(prefix + START_PROMPT) + + async def cmd_confirm(self, arg: str, user_id: int) -> str: + setup, expired = self._active_setup(user_id) + if setup is None: + note = "Your previous setup expired. " if expired else "" + return f"{note}There is no upload task waiting for confirmation. Start one with /upload." + if setup.status != TaskStatus.AWAITING_CONFIRMATION: + return ( + "Both the start and end links must be validated before /confirm. " + f"Task #{setup.id} is {_STEP[setup.status]}." + ) + task = self.queue.submit_task(setup.id) + ahead = [t for t in self.db.tasks_with_status(*ACTIVE_TASK_STATUSES) if t.id < task.id] + lines = [f"🚀 Upload Task #{task.id} confirmed", f"Range: {task.range_text}"] + if self.queue.paused_flag: + lines.append("The queue is paused; send /resume to start processing.") + elif ahead: + lines.append("Queued after task(s) " + ", ".join(f"#{t.id}" for t in ahead) + ".") + else: + lines.append("Processing has started. Use /taskstatus to follow it.") + return HtmlReply("\n".join(lines)) + + async def cmd_cancel(self, arg: str, user_id: int) -> str: + if arg: + if not arg.isdigit(): + return "Usage: /cancel or /cancel TASK_ID" + return self._cancel_reply(int(arg), await self.queue.cancel_task(int(arg))) + setup, _ = self._active_setup(user_id) + if setup is not None: + await self.queue.cancel_task(setup.id, reason="setup cancelled") + return f"Upload setup #{setup.id} cancelled. Nothing was downloaded or uploaded." + own = self.db.tasks_with_status(*ACTIVE_TASK_STATUSES, admin_user_id=user_id) + if own: + running = [t for t in own if t.status == TaskStatus.RUNNING] + target = running[0] if running else own[-1] + return self._cancel_reply(target.id, await self.queue.cancel_task(target.id)) + others = self.db.tasks_with_status(*ACTIVE_TASK_STATUSES) + if others: + ids = ", ".join(f"#{t.id}" for t in others) + return f"You have nothing to cancel. Active task(s) {ids} belong to another administrator; use /cancel ID to cancel one explicitly." + return "Nothing to cancel." + + @staticmethod + def _cancel_reply(task_id: int, outcome: str) -> str: + return { + "unknown": f"Task #{task_id} does not exist.", + "finished": f"Task #{task_id} has already finished.", + "setup_cancelled": f"Upload setup #{task_id} cancelled.", + "cancelled": ( + f"🛑 Task #{task_id} cancelled at a safe checkpoint. Completed uploads are kept; " + "the remaining videos were not processed and their temporary files were removed." + ), + }[outcome] + + async def cmd_taskstatus(self, arg: str, user_id: int) -> str: + if arg: + if not arg.isdigit(): + return "Usage: /taskstatus or /taskstatus TASK_ID" + task = self.db.get_task(int(arg)) + return self._describe_task(task) if task else f"Task #{arg} does not exist." + parts = [] + setup, expired = self._active_setup(user_id) + if setup is not None: + parts.append( + f"Your setup: Task #{setup.id}, {_STEP[setup.status]}" + + (f" (start {setup.start_message_id})" if setup.start_message_id else "") + ) + elif expired: + parts.append("Your previous setup expired.") + active = self.db.tasks_with_status(*ACTIVE_TASK_STATUSES) + parts += [self._describe_task(t) for t in active] + if not active: + recent = self.db.recent_tasks(1) + parts.append("No active upload tasks." + (f"\n\nLast task:\n{self._describe_task(recent[0])}" if recent else "")) + return HtmlReply("\n\n".join(parts)) + + def _describe_task(self, task: UploadTask) -> HtmlReply: + lines = [ + f"Task #{task.id} — {esc(task.status.value)}", + f"Range: {task.range_text}", + f"Channel: {esc(self._channel_label(task)) if task.channel_id else '-'}", + f"Requested by: {task.admin_user_id}", + ] + if task.status not in SETUP_TASK_STATUSES: + lines += [ + f"Eligible videos: {task.eligible_count} · Uploaded: {task.uploaded_count} · " + f"Duplicates: {task.duplicate_count} · Failed: {task.failed_count}", + f"Non-video skipped: {task.non_video_count} · Other skipped: {task.skipped_count} · " + f"Missing: {task.missing_count}", + f"Temporary files deleted: {task.deleted_files_count}", + ] + if task.current_message_id and task.status == TaskStatus.RUNNING: + lines.append(f"Current message: {task.current_message_id}") + board = self.queue.reporter.board if self.queue.reporter else None + if board and board.task and board.task.get("id") == task.id: + lines.append(f"Video: {esc(board.title)} · {esc(board.status)}") + if board.upload: + lines.append(f"Upload: {esc(board.upload.describe())}") + elif board.download: + lines.append(f"Download: {esc(board.download.describe())}") + for label, value in (("Created", task.created_at), ("Started", task.started_at), ("Finished", task.completed_at)): + if value: + lines.append(f"{label}: {value}") + if task.last_error: + lines.append(f"Note: {esc(_short(task.last_error))}") + return HtmlReply("\n".join(lines)) + + async def cmd_uploadhistory(self, arg: str, user_id: int) -> str: + tasks = self.db.recent_tasks(10) + if not tasks: + return "No upload tasks yet. Start one with /upload." + lines = ["Recent upload tasks"] + for t in tasks: + lines.append( + f"#{t.id} {t.range_text} {esc(t.status.value)} — uploaded {t.uploaded_count}, " + f"duplicates {t.duplicate_count}, failed {t.failed_count}, non-video {t.non_video_count} " + f"(by {t.admin_user_id})" + ) + return HtmlReply("\n".join(lines)) + + # -- general commands --------------------------------------------------------------------- + + async def cmd_start(self, arg: str, user_id: int) -> str: + return ( + "Telegram -> Google Drive video organizer.\n" + "Videos are uploaded into Drive folders named after the caption Topic, " + "using the caption Title as the file name.\n\n" + "Process a message range: /upload, send the start and end links, then /confirm.\n" + "Process the whole channel: /sync then /run.\n\n" + HELP + ) + + async def cmd_help(self, arg: str, user_id: int) -> str: + return HELP + + async def cmd_status(self, arg: str, user_id: int) -> str: + counts = self.db.counts() + lines = [f"Queue: {self.queue.status}"] + if self.queue.halt_reason: + lines.append(f"Halted because: {self.queue.halt_reason}") + active = self.db.tasks_with_status(*ACTIVE_TASK_STATUSES) + if active: + lines.append("Upload tasks: " + ", ".join(f"#{t.id} {t.status.value} ({t.range_text})" for t in active)) + current = self.queue.current + if current and self.queue.reporter: + lines.append("") + lines.append(re.sub(r"", "", self.queue.reporter.board.render())) + elif current: + lines.append(f"Current: message {current.message_id} ({current.title or 'untitled'})") + lines.append("") + lines.append( + f"Completed {counts['COMPLETED']} · Pending {counts['PENDING'] + counts['RETRY_WAIT']} · " + f"Failed {counts['FAILED']} · Skipped {counts['SKIPPED']}" + ) + lines.append(f"Whole-channel processing (/run): {'on' if self.queue.global_enabled else 'off'}") + lines.append(f"Live channel listener: {'on' if self.queue.listening else 'off'}") + return "\n".join(lines) + + async def cmd_progress(self, arg: str, user_id: int) -> str: + c = self.db.counts() + total = sum(c.values()) + processed = c["COMPLETED"] + remaining = self.queue.remaining() + considered = total - c["SKIPPED"] + pct = f"{100 * processed / considered:.1f}%" if considered else "n/a" + return ( + f"Total discovered: {total}\n" + f"Processed (uploaded & verified): {processed} ({pct})\n" + f"Pending: {c['PENDING']}\n" + f"Waiting to retry: {c['RETRY_WAIT']}\n" + f"In progress: {c['DOWNLOADING'] + c['DOWNLOADED'] + c['UPLOADING']}\n" + f"Failed: {c['FAILED']}\n" + f"Skipped: {c['SKIPPED']}\n" + f"Remaining: {remaining}" + ) + + async def cmd_run(self, arg: str, user_id: int) -> str: + if self.queue.start(): + pending = self.queue.remaining() + hint = "" if pending else "\nThe queue is empty - use /sync to discover channel videos." + return f"Processing started ({pending} item(s) waiting).{hint}" + return f"Processing is already {self.queue.status}." + + async def cmd_pause(self, arg: str, user_id: int) -> str: + if self.queue.pause(): + return "Pausing: the current transfer stops at the next resumable checkpoint. Use /resume to continue." + return f"Nothing to pause (queue is {self.queue.status})." + + async def cmd_resume(self, arg: str, user_id: int) -> str: + if self.queue.resume(): + return "Processing resumed." + return f"Nothing to resume (queue is {self.queue.status})." + + async def cmd_stop(self, arg: str, user_id: int) -> str: + await self.queue.stop(timeout=60) + return ( + "Stopped. Interrupted work was checkpointed; a running upload task was paused. " + "Use /resume to continue upload tasks or /run for whole-channel processing." + ) + + async def cmd_retry(self, arg: str, user_id: int) -> str: + count = self.queue.retry_failed() + if not count: + return "No failed items to retry." + started = self.queue.start() + return f"Re-queued {count} failed item(s)." + (" Processing started." if started else "") + + async def cmd_failed(self, arg: str, user_id: int) -> str: + items = self.db.list_by_state(State.FAILED, limit=20) + if not items: + return "No failed items." + lines = [f"Failed items (showing {len(items)}):"] + for rec in items: + lines.append(f"• #{rec.message_id} {_short(rec.title, 60)} — {_short(rec.last_error)}") + lines.append("Use /retry to re-queue them, /check ID for details.") + return "\n".join(lines) + + async def cmd_folders(self, arg: str, user_id: int) -> str: + topics = self.db.topic_counts() + counts = self.db.counts() + lines = [f"Drive folders created/reused: {self.db.folder_count()}", f"Files uploaded: {counts['COMPLETED']}"] + if topics: + lines.append("") + lines += [f"• {topic or self.config.fallback_topic}: {n}" for topic, n in topics] + return "\n".join(lines) + + async def cmd_check(self, arg: str, user_id: int) -> str: + if not arg.lstrip("-").isdigit(): + return "Usage: /check MESSAGE_ID" + records = self.db.find_by_message_id(int(arg)) + if not records: + return f"Message {arg} is not in the database. Use /sync or /upload to discover messages." + return "\n\n".join(self._describe(r) for r in records) + + async def cmd_sync(self, arg: str, user_id: int) -> str: + result = await self.queue.sync(rescan=arg.lower() == "all") + if self.queue.running: + self.queue.wake() + return "Sync complete.\n" + result.summary() + + def _describe(self, rec: VideoRecord) -> str: + lines = [ + f"Message {rec.message_id} — {rec.state}", + f"Title: {rec.title or '-'}", + f"Topic: {rec.topic or '- (uses ' + self.config.fallback_topic + ')'}", + f"Batch: {rec.batch or '-'}", + f"Index: {', '.join(rec.indexes) or '-'}", + f"Extracted by: {rec.extracted_by or '-'}", + f"Download: {rec.download_status or '-'} · Upload: {rec.upload_status or '-'}", + f"Size: {rec.expected_size or rec.local_size or '?'} bytes", + f"Attempts: {rec.attempts}", + ] + if rec.task_id: + lines.append(f"Upload task: #{rec.task_id}") + if rec.final_name: + lines.append(f"Drive name: {rec.final_name}") + if rec.drive_file_id: + lines.append(f"Drive file ID: {rec.drive_file_id}") + if rec.next_retry_at and rec.state == State.RETRY_WAIT: + lines.append(f"Next retry: {rec.next_retry_at}") + if rec.skip_reason: + lines.append(f"Skip reason: {_short(rec.skip_reason)}") + if rec.last_error: + lines.append(f"Last error: {_short(rec.last_error)}") + if rec.warnings: + lines.append(f"Warnings: {_short('; '.join(rec.warnings), 300)}") + return "\n".join(lines) + + +def _parse_mode(text: str) -> str | None: + return "html" if isinstance(text, HtmlReply) else None + + +class TelethonNotifier: + def __init__(self, client): + self.client = client + + async def send(self, chat_id: int, text: str) -> int | None: + message = await self.client.send_message(chat_id, str(text), parse_mode=_parse_mode(text), link_preview=False) + return message.id + + async def edit(self, chat_id: int, message_id: int, text: str) -> None: + from telethon import errors + + try: + await self.client.edit_message(chat_id, message_id, str(text), parse_mode=_parse_mode(text), link_preview=False) + except errors.MessageNotModifiedError: + pass + + +def register_command_handlers(client, router: CommandRouter) -> None: + from telethon import events + + @client.on(events.NewMessage(incoming=True, func=lambda e: e.is_private)) + async def _on_private_message(event) -> None: # pragma: no cover - needs a live Telegram connection + reply = await router.handle(event.sender_id, event.raw_text) + if not reply: + return + if isinstance(reply, HtmlReply): + await event.respond(str(reply), parse_mode="html", link_preview=False) + return + for start in range(0, len(reply), 4000): + await event.respond(reply[start : start + 4000], parse_mode=None, link_preview=False) diff --git a/caption_parser.py b/caption_parser.py new file mode 100644 index 0000000..483b8ca --- /dev/null +++ b/caption_parser.py @@ -0,0 +1,108 @@ +"""Extract structured metadata from a Telegram video caption. + +Expected format (fields may appear in any order, with extra blank lines, emoji or +bullet decorations, and varying whitespace):: + + Index: 001 + Index: 002 + Title: Class 01 Aman sir.mp4 + Topic: Misc• Grammar Practice + Batch: English ( SSC MAINS ) + Extracted By: @PUTINxINDIA + +Parsing is line based, so the value of one field can never swallow the next field. +If a single line holds several known fields (captions flattened by a forwarder), +the line is split at each known label. +""" +from __future__ import annotations + +import re + +from models import ParsedCaption + +REQUIRED_FIELDS = ("title", "topic") + +_LABEL = r"(?P