4
\$\begingroup\$

I wrote this code which has the ability to download images and videos from a specific Instagram profile.

Using multiprocessing and threading I managed to speed up the extraction of data.

My goal is to achieve:

  1. Make it faster (if it is possible)
  2. Writing less code (if it is possible)
  3. Using better methods (if it is possible)
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 threading

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


class Errors:
    """Checking Instagram Profiles"""

    def __init__(self, link, cookies=None):
        self.link = urllib.parse.urljoin(link, "?__a=1")
        self.cookies = cookies
        if self.cookies is not None:
            self.cookies = cookies

    def availability(self):
        """
        Check The Profile Availability
        From status_code and from graphql json that provides the link https://www.instagram.com/{username}/?__a=1
        :return: True, If it's not private or its available
        """

        search = requests.get(self.link, self.cookies)
        if search.status_code == 404:
            return "Sorry, this page isn't available."
        elif search.json()["graphql"]["user"]["is_private"] is True:
            return "This Account is Private"
        else:
            return True


class fetch_urls(threading.Thread):

    def __init__(self, url, cookies=None):
        threading.Thread.__init__(self)
        self.cookies = cookies
        if self.cookies is not None:
            self.cookies = cookies
        self.url = url

    def run(self):
        """Extract Images or Videos From Every Url Using json and graphql"""
        logging_page_id = requests.get(self.url.split()[0], cookies=COOKIES).json()
        try:
            """Taking Url from 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 url from 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 Instagram_pv:

    def close(self):
        self.driver.close()

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

        :param username: The username
        :param password: The password
        :param folder: The folder name that images and videos will be saved
        :param name: The instagram name that will search
        """
        self.username = username
        self.password = password
        self.name = name
        self.folder = folder
        try:
            self.driver = webdriver.Chrome()
        except WebDriverException as e:
            print(str(e))
            sys.exit(1)

    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:
            self.close()
            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 requests If the Profile is Private and you are following, 
        otherwise the data from graphql will be incomplete"""
        cookies = self.driver.get_cookies()
        needed_cookies = ['csrftoken', 'ds_user_id', 'ig_did', 'mid', 'sessionid']
        global COOKIES
        COOKIES = {cookies[i]['name']: cookies[i]['value'] for i in range(len(cookies)) if
                   cookies[i]['name'] in needed_cookies}

        self.driver.get("https://www.instagram.com/{name}/".format(name=self.name))
        """From The Class <Errors> Checking the Profile Availability"""
        error = Errors("https://www.instagram.com/{name}/".format(name=self.name), COOKIES).availability()
        if error is not True:
            print(error)
            self.close()
            sys.exit(1)
        else:
            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.close()
                self.extraction_url()
                break

    def extraction_url(self):
        """Gathering Images and Videos Using Threads From Class <fetch_urls>"""
        links = list(set(LINKS))
        print("[!] Ready for video - images".title())
        print("[*] extracting {links} posts , please wait...".format(links=len(links)).title())
        for url in LINKS:
            new_link = urllib.parse.urljoin(url, '?__a=1')
            fetch_urls(new_link).start()
        for thread in threading.enumerate():
            if thread is not threading.currentThread():
                thread.join()

    def content_of_url(self, url):
        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())


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()

    ipv = Instagram_pv(args.username, args.password, args.filename, args.name)
    ipv.control()
    ipv.login()
    ipv.downloading_video_images()

Usage of code:

myfile.py -u example@hotmail.com -p mypassword -f myfile -n stackoverjoke
\$\endgroup\$
  • \$\begingroup\$ Can you explain what the code is doing? I see no documentation, so having to reverse engineer the code is going to be annoying. \$\endgroup\$ – AMC Mar 13 at 23:10
  • \$\begingroup\$ @AMC You are right. I already updated! I tried to write how it works, i hope will not need more. Thanks for the observation! \$\endgroup\$ – AlexDotis Mar 14 at 1:50
2
\$\begingroup\$

Error management

This:

        if search.status_code == 404:
            return "Sorry, this page isn't available."
        elif search.json()["graphql"]["user"]["is_private"] is True:
            return "This Account is Private"
        else:
            return True

is problematic. First of all, you're mixing return types (boolean and string). More importantly: returning "string if error or true-boolean otherwise" is a nasty mix of in-band error signalling, mixing of user display vs. business logic concerns, and tight coupling.

Instead of this Errors class, you could consider writing a method check_availability, which

  • accepts the same parameters as your __init__
  • fires off the same request
  • calls search.raise_for_status(), potentially catching and wrapping any exception that arises - this will cover your 404 and dozens of other HTTP errors
  • does the JSON load-and-check, and if there are any issues, raise an exception. It's important to note that your ["graphql"]["user"]["is_private"] is fragile, so any key errors will currently be thrown with no additional information. Either catch and wrap that key error, or be more careful and use get on those nested dictionaries.
  • If nothing is wrong, do not throw an exception and simply return.

The above, combined with custom exception types, will make your program much more programmer-friendly and will improve the structure of your code.

Case conventions

fetch_urls should be FetchURLs. That said, its name makes it sound like a function when it's currently a class. So either:

  • Just make it a function; it's not really useful as a class anyway; or
  • Name it something like URLFetcher.

Puzzling None logic

        self.cookies = cookies
        if self.cookies is not None:
            self.cookies = cookies

Not sure what you were going for here. The if can be dropped altogether, as it doesn't affect what will land in self.cookies.

Cookie management

This:

        global COOKIES
        COOKIES = {cookies[i]['name']: cookies[i]['value'] for i in range(len(cookies)) if
                   cookies[i]['name'] in needed_cookies}

is also problematic. First of all, globals are a code smell, particularly when they're set from outside of global scope like this. Second of all, rather than manipulating these cookies yourself, you may want to set up a Requests session object and pass it around to those who need to use requests with that cookie jar. Advantages are that any additional cookies modified in the following web traffic will be obeyed. Disadvantages are that you may end up carrying around state that (a) you don't care about, or (b) actively harms your workflow; but these are unlikely.

Method order

    def close(self):
        self.driver.close()

should appear after __init__, which should almost always appear first. Also: since you have a driver that needs closing, you should make InstagramPV a context manager and refer to it using a with block so that the driver gets closed regardless of any exceptions that take place.

| improve this answer | |
\$\endgroup\$
  • \$\begingroup\$ Perfect explanation!!! I have only one question. What do you mean by "Just make it a function; it's not really useful as a class anyway;" Also, can i update my post using your advises ? Thank you so much! \$\endgroup\$ – AlexDotis Mar 14 at 3:44
  • \$\begingroup\$ can i update my post using your advises - Please don't; CR policy is that you open a new question after each round of review. \$\endgroup\$ – Reinderien Mar 14 at 3:53
  • 1
    \$\begingroup\$ What do you mean by "Just make it a function" - You haven't really gained anything by making fetch_urls a class. It's a fairly reasonable implementation of threading.Thread, but it could just as well be spawned as a function passed to thread.start_new_thread. \$\endgroup\$ – Reinderien Mar 14 at 3:58
  • 1
    \$\begingroup\$ n.b. rather than start_new_thread, if you're in Python 3, you can pass a function to the target parameter of Thread. \$\endgroup\$ – Reinderien Mar 14 at 4:00
  • 1
    \$\begingroup\$ The best place to show me is in a new question :) \$\endgroup\$ – Reinderien Mar 14 at 19:09

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.