diff --git a/ENV_SETUP.md b/ENV_SETUP.md index eabfe00..0ec9a32 100644 --- a/ENV_SETUP.md +++ b/ENV_SETUP.md @@ -114,6 +114,8 @@ Another way: forward a channel post to [@JsonDumpBot](https://t.me/JsonDumpBot) SOURCE_CHANNEL_ID=-1001234567890 ``` +> The same post links are what you send to the bot's `/upload` command to pick a message range. + > 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. @@ -351,6 +353,11 @@ These are preferences, not credentials. Defaults are shown; full descriptions ar | `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 | +| `UPLOAD_RESTRICT_TO_SOURCE_CHANNEL` | `true` | `/upload` links must point to `SOURCE_CHANNEL_ID`; `false` allows any channel the account/bot can access | +| `UPLOAD_SETUP_TIMEOUT` | `30` | Minutes before an unfinished `/upload` setup expires | +| `UPLOAD_MAX_RANGE` | `5000` | Largest number of message IDs in one `/upload` task | +| `KEEP_FAILED_DOWNLOADS` | `false` | Keep the local file of a FAILED video instead of deleting it | +| `TEMP_CLEANUP_INTERVAL` | `3600` | Seconds between sweeps that delete abandoned temporary files | --- diff --git a/README.md b/README.md index a147b17..f1779f9 100644 --- a/README.md +++ b/README.md @@ -8,7 +8,8 @@ > 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 +uploads it to Google Drive. You can process the whole channel (`/sync` + `/run`) or just a range of +posts with the interactive **`/upload`** command (send a start link and an end link, then `/confirm`). 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. @@ -51,6 +52,11 @@ the temp file → next video.** - 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. +- **Interactive `/upload` tasks:** pick an inclusive message range by pasting two post links, + confirm, and watch a live task progress message. Tasks are stored in the database and survive restarts. +- **Automatic temp-file cleanup:** each video's local file is deleted right after its upload is + verified in Drive *and* recorded in the database. Abandoned temp files are swept at startup, + periodically, and whenever disk space runs low. - 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. @@ -216,6 +222,7 @@ The Heroku button only works with **public** repositories. |---|---|---| | 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`) | +| `/upload` message ranges | ✅ | ✅ messages are fetched by ID, which bots that are channel admins may do | | 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 | @@ -302,6 +309,11 @@ Set these in the dashboard (*Settings → Reveal Config Vars*) or with | `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 | +| `UPLOAD_RESTRICT_TO_SOURCE_CHANNEL` | no (`true`) | `/upload` links must point to `SOURCE_CHANNEL_ID` | +| `UPLOAD_SETUP_TIMEOUT` | no (`30`) | minutes before an unfinished `/upload` setup expires | +| `UPLOAD_MAX_RANGE` | no (`5000`) | largest number of message IDs per task | +| `KEEP_FAILED_DOWNLOADS` | no (`false`) | keep local files of FAILED videos | +| `TEMP_CLEANUP_INTERVAL` | no (`3600`) | seconds between abandoned-temp-file sweeps | 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). @@ -332,8 +344,9 @@ Production state lives in **PostgreSQL**: 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. +3. Either send `/upload` to process a message range (see [§13](#13-telegram-commands)), or send + `/sync` to discover all channel videos and then `/run`. Watch `/taskstatus` or `/progress`. +4. Running tasks and the `/run` choice are stored. After any dyno restart the bot resumes on its own. `/pause` and `/stop` are also remembered. ## 13. Telegram commands @@ -354,8 +367,82 @@ Send these to the bot in a **private chat**. Users not in `ADMIN_USER_IDS` are r | `/failed` | Failed items with short error summaries | | `/folders` | Folder count and uploads per topic | | `/check ID` | Full record for one Telegram message ID | +| `/upload` | Start an interactive upload task for a message range | +| `/confirm` | Start the prepared task (only after both links were validated) | +| `/cancel` | Discard your setup, or cancel your active task at a safe checkpoint. `/cancel ID` targets a specific task. | +| `/taskstatus` | Your setup step, plus active tasks with range, counters and live progress. `/taskstatus ID` for one task. | +| `/uploadhistory` | The 10 most recent tasks and their results | -Progress is a single message, edited every `PROGRESS_INTERVAL` seconds: +`/pause`, `/resume` and `/stop` also apply to upload tasks. `/stop` parks a running task as PAUSED; +`/resume` continues it. + +### Processing a message range with `/upload` + +1. Send `/upload`. The bot asks for the **start link**: in Telegram, right-click (or long-press) a + post → **Copy Post Link**, e.g. `https://t.me/c/3909337424/14`. +2. Send the link. The bot asks for the **end link**, e.g. `https://t.me/c/3909337424/20`. +3. The bot shows *Upload Task Ready* with both message IDs and the channel. +4. Send `/confirm` to start, or `/cancel` to discard. Nothing is downloaded before `/confirm`. + +Link rules: +- Accepted formats: `t.me/c//` (private), `t.me//` (public), and + `t.me/s//`. `www.`, `telegram.me`, a trailing slash and `?single` are fine. +- Invite links and links without a message ID are rejected with an explanation. +- The `/c/` number is Telegram's internal channel ID. It is converted to the API ID + `-(10¹² + id)` and then **resolved through the bot's Telegram account**. A channel that account + can't access is refused; nothing bypasses membership or permissions. +- Both links must be from the same channel, the end ID must be ≥ the start ID, the range is + limited by `UPLOAD_MAX_RANGE`, and both messages must exist. +- By default the links must point to `SOURCE_CHANNEL_ID`. Set + `UPLOAD_RESTRICT_TO_SOURCE_CHANNEL=false` to allow any channel the account (or admin bot) can read. +- If the end link is invalid, the start link is kept; just send another end link. + +How a task runs: +- **The range is inclusive** (14–20 means 14, 15, …, 20), processed in ascending order, one video at a time. +- **Text, photos, stickers and deleted messages** are counted as skipped, never as failures. +- **Videos already uploaded earlier** are counted as duplicates and are not uploaded again. + Repeating a finished task therefore uploads nothing new. +- **Caption metadata** (Title, Topic, …) is used exactly as for whole-channel processing. +- **Each local file is deleted** only after the Drive upload is verified *and* its COMPLETED state is committed. +- **Failed videos** are recorded, never marked completed, and can be re-queued with `/retry`. + Their local file is deleted (unless `KEEP_FAILED_DOWNLOADS=true`) because it can be downloaded again. +- **`/cancel` during processing** stops at the next safe checkpoint: + - finished uploads stay in Drive; + - the remaining videos are marked *cancelled* and their temp files are removed; + - a later task covering the same range picks the cancelled videos up again. +- **Setups are per administrator.** Two admins can prepare tasks at the same time without + affecting each other. A second `/upload` never overwrites your unfinished setup. A setup expires + after `UPLOAD_SETUP_TIMEOUT` minutes. +- **Confirmed tasks run one after another** in the order they were confirmed. Running and queued + tasks resume automatically after a restart or redeploy. + +Task progress is one message, edited in place: + +``` +🚀 Upload Task Running + +Task: #12 +Range: 14–20 +Current Message: 16 +Current Video: Class 01 Aman sir.mp4 +Topic: Misc• Grammar Practice +Status: Uploading to Google Drive +Upload: 45% (540.0 MB / 1.2 GB) · 18.3 MB/s · ETA 36s + +Completed: 1 +Skipped: 1 +Failed: 0 +``` + +When the task finishes, the bot sends a summary covering: +- the range; +- messages examined; +- eligible videos; +- uploads, duplicates, non-video messages and failures; +- temp files deleted; +- the Drive folders used. + +For whole-channel processing, progress is a single message, edited every `PROGRESS_INTERVAL` seconds: ``` Processing video 12 of 150 @@ -437,6 +524,10 @@ pipeline is intentionally sequential. 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. +- **Temporary files:** downloads live only in `DOWNLOAD_DIR`. The bot refuses to use `/`, `/tmp`, + your home directory or the project directory itself. Cleanup only removes files named + `_.part|.video` directly inside that directory, never follows symlinks, and + re-checks the database first. Files of unfinished items are kept for resuming. - **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. @@ -449,7 +540,7 @@ pipeline is intentionally sequential. ### 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. +- **Automated tests (259):** 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 @@ -488,6 +579,15 @@ both backends. Without it, only SQLite is tested. The suite covers: - a successful upload followed by a database-write failure; - sequential processing, pause/resume and graceful shutdown; - admin-only commands; +- the `/upload` workflow: + - link parsing and validation; + - per-admin setups, persistence and expiry; + - inclusive ascending ranges, and skipping non-video messages; + - delete-only-after-verified-upload, and failed uploads never completed; + - repeat tasks without duplicates; + - cancellation; + - crash/restart recovery and pause/resume; +- temp-file cleanup safety; - configuration and Heroku validation, env-var credentials, log redaction; - `app.json` / Procfile consistency. @@ -508,3 +608,11 @@ both backends. Without it, only SQLite is tested. The suite covers: 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`. +7. `/upload`: send the link of the first test post, then the last one, then `/confirm`. Check that: + - only that range is processed; + - a text post inside the range is reported as a skipped non-video; + - the summary counts are correct. + Run the same range again and confirm it reports duplicates and uploads nothing. +8. Start another `/upload` task and send `/cancel` while a video is downloading. Check that: + - finished uploads remain in Drive; + - `downloads/` is empty. diff --git a/config.py b/config.py index 6fdf3b9..556e258 100644 --- a/config.py +++ b/config.py @@ -301,8 +301,11 @@ class Config: config.sequence_prefix_format.format(seq=1) except (KeyError, IndexError, ValueError): r.problems.append("SEQUENCE_PREFIX_FORMAT must be a format string using {seq}, e.g. '{seq:03d} - '") + import tempfile + resolved = config.download_dir.resolve() - if resolved in {Path("/").resolve(), Path.home().resolve(), PROJECT_DIR.resolve()} or len(resolved.parts) < 3: + shared = {Path(p).resolve() for p in ("/", "/tmp", "/var/tmp", tempfile.gettempdir(), Path.home(), PROJECT_DIR)} + if resolved in shared or len(resolved.parts) < 3: r.problems.append( f"DOWNLOAD_DIR={config.download_dir} is not a dedicated directory; use e.g. 'downloads' or '/tmp/telegram_drive_bot'" ) diff --git a/tests/test_database.py b/tests/test_database.py index 3607d49..6c5a6e2 100644 --- a/tests/test_database.py +++ b/tests/test_database.py @@ -153,3 +153,31 @@ def test_instance_lock_blocks_second_worker(config, db): third = InstanceLock(config) third.acquire() third.release() + + +def test_upgrade_from_schema_v1_keeps_data(config, db_url): + """An existing v1 deployment (before /upload tasks) migrates in place without data loss.""" + import database as dbmod + + v2 = dbmod.MIGRATIONS[1] + dbmod.MIGRATIONS.pop() # simulate the old code base + try: + old = Database.from_config(config) + with old._tx() as conn: # a row exactly as the v1 code wrote it + conn.execute( + "INSERT INTO videos (channel_id, message_id, state, indexes, title, discovered_at, updated_at) " + "VALUES (?,?,?,?,?,?,?)", + (CHANNEL_ID, 5, "DOWNLOADING", '["001", "002"]', "Class 01 Aman sir.mp4", "2026-01-01T00:00:00+00:00", + "2026-01-01T00:00:00+00:00"), + ) + assert old.get_kv("schema_version") == "1" + old.close() + finally: + dbmod.MIGRATIONS.append(v2) + new = Database.from_config(config) + assert new.get_kv("schema_version") == "2" + rec = new.get((CHANNEL_ID, 5)) + assert rec.state == State.DOWNLOADING and rec.task_id is None and rec.indexes == ["001", "002"] + task = new.create_task(111) + assert task.id == 1 and new.next_item(task_id=task.id) is None + new.close() diff --git a/tests/test_storage_cleanup.py b/tests/test_storage_cleanup.py new file mode 100644 index 0000000..967bd2f --- /dev/null +++ b/tests/test_storage_cleanup.py @@ -0,0 +1,134 @@ +"""Temporary-storage safety: cleanup never touches anything outside the app's temp files.""" +import os +from pathlib import Path + +import pytest + +from caption_parser import parse_caption +from config import ConfigError +from conftest import CHANNEL_ID, FakeDrive, FakeSource, build_queue, make_config, run, video_bytes +from downloader import Downloader +from models import State + +PREFIX = abs(CHANNEL_ID) + + +def _record(db, source, message_id, state): + msg = source.add(message_id, f"Title: V{message_id}.mp4\nTopic: T") + db.insert_video(msg, parse_caption(msg.caption)) + path = { + State.COMPLETED: [State.DOWNLOADING, State.DOWNLOADED, State.UPLOADING, State.COMPLETED], + State.FAILED: [State.FAILED], + State.SKIPPED: [State.SKIPPED], + State.DOWNLOADED: [State.DOWNLOADING, State.DOWNLOADED], + State.UPLOADING: [State.DOWNLOADING, State.DOWNLOADED, State.UPLOADING], + State.PENDING: [], + }[state] + for step in path: + db.transition((CHANNEL_ID, message_id), step) + + +def test_cleanup_only_removes_finished_app_temp_files(config, db, source, tmp_path): + downloader = Downloader(source, config.download_dir) + d = config.download_dir + for message_id, state in [(1, State.COMPLETED), (2, State.SKIPPED), (3, State.FAILED), (4, State.PENDING), + (5, State.DOWNLOADED), (6, State.UPLOADING)]: + _record(db, source, message_id, state) + files = { + "done": d / f"{PREFIX}_1.video", + "skipped": d / f"{PREFIX}_2.part", + "failed": d / f"{PREFIX}_3.video", + "pending_partial": d / f"{PREFIX}_4.part", + "downloaded": d / f"{PREFIX}_5.video", + "downloaded_stale_part": d / f"{PREFIX}_5.part", + "uploading": d / f"{PREFIX}_6.video", + "unknown_record": d / f"{PREFIX}_99.video", + "user_file": d / "notes.txt", + "lookalike": d / f"{PREFIX}_1.video.bak", + } + for path in files.values(): + path.write_bytes(b"x") + outside = tmp_path / "outside" / f"{PREFIX}_1.video" + outside.parent.mkdir() + outside.write_bytes(b"precious") + (d / f"{PREFIX}_7.video").symlink_to(outside) # matching name, but a symlink pointing outside + subdir = d / f"{PREFIX}_8.video" + subdir.mkdir() # directories are never removed + + deleted = downloader.cleanup_orphans(db) + + gone = {k for k, p in files.items() if not p.exists()} + assert gone == {"done", "skipped", "failed", "downloaded_stale_part", "unknown_record"} + assert deleted == 5 + assert outside.read_bytes() == b"precious" + assert (d / f"{PREFIX}_7.video").is_symlink() and subdir.is_dir() + + +def test_failed_files_can_be_kept(config, db, source): + downloader = Downloader(source, config.download_dir) + _record(db, source, 3, State.FAILED) + path = config.download_dir / f"{PREFIX}_3.video" + path.write_bytes(b"x") + assert downloader.cleanup_orphans(db, keep_failed=True) == 0 and path.exists() + + +def test_safe_unlink_refuses_anything_outside_temp_dir(config, source, tmp_path): + downloader = Downloader(source, config.download_dir) + outside = tmp_path / f"{PREFIX}_1.video" + outside.write_bytes(b"keep me") + assert downloader.safe_unlink(outside) is False and outside.exists() + traversal = config.download_dir / ".." / f"{PREFIX}_1.video" + assert downloader.safe_unlink(traversal) is False and outside.exists() + odd = config.download_dir / "important.db" + odd.write_bytes(b"x") + assert downloader.safe_unlink(odd) is False and odd.exists() + + +def test_low_disk_space_reclaims_abandoned_files_before_failing(config, db, source): + drive = FakeDrive() + _record(db, source, 1, State.COMPLETED) + orphan = config.download_dir + orphan.mkdir(parents=True, exist_ok=True) + orphan = orphan / f"{PREFIX}_1.video" + orphan.write_bytes(b"x" * 10) + source.add(2, "Title: New.mp4\nTopic: T", video_bytes(600_000, 2)) + + def disk_usage(_): + free = 10**12 if not orphan.exists() else 0 # only enough space once the orphan is gone + return (10**12, 10**12 - free, free) + + queue = build_queue(config, db, source, drive, disk_usage=disk_usage) + + async def go(): + await queue.sync() + queue.start() + await queue.wait() + + run(go()) + assert not orphan.exists() + assert db.get((CHANNEL_ID, 2)).state == State.COMPLETED + + +def test_permanent_download_failure_removes_partial(config, db, source): + from errors import PermanentError + + drive = FakeDrive() + source.add(1, "Title: A.mp4\nTopic: T", video_bytes(3 * 1024 * 1024, 1)) + source.failures[1] = [("after", 2 * 1024 * 1024, PermanentError("message deleted mid-download"))] + queue = build_queue(config, db, source, drive) + + async def go(): + await queue.sync() + queue.start() + await queue.wait() + + run(go()) + assert db.get((CHANNEL_ID, 1)).state == State.FAILED + assert list(config.download_dir.iterdir()) == [] + + +@pytest.mark.parametrize("bad", ["/", str(Path.home()), "/tmp"]) +def test_download_dir_must_be_dedicated(tmp_path, bad): + with pytest.raises(ConfigError) as err: + make_config(tmp_path, DOWNLOAD_DIR=bad) + assert "DOWNLOAD_DIR" in str(err.value)