3
\$\begingroup\$

I made a very simple Instagram Bot that can download images and videos of the user, like Gallery with photos or videos. It saves the data in the folder.

How it works:

  • Creating directory for saving images and videos

  • Using Selenium for links extraction

  • Check the availability of Instagram profile if it's private or existing

  • Using threads and multiprocessing improve execution speed

My code:

import string
import requests
import os
import time
from selenium import webdriver
from selenium.common.exceptions import WebDriverException
import sys
from multiprocessing.dummy import Pool
import random
import urllib.parse
import argparse
import re
from concurrent.futures import ThreadPoolExecutor

LINKS = []
PICTURES = []
VIDEO = []


def check_availability(link, cookies):
    """
    This function checks the availability of profile and the status code
    :param link:  link that searching for and includes the profile name
    :param cookies: cookies from class <MyHttpBase>
    :return: False if the <privacy> is True and <followed_by_viewer> is False
    """
    search = requests.get(urllib.parse.urljoin(link, "?__a=1"), cookies)
    if search.ok:
        load_and_check = search.json()
        privacy = load_and_check.get("graphql").get("user").get("is_private")
        followed_by_viewer = load_and_check.get("graphql").get("user").get("followed_by_viewer")
        if privacy and not followed_by_viewer:
            return False
    else:
        search.raise_for_status()


def URL_fetcher(url, cookies):
    """
    This function extracts images and videos
    :param url: Taking the url of array LINKS
    :param cookies: cookies from class <MyHttpBase>
    :return:
    """
    logging_page_id = requests.get(url.split()[0], cookies=cookies).json()
    try:
        """Taking Gallery Photos or Videos"""
        for i in range(len(logging_page_id['graphql']['shortcode_media']['edge_sidecar_to_children']['edges'])):
            video = \
                logging_page_id['graphql']['shortcode_media']['edge_sidecar_to_children']['edges'][i]['node'][
                    "is_video"]
            if video is True:
                video_url = \
                    logging_page_id['graphql']['shortcode_media']['edge_sidecar_to_children']['edges'][i][
                        'node'][
                        "video_url"]
                if video_url not in VIDEO:
                    VIDEO.append(video_url)

            else:
                image = \
                    logging_page_id['graphql']['shortcode_media']['edge_sidecar_to_children']['edges'][i][
                        'node'][
                        'display_url']
                if image not in PICTURES:
                    PICTURES.append(image)
    except KeyError:
        """Unique photo or Video"""
        image = logging_page_id['graphql']['shortcode_media']['display_url']
        if image not in PICTURES:
            PICTURES.append(image)

        if logging_page_id['graphql']['shortcode_media']["is_video"] is True:
            videos = logging_page_id['graphql']['shortcode_media']["video_url"]
            if videos not in VIDEO:
                VIDEO.append(videos)


class MyHttpBase:
    """
    Setting up a Requests session and pass it around
    """
    s = requests.Session()

    def setupCookies(self, COOKIES):
        for cookie in COOKIES:
            c = {cookie["name"]: cookie["value"]}
            self.s.cookies.update(c)

    def cookieJar(self):
        return self.s.cookies

    def close_session(self):
        return self.s.close()


