Instagram Post Downloader

# Install
pip install yt-dlp

# Download a Reel
yt-dlp "https://www.instagram.com/reel/XXXX/" -o "%(title)s.%(ext)s"

# Download to a specific folder
yt-dlp "https://www.instagram.com/reel/XXXX/" \
  -o "./instagram-downloads/%(uploader)s_%(id)s.%(ext)s"

# Download best quality
yt-dlp -f "bestvideo+bestaudio" "https://www.instagram.com/reel/XXXX/"

#!/usr/bin/env python3
"""
Instagram Post Downloader
Fetches high-res images from public Instagram posts and carousels.
Requires: pip install requests Pillow
"""

import os
import re
import sys
import json
import time
import requests
from datetime import datetime, timezone
from pathlib import Path

try:
    from PIL import Image
    PILLOW_AVAILABLE = True
except ImportError:
    PILLOW_AVAILABLE = False
    print("Warning: Pillow not installed. PDF stitching disabled. Run: pip install Pillow")


HEADERS = {
    "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 "
                  "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
    "Accept-Language": "en-US,en;q=0.9",
    "Referer": "https://www.instagram.com/",
}


def extract_shortcode(url: str) -> str:
    match = re.search(r"instagram\.com/(?:p|reel|tv)/([A-Za-z0-9_-]+)", url)
    if not match:
        raise ValueError(f"Cannot extract shortcode from URL: {url}")
    return match.group(1)


def fetch_post_data(shortcode: str) -> dict:
    """Try multiple endpoints to get post JSON data."""
    # Attempt A: JSON endpoint
    try:
        url = f"https://www.instagram.com/p/{shortcode}/?__a=1&__d=dis"
        r = requests.get(url, headers=HEADERS, timeout=15)
        if r.status_code == 200:
            data = r.json()
            media = (data.get("graphql", {}).get("shortcode_media") or
                     data.get("data", {}).get("shortcode_media"))
            if media:
                return media
    except Exception:
        pass

    # Attempt B: Embed page
    try:
        url = f"https://www.instagram.com/p/{shortcode}/embed/captioned/"
        r = requests.get(url, headers=HEADERS, timeout=15)
        html = r.text
        # Look for JSON blob in script tags
        matches = re.findall(r'window\.__additionalDataLoaded\([^,]+,(\{.+?\})\);', html)
        for blob in matches:
            try:
                data = json.loads(blob)
                media = (data.get("graphql", {}).get("shortcode_media") or
                         data.get("data", {}).get("shortcode_media"))
                if media:
                    return media
            except json.JSONDecodeError:
                continue
    except Exception:
        pass

    return {}


def get_cdn_urls(media: dict) -> list[tuple[str, str]]:
    """Return list of (filename, cdn_url) tuples."""
    results = []
    media_type = media.get("__typename", "")

    if media_type == "GraphSidecar":
        edges = media.get("edge_sidecar_to_children", {}).get("edges", [])
        for i, edge in enumerate(edges, start=1):
            node = edge.get("node", {})
            resources = node.get("display_resources", [])
            url = (max(resources, key=lambda r: r.get("config_width", 0)).get("src")
                   if resources else node.get("display_url", ""))
            if url:
                ext = "jpg" if "jpg" in url.lower() else "webp"
                filename = f"slide_{i:02d}.{ext}"
                results.append((filename, url))
    else:
        resources = media.get("display_resources", [])
        url = (max(resources, key=lambda r: r.get("config_width", 0)).get("src")
               if resources else media.get("display_url", ""))
        if url:
            ext = "jpg" if "jpg" in url.lower() else "webp"
            results.append((f"image.{ext}", url))

    return results


def sanitise_folder_name(caption: str, shortcode: str) -> str:
    truncated = caption[:40] if caption else ""
    cleaned = re.sub(r"[^a-zA-Z0-9 \-]", "", truncated)
    underscored = re.sub(r"[\s\-]+", "_", cleaned).strip("_").lower()
    return underscored if underscored else shortcode


def download_file(url: str, dest_path: str) -> bool:
    r = requests.get(url, headers=HEADERS, stream=True, timeout=30)
    r.raise_for_status()
    with open(dest_path, "wb") as f:
        for chunk in r.iter_content(chunk_size=8192):
            f.write(chunk)
    return True


