An Instagram Bot which downloads the posts from profile
I have to mention my previous posts:
- Instagram scraper Posts (Videos and Photos)
- Scraping Instagram with selenium, extract URLs, download posts
My code:
import requests
import os
import time
from selenium import webdriver
from selenium.common.exceptions import NoSuchElementException
from multiprocessing.dummy import Pool
import urllib.parse
import argparse
import re
from concurrent.futures import ThreadPoolExecutor
LINKS = []
PICTURES = []
VIDEO = []
chromedriver_path = None
def check_availability(link, session_base):
"""
This function checks the availability of profile and the status code
:param session_base: The requests session
:param link: link that searching for and includes the profile name
:return: raise Exception if <privacy> is True and <followed_by_viewer> is False
"""
search = session_base.get(urllib.parse.urljoin(link, "?__a=1"))
search.raise_for_status()
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:
raise Exception("[!] Account is private")
def fetch_url(url, session_base):
"""
This function extracts images and videos
:param session_base: The requests session
:param url: Taking the url of array LINKS
"""
logging_page_id = session_base.get(url.split()[0]).json()
try:
"""Taking Gallery Photos or Videos"""
for log_pages in logging_page_id['graphql']['shortcode_media']['edge_sidecar_to_children']['edges']:
video = log_pages.get("node").get("is_video")
if video:
video_url = log_pages.get("node").get("video_url")
VIDEO.append(video_url)
else:
image = log_pages.get("node").get("display_url")
PICTURES.append(image)
except KeyError:
"""Unique photo or Video"""
image = logging_page_id.get('graphql').get('shortcode_media').get('display_url')
PICTURES.append(image)
if logging_page_id.get('graphql').get('shortcode_media').get("is_video"):
videos = logging_page_id.get('graphql').get('shortcode_media').get("video_url")
VIDEO.append(videos)
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
self.HttpBase = requests.Session()
"""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
if chromedriver_path is not None:
self.driver = webdriver.Chrome(chromedriver_path)
else:
self.driver = webdriver.Chrome()
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("[*] Already 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)
"""Check For Invalid Credentials"""
try:
var_error = self.driver.find_element_by_class_name("eiCW-").text
raise ValueError("[!] Invalid Credentials")
except NoSuchElementException:
pass
try:
"""Close Notifications"""
self.driver.find_element_by_xpath('//button[text()="Not Now"]').click()
except NoSuchElementException:
pass
time.sleep(2)
"""Taking Cookies and update the self.HttpBase"""
cookies = self.driver.get_cookies()
for cookie in cookies:
c = {cookie["name"]: cookie["value"]}
self.HttpBase.cookies.update(c)
self.driver.get("https://www.instagram.com/{name}/".format(name=self.search_name))
"""Checking for availability"""
check_availability("https://www.instagram.com/{name}/".format(name=self.search_name), self.HttpBase)
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 <fetch_url> """
links = list(set(LINKS))
print("[!] Ready for video - images".title())
print(f"[*] extracting {len(links)} posts , please wait...".title())
new_links = [urllib.parse.urljoin(link, "?__a=1") for link in links]
with ThreadPoolExecutor(max_workers=8) as executor:
[executor.submit(fetch_url, link, self.HttpBase) for link in new_links]
def _download_video(self, new_videos):
"""
Saving the content of video in the file
"""
number, link = new_videos
with open(os.path.join(self.folder, f"Video{number}.mp4"), "wb") as f:
content_of_video = InstagramPV.content_of_url(link)
f.write(content_of_video)
def _images_download(self, new_pictures):
"""Saving the content of picture in the file"""
number, link = new_pictures
with open(os.path.join(self.folder, f"Image{number}.jpg"), "wb") as f:
content_of_picture = InstagramPV.content_of_url(link)
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))
picture_data = [i for i in enumerate(new_pictures)]
video_data = [i for i in enumerate(new_videos)]
pool = Pool(8)
pool.map(self._images_download, picture_data)
pool.map(self._download_video, video_data)
print("[+] Done")
def __exit__(self, exc_type, exc_val, exc_tb):
self.HttpBase.close()
self.driver.close()
@staticmethod
def content_of_url(url):
req = requests.get(url)
return req.content
def 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()
if __name__ == '__main__':
main()
```
