-
Notifications
You must be signed in to change notification settings - Fork 1
/
fetch_data.py
153 lines (115 loc) · 5.77 KB
/
fetch_data.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
from rich.console import Console
from PIL import Image
import requests
import json
import os
def fetch_username_by_id(session: requests.session, user_id: str) -> str:
"""
Function that fetch the IG username linked to the ID.
:param session: Session that contain cookies and headers.
:param user_id: Instagram User ID.
:return: The username linked to the user ID.
"""
username = None
data = {}
if os.path.isfile(os.path.join('cache', 'saved_usernames.json')):
# check if cache file isn't empty.
if os.stat(os.path.join('cache', 'saved_usernames.json')).st_size != 0 :
with open(os.path.join('cache', 'saved_usernames.json'), 'r') as save_file:
data = json.load(save_file)
# Check if the username is in the cache if not request the username and save it.
username = data.get(str(user_id))
# delete empty cache file if it's empty.
else: os.remove(os.path.join('cache', 'saved_usernames.json'))
if username is None:
req = session.get(f'https://i.instagram.com/api/v1/users/{user_id}/info/')
# Check if the request is successful.
if req.status_code == 200:
# Fetch the username and save it into the cache/saved_usernames.json file.
username = req.json()['user'].get('username')
data[str(user_id)] = username
with open(os.path.join('cache', 'saved_usernames.json'), 'w') as save_file:
json.dump(data, save_file)
else:
# The connection was not successful.
print(f' -> [{req.status_code}] Error when fetching instagram server.')
return username
def fetch_profile_pic_by_id(session: requests.Session, user_id: str) -> str:
"""
Function that fetch the IG profile picture linked to the ID.
:param session: Session that contain cookies and headers.
:param user_id: Instagram User ID.
:return: The 4 colors of the profile picture.
"""
colors = None
data = {}
if os.path.isfile(os.path.join('cache', 'saved_profile_pic.json')):
if os.stat(os.path.join('cache', 'saved_profile_pic.json')).st_size != 0 : # check if cache file isn't empty
with open(os.path.join('cache', 'saved_profile_pic.json'), 'r') as save_file:
data = json.load(save_file)
# Check if the profile picture is in the cache if not request it and save it.
colors = data.get(str(user_id))
else:
os.remove(os.path.join('cache', 'saved_profile_pic.json')) # delete empty cache file
if colors is None:
colors = []
req = session.get(f'https://i.instagram.com/api/v1/users/{user_id}/info/')
# Check if the request is successful.
if req.status_code == 200:
# Fetch the profile picture url and request it.
profile_pic_url = req.json()['user'].get('profile_pic_url')
req = session.get(profile_pic_url, stream=True)
# Check if the request is successful.
if req.status_code == 200:
profile_pic = Image.open(req.raw).resize((2, 2))
for index_width in range(profile_pic.width):
for index_height in range(profile_pic.height):
r, g, b = profile_pic.getpixel((index_width, index_height))
colors.append('#{:02x}{:02x}{:02x}'.format(r, g, b))
data[str(user_id)] = colors
with open(os.path.join('cache', 'saved_profile_pic.json'), 'w') as save_file:
json.dump(data, save_file)
else:
# The request failed, set every PP to red.
colors = ['#ff0000', '#ff0000', '#ff0000', '#ff0000']
return f'[{colors[0]} on {colors[1]}]▄[/][{colors[2]} on {colors[3]}]▄[/]'
def fetch_img_from_url(session: requests.Session, console: Console, url: str) -> str:
"""
Function that fetch an image from a URL and format it to be able to render it in the _console.
:param session: Session that contain cookies and headers.
:param console: Console object that contain every info on the current console.
:param url: URL of the image.
:return: The formatted image ready to be rendered.
"""
# Fetch raw data of the image.
req = session.get(str(url), stream=True)
# check if the request is successful.
if req.status_code == 200:
# Open and resize the image.
img = Image.open(req.raw)
img_ratio = img.width / img.height
# The resize depend on if the image is portrait or landscape.
if img.height > img.width:
# Image is portrait, define a new height and find the new width using the image ratio.
new_height = console.height // 2
new_width = int(img_ratio * new_height) * 2
else:
# Image is landscape or square, define a new width and find the new height using the image ratio.
new_width = console.width // 2
new_height = int(img_ratio * new_width) // 2
img = img.resize((new_width, new_height))
# Format the image into a string.
formatted_img = ''
for line in range(img.height - 1):
for pixel in range(img.width):
r1, g1, b1 = img.getpixel((pixel, line + 1))
r2, g2, b2 = img.getpixel((pixel, line))
formatted_img += '[#{:02x}{:02x}{:02x} on #{:02x}{:02x}{:02x}]▄[/]'.format(r1, g1, b1, r2, g2, b2)
formatted_img += '\n'
else:
# If the request was not successful, return and error message.
formatted_img = '[red]Cannot be loaded...[/]'
# return the formatted image.
return formatted_img
if __name__ == '__main__':
print('This code is intended to be imported...')