def stitch_pdf(image_paths: list[str], output_path: str) -> None:
    if not PILLOW_AVAILABLE:
        return
    images = [Image.open(p).convert("RGB") for p in sorted(image_paths)]
    if images:
        images[0].save(output_path, format="PDF", save_all=True,
                       append_images=images[1:], resolution=150.0)


def process_url(post_url: str, base_dir: str, stitch_pdf_flag: bool) -> dict:
    result = {"url": post_url, "status": "ok", "files": [], "error": None}
    try:
        shortcode = extract_shortcode(post_url)
        media = fetch_post_data(shortcode)

        caption = ""
        username = ""
        if media:
            caption_edges = media.get("edge_media_to_caption", {}).get("edges", [])
            caption = caption_edges[0]["node"]["text"] if caption_edges else ""
            owner = media.get("owner", {})
            username = owner.get("username", "")

        folder_name = sanitise_folder_name(caption, shortcode)
        post_dir = os.path.join(base_dir, folder_name)
        if os.path.exists(post_dir):
            post_dir = f"{post_dir}_{shortcode}"
        os.makedirs(post_dir, exist_ok=True)

        cdn_urls = get_cdn_urls(media) if media else []
        if not cdn_urls:
            # Fallback: oEmbed
            oembed_url = f"https://api.instagram.com/oembed/?url={post_url}&omitscript=true"
            r = requests.get(oembed_url, headers=HEADERS, timeout=10)
            if r.status_code == 200:
                thumb = r.json().get("thumbnail_url", "")
                if thumb:
                    cdn_urls = [("image.jpg", thumb)]
                    username = r.json().get("author_name", "")

        downloaded_paths = []
        cdn_map = {}
        for filename, url in cdn_urls:
            dest = os.path.join(post_dir, filename)
            download_file(url, dest)
            downloaded_paths.append(dest)
            cdn_map[filename] = url
            result["files"].append(filename)

        if stitch_pdf_flag and len(downloaded_paths) > 1 and PILLOW_AVAILABLE:
            pdf_path = os.path.join(post_dir, "carousel.pdf")
            stitch_pdf(downloaded_paths, pdf_path)
            result["files"].append("carousel.pdf")

        post_type = "carousel" if len(cdn_urls) > 1 else "single_image"
        write_metadata(post_dir, post_url, shortcode, post_type, caption, username, cdn_map)
        result["files"].append("metadata.txt")

    except Exception as e:
        result["status"] = "error"
        result["error"] = str(e)

    return result


def write_metadata(post_dir, post_url, shortcode, post_type, caption, username, cdn_map):
    lines = [
        f"Post URL:       {post_url}",
        f"Shortcode:      {shortcode}",
        f"Type:           {post_type}",
    ]
    if post_type == "carousel":
        lines.append(f"Slide count:    {len([k for k in cdn_map if 'slide' in k])}")
    lines += [
        f"Caption:        {caption}",
        f"Username:       @{username}",
        f"Fetched at:     {datetime.now(timezone.utc).isoformat()}",
        "CDN URLs:",
    ]
    for fn, url in cdn_map.items():
        lines.append(f"  {fn:<18} {url}")
    with open(os.path.join(post_dir, "metadata.txt"), "w", encoding="utf-8") as f:
        f.write("\n".join(lines) + "\n")


def main(urls: list[str], base_dir: str = "./instagram-downloads", stitch: bool = True):
    os.makedirs(base_dir, exist_ok=True)
    results = []
    for url in urls:
        url = url.strip()
        if not url:
            continue
        print(f"Processing: {url}")
        r = process_url(url, base_dir, stitch)
        results.append(r)
        time.sleep(1)  # polite delay between requests

    # Summary
    ok = [r for r in results if r["status"] == "ok"]
    err = [r for r in results if r["status"] == "error"]
    total_files = sum(len(r["files"]) for r in ok)
    print("\nInstagram Post Downloader — Batch Complete")
    print("==========================================")
    print(f"URLs processed:   {len(results)}")
    print(f"Posts saved:      {len(ok)}")
    print(f"Total files:      {total_files}")
    print(f"Errors:           {len(err)}")
    print(f"Output dir:       {os.path.abspath(base_dir)}\n")
    for r in results:
        if r["status"] == "ok":
            print(f"  OK  {r['url']}")
        else:
            print(f"  ERR {r['url']}  — {r['error']}")


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: python instagram_downloader.py <url1> [url2] ...")
        sys.exit(1)
    main(sys.argv[1:])