November 10

instagam down


🧠 Umumiy ma’lumot

Bu bot:

  • Foydalanuvchi yuborgan Instagram havolalari orqali public post, reels, storylarni yuklab beradi.
  • Private yoki o‘chirilgan kontentni yuklamaydi.
  • Asosiy yuklab olish ishini yt-dlp kutubxonasi orqali bajaradi.
  • Bot esa Aiogram 3.22 orqali Telegram interfeysini boshqaradi.

🔧 Asosiy kutubxonalar

import os, asyncio, tempfile, shutil
from concurrent.futures import ThreadPoolExecutor
from typing import List
from yt_dlp import YoutubeDL, DownloadError
from aiogram import Bot, Dispatcher, types
from aiogram.filters import Command
from aiogram.types import FSInputFile

Izoh:

  • yt_dlp — videolarni yuklab olish uchun qudratli kutubxona (Instagram, YouTube, TikTok va boshqalar uchun ishlaydi).
  • asyncio — asinxron ishlarni boshqarish uchun.
  • ThreadPoolExecutor — og‘ir bloklovchi ishlarni (masalan, yuklab olish) alohida oqimda bajarish uchun.
  • aiogram — Telegram bot yaratish uchun zamonaviy framework.

⚙️ Token va botni ishga tushirish

BOT_TOKEN = "7627050170:AAF2w-NYia_7u1GaTD6lW5wSLwEx4C9Tepc"
bot = Bot(token=BOT_TOKEN)
dp = Dispatcher()
executor = ThreadPoolExecutor(max_workers=2)

Izoh:

  • BOT_TOKEN — bu sizning Telegram bot tokeningiz.
  • ThreadPoolExecutor — fon ishlarini 2 ta oqimda bajarish imkonini beradi, shunda bot “osilib qolmaydi”.

🎬 yt-dlp sozlamalari

YDL_OPTS_COMMON = {
    "quiet": True,
    "no_warnings": True,
    "noplaylist": False,
    "merge_output_format": "mp4",
}

Izoh:

Bu sozlamalar yt-dlpga shovqinli loglarni chiqarishni to‘xtatadi va chiqish faylini MP4 formatda birlashtiradi.


🔍 URL tekshirish funksiyasi

def is_instagram_url(text: str) -> bool:
    if not text:
        return False
    t = text.strip()
    return ("instagram.com" in t) or ("instagr.am" in t)

Izoh:

Bu funksiya yuborilgan matnda Instagram havolasi borligini tekshiradi.


🧰 Yuklab olish yordamchilari

def ytdlp_extract_info(url: str) -> dict:
    opts = YDL_OPTS_COMMON.copy()
    opts["outtmpl"] = "%(id)s.%(ext)s"
    with YoutubeDL(opts) as ydl:
        info = ydl.extract_info(url, download=False)
    return info
  • Bu funksiya faqat metama’lumot (title, description, views va h.k.) oladi, faylni yuklamaydi.

def ytdlp_download(url: str, out_dir: str) -> List[str]:
    opts = YDL_OPTS_COMMON.copy()
    opts["outtmpl"] = os.path.join(out_dir, "%(id)s.%(ext)s")
    with YoutubeDL(opts) as ydl:
        ydl.download([url])
    files = [os.path.join(out_dir, f) for f in os.listdir(out_dir)]
    return sorted(files)
  • Bu funksiya esa fayllarni yuklab olishni amalga oshiradi va yuklangan fayl nomlarini qaytaradi.

🤖 /start komandasi

@dp.message(Command("start"))
async def cmd_start(message: types.Message):
    await message.reply("👋 Salom! ...")
  • Bot ishga tushganda foydalanuvchiga qanday ishlashni tushuntiradi.

👤 /profile komandasi

@dp.message(Command("profile"))
async def cmd_profile(message: types.Message):
    ...
  • Bu komandada foydalanuvchi username yoki link yuboradi.
  • Bot yt-dlp yordamida profil haqida ma’lumot (bio, link, likes, views) olishga harakat qiladi.
  • Private yoki mavjud bo‘lmagan profilga kirsangiz, xato xabari chiqadi.