class InstagramPV:

    def __init__(self, username, password, folder, search_name):
        """

        :param username: username
        :param password: password
        :param folder: folder name
        :param search_name: the name what will search
        """
        self.username = username
        self.password = password
        self.folder = folder

        """To avoid any errors, with regex find the url and taking the name <search_name>"""
        find_name = "".join(re.findall(r"(?P<url>https?://[^\s]+)", search_name))
        if find_name.startswith("https"):
            self.search_name = urllib.parse.urlparse(find_name).path.split("/")[1]
        else:
            self.search_name = search_name

        try:
            self.driver = webdriver.Chrome()
        except WebDriverException as e:
            print(str(e))
            sys.exit(1)

    def __enter__(self):
        return self

    def control(self):
        """
        Create the folder name and raises an error if already exists
        """
        if not os.path.exists(self.folder):
            os.mkdir(self.folder)
        else:
            raise FileExistsError("[*] Alredy Exists This Folder")

    def login(self):
        """Login To Instagram"""
        self.driver.get("https://www.instagram.com/accounts/login")
        time.sleep(3)
        self.driver.find_element_by_name('username').send_keys(self.username)
        self.driver.find_element_by_name('password').send_keys(self.password)
        submit = self.driver.find_element_by_tag_name('form')
        submit.submit()
        time.sleep(3)
        try:
            """Check For Invalid Credentials"""
            var_error = self.driver.find_element_by_class_name("eiCW-").text
            if len(var_error) > 0:
                print(var_error)
                sys.exit(1)
        except WebDriverException:
            pass

        try:
            self.driver.find_element_by_xpath('//button[text()="Not Now"]').click()
        except WebDriverException:
            pass
        time.sleep(2)
        """Taking Cookies To pass it in class <MyHttpBase>"""
        cookies = self.driver.get_cookies()

        MyHttpBase().setupCookies(cookies)
        COOKIES = MyHttpBase().cookieJar()

        self.driver.get("https://www.instagram.com/{name}/".format(name=self.search_name))

        """Checking the availability"""
        if not check_availability("https://www.instagram.com/{name}/".format(name=self.search_name), COOKIES):
            return self.scroll_down()

    def _get_href(self):
        elements = self.driver.find_elements_by_xpath("//a[@href]")
        for elem in elements:
            urls = elem.get_attribute("href")
            if "p" in urls.split("/"):
                LINKS.append(urls)

    def scroll_down(self):
        """Taking hrefs while scrolling down"""
        end_scroll = []
        while True:
            self.driver.execute_script("window.scrollTo(0, document.body.scrollHeight);")
            time.sleep(2)
            self._get_href()
            time.sleep(2)
            new_height = self.driver.execute_script("return document.body.scrollHeight")
            end_scroll.append(new_height)
            if end_scroll.count(end_scroll[-1]) > 4:
                self.extraction_url()
                break

    def extraction_url(self):
        """Gathering Images and Videos Using ThreadPoolExecutor and pass to function <URL_fetcher> """
        links = list(set(LINKS))
        print("[!] Ready for video - images".title())
        print("[*] extracting {links} posts , please wait...".format(links=len(links)).title())
        cookies = MyHttpBase().cookieJar()
        new_links = [urllib.parse.urljoin(link, "?__a=1") for link in links]
        with ThreadPoolExecutor(max_workers=8) as executor:
            [executor.submit(URL_fetcher, link, cookies) for link in new_links]

    def content_of_url(self, url):
        """
        :param url: the url
        :return: the content
        """
        re = requests.get(url)
        return re.content

    def _download_video(self, new_videos):
        """
        Saving the content of video in the file
        """
        with open(
                os.path.join(self.folder, "Video{}.mp4").format(
                    "".join([random.choice(string.digits) for i in range(20)])),
                "wb") as f:
            content_of_video = self.content_of_url(new_videos)
            f.write(content_of_video)

    def _images_download(self, new_pictures):
        """Saving the content of picture in the file"""
        with open(
                os.path.join(self.folder, "Image{}.jpg").format(
                    "".join([random.choice(string.digits) for i in range(20)])),
                "wb") as f:
            content_of_picture = self.content_of_url(new_pictures)
            f.write(content_of_picture)

    def downloading_video_images(self):
        """Using multiprocessing for Saving Images and Videos"""
        print("[*] ready for saving images and videos!".title())
        new_pictures = list(set(PICTURES))
        new_videos = list(set(VIDEO))
        pool = Pool(8)
        pool.map(self._images_download, new_pictures)
        pool.map(self._download_video, new_videos)
        print("[+] done".title())
        MyHttpBase().close_session()

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.driver.close()


if __name__ == '__main__':
    parser = argparse.ArgumentParser()

    parser.add_argument("-u", "--username", help='Username or your email of your account', action="store",
                        required=True)
    parser.add_argument("-p", "--password", help='Password of your account', action="store", required=True)
    parser.add_argument("-f", "--filename", help='Filename for storing data', action="store", required=True)
    parser.add_argument("-n", "--name", help='Name to search', action="store", required=True)
    args = parser.parse_args()

    with InstagramPV(args.username, args.password, args.filename, args.name) as pv:
        pv.control()
        pv.login()
        pv.downloading_video_images()

Simple Usage:

myfile.py -u example@hotmail.com -p mypassword -f myfile -n stackoverjoke

Even though it was just a way to learn a bit the Selenium, the main thing it was the data scraping, but became a simple 'download posts' bot.

Instagram scraper Posts (Videos and Photos) is the previous related question.

\$\endgroup\$
4
\$\begingroup\$

Requests raising

This pattern:

if search.ok:
    ...
else:
    search.raise_for_status()

is redundant. Just call raise_for_status(), which will be a no-op if the response is OK.

check_availability still has a confused return. If you're returning False, is there ever a point where it could return True? If not, why return a value at all? This boils down to the same problem that you had in the first question: either you should catch the set of exceptions that you expect from a request and return either True or False; or you should do what raise_for_status does - no-op if successful, raise if unsuccessful.

Temporary variables

You should use one for logging_page_id['graphql']['shortcode_media']['edge_sidecar_to_children']['edges']. Also, this pattern:

for i in range(len(things)):
    .. use things[i]

is counter to idiomatic Python iteration. Instead,

for thing in things:

Uniqueness

If you don't care about the order of PICTURES, make it a set. That way, you don't need this if:

        if videos not in VIDEO:
            VIDEO.append(videos)

The same applies to your images list.

Nomenclature

URL_fetcher (more specifically URLFetcher) would have been appropriate as a class name, but it's no longer a class, it's a function. So call it fetch_url.

Cookie handling

Your handling of cookies is certainly better than last time, but I still think you should take this a step further and try assigning them to a session and passing the session around instead. Then, instead of requests.get(url, cookies), you can simply write session.get(url).

Looking further down - you have this MyHttpBase with a session in it. First of all, you've made s effectively a class static, which you shouldn't - it should be in instance scope. That aside, I don't think MyHttpBase should exist at all. Have a read through https://2.python-requests.org/en/master/api/#api-cookies

Exception handling

This:

    try:
        self.driver = webdriver.Chrome()
    except WebDriverException as e:
        print(str(e))
        sys.exit(1)

should really not be done in the scope of a class init function. If you want to print exceptions, fine; do it at the top level in main.

Also, this pattern:

    try:
        self.driver.do_something()
    except WebDriverException:
        pass

is almost certainly not what you actually want to happen. If it's actually somewhat OK for the driver to explode, at the least you'd want to print a warning about it. But why is it OK for the driver to explode?

f-strings

"[*] extracting {links} posts , please wait...".format(links=len(links))

can be

f"[*] extracting {len(links)} posts; please wait..."

Typo

Alredy = Already

Booleans

Replace if video is True: with if video:.

Static methods

content_of_url shouldn't exist. Even if it did need to exist, it should be made a static method, or more likely a utility function outside of the class.

Random filenames

            os.path.join(self.folder, "Image{}.jpg").format(
                "".join([random.choice(string.digits) for i in range(20)])),

I'm sure given the information you're scraping from IG that you can do better than this. Even if you can't, at least use something like a timestamp, which is both reasonably guaranteed to be unique as well as meaningful to the user.

Context manager

You've done a good job in implementing a context manager to close your driver. However, this:

    MyHttpBase().close_session()

is (a) done in the wrong function - it should be done in __exit__; and (b) should simply be manipulating a Requests session object directly.

Main method

Put the last 11-ish lines of your program into a main method. Currently they're in global scope.

| improve this answer | |
\$\endgroup\$
  • \$\begingroup\$ I don't know how to thank you. Now, in the check_availability i prefer to raise an error if its unsuccessful. My first thought to pass around the cookies was to create a class. I don't understand very good the part "try assigning them to a session and passing the session around instead". You mean to replace? and second, would be better if i let the driver to "explode" without control the exceptions? and why? THANKS!!!!!! \$\endgroup\$ – AlexDotis Mar 15 at 16:02
  • 1
    \$\begingroup\$ I don't know how to thank you - Return the favour; stay an active member in this community :) \$\endgroup\$ – Reinderien Mar 15 at 16:21
  • 1
    \$\begingroup\$ I don't understand very good the part "try assigning them to a session and passing the session around instead". - Where you currently pass the cookies in as a parameter, instead pass the session itself. \$\endgroup\$ – Reinderien Mar 15 at 16:22
  • 1
    \$\begingroup\$ would be better if i let the driver to "explode" without control the exceptions? - That depends on a lot of things. What particular exceptions are coming out of this section of the code? If a driver call fails, shouldn't you know about it? And is the failure fatal (i.e. it indicates that we probably won't be able to process the rest of the page), or not? At the least, it should probably somehow be printed on the terminal. \$\endgroup\$ – Reinderien Mar 15 at 16:24
  • \$\begingroup\$ I am sorry for asking all the time. One last question. I am thinking to open a session in the __init__ while i am passing the cookies, passing around the session and close it in __exit__ . Is it inappropriate? I mean like "code smell" \$\endgroup\$ – AlexDotis Mar 15 at 16:57

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.