Upload 5 files

Dockerfile

@@ -0,0 +1,28 @@
+FROM python:3.11-slim-bookworm
+
+ENV PYTHONUNBUFFERED=1
+ENV TZ=Asia/Kolkata
+ARG DEBIAN_FRONTEND=noninteractive
+
+RUN apt-get update && \
+    apt-get install -y --no-install-recommends \
+    build-essential tzdata libssl-dev libffi-dev && \
+    ln -snf "/usr/share/zoneinfo/$TZ" /etc/localtime && \
+    echo "$TZ" > /etc/timezone && \
+    rm -rf /var/lib/apt/lists/*
+
+RUN pip install --no-cache-dir -U pip
+
+WORKDIR /app
+
+COPY requirements.txt .
+RUN pip install -U -r requirements.txt
+
+COPY main.py .
+COPY fileManager.py .
+
+# Storage dirs
+RUN mkdir -p fl
+
+EXPOSE 7860
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "7860"]

config.env

@@ -0,0 +1,13 @@
+# Telegram API credentials — https://my.telegram.org
+API_ID=
+API_HASH=
+
+# Bot token from @BotFather
+BOT_TOKEN=
+
+# Your HuggingFace Space URL (this space itself)
+# e.g. https://yourname-linkgenbot.hf.space
+SPACE_URL=
+
+# How long files are kept in minutes (0 = keep forever)
+FILE_EXPIRE_MINUTES=0

fileManager.py

@@ -0,0 +1,108 @@
+from threading import Thread
+from copy import deepcopy
+from uuid import uuid4
+from time import sleep
+import json
+import time
+import re
+import os
+
+
+def syncmethod(func):
+    def function(*args, **kwargs):
+        th = Thread(target=func, args=args, kwargs=kwargs)
+        th.start()
+    return function
+
+
+@syncmethod
+def interval(fc, t):
+    while True:
+        sleep(t)
+        fc()
+
+
+class FileManager:
+    def __init__(self, base_url: str, expires_minutes: int = 60):
+        self.data_file = "collector_data.json"
+        self.base_url  = base_url.rstrip("/")
+        self.expires_minutes = expires_minutes
+        os.makedirs("fl", exist_ok=True)
+        if not os.path.exists(self.data_file):
+            self._write_json()
+        else:
+            self.check_files()
+
+    # ------------------------------------------------------------------
+    def _format_path(self, path: str) -> str:
+        p = re.sub(r'\s+', '-', path)
+        p = re.sub(r'[!?#*Ççêwrong_char]', '', p)
+        return p
+
+    def _read_json(self) -> list:
+        with open(self.data_file, "r") as f:
+            return json.load(f)
+
+    def _write_json(self, data: list = None):
+        with open(self.data_file, "w") as f:
+            json.dump(data or [], f, indent=2)
+
+    # ------------------------------------------------------------------
+    def start(self, gap_minutes: int = 5):
+        interval(self.check_files, gap_minutes * 60)
+
+    def check_files(self):
+        data = self._read_json()
+        now  = time.time()
+        keep = []
+        for entry in data:
+            if entry["expiresAt"] < now:
+                try:
+                    os.remove(entry["path"])
+                except Exception as e:
+                    print(f"[cleanup] {e}")
+            else:
+                keep.append(entry)
+        self._write_json(keep)
+
+    # ------------------------------------------------------------------
+    def _revalidate(self, path: str):
+        data = self._read_json()
+        for entry in data:
+            if entry["path"] == path:
+                entry["expiresAt"] = time.time() + self.expires_minutes * 60
+        self._write_json(data)
+
+    def _add_path(self, path: str) -> dict:
+        new_path = self._format_path(path)
+        if path != new_path:
+            os.rename(path, new_path)
+        data = self._read_json()
+        data = [e for e in data if e["path"] != new_path]
+        entry = {
+            "path":      new_path,
+            "filename":  new_path.split("/")[-1],
+            "id":        str(uuid4()),
+            "expiresAt": time.time() + self.expires_minutes * 60,
+        }
+        data.append(entry)
+        self._write_json(data)
+        return entry
+
+    # ------------------------------------------------------------------
+    def save_file(self, data: bytes, filename: str) -> dict:
+        filename  = self._format_path(filename)
+        file_path = f"fl/{uuid4()}_{filename}"
+        with open(file_path, "wb") as f:
+            f.write(data)
+        return self._add_path(file_path)
+
+    def get_url(self, entry: dict) -> str:
+        return f"{self.base_url}/file=./{entry['path']}"
+
+    def get_entry_by_id(self, uid: str) -> dict | None:
+        for entry in self._read_json():
+            if entry["id"] == uid:
+                self._revalidate(entry["path"])
+                return entry
+        return None

main.py

@@ -0,0 +1,309 @@
+# Link Generator Bot
+# Webhook-only | HuggingFace Spaces compatible
+#
+# Flow:
+#   User sends any file / photo / video / audio / document to the bot
+#   → Bot downloads it via Pyrogram MTProto (no Bot API HTTP)
+#   → Bot saves it to local HF Space storage
+#   → Bot replies with a permanent public https://...hf.space/file=./... link
+
+import os
+import sys
+import logging
+import asyncio
+import traceback
+from contextlib import asynccontextmanager
+from logging.handlers import RotatingFileHandler
+
+from dotenv import load_dotenv
+from fastapi import FastAPI, Request
+from fastapi.responses import JSONResponse
+from pyrogram import Client
+from pyrogram.enums import ParseMode
+from pyrogram.errors import FloodWait
+
+from fileManager import FileManager
+
+# ============================================================
+# Logging
+# ============================================================
+try:
+    os.remove("logs.txt")
+except Exception:
+    pass
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="[%(asctime)s - %(levelname)s] - %(name)s - %(message)s",
+    datefmt="%d-%b-%y %I:%M:%S %p",
+    handlers=[
+        RotatingFileHandler("logs.txt", mode="w+", maxBytes=5_000_000, backupCount=10),
+        logging.StreamHandler(),
+    ],
+)
+logging.getLogger("pyrogram").setLevel(logging.ERROR)
+
+def LOGGER(name: str) -> logging.Logger:
+    return logging.getLogger(name)
+
+# ============================================================
+# Config
+# ============================================================
+try:
+    load_dotenv("config.env.local")
+    load_dotenv("config.env")
+except Exception:
+    pass
+
+_required = ["BOT_TOKEN", "API_ID", "API_HASH", "SPACE_URL"]
+_missing  = [v for v in _required if not os.getenv(v)]
+if _missing:
+    print(f"ERROR: Missing required env vars: {', '.join(_missing)}")
+    sys.exit(1)
+
+API_ID    = int(os.getenv("API_ID"))
+API_HASH  = os.getenv("API_HASH")
+BOT_TOKEN = os.getenv("BOT_TOKEN")
+SPACE_URL = os.getenv("SPACE_URL")          # e.g. https://yourname-spacename.hf.space
+FILE_EXPIRE_MINUTES = int(os.getenv("FILE_EXPIRE_MINUTES", "0"))  # 0 = never expire
+
+# ============================================================
+# FileManager
+# ============================================================
+os.makedirs("fl", exist_ok=True)
+fm = FileManager(base_url=SPACE_URL, expires_minutes=FILE_EXPIRE_MINUTES or 999_999)
+fm.start()
+
+# ============================================================
+# Pyrogram bot client (MTProto only — no Bot API HTTP)
+# ============================================================
+bot = Client(
+    "link_gen_bot",
+    api_id=API_ID,
+    api_hash=API_HASH,
+    bot_token=BOT_TOKEN,
+    workers=50,
+    parse_mode=ParseMode.MARKDOWN,
+    max_concurrent_transmissions=4,
+    sleep_threshold=30,
+)
+
+# ============================================================
+# Global state
+# ============================================================
+RUNNING_TASKS: set = set()
+
+def track_task(coro):
+    task = asyncio.create_task(coro)
+    RUNNING_TASKS.add(task)
+    task.add_done_callback(RUNNING_TASKS.discard)
+    return task
+
+# ============================================================
+# Webhook response helpers
+# ============================================================
+def _msg(chat_id: int, text: str, parse_mode: str = "Markdown",
+         reply_markup: dict = None, disable_web_page_preview: bool = False) -> dict:
+    payload = {"method": "sendMessage", "chat_id": chat_id,
+               "text": text, "parse_mode": parse_mode}
+    if reply_markup:
+        payload["reply_markup"] = reply_markup
+    if disable_web_page_preview:
+        payload["disable_web_page_preview"] = True
+    return payload
+
+# ============================================================
+# Instant command handlers
+# ============================================================
+def handle_start(chat_id: int) -> dict:
+    return _msg(chat_id,
+        "👋 **Welcome to Link Generator Bot!**\n\n"
+        "Just send me any file, photo, video, audio or document\n"
+        "and I'll give you a **public shareable link** for it.\n\n"
+        "📎 Supports: documents, photos, videos, audio, voice, stickers\n"
+        "🔗 Links are permanent and accessible by anyone.\n\n"
+        "Just send a file to get started!")
+
+def handle_help(chat_id: int) -> dict:
+    return _msg(chat_id,
+        "💡 **Link Generator Bot Help**\n\n"
+        "➤ Send any file/media → get a public link\n"
+        "➤ `/start` – welcome message\n"
+        "➤ `/help`  – this message\n"
+        "➤ `/stats` – bot stats\n\n"
+        "**Supported types:**\n"
+        "• Documents, PDFs, ZIPs, etc.\n"
+        "• Photos\n"
+        "• Videos\n"
+        "• Audio / Voice messages\n"
+        "• Stickers\n"
+        "• Any other file type\n\n"
+        "Just send the file — no commands needed!")
+
+def handle_stats(chat_id: int) -> dict:
+    import shutil, psutil
+    total, used, free = shutil.disk_usage(".")
+    proc = psutil.Process(os.getpid())
+
+    def fmt(b):
+        for u in ["B","KB","MB","GB"]:
+            if b < 1024: return f"{b:.1f} {u}"
+            b /= 1024
+        return f"{b:.1f} TB"
+
+    fl_count = len(os.listdir("fl")) if os.path.isdir("fl") else 0
+    return _msg(chat_id,
+        "**📊 Bot Stats**\n\n"
+        f"**➜ Disk:** `{fmt(used)}` used / `{fmt(total)}` total\n"
+        f"**➜ Free:** `{fmt(free)}`\n"
+        f"**➜ Memory:** `{round(proc.memory_info()[0]/1024**2)} MiB`\n"
+        f"**➜ CPU:** `{psutil.cpu_percent(interval=0.2)}%`\n"
+        f"**➜ Stored files:** `{fl_count}`")
+
+# ============================================================
+# Detect media in a raw Telegram message dict
+# ============================================================
+MEDIA_TYPES = [
+    "document", "video", "audio", "voice",
+    "photo", "animation", "sticker", "video_note"
+]
+
+def get_media_info(msg: dict) -> tuple[str, str] | tuple[None, None]:
+    """Returns (media_type, file_name) or (None, None) if no media."""
+    for mtype in MEDIA_TYPES:
+        if mtype not in msg:
+            continue
+        obj = msg[mtype]
+        if mtype == "photo":
+            # photo is a list — pick largest
+            obj = msg["photo"][-1] if isinstance(msg["photo"], list) else msg["photo"]
+            return mtype, f"{obj.get('file_unique_id', 'photo')}.jpg"
+        fname = (
+            obj.get("file_name")
+            or obj.get("file_unique_id", mtype)
+        )
+        # ensure extension
+        if "." not in fname.split("/")[-1]:
+            ext_map = {
+                "video": "mp4", "audio": "mp3", "voice": "ogg",
+                "animation": "gif", "sticker": "webp", "video_note": "mp4"
+            }
+            fname = f"{fname}.{ext_map.get(mtype, 'bin')}"
+        return mtype, fname
+    return None, None
+
+# ============================================================
+# Async media handler
+# ============================================================
+async def handle_media(chat_id: int, msg_id: int, tg_msg_id: int, filename: str):
+    """Download file via Pyrogram MTProto, save to HF space, reply with link."""
+    progress_msg = await bot.send_message(
+        chat_id, "⏳ **Processing your file...**",
+        reply_to_message_id=tg_msg_id)
+    try:
+        # Get the actual Pyrogram message object
+        pyro_msg = await bot.get_messages(chat_id, tg_msg_id)
+        if not pyro_msg or not pyro_msg.media:
+            await progress_msg.edit("❌ **Could not read the file. Please try again.**")
+            return
+
+        # Download to memory-mapped temp path
+        tmp_path = f"fl/tmp_{tg_msg_id}_{filename}"
+        for attempt in range(2):
+            try:
+                await pyro_msg.download(file_name=tmp_path)
+                break
+            except FloodWait as e:
+                wait_s = int(getattr(e, "value", 0) or 0)
+                LOGGER(__name__).warning(f"FloodWait: {wait_s}s")
+                if attempt == 0 and wait_s > 0:
+                    await asyncio.sleep(wait_s + 1)
+                    continue
+                raise
+
+        if not os.path.exists(tmp_path) or os.path.getsize(tmp_path) == 0:
+            await progress_msg.edit("❌ **Download failed. Please try again.**")
+            try: os.remove(tmp_path)
+            except: pass
+            return
+
+        # Read and save via FileManager
+        with open(tmp_path, "rb") as f:
+            file_data = f.read()
+        os.remove(tmp_path)
+
+        entry = fm.save_file(file_data, filename)
+        link  = fm.get_url(entry)
+
+        file_size = len(file_data)
+        size_str  = f"{file_size/1024/1024:.2f} MB" if file_size > 1024*1024 else f"{file_size/1024:.1f} KB"
+
+        await progress_msg.edit(
+            f"✅ **Link Generated!**\n\n"
+            f"🔗 **Link:** `{link}`\n\n"
+            f"📄 **File:** `{entry['filename']}`\n"
+            f"📦 **Size:** `{size_str}`\n\n"
+            f"_Link is public and accessible by anyone._"
+        )
+        LOGGER(__name__).info(f"Generated link for {filename}: {link}")
+
+    except Exception as e:
+        LOGGER(__name__).error(traceback.format_exc())
+        try:
+            await progress_msg.edit(f"❌ **Error:** `{str(e)}`")
+        except Exception:
+            pass
+
+# ============================================================
+# FastAPI lifespan
+# ============================================================
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    LOGGER(__name__).info("Starting bot client…")
+    await bot.start()
+    LOGGER(__name__).info("Bot ready — waiting for webhook updates.")
+    yield
+    LOGGER(__name__).info("Shutting down…")
+    await bot.stop()
+
+app = FastAPI(lifespan=lifespan)
+
+# ============================================================
+# Webhook endpoint
+# ============================================================
+@app.post("/webhook")
+async def telegram_webhook(request: Request):
+    update = await request.json()
+    LOGGER(__name__).debug(f"Update: {update}")
+
+    if "message" not in update:
+        return JSONResponse({"status": "ok"})
+
+    msg     = update["message"]
+    chat_id = msg["chat"]["id"]
+    text    = msg.get("text", "").strip()
+    msg_id  = msg["message_id"]
+
+    # ── commands ──────────────────────────────────────────
+    if text.startswith("/start"):
+        return JSONResponse(handle_start(chat_id))
+
+    if text.startswith("/help"):
+        return JSONResponse(handle_help(chat_id))
+
+    if text.startswith("/stats"):
+        return JSONResponse(handle_stats(chat_id))
+
+    # ── media ─────────────────────────────────────────────
+    media_type, filename = get_media_info(msg)
+    if media_type:
+        track_task(handle_media(chat_id, msg_id, msg_id, filename))
+        return JSONResponse({"status": "ok"})
+
+    # ── unknown ───────────────────────────────────────────
+    if text and not text.startswith("/"):
+        return JSONResponse(_msg(chat_id,
+            "📎 **Please send a file, photo, video or audio.**\n"
+            "I'll generate a public link for it!"))
+
+    return JSONResponse({"status": "ok"})

requirements.txt

@@ -0,0 +1,6 @@
+pyrofork
+tgcrypto
+python-dotenv
+fastapi
+uvicorn[standard]
+psutil