Your Name
115 lines
4.5 KiB
Python
115 lines
4.5 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import html
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from aiogram.types import FSInputFile
|
|
|
|
from vk_parser_app.constants import MEDIA_STATUS_FAILED, MEDIA_STATUS_LINK_ONLY
|
|
from vk_parser_app.db import get_pool
|
|
from vk_parser_app.workers.vk_storage_uploader import TMP_DIR, TMP_PREFIX, TelegramStorageUploader
|
|
|
|
|
|
def jlog(event: str, **payload: Any) -> None:
|
|
print(json.dumps({"event": event, **payload}, ensure_ascii=False, default=str), flush=True)
|
|
|
|
|
|
async def load_pending_videos(worker: TelegramStorageUploader, limit: int) -> list[dict[str, Any]]:
|
|
rows = await worker.pool.fetch(
|
|
"""
|
|
SELECT m.*, rp.original_url AS post_url
|
|
FROM raw_post_media m
|
|
JOIN raw_posts rp ON rp.id=m.raw_post_id
|
|
WHERE m.media_type='video'
|
|
AND m.status='pending'
|
|
AND COALESCE(m.tg_file_id, m.storage_attachment_id, '') = ''
|
|
AND COALESCE(m.error, '') ILIKE '%timeout%'
|
|
AND COALESCE(m.duration_sec, 0) <= $1
|
|
ORDER BY COALESCE(rp.publication_status, 'pending')='published', m.id DESC
|
|
LIMIT $2
|
|
""",
|
|
worker.video_max_duration_sec,
|
|
limit,
|
|
)
|
|
return [dict(row) for row in rows]
|
|
|
|
|
|
async def send_video_to_storage(worker: TelegramStorageUploader, media: dict[str, Any], path: str) -> None:
|
|
caption = (
|
|
f"Backfill video media #{int(media['id'])} for raw_post #{int(media['raw_post_id'])}\n"
|
|
f'<a href="{html.escape(str(media.get("post_url") or ""), quote=True)}">source post</a>'
|
|
)
|
|
message = await worker.tg_retry(
|
|
lambda: worker.bot.send_video(
|
|
video=FSInputFile(path, filename=f"video_{int(media['id'])}.mp4"),
|
|
caption=caption[: worker.caption_limit],
|
|
parse_mode="HTML",
|
|
**worker.chat_kwargs(),
|
|
)
|
|
)
|
|
if not message.video:
|
|
raise RuntimeError("Telegram accepted message without video payload")
|
|
await worker.mark_media_uploaded(int(media["id"]), message.video.file_id, message.video.file_unique_id)
|
|
|
|
|
|
async def main() -> None:
|
|
limit = int(os.getenv("BACKFILL_LIMIT", "200"))
|
|
worker = TelegramStorageUploader()
|
|
await worker.init()
|
|
videos = await load_pending_videos(worker, limit)
|
|
jlog("start", count=len(videos), limit=limit)
|
|
ok = 0
|
|
failed = 0
|
|
link_only = 0
|
|
try:
|
|
for media in videos:
|
|
media_id = int(media["id"])
|
|
raw_post_id = int(media["raw_post_id"])
|
|
path = str(TMP_DIR / f"{TMP_PREFIX}backfill_{raw_post_id}_{media_id}.mp4")
|
|
jlog(
|
|
"video_start",
|
|
media_id=media_id,
|
|
raw_post_id=raw_post_id,
|
|
url=media.get("original_url"),
|
|
duration_sec=media.get("duration_sec"),
|
|
)
|
|
try:
|
|
info = await worker.download_video(str(media.get("original_url") or ""), path)
|
|
if not info:
|
|
await worker.mark_media_failed_attempt(media_id, "video download failed")
|
|
failed += 1
|
|
jlog("video_failed", media_id=media_id, raw_post_id=raw_post_id, error="video download failed")
|
|
continue
|
|
if info.get("error"):
|
|
if info.get("permanent"):
|
|
await worker.mark_media_link_only(media_id, str(info["error"]))
|
|
link_only += 1
|
|
else:
|
|
await worker.mark_media_failed_attempt(media_id, str(info["error"]))
|
|
failed += 1
|
|
jlog("video_failed", media_id=media_id, raw_post_id=raw_post_id, error=info.get("error"))
|
|
continue
|
|
await send_video_to_storage(worker, media, path)
|
|
ok += 1
|
|
jlog("video_uploaded", media_id=media_id, raw_post_id=raw_post_id, size_bytes=info.get("size_bytes"))
|
|
await asyncio.sleep(worker.media_upload_delay_sec)
|
|
except Exception as exc:
|
|
await worker.mark_media_failed_attempt(media_id, str(exc))
|
|
failed += 1
|
|
jlog("video_failed", media_id=media_id, raw_post_id=raw_post_id, error=str(exc))
|
|
finally:
|
|
Path(path).unlink(missing_ok=True)
|
|
finally:
|
|
await worker.close()
|
|
if worker.pool:
|
|
await worker.pool.close()
|
|
jlog("done", ok=ok, failed=failed, link_only=link_only, total=len(videos))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|