📖 /story komandasi

@dp.message(Command("story"))
async def cmd_story(message: types.Message):
    ...
  • Foydalanuvchi username yuboradi.
  • Bot https://www.instagram.com/stories/{username}/ sahifasini yt-dlp orqali tahlil qiladi.
  • Agar foydalanuvchi public bo‘lsa, story’larni yuklab, foydalanuvchiga yuboradi.
  • Fayllarni vaqtinchalik papkada saqlaydi (tempfile.mkdtemp) va keyin o‘chiradi.

📎 URL yuborish orqali yuklab olish

@dp.message()
async def handle_any_message(message: types.Message):
    text = (message.text or message.caption or "").strip()
    if not is_instagram_url(text):
        return
  • Bu universal handler.
  • Agar foydalanuvchi Instagram havolasini yuborsa, avtomatik media yuklab olib, Telegramga yuboradi.
  • Har xil fayl turlarini (foto, video, hujjat) ajratib yuboradi.

🧹 Fayllarni tozalash

shutil.rmtree(tmpdir, ignore_errors=True)
  • Yuklab olingan vaqtinchalik fayllar ish tugagandan so‘ng avtomatik o‘chiriladi.

🚀 Botni ishga tushirish

async def main():
    print("🤖 Bot ishga tushmoqda...")
    await dp.start_polling(bot)
  • Asosiy polling jarayonini ishga tushiradi.
  • Bu asyncio.run(main()) bilan bajariladi.

📘 Yakuniy xulosa

Bo‘lim Vazifasi yt_dlp Instagram public kontentni yuklash ThreadPoolExecutor Og‘ir vazifalarni asinxron bajarish Aiogram Telegram bilan ishlash tempfile va shutil Fayllarni vaqtincha saqlash va tozalash /profile, /story, URL Foydalanuvchi komandalarini bajarish


Kod:


import os
import asyncio
import tempfile
import shutil
from concurrent.futures import ThreadPoolExecutor
from typing import List, Optional

from yt_dlp import YoutubeDL, DownloadError
from aiogram import Bot, Dispatcher, types
from aiogram.filters import Command
from aiogram.types import FSInputFile

# =========================
#  TOKENNI BU YERGA JOYLASH
# =========================
BOT_TOKEN = "1234567890:ABCDEF-your-telegram-bot-token"  # <-- o'zgartiring

# =========================
#  Aiogram init
# =========================
bot = Bot(token=BOT_TOKEN)
dp = Dispatcher()

# ThreadPoolExecutor for blocking yt-dlp work
executor = ThreadPoolExecutor(max_workers=2)

# yt-dlp umumiy opsiyalari
YDL_OPTS_COMMON = {
    "quiet": True,
    "no_warnings": True,
    "noplaylist": False,  # playlistlarni ham olishga urinish
    "merge_output_format": "mp4",
    # "cookiefile": "cookies.txt"  # agar keyin login kerak bo'lsa
}

# yordamchi: Instagram URL ekanligini tekshirish
def is_instagram_url(text: str) -> bool:
    if not text:
        return False
    t = text.strip()
    return ("instagram.com" in t) or ("instagr.am" in t)

# =========================
#  yt-dlp yordamchi funksiyalari
# =========================
def ytdlp_extract_info(url: str) -> dict:
    """yt-dlp bilan metadata olish (download=False).
    Xatolik bo'lsa exception tashlanadi.
    """
    opts = YDL_OPTS_COMMON.copy()
    opts["outtmpl"] = "%(id)s.%(ext)s"
    with YoutubeDL(opts) as ydl:
        info = ydl.extract_info(url, download=False)
    return info

def ytdlp_download(url: str, out_dir: str) -> List[str]:
    """yt-dlp bilan yuklab olish (bloklovchi). Natija: fayllar ro'yxati."""
    files = []
    opts = YDL_OPTS_COMMON.copy()
    opts["outtmpl"] = os.path.join(out_dir, "%(id)s.%(ext)s")
    with YoutubeDL(opts) as ydl:
        # yt-dlp o'zi ichidagi playlist/entrylarni qaytaradi
        ydl.download([url])
    # yig'ish
    for fname in os.listdir(out_dir):
        files.append(os.path.join(out_dir, fname))
    files.sort()
    return files

# =========================
#  Aiogram handlers
# =========================
@dp.message(Command("start"))
async def cmd_start(message: types.Message):
    await message.reply(
        "👋 Salom!\n\n"
        "Men Instagram dan public media (post, reel, story) yuklaydigan botman.\n\n"
        "Foydalanish:\n"
        "• Instagram havolasini yuboring (post/reel/story).\n"
        "• /profile <username_or_link> — profil haqida ma'lumot.\n"
        "• /story <username> — hozirgi (public) storylarni yuklab ko'rishga urinish.\n\n"
        "Eslatma: faqat public kontent. Private yoki o'chirilgan kontentni yuklashga yordam bermayman."
    )

@dp.message(Command("profile"))
async def cmd_profile(message: types.Message):
    args = message.get_args().strip()
    if not args:
        await message.reply("❗ /profile komandasidan so'ng username yoki profil havolasini yozing.\nMasalan: /profile natgeo yoki /profile https://www.instagram.com/natgeo/")
        return

    # normalize: agar link bo'lsa uni profil linkiga aylantirish
    if is_instagram_url(args):
        url = args
    else:
        url = f"https://www.instagram.com/{args}/"

    status_msg = await message.reply("🔎 Profil ma'lumotlari olinmoqda...")

    loop = asyncio.get_event_loop()
    try:
        info = await loop.run_in_executor(executor, ytdlp_extract_info, url)
    except DownloadError:
        await status_msg.edit_text("❌ Profilni o'qib bo'lmadi — ehtimol private yoki mavjud emas.")
        return
    except Exception as e:
        await status_msg.edit_text(f"❌ Xatolik: {e}")
        return

    # yt-dlp profiling strukturasiga qarab turli natija bo'lishi mumkin
    # Urinish: asosiy ma'lumotlarni chiqarish
    try:
        # agar profiler playlist (e.g., entries) bo'lsa, uploader info olinadi
        uploader = info.get("uploader") or info.get("uploader_id") or info.get("channel") or info.get("title")
        webpage_url = info.get("webpage_url") or url
        # ba'zi hollarda 'entries' bo'ladi - bu profilga tegishli postlar
        entries = info.get("entries")
        count = len(entries) if entries else None

        text_lines = []
        if uploader:
            text_lines.append(f"👤 Profil: {uploader}")
        text_lines.append(f"🔗 URL: {webpage_url}")
        if info.get("description"):
            text_lines.append(f"📝 Bio: {info.get('description')[:400]}")  # qisqaroq Bio
        if info.get("view_count") is not None:
            text_lines.append(f"👁️ Views: {info.get('view_count')}")
        if info.get("like_count") is not None:
            text_lines.append(f"❤️ Likes: {info.get('like_count')}")
        if count is not None:
            text_lines.append(f"📦 Ko'rilgan postlar soni (taxmin): {count}")

        # fallback: agar info bo'sh bo'lsa, butun info dictning bir necha maydonini ko'rsatamiz
        if not text_lines:
            text_lines.append("⚠️ Profil haqida batafsil ma'lumot topilmadi.")
        await status_msg.edit_text("\n".join(text_lines))
    except Exception as e:
        await status_msg.edit_text(f"⚠️ Ma'lumotni qayta ishlashda xato: {e}")

@dp.message(Command("story"))
async def cmd_story(message: types.Message):
    args = message.get_args().strip()
    if not args:
        await message.reply("❗ /story komandasidan so'ng username yozing.\nMasalan: /story natgeo")
        return

    username = args.split()[-1].strip().lstrip("@")
    url = f"https://www.instagram.com/stories/{username}/"

    status_msg = await message.reply("🔄 Storylar olinmoqda (public bo'lsa)...")

    tmpdir = tempfile.mkdtemp(prefix="insta_story_")
    loop = asyncio.get_event_loop()

    try:
        # avval info so'raymiz — ba'zan story~lar uchun extract_info yetadi
        info = await loop.run_in_executor(executor, ytdlp_extract_info, url)
    except DownloadError:
        # qisqaroq xabar
        await status_msg.edit_text("❌ Storylarni o'qib bo'lmadi — ehtimol private yoki mavjud emas.")
        shutil.rmtree(tmpdir, ignore_errors=True)
        return
    except Exception as e:
        # davom etishga urinamiz: ba'zida extract_info ishlamaydi, lekin download mumkin
        # lekin umumiy holda biz downloadni sinab ko'ramiz
        pass

    # endi yuklab ko'ring
    try:
        files = await loop.run_in_executor(executor, ytdlp_download, url, tmpdir)
    except Exception as e:
        await status_msg.edit_text(f"❌ Storylarni yuklashda xatolik: {e}")
        shutil.rmtree(tmpdir, ignore_errors=True)
        return

    if not files:
        await status_msg.edit_text("⚠️ Storylar topilmadi yoki ular public emas.")
        shutil.rmtree(tmpdir, ignore_errors=True)
        return

    # yuborish
    sent_any = False
    for path in files:
        ext = os.path.splitext(path)[1].lower()
        try:
            if ext in [".jpg", ".jpeg", ".png", ".webp"]:
                await message.reply_photo(FSInputFile(path))
            elif ext in [".mp4", ".mov", ".mkv", ".webm"]:
                await message.reply_video(FSInputFile(path))
            else:
                await message.reply_document(FSInputFile(path))
            sent_any = True
        except Exception:
            continue

    if sent_any:
        await status_msg.edit_text("✅ Storylar yuklandi va yuborildi.")
    else:
        await status_msg.edit_text("⚠️ Fayllarni yuborib bo'lmadi.")

    shutil.rmtree(tmpdir, ignore_errors=True)

@dp.message()
async def handle_any_message(message: types.Message):
    """Agar foydalanuvchi oddiy matn yuborsa — instagram havolasi bo'lsa yuklab olamiz."""
    text = (message.text or message.caption or "").strip()
    if not is_instagram_url(text):
        # boshqa so'zlar uchun kichik yo'l-yo'riq
        return

    wait_msg = await message.reply("🔄 URL qabul qilindi. Yuklab olinmoqda...")

    tmpdir = tempfile.mkdtemp(prefix="insta_dl_")
    loop = asyncio.get_event_loop()

    try:
        files = await loop.run_in_executor(executor, ytdlp_download, text, tmpdir)
    except DownloadError:
        await wait_msg.edit_text("❌ Yuklab olishda xatolik: ehtimol post private yoki mavjud emas.")
        shutil.rmtree(tmpdir, ignore_errors=True)
        return
    except Exception as e:
        await wait_msg.edit_text(f"❌ Xatolik: {e}")
        shutil.rmtree(tmpdir, ignore_errors=True)
        return

    if not files:
        await wait_msg.edit_text("⚠️ Hech qanday fayl yuklanmadi.")
        shutil.rmtree(tmpdir, ignore_errors=True)
        return

    # Fayllarni yuborish
    sent = False
    # Agar rasmlar/videolar bo'lsa, birgalikda albumga qo'shish mumkin — ammo soddaligi uchun alohida yuboramiz
    for path in files:
        ext = os.path.splitext(path)[1].lower()
        try:
            if ext in [".jpg", ".jpeg", ".png", ".webp"]:
                await message.reply_photo(FSInputFile(path))
            elif ext in [".mp4", ".mov", ".mkv", ".webm"]:
                await message.reply_video(FSInputFile(path))
            else:
                await message.reply_document(FSInputFile(path))
            sent = True
        except Exception:
            continue

    if sent:
        await wait_msg.edit_text("✅ Yuklab olindi va yuborildi.")
    else:
        await wait_msg.edit_text("⚠️ Fayllarni Telegramga yuborishda muammo bo'ldi.")

    shutil.rmtree(tmpdir, ignore_errors=True)

# =========================
#  Start polling
# =========================
async def main():
    print("🤖 Bot ishga tushmoqda...")
    try:
        await dp.start_polling(bot)
    finally:
        await bot.session.close()

if __name__ == "__main__":
    asyncio.run(main())