feat: add channel_last_message_id_overrides for one-time sync cursor override
This commit is contained in:
18
README.md
18
README.md
@@ -128,9 +128,25 @@ docker compose logs -f telegram-backup
|
||||
| `status_port` | No | `0` | HTTP port; `0` disables the server. |
|
||||
| `db_path` | No | `telegram_state.db` | SQLite state database path. |
|
||||
| `channel_overrides` | No | `{}` | Per-channel controls. Only `turnon` is supported. Use `turnon: false` to skip a channel without removing it from `channels`. Example: `{"-100321012345":{"turnon":false}}`. Change is detected by config watcher — no restart needed. |
|
||||
| `channel_last_message_id_overrides` | No | `{}` | One-time per-channel sync cursor overrides. With `updateonce: true`, sets the channel's SQLite `last_message_id` to the specified value, then automatically changes `updateonce` to `false` in `config.json`. Use numeric Telegram peer IDs that are included in `channels`. |
|
||||
| `manual_downloads` | No | `{}` | Message IDs to prioritize, keyed by channel ID. Useful to force-retry specific messages. Example: `{"-1001":[42,43,44]}`. Change is detected by config watcher — no restart needed. |
|
||||
|
||||
`config.json` is watched every 10 seconds. Changes to channels, overrides, and manual downloads are applied without restarting. Do not set `status_port`, `db_path`, worker count, or API credentials expecting a live process to rebind/recreate those resources; restart after changing them.
|
||||
`config.json` is watched every 10 seconds. Changes to channels, overrides, cursor overrides, and manual downloads are applied without restarting. Do not set `status_port`, `db_path`, worker count, or API credentials expecting a live process to rebind/recreate those resources; restart after changing them.
|
||||
|
||||
### One-Time Sync Cursor Override
|
||||
|
||||
Use `channel_last_message_id_overrides` to resume a channel from a known message ID or re-download messages after a cursor correction. The channel must be present in `channels`. When the running downloader detects `updateonce: true`, it updates the database cursor and writes `updateonce: false` only after the database change succeeds.
|
||||
|
||||
```json
|
||||
"channel_last_message_id_overrides": {
|
||||
"-1004295572354": {
|
||||
"updateonce": true,
|
||||
"last_message_id": 21321
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
The next sync starts from a small overlap before that cursor, so existing message and media records prevent duplicate output.
|
||||
|
||||
## HTTP API
|
||||
|
||||
|
||||
@@ -4,7 +4,7 @@
|
||||
"api_hash": "your_api_hash_here",
|
||||
"phone_number": "+1234567890",
|
||||
"session_name": "telegram_session",
|
||||
"channels": ["@channel_username", "-1001234567890"],
|
||||
"channels": ["@channel_username", "-1001231231231", "-1003213213213"],
|
||||
"parallel_downloads": 3,
|
||||
"download_timeout_seconds": 600,
|
||||
"download_retry_count": 3,
|
||||
@@ -16,7 +16,20 @@
|
||||
"retry_drop_log": true,
|
||||
"resync_interval_minutes": 60,
|
||||
"status_port": 8080,
|
||||
"db_path": "data/telegram_state.db",
|
||||
"channel_overrides": {},
|
||||
"db_path": "/app/data/telegram_state.db",
|
||||
"channel_overrides": {
|
||||
"-1001231231231":{"turnon":false},
|
||||
"-1003213213213":{"turnon":true}
|
||||
},
|
||||
"channel_last_message_id_overrides": {
|
||||
"-1001231231231": {
|
||||
"updateonce": false,
|
||||
"last_message_id": 123123
|
||||
},
|
||||
"-1003213213213": {
|
||||
"updateonce": true,
|
||||
"last_message_id": 321321
|
||||
}
|
||||
},
|
||||
"manual_downloads": {}
|
||||
}
|
||||
}
|
||||
@@ -1,13 +1,14 @@
|
||||
version: "3.9"
|
||||
|
||||
services:
|
||||
telegram-backup:
|
||||
build: .
|
||||
image: hoelee/telegram-backup-downloader:latest
|
||||
container_name: telegram-backup
|
||||
restart: unless-stopped
|
||||
user: root
|
||||
volumes:
|
||||
- ./config.json:/app/config.json:ro
|
||||
- ./telegram_session.session:/app/telegram_session.session
|
||||
- ./data:/app/data
|
||||
- ./channels:/app/channels
|
||||
- ./logs:/app/logs
|
||||
ports:
|
||||
@@ -19,4 +20,4 @@ services:
|
||||
interval: 30s
|
||||
timeout: 5s
|
||||
retries: 3
|
||||
start_period: 15s
|
||||
start_period: 15s
|
||||
@@ -125,6 +125,17 @@ def validate_config(cfg):
|
||||
raise ValueError("Config key 'api_id' must be a positive integer")
|
||||
if not all(isinstance(channel, (str, int)) for channel in cfg["channels"]):
|
||||
raise ValueError("Config key 'channels' must contain strings or integers")
|
||||
overrides = cfg.get("channel_last_message_id_overrides", {})
|
||||
if not isinstance(overrides, dict):
|
||||
raise ValueError("Config key 'channel_last_message_id_overrides' must be an object")
|
||||
for channel_id, override in overrides.items():
|
||||
if not isinstance(channel_id, str) or not isinstance(override, dict):
|
||||
raise ValueError("Each channel_last_message_id_overrides entry must have a string channel ID and object value")
|
||||
if not isinstance(override.get("updateonce"), bool):
|
||||
raise ValueError(f"Override for channel '{channel_id}' must include boolean 'updateonce'")
|
||||
message_id = override.get("last_message_id")
|
||||
if not isinstance(message_id, int) or isinstance(message_id, bool) or message_id < 0:
|
||||
raise ValueError(f"Override for channel '{channel_id}' must include a non-negative integer 'last_message_id'")
|
||||
|
||||
|
||||
def sanitize_filename(name):
|
||||
@@ -192,6 +203,27 @@ async def update_last_message_id(channel_id, message_id):
|
||||
await db.commit()
|
||||
|
||||
|
||||
async def apply_last_message_id_overrides():
|
||||
overrides = CONFIG.get("channel_last_message_id_overrides", {})
|
||||
changed = False
|
||||
for channel_id, override in overrides.items():
|
||||
if not override["updateonce"]:
|
||||
continue
|
||||
if channel_id not in MONITORED_CHANNEL_IDS:
|
||||
logger.warning("[CURSOR OVERRIDE] Channel %s is not resolved; will retry", channel_id)
|
||||
continue
|
||||
message_id = override["last_message_id"]
|
||||
await db.execute("""INSERT INTO channel_state(channel_id,last_message_id) VALUES (?,?)
|
||||
ON CONFLICT(channel_id) DO UPDATE SET last_message_id=excluded.last_message_id""",
|
||||
(channel_id, message_id))
|
||||
await db.commit()
|
||||
override["updateonce"] = False
|
||||
changed = True
|
||||
logger.info("[CURSOR OVERRIDE] Set channel %s last_message_id to %d", channel_id, message_id)
|
||||
if changed:
|
||||
await save_config()
|
||||
|
||||
|
||||
async def media_exists(channel_id, message_id):
|
||||
async with db.execute("SELECT 1 FROM downloaded_media WHERE channel_id=? AND message_id=?", (channel_id, message_id)) as cur:
|
||||
return await cur.fetchone() is not None
|
||||
@@ -319,6 +351,7 @@ async def reload_config():
|
||||
RETRY_DROP_LOG = CONFIG.get("retry_drop_log", True)
|
||||
apply_channel_overrides()
|
||||
await resolve_channels()
|
||||
await apply_last_message_id_overrides()
|
||||
if manual_changed:
|
||||
await process_manual_downloads()
|
||||
logger.info("[CONFIG] Reloaded")
|
||||
@@ -339,6 +372,8 @@ async def config_watcher():
|
||||
if mtime != last_mtime:
|
||||
last_mtime = mtime
|
||||
await reload_config()
|
||||
else:
|
||||
await apply_last_message_id_overrides()
|
||||
except asyncio.CancelledError:
|
||||
break
|
||||
except Exception as error:
|
||||
@@ -627,6 +662,7 @@ async def connection_supervisor():
|
||||
try:
|
||||
await client.connect()
|
||||
await resolve_channels()
|
||||
await apply_last_message_id_overrides()
|
||||
if sync_task and not sync_task.done():
|
||||
sync_task.cancel()
|
||||
await asyncio.gather(sync_task, return_exceptions=True)
|
||||
@@ -751,6 +787,7 @@ async def main():
|
||||
me = await client.get_me()
|
||||
logger.info("[CONNECTED] Logged in as %s", me.first_name)
|
||||
await resolve_channels()
|
||||
await apply_last_message_id_overrides()
|
||||
await process_manual_downloads()
|
||||
for index in range(PARALLEL_DOWNLOADS):
|
||||
worker_tasks.append(asyncio.create_task(worker(index + 1)))
|
||||
|
||||
Reference in New Issue
Block a user