Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- def get_requests_session(i_auth=None):
- link = 'https://www.instagram.com/accounts/login/'
- login_url = 'https://www.instagram.com/accounts/login/ajax/'
- s = requests.Session()
- incorrect_auth_pk = set()
- if i_auth is None:
- i_auth = InstagramAuth()
- iter_counter = len(i_auth.all_auth_list) or 1
- while iter_counter > -1:
- if i_auth.proxy:
- proxy = i_auth.proxy.proxies
- else:
- proxy = {}
- time = int(datetime.datetime.now().timestamp())
- payload = {
- 'username': i_auth.username,
- 'enc_password': f'#PWD_INSTAGRAM_BROWSER:0:{time}:{i_auth.password}',
- # <-- note the '0' - that means we want to use plain passwords
- 'queryParams': {},
- 'optIntoOneTap': 'false'
- }
- r = s.get(link)
- csrf = re.findall(r"csrf_token\":\"(.*?)\"", r.text)[0]
- logger.info(f"Try login with username: {i_auth.username} password: {i_auth.password}")
- r = s.post(login_url, data=payload, proxies=proxy, headers={
- "user-agent": "Mozilla/5.0 (Windows NT 6.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/77.0.3865.120 Safari/537.36",
- "x-requested-with": "XMLHttpRequest",
- "referer": "https://www.instagram.com/accounts/login/",
- "x-csrftoken": csrf
- })
- try:
- answer = r.json()
- if answer['authenticated']:
- logger.info("Success auth!")
- return s
- except:
- answer = {'authenticated': False}
- logger.info(answer)
- incorrect_auth_pk.add(i_auth.pk)
- logger.info(f"Auth data {i_auth.username} is broken. Try next. Has {len(i_auth.all_auth_list)} values")
- i_auth.from_redis()
- if incorrect_auth_pk == set(i_auth.all_auth_list):
- raise AllInstagramDatasBroken('No more datas')
- logger.info(f"Try auth with {i_auth.username}")
- iter_counter -= 1
- def requests_session_get(link: str, i_auth: InstagramAuth, session: Session):
- i_auth.get_headers_from_string()
- logger.info(f'Parse requests with {i_auth.username} login. link - {link}')
- if i_auth.proxy:
- proxy = i_auth.proxy.proxies
- else:
- proxy = {}
- logger.info(proxy)
- # session = requests_login(session=session, auth=auth)
- logger.info(f"Session after: {session}")
- r = session.get(url=link, proxies=proxy, allow_redirects=True)
- logger.info(f"Request ends with {r.status_code}")
- logger.info(f"Answer is: \n {r.content[:100]}")
- answer = {}
- try:
- answer = r.json()
- except simplejson.errors.JSONDecodeError as err:
- logger.error(f"Error in parsing {r.url}")
- logger.error(err)
- i_auth.state.downgrade()
- i_auth.to_redis()
- try:
- session.close()
- sleep(3)
- i_auth.from_redis()
- session = get_requests_session(i_auth)
- r = session.get(url=link, proxies=proxy, allow_redirects=True)
- logger.info(f"Request ends with {r.status_code}")
- logger.info(f"Answer is: \n {r.content[:100]}")
- answer = r.json()
- except AllInstagramDatasBroken as err:
- logger.error(f"All instagram datas is broken")
- logger.error(err)
- raise InstagramAuthDataError(f"Broken auth data")
- except Exception as err:
- logger.error(f"answer url is {r.url} reason {r.reason}")
- logger.error(err)
- sleep(1)
- return answer, session
- def get_comment(victim: str, auth: InstagramAuth, session: Session):
- victim_hash = victim.split('/')[-1]
- comment_hash = auth.comments_hash
- logger.info(f"Comment hash {comment_hash}")
- url_base = 'https://www.instagram.com/graphql/query/?'
- after = None
- error_counter = 0
- comments = []
- while True:
- logger.info(f'After is {after}')
- after_value = f',"after":"{after}"' if after else ''
- after_value = after_value.replace(': ', ':').replace(", ", ",")
- variables = f'{{"shortcode":"{victim_hash}","first":20{after_value}}}'
- get_params = {
- 'query_hash': comment_hash,
- 'variables': variables
- }
- current_link = url_base + urlencode(get_params)
- try:
- data, session = requests_session_get(current_link, i_auth=auth, session=session)
- # data = curl_get(current_link, auth=auth)
- logger.info(f"Comment data is {data}")
- test_data = data['data']
- except KeyError as err:
- error_counter += 1
- logger.error(err)
- logger.error(f"incorrect parsing link {unquote(current_link)}\n {error_counter} try to sleep ")
- sleep(1)
- if error_counter < 5:
- continue
- return comments, session
- except InstagramAuthDataError as err:
- raise InstagramAuthDataError(err)
- else:
- for node in data["data"]["shortcode_media"]["edge_media_to_parent_comment"]['edges']:
- logger.info(f"{node['node']['text']} from {node['node']['owner']['username']}")
- template = f"{node['node']['owner']['username']} написал: {node['node']['text']}"
- comments.append(template)
- else:
- if data["data"]["shortcode_media"]["edge_media_to_parent_comment"]['page_info']['has_next_page']:
- after = data["data"]["shortcode_media"]["edge_media_to_parent_comment"]['page_info']['end_cursor']
- sleep(2)
- continue
- else:
- return comments, session
- return comments, session
- def parse(link: str):
- if not link.startswith('https://www.instagram.com/p/') and not link.startswith('https://www.instagram.com/tv/'):
- raise ValueError(f'Incorrect link {link}')
- if link.count("?"):
- link = link.split("?")[0]
- post_hash = link.split('/')[-1]
- i_auth = InstagramAuth()
- session = get_requests_session(i_auth)
- comments = get_comment(victim=post_hash, auth=i_auth, session=session)
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement