Skip to content

Instantly share code, notes, and snippets.

@jpatrickdill
Last active July 21, 2020 02:40
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save jpatrickdill/097cfd1f66d0e07ebde888f026563803 to your computer and use it in GitHub Desktop.
Save jpatrickdill/097cfd1f66d0e07ebde888f026563803 to your computer and use it in GitHub Desktop.
indexes all images from an instagram thread
from selenium import webdriver
import requests
import figa
from redis import Redis
from pprint import pp
@figa.config
class Config:
default = "config.json"
def get_instagram_session():
opts = webdriver.ChromeOptions()
opts.add_argument("headless")
opts.add_argument('window-size=1280x720')
driver = webdriver.Chrome() # chrome_options=opts)
driver.implicitly_wait(3)
driver.get("https://instagram.com/")
input("Press enter when logged in.")
session = requests.Session()
for cookie in driver.get_cookies():
session.cookies.set(cookie["name"], cookie["value"])
session.headers.update({
"referer": "https://www.instagram.com/direct/inbox",
"user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/81.0.40"
"44.138 Safari/537.36 OPR/68.0.3618.173",
})
driver.quit()
return session
if __name__ == "__main__":
cfg = Config("default")
db = Redis(cfg.redis.host, cfg.redis.port, password=cfg.redis.password, decode_responses=True)
session = get_instagram_session()
session.headers.update({
"x-ig-app-id": cfg.app_id,
"x-ig-www-claim": "hmac.AR2Fmty-ITmxPw22Wm2U2sYqm92w3eYApVAlxSj2beCUI8xM",
"x-requested-with": "XMLHttpRequest"
})
threads = session.get(
"https://www.instagram.com/direct_v2/web/inbox/?persistentBadging=true&folder=&limit=10&thread_message_limit=10"
)
thread = None
for t in threads.json()["inbox"]["threads"]:
if t["thread_id"] == cfg.thread_id:
thread = t
break
if thread is None:
raise KeyError("Thread not found")
cursor = thread["newest_cursor"]
while True:
resp = session.get("https://www.instagram.com/direct_v2/web/threads/{}/".format(cfg.thread_id),
params={"cursor": cursor}).json()
for message in resp["thread"]["items"]:
if message["item_type"] == "media":
if "image_versions2" in message["media"]:
versions = sorted(message["media"]["image_versions2"]["candidates"],
key=lambda c: c["width"] * c["height"])
largest = versions[-1]
db.set("images:{}".format(message["media"]["id"]), largest["url"])
else:
pp(message["media"])
cursor = resp["thread"]["prev_cursor"]
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment