Browse Source

Update 'pluralsight.py'

master
parent
commit
6d5af68bff
1 changed files with 78 additions and 62 deletions
  1. 78
    62
      pluralsight.py

+ 78
- 62
pluralsight.py View File

@@ -33,7 +33,10 @@ SELENIUM_TIMEOUT_IN_SECONDS = 30
MAX_PARALLEL_COURSE_DOWNLOAD_COUNT = 3

# Maximum number of retry count for downloads
DOWNLOAD_RETRY_COUNT = 3
DOWNLOAD_RETRY_COUNT = 100

# Maximum number of account registrations allowed per IP
MAX_ACCOUNT_PER_IP = 5

# Minimum number of seconds to wait between consecutive video downloads
MIN_VIDEO_DOWNLOAD_DELAY = 10
@@ -71,13 +74,10 @@ ydl_options = {
'nooverwrites': True,
'sleep_interval': MIN_VIDEO_DOWNLOAD_DELAY,
'max_sleep_interval': MAX_VIDEO_DOWNLOAD_DELAY,
'retries': 100,
'download_archive': ARCHIVE_FILE_PATH
}

# List of links of already downloaded courses
DOWNLOADED_COURSE_LINKS_LIST = []


# endregion


@@ -118,7 +118,6 @@ class DisposableMail:
mail_id = result[0]

return mail_id

except:
pass

@@ -193,8 +192,6 @@ class Pluralsight:
.until(expected_conditions
.presence_of_element_located((By.XPATH, "//*[contains(text(), 'I agree, activate benefit')]")))

# time.sleep(SELENIUM_SLEEP_DURATION)

create_account_button_element = self.driver.find_element_by_xpath(
"//*[contains(text(), 'I agree, activate benefit')]")
create_account_button_element.click()
@@ -203,13 +200,9 @@ class Pluralsight:
.until(expected_conditions
.presence_of_element_located((By.CLASS_NAME, "cancelButton---CKAut")))

# time.sleep(SELENIUM_SLEEP_DURATION*3)

cancel_button_element = self.driver.find_element_by_class_name("cancelButton---CKAut")
cancel_button_element.click()

# time.sleep(SELENIUM_SLEEP_DURATION)

def set_password(self, verification_link: str) -> None:
"""
Sets password in the given verification link
@@ -224,8 +217,6 @@ class Pluralsight:
.until(expected_conditions
.presence_of_element_located((By.ID, "Password")))

# time.sleep(SELENIUM_SLEEP_DURATION)

password_input_element = self.driver.find_element_by_id("Password")
password_confirm_input_element = self.driver.find_element_by_id("PasswordConfirmation")
save_button_element = self.driver.find_element_by_class_name("psds-button--appearance-primary")
@@ -239,8 +230,6 @@ class Pluralsight:
.until(expected_conditions
.presence_of_element_located((By.ID, "Password")))

# time.sleep(SELENIUM_SLEEP_DURATION)


# endregion

@@ -335,6 +324,20 @@ def get_directory_full_path(root_directory_path: str, pluralsight_path: Pluralsi
return directory_full_path


def save_progress(course_link: str) -> None:
with open(DOWNLOADED_COURSES_FILE_LOCATION, 'a+') as downloaded_file:
downloaded_file.write(f"{course_link}\n")


def is_downloaded(course_link: str) -> bool:
if not os.path.isfile(DOWNLOADED_COURSES_FILE_LOCATION):
return False

downloaded_courses_list = list(filter(None, Path(DOWNLOADED_COURSES_FILE_LOCATION).read_text().split('\n')))

return course_link in downloaded_courses_list


def download_single_course(course_link: str, username: str, password: str, save_directory_path: str) -> bool:
"""
Download the given course using the provided credential
@@ -348,9 +351,7 @@ def download_single_course(course_link: str, username: str, password: str, save_
Returns: True/False bool value denoting the success status of the download
"""

global DOWNLOADED_COURSE_LINKS_LIST

if course_link in DOWNLOADED_COURSE_LINKS_LIST:
if is_downloaded(course_link):
return True

retry_count = 0
@@ -369,11 +370,15 @@ def download_single_course(course_link: str, username: str, password: str, save_
with youtube_dl.YoutubeDL(ydl_options) as ydl:
ydl.download([course_link])

DOWNLOADED_COURSE_LINKS_LIST.append(course_link)
save_progress(course_link)
return True
except KeyboardInterrupt:
break
except:
except Exception as e:
if "http error 429" in str(e).lower():
time.sleep(300)
elif "http error 403" in str(e).lower():
return False
retry_count += 1

return False
@@ -388,60 +393,71 @@ def download_batch_course(course_links_list: List[str], username: str, password:
with Pool(MAX_PARALLEL_COURSE_DOWNLOAD_COUNT) as p:
download_results = map(lambda link: link, p.map(download_single_course_function, course_links_list))

return False in download_results
return False not in download_results


def download_pluralsight_path(pluralsight_paths_list: List[PluralsightPath], path_id: int) -> None:
credential_dict = create_pluralsight_account()
def download_pluralsight_path(pluralsight_paths_list: List[PluralsightPath], path_id: int) -> bool:
try:
retry_count = 0
while retry_count < MAX_ACCOUNT_PER_IP:
credential_dict = create_pluralsight_account()

for pluralsight_path in pluralsight_paths_list:
if pluralsight_path.id == path_id:
save_directory_path = get_directory_full_path(SAVE_PATHS_DIRECTORY_PATH, pluralsight_path)
for pluralsight_path in pluralsight_paths_list:
if pluralsight_path.id == path_id:
save_directory_path = get_directory_full_path(SAVE_PATHS_DIRECTORY_PATH, pluralsight_path)

download_batch_course(course_links_list=pluralsight_path.course_links,
username=credential_dict['email'],
password=credential_dict['password'],
save_directory_path=save_directory_path)
result = download_batch_course(course_links_list=pluralsight_path.course_links,
username=credential_dict['email'],
password=credential_dict['password'],
save_directory_path=save_directory_path)

if result:
return True

def download_independent_courses(course_list: List[str]) -> None:
credential_dict = create_pluralsight_account()
retry_count += 1
except Exception as e:
print(e)
return False

download_batch_course(course_links_list=course_list,
username=credential_dict['email'],
password=credential_dict['password'],
save_directory_path=SAVE_COURSES_DIRECTORY_PATH)

def download_independent_courses(course_list: List[str]) -> bool:
try:
retry_count = 0
while retry_count < MAX_ACCOUNT_PER_IP:
credential_dict = create_pluralsight_account()

def main():
global DOWNLOADED_COURSE_LINKS_LIST
result = download_batch_course(course_links_list=course_list,
username=credential_dict['email'],
password=credential_dict['password'],
save_directory_path=SAVE_COURSES_DIRECTORY_PATH)

while True:
try:
if os.path.isfile(DOWNLOADED_COURSES_FILE_LOCATION):
DOWNLOADED_COURSE_LINKS_LIST = list(
filter(None, Path(DOWNLOADED_COURSES_FILE_LOCATION).read_text().split('\n')))
if result:
return True

paths_list, individual_courses_list = get_paths_and_independent_courses(PATHS_JSON_LOCATION,
INDEPENDENT_COURSES_FILE_LOCATION)
retry_count += 1
except Exception as e:
print(e)
return False

if len(sys.argv) == 1:
download_independent_courses(individual_courses_list)
else:
for arg in sys.argv[1:]:
if 1 <= int(arg) <= len(paths_list):
download_pluralsight_path(paths_list, int(arg))

break
except Exception as exception:
print(exception)
retry = input("Retry (Y/y for yes): ").lower()

if retry != 'y':
break
finally:
with open(DOWNLOADED_COURSES_FILE_LOCATION, 'w+') as downloaded_file:
downloaded_file.write('\n'.join(set(DOWNLOADED_COURSE_LINKS_LIST)))
def main():
try:
paths_list, individual_courses_list = get_paths_and_independent_courses(PATHS_JSON_LOCATION,
INDEPENDENT_COURSES_FILE_LOCATION)

if len(sys.argv) == 1:
download_independent_courses(individual_courses_list)
else:
for arg in sys.argv[1:]:
if 1 <= int(arg) <= len(paths_list):
result = download_pluralsight_path(paths_list, int(arg))

if result:
print("SUCCESS")
else:
print("ERROR OCCURRED!!")
except Exception as exception:
print(exception)


if __name__ == '__main__':

Loading…
Cancel
Save