MasterHttpRelayVPN-RUST/.github/scripts/telegram_publish_files.py

#!/usr/bin/env python3
"""Post each release artifact individually to a Telegram channel.

Used by .github/workflows/telegram-publish-files.yml. Reads files from
--assets-dir, picks a Persian caption per filename, posts via the
Telegram Bot API `sendDocument` endpoint with --hashtag appended.

Files larger than the Telegram Bot API's 50 MB ceiling are split into
~45 MB byte chunks via Python (no `split` shell dep) and posted as
`<name>.part_aa`, `.part_ab`, ... — recipients reassemble with
`cat <name>.part_* > <name>`.

Re-runnable: posts every file every time. Use carefully when re-running
for the same version (the channel will get duplicate posts).
"""

from __future__ import annotations

import argparse
import hashlib
import os
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
import json
from pathlib import Path

# Telegram Bot API uploads cap at 50 MB. Pick 45 MB for chunks so the
# multipart envelope + caption + Telegram's own overhead don't push us
# over. Bigger chunks (e.g. 49 MB) sometimes hit "Request Entity Too
# Large" depending on caption length.
CHUNK_LIMIT_BYTES = 45 * 1024 * 1024

# Sleep between uploads. Telegram's documented rate limit is 1 msg/sec
# to the same chat, plus a soft "burst" allowance. 1.5s is conservative
# and means a 20-file release publishes in ~30 s.
INTER_UPLOAD_SLEEP_SECS = 1.5

# Filename-substring → Persian caption. Order matters: longest /
# most-specific patterns first, since a shorter pattern (e.g.
# "android-x86") can match a more-specific filename ("android-x86_64").
# Match is `pattern in filename`.
CAPTIONS: list[tuple[str, str]] = [
    # Android — universal first (the recommended default for non-technical users).
    ("android-universal", "نسخه اندروید (universal) — برای همه دستگاه‌ها"),
    ("android-arm64-v8a", "نسخه اندروید (arm64-v8a) — گوشی‌های مدرن ۶۴ بیتی"),
    ("android-armeabi-v7a", "نسخه اندروید (armv7) — گوشی‌های قدیمی‌تر ۳۲ بیتی"),
    ("android-x86_64", "نسخه اندروید (x86_64) — شبیه‌ساز ۶۴ بیتی"),
    ("android-x86", "نسخه اندروید (x86) — شبیه‌ساز"),
    # Windows.
    ("windows-amd64", "نسخه ویندوز x64 (۶۴ بیتی)"),
    ("windows-i686", "نسخه ویندوز x86 (۳۲ بیتی، Win7+)"),
    # macOS — .app bundles before plain CLI tarballs.
    ("macos-arm64-app", "نسخه macOS (Apple Silicon) — برنامه گرافیکی .app"),
    ("macos-amd64-app", "نسخه macOS (Intel) — برنامه گرافیکی .app"),
    ("macos-arm64", "نسخه macOS (Apple Silicon) — CLI"),
    ("macos-amd64", "نسخه macOS (Intel) — CLI"),
    # Linux — musl static first, glibc second.
    ("linux-musl-amd64", "نسخه لینوکس amd64 (musl static) — Alpine / OpenWRT-x86"),
    ("linux-musl-arm64", "نسخه لینوکس arm64 (musl static)"),
    ("linux-amd64", "نسخه لینوکس amd64 (glibc)"),
    ("linux-arm64", "نسخه لینوکس arm64 (glibc)"),
    # Embedded targets.
    ("openwrt-mipsel-softfloat", "نسخه OpenWRT (mipsel softfloat) — روتر MT7621"),
    ("raspbian-armhf", "نسخه Raspbian (armhf) — رزبری پای ۳۲ بیتی"),
]


def caption_for(filename: str) -> str:
    """Return the Persian caption for a filename, falling back to the
    bare filename if nothing matches."""
    for pattern, persian in CAPTIONS:
        if pattern in filename:
            return persian
    return f"نسخه `{filename}`"


def order_files(files: list[Path]) -> list[Path]:
    """Sort release files in CAPTIONS order (Android first, then
    Windows, macOS, Linux, embedded). Files not matching any pattern
    fall to the end in alphabetical order."""
    order_map: dict[str, int] = {pattern: idx for idx, (pattern, _) in enumerate(CAPTIONS)}

    def key(p: Path) -> tuple[int, str]:
        for pattern, idx in order_map.items():
            if pattern in p.name:
                return (idx, p.name)
        # Unknown patterns: push to end, alphabetize among themselves.
        return (len(CAPTIONS), p.name)

    return sorted(files, key=key)


def split_file(path: Path, chunk_bytes: int) -> list[Path]:
    """Split `path` into chunks of at most `chunk_bytes` bytes. Returns
    the list of chunk paths, named `<orig>.part_aa`, `.part_ab`, ...
    Mimics `split -b <chunk_bytes>`. Reassembled via
    `cat <name>.part_* > <name>`.

    Skips work if existing parts are already present (idempotent re-run)."""
    parts: list[Path] = []

    def part_name(idx: int) -> str:
        # 26-letter base: aa..az, ba..bz, ... mirroring split's default.
        first = chr(ord("a") + idx // 26)
        second = chr(ord("a") + idx % 26)
        return f"{path.name}.part_{first}{second}"

    idx = 0
    with path.open("rb") as src:
        while True:
            buf = src.read(chunk_bytes)
            if not buf:
                break
            part_path = path.parent / part_name(idx)
            with part_path.open("wb") as dst:
                dst.write(buf)
            parts.append(part_path)
            idx += 1
    return parts


def send_document(
    bot_token: str,
    chat_id: str,
    file_path: Path,
    caption: str,
) -> dict:
    """POST a single file via the Telegram Bot API sendDocument endpoint.
    Returns the parsed JSON response. Raises on HTTP error.

    Uses urllib + a hand-rolled multipart/form-data encoder so we don't
    pull `requests` (the workflow runs on stock GitHub-hosted runners
    where stdlib-only is preferable for cold-start speed)."""
    url = f"https://api.telegram.org/bot{bot_token}/sendDocument"
    boundary = "----mhrvUploadBoundary" + str(int(time.time() * 1000))
    body = build_multipart(
        boundary,
        fields={
            "chat_id": chat_id,
            "caption": caption,
            "parse_mode": "HTML",
            # Disable preview to keep the channel tidy.
            "disable_notification": "false",
        },
        files={"document": (file_path.name, file_path.read_bytes(), "application/octet-stream")},
    )
    req = urllib.request.Request(
        url,
        data=body,
        headers={"Content-Type": f"multipart/form-data; boundary={boundary}"},
        method="POST",
    )
    # 5 minute timeout for the actual upload — Telegram occasionally
    # takes a while to process 40+ MB documents.
    with urllib.request.urlopen(req, timeout=300) as resp:
        return json.loads(resp.read().decode("utf-8"))


def build_multipart(
    boundary: str,
    fields: dict[str, str],
    files: dict[str, tuple[str, bytes, str]],
) -> bytes:
    """Build a multipart/form-data body. `files` is name → (filename,
    bytes, mime). Plain stdlib so we don't need `requests`."""
    parts: list[bytes] = []
    crlf = b"\r\n"
    bnd = f"--{boundary}".encode()

    for name, value in fields.items():
        parts.append(bnd)
        parts.append(f'Content-Disposition: form-data; name="{name}"'.encode())
        parts.append(b"")
        parts.append(value.encode("utf-8"))

    for name, (filename, data, mime) in files.items():
        parts.append(bnd)
        parts.append(
            f'Content-Disposition: form-data; name="{name}"; filename="{filename}"'.encode()
        )
        parts.append(f"Content-Type: {mime}".encode())
        parts.append(b"")
        parts.append(data)

    parts.append(f"--{boundary}--".encode())
    parts.append(b"")
    return crlf.join(parts)


def html_escape(s: str) -> str:
    return s.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")


def sha256_hex(path: Path) -> str:
    """Stream-hash the file in 1 MiB chunks. Avoids loading 40+ MB APKs
    into RAM twice (once for hashing, once for upload)."""
    h = hashlib.sha256()
    with path.open("rb") as f:
        for chunk in iter(lambda: f.read(1 << 20), b""):
            h.update(chunk)
    return h.hexdigest()


def post_file(
    bot_token: str,
    chat_id: str,
    file_path: Path,
    base_caption: str,
    hashtag: str,
) -> bool:
    """Post one file. If too big, split + post each part. Returns True
    on success of all parts, False on any failure.

    Each caption ends with the file's SHA-256 in hex under a Persian
    "تایید اصالت" (authenticity verification) label, so recipients can
    `sha256sum <file>` after download and confirm it matches what the
    channel posted — defends against modified copies if the channel is
    ever compromised or relayed through a third party."""
    size = file_path.stat().st_size

    # Compute the original-file hash regardless of whether we'll chunk
    # it. For chunked uploads, every part's caption shows this hash so
    # the user can verify the full file once reassembled with `cat`.
    print(f"  hashing {file_path.name}...", flush=True)
    full_sha = sha256_hex(file_path)

    if size <= CHUNK_LIMIT_BYTES:
        caption = (
            f"<b>{html_escape(base_caption)}</b>\n"
            f"<code>{html_escape(file_path.name)}</code>\n"
            f"\nتایید اصالت (SHA-256):\n"
            f"<code>{full_sha}</code>\n"
            f"\n{hashtag}"
        )
        print(f"  uploading {file_path.name} ({size / 1_048_576:.1f} MB)...", flush=True)
        try:
            resp = send_document(bot_token, chat_id, file_path, caption)
            if not resp.get("ok"):
                print(f"    !! Telegram returned not-ok: {resp}", flush=True)
                return False
            print(f"    ok (message_id={resp['result']['message_id']})", flush=True)
            return True
        except urllib.error.HTTPError as e:
            err_body = e.read().decode("utf-8", errors="replace")[:500]
            print(f"    !! HTTP {e.code}: {err_body}", flush=True)
            return False
        except Exception as e:
            print(f"    !! exception: {e}", flush=True)
            return False
        finally:
            time.sleep(INTER_UPLOAD_SLEEP_SECS)

    # Too big — split and post each part.
    print(
        f"  splitting {file_path.name} ({size / 1_048_576:.1f} MB > "
        f"{CHUNK_LIMIT_BYTES / 1_048_576:.0f} MB ceiling)...",
        flush=True,
    )
    parts = split_file(file_path, CHUNK_LIMIT_BYTES)
    if not parts:
        print(f"    !! split produced 0 parts (empty file?)", flush=True)
        return False

    n = len(parts)
    all_ok = True
    for idx, part_path in enumerate(parts, start=1):
        # Hash the individual part too — lets the user verify each
        # downloaded chunk before bothering to reassemble.
        part_sha = sha256_hex(part_path)
        part_caption = (
            f"<b>{html_escape(base_caption)} — قسمت {idx}/{n}</b>\n"
            f"<code>{html_escape(part_path.name)}</code>\n"
            f"\nبرای بازسازی فایل اصلی:\n"
            f"<code>cat {html_escape(file_path.name)}.part_* &gt; "
            f"{html_escape(file_path.name)}</code>\n"
            f"\nتایید اصالت این قسمت (SHA-256):\n"
            f"<code>{part_sha}</code>\n"
            f"\nتایید اصالت فایل کامل پس از بازسازی (SHA-256):\n"
            f"<code>{full_sha}</code>\n"
            f"\n{hashtag}"
        )
        psize = part_path.stat().st_size
        print(
            f"    uploading part {idx}/{n}: {part_path.name} ({psize / 1_048_576:.1f} MB)...",
            flush=True,
        )
        try:
            resp = send_document(bot_token, chat_id, part_path, part_caption)
            if not resp.get("ok"):
                print(f"      !! Telegram returned not-ok: {resp}", flush=True)
                all_ok = False
            else:
                print(
                    f"      ok (message_id={resp['result']['message_id']})", flush=True
                )
        except urllib.error.HTTPError as e:
            err_body = e.read().decode("utf-8", errors="replace")[:500]
            print(f"      !! HTTP {e.code}: {err_body}", flush=True)
            all_ok = False
        except Exception as e:
            print(f"      !! exception: {e}", flush=True)
            all_ok = False
        finally:
            time.sleep(INTER_UPLOAD_SLEEP_SECS)
            # Tidy up the part file once posted.
            try:
                part_path.unlink()
            except OSError:
                pass

    return all_ok


def files_channel_post_link(chat_id: str, message_id: int) -> str:
    """Build a `t.me` link to a specific message in the files channel.

    For private supergroups/channels (negative ID with `-100` prefix),
    Telegram exposes posts at `https://t.me/c/<id>/<msg>` where `<id>`
    is the chat ID with the `-100` stripped. This link works for users
    who are members of the channel.

    If `FILES_CHANNEL_USERNAME` is set in env (e.g. `mhrv_files`), uses
    the public-channel form `https://t.me/<username>/<msg>` instead,
    which is clickable for everyone."""
    username = os.environ.get("FILES_CHANNEL_USERNAME", "").strip().lstrip("@")
    if username:
        return f"https://t.me/{username}/{message_id}"
    cid = chat_id
    if cid.startswith("-100"):
        cid = cid[4:]
    elif cid.startswith("-"):
        cid = cid[1:]
    return f"https://t.me/c/{cid}/{message_id}"


def post_main_channel_pointer(
    bot_token: str,
    main_chat_id: str,
    files_channel_post_link: str,
    version: str,
    hashtag: str,
    channel_username_link: str = "",
    channel_invite_link: str = "",
) -> bool:
    """Post a short cross-link to the main announcement channel pointing
    at the anchor post in the files channel. Replaces the previous
    behaviour of posting the universal APK + full changelog directly
    to the main channel — the main channel becomes a discovery surface
    while the files channel hosts the actual artifacts.

    Includes channel-join links (public username + invite hash) at the
    bottom so recipients who aren't yet members can subscribe before
    clicking through to the specific release post.
    """
    parts = [
        f"<b>📦 mhrv-rs v{html_escape(version)} منتشر شد</b>",
        "",
        f"برای دانلود فایل‌ها (Android، Windows، macOS، Linux و ...) "
        f"به کانال فایل‌ها مراجعه کنید:",
        "",
        f"👉 <a href=\"{html_escape(files_channel_post_link)}\">"
        f"v{html_escape(version)} — همه فایل‌ها + SHA-256</a>",
    ]
    # Channel-join links. Two forms handle different states of the
    # files channel: the `t.me/<username>` form works for public
    # channels and is the prettier link; the `t.me/+<hash>` invite
    # link works regardless of whether the channel is public, and is
    # the only path in for private/restricted channels. Showing both
    # is forgiving — recipients click whichever works for them.
    if channel_username_link or channel_invite_link:
        parts.append("")
        parts.append("لینک کانال:")
        if channel_username_link:
            # Render as plain URL (not HTML <a>) so the text shows the
            # link itself — useful when users share the message via
            # screenshot or copy-paste outside Telegram, which would
            # strip the <a href> wrapper.
            parts.append(html_escape(channel_username_link))
        if channel_invite_link:
            parts.append(f"و یا: {html_escape(channel_invite_link)}")
    parts.append("")
    parts.append(hashtag)
    text = "\n".join(parts)
    url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
    data = urllib.parse.urlencode({
        "chat_id": main_chat_id,
        "text": text,
        "parse_mode": "HTML",
        "disable_web_page_preview": "false",
    }).encode()
    print(f"  posting cross-link to main channel {main_chat_id}...", flush=True)
    try:
        with urllib.request.urlopen(
            urllib.request.Request(url, data=data, method="POST"), timeout=30
        ) as resp:
            r = json.loads(resp.read().decode("utf-8"))
            if not r.get("ok"):
                print(f"    !! main-channel post failed: {r}", flush=True)
                return False
            print(
                f"    ok (message_id={r['result']['message_id']})", flush=True
            )
            return True
    except urllib.error.HTTPError as e:
        err_body = e.read().decode("utf-8", errors="replace")[:500]
        print(f"    !! HTTP {e.code}: {err_body}", flush=True)
        return False
    except Exception as e:
        print(f"    !! exception: {e}", flush=True)
        return False


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--assets-dir", required=True, type=Path)
    parser.add_argument("--version", required=True, help="e.g. 1.8.0")
    parser.add_argument("--hashtag", required=True, help="e.g. #v180")
    args = parser.parse_args()

    bot_token = os.environ.get("BOT_TOKEN")
    chat_id = os.environ.get("CHAT_ID")
    if not bot_token or not chat_id:
        print("BOT_TOKEN and CHAT_ID env vars required", file=sys.stderr)
        return 2

    if not args.assets_dir.is_dir():
        print(f"--assets-dir {args.assets_dir} not a directory", file=sys.stderr)
        return 2

    # Collect all regular files in the directory (no recursion). Skip
    # split-part leftovers from a previous run of this script if any
    # exist — we'll regenerate cleanly.
    raw_files = [
        p for p in args.assets_dir.iterdir()
        if p.is_file() and ".part_" not in p.name
    ]
    if not raw_files:
        print(f"no files found in {args.assets_dir}", file=sys.stderr)
        return 2

    files = order_files(raw_files)
    print(f"publishing {len(files)} file(s) to Telegram chat {chat_id} for v{args.version}:")
    for f in files:
        print(f"  - {f.name}")
    print()

    # Leading announcement in the files channel. Captured `message_id`
    # is the anchor that the main-channel cross-link points at — the
    # main channel doesn't carry files anymore, just a single message
    # saying "new release, click here." Recipients land on this anchor
    # and scroll down to see all the platform-specific files.
    announce = (
        f"<b>📦 mhrv-rs {html_escape('v' + args.version)} منتشر شد</b>\n"
        f"\nفایل‌ها در ادامه به ترتیب پلتفرم ارسال می‌شن.\n"
        f"هر فایل با SHA-256 (تایید اصالت) همراه هست.\n"
        f"\n{args.hashtag}"
    )
    announce_msg_id: int | None = None
    try:
        url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
        data = urllib.parse.urlencode({
            "chat_id": chat_id,
            "text": announce,
            "parse_mode": "HTML",
            "disable_web_page_preview": "true",
        }).encode()
        with urllib.request.urlopen(
            urllib.request.Request(url, data=data, method="POST"), timeout=30
        ) as resp:
            r = json.loads(resp.read().decode("utf-8"))
            if not r.get("ok"):
                print(f"  !! announcement failed: {r}", flush=True)
            else:
                announce_msg_id = r["result"]["message_id"]
                print(
                    f"  announcement posted (message_id={announce_msg_id})",
                    flush=True,
                )
    except Exception as e:
        # Non-fatal for the file uploads, but cross-link to the main
        # channel below will be skipped — without the anchor message_id
        # there's nothing to point at.
        print(f"  !! announcement exception: {e}", flush=True)
    time.sleep(INTER_UPLOAD_SLEEP_SECS)

    failures = 0
    for f in files:
        base = caption_for(f.name)
        ok = post_file(bot_token, chat_id, f, base, args.hashtag)
        if not ok:
            failures += 1

    # Cross-link to the main announcement channel. Skipped if MAIN_CHAT_ID
    # is unset (development / private testing) or if the files-channel
    # announcement didn't post (no anchor to link to).
    main_chat_id = os.environ.get("MAIN_CHAT_ID", "").strip()
    if main_chat_id and announce_msg_id is not None:
        link = files_channel_post_link(chat_id, announce_msg_id)
        # Optional channel-join links rendered alongside the cross-link.
        # `FILES_CHANNEL_USERNAME` is the public-username form (clean
        # `t.me/<name>` URL — clickable for everyone). `FILES_CHANNEL_INVITE`
        # is the `t.me/+<hash>` invite link, the only join path for
        # private channels. Either or both can be set; both render in
        # the body as separate lines.
        username = os.environ.get("FILES_CHANNEL_USERNAME", "").strip().lstrip("@")
        username_link = f"https://t.me/{username}" if username else ""
        invite_link = os.environ.get("FILES_CHANNEL_INVITE", "").strip()
        print()
        print(f"posting cross-link to main channel:")
        print(f"  post link: {link}")
        if username_link:
            print(f"  channel username link: {username_link}")
        if invite_link:
            print(f"  channel invite link:   {invite_link}")
        ok = post_main_channel_pointer(
            bot_token,
            main_chat_id,
            link,
            args.version,
            args.hashtag,
            channel_username_link=username_link,
            channel_invite_link=invite_link,
        )
        if not ok:
            failures += 1
    elif main_chat_id and announce_msg_id is None:
        print()
        print(
            "  !! MAIN_CHAT_ID is set but announcement message_id is None — "
            "skipping cross-link (no anchor to point at).",
            flush=True,
        )
        failures += 1
    else:
        print()
        print("  MAIN_CHAT_ID not set, skipping cross-link", flush=True)

    print()
    if failures:
        print(f"DONE with {failures} failure(s) out of {len(files)}", flush=True)
        return 1
    print(f"DONE — {len(files)} files posted successfully", flush=True)
    return 0


if __name__ == "__main__":
    sys.exit(main())