|
6 | 6 | from instagrapi import Client
|
7 | 7 | from cryptography.fernet import Fernet
|
8 | 8 | import os
|
| 9 | +from dataclasses import dataclass |
| 10 | +from typing import Optional |
| 11 | + |
| 12 | +@dataclass |
| 13 | +class Settings: |
| 14 | + uuids: dict |
| 15 | + mid: str |
| 16 | + ig_u_rur: Optional[str] |
| 17 | + ig_www_claim: Optional[str] |
| 18 | + authorization_data: dict |
| 19 | + cookies: dict |
| 20 | + last_login: float |
| 21 | + device_settings: dict |
| 22 | + user_agent: str |
| 23 | + country: str |
| 24 | + country_code: int |
| 25 | + locale: str |
| 26 | + timezone_offset: int |
| 27 | + |
| 28 | + @classmethod |
| 29 | + def from_dict(cls, data: dict) -> 'Settings': |
| 30 | + return cls( |
| 31 | + uuids=data.get('uuids', {}), |
| 32 | + mid=data.get('mid', ''), |
| 33 | + ig_u_rur=data.get('ig_u_rur'), |
| 34 | + ig_www_claim=data.get('ig_www_claim'), |
| 35 | + authorization_data=data.get('authorization_data', {}), |
| 36 | + cookies=data.get('cookies', {}), |
| 37 | + last_login=data.get('last_login', 0.0), |
| 38 | + device_settings=data.get('device_settings', {}), |
| 39 | + user_agent=data.get('user_agent', ''), |
| 40 | + country=data.get('country', ''), |
| 41 | + country_code=data.get('country_code', 0), |
| 42 | + locale=data.get('locale', ''), |
| 43 | + timezone_offset=data.get('timezone_offset', 0) |
| 44 | + ) |
| 45 | + |
| 46 | + def to_dict(self) -> dict: |
| 47 | + return { |
| 48 | + 'uuids': self.uuids, |
| 49 | + 'mid': self.mid, |
| 50 | + 'ig_u_rur': self.ig_u_rur, |
| 51 | + 'ig_www_claim': self.ig_www_claim, |
| 52 | + 'authorization_data': self.authorization_data, |
| 53 | + 'cookies': self.cookies, |
| 54 | + 'last_login': self.last_login, |
| 55 | + 'device_settings': self.device_settings, |
| 56 | + 'user_agent': self.user_agent, |
| 57 | + 'country': self.country, |
| 58 | + 'country_code': self.country_code, |
| 59 | + 'locale': self.locale, |
| 60 | + 'timezone_offset': self.timezone_offset |
| 61 | + } |
| 62 | + |
9 | 63 |
|
10 | 64 | class Authorization:
|
11 | 65 | def __init__(
|
12 | 66 | self,
|
13 | 67 | username: str = None,
|
14 | 68 | password: str = None,
|
15 | 69 | token_path: str = "",
|
| 70 | + settings: Settings = None, |
16 | 71 | ):
|
17 | 72 | self.username = username
|
18 | 73 | self.password = password
|
19 | 74 | self.token_path = token_path if token_path else "threads_token.bin"
|
20 |
| - |
| 75 | + self.settings = settings |
21 | 76 | self.headers = get_default_headers()
|
22 | 77 |
|
23 | 78 | def generate_key_from_password(self, password):
|
@@ -47,20 +102,33 @@ def _retrieve_token(self):
|
47 | 102 |
|
48 | 103 |
|
49 | 104 | def get_instagram_api_token(self, refresh: bool = False) -> Union[str, None]:
|
50 |
| - token = self._retrieve_token() |
| 105 | + # token = self._retrieve_token() |
| 106 | + token = None |
51 | 107 | if token is not None and not refresh:
|
52 | 108 | return token
|
53 | 109 |
|
54 | 110 | try:
|
55 | 111 | iapi = Client()
|
56 |
| - iapi.login(self.username, self.password) |
| 112 | + if self.settings is not None: |
| 113 | + print("cached") |
| 114 | + iapi.set_settings(self.settings.to_dict()) |
| 115 | + iapi.login(self.username, self.password) |
| 116 | + else: |
| 117 | + print("not cached") |
| 118 | + iapi.login(self.username, self.password) |
| 119 | + self.settings = Settings.from_dict(iapi.get_settings()) |
| 120 | + |
57 | 121 | token = iapi.private.headers['Authorization'].split("Bearer IGT:2:")[1]
|
| 122 | + print(token) |
58 | 123 | self._store_token(token)
|
59 | 124 | return token
|
60 | 125 | except Exception as e:
|
61 | 126 | print(e)
|
62 | 127 | raise
|
63 | 128 |
|
| 129 | + def get_settings(self) -> Settings: |
| 130 | + return self.settings |
| 131 | + |
64 | 132 | def get_public_api_token(self) -> str:
|
65 | 133 | response = requests.get(
|
66 | 134 | url='https://www.instagram.com/instagram',
|
|
0 commit comments