diff --git a/athleticotogo/__init__.py b/athletico/__init__.py similarity index 100% rename from athleticotogo/__init__.py rename to athletico/__init__.py diff --git a/athletico/athletico.py b/athletico/athletico.py new file mode 100644 index 0000000..2300a5f --- /dev/null +++ b/athletico/athletico.py @@ -0,0 +1,138 @@ +import json +from pathlib import Path + +import jinja2 +import requests +from requests import RequestException + + +class Athletico: + def fetch_episodes_for_access_code(self, access_code: str) -> dict: + """ + Given a home access code, fetch exercise information + :param access_code: the access code to a home exercise program from Athletico + :return: a dict containing the exercise information + """ + session = requests.Session() + home_url = "https://athleticopt.medbridgego.com" + fetch_episodes_url = "https://athleticopt.medbridgego.com/api/v4/lite/episodes/" + + session.get(home_url) + csrf_token = session.cookies.get("csrf_cookie_name") + + session.post( + url="https://athleticopt.medbridgego.com/register_token", + data={ + "token": access_code, + "X-CSRF-Token": csrf_token, + "verify_access_code": "Verify+Access+Code", + }, + ) + + response = session.get(fetch_episodes_url) + + if response.ok and response.json() and response.json().get("status"): + return response.json().get("episodes") + else: + raise RequestException( + f"Request Failed, {response.status_code}: {response.reason}" + ) + + def render(self, template: str, output: str, context: dict) -> str: + """ + Renders an HTML page. + :param template: Path to Jinja template + :param output: Destination for .html file + :param context: The context to be passed to the render + :return: Path to html file + """ + + output_fpath = Path(output) + template_fpath = Path(output) + environment = jinja2.Environment(loader=jinja2.FileSystemLoader()) + template = environment.get_template(str(template_fpath.absolute())) + page = template.render(**context) + + output_fpath.mkdir(parents=True, exist_ok=True) + with output_fpath.open("w") as f: + f.write(page) + + return str(output_fpath) + + def save_episodes(self, episodes: list, destination: str = ".") -> None: + """ + Outputs episodes to disk in a folder structure + ├── exercises/ + │ ├── {EXERCISE_ID}/ + │ │ ├── {EXERCISE_ID}.json + │ │ ├── {EXERCISE_ID}.txt + │ │ ├── description.html + │ │ └── video_file.m3u8 + │ └── {EXERCISE_ID}/ + │ ├── {EXERCISE_ID}.json + │ ├── {EXERCISE_ID}.txt + │ ├── description.html + │ └── video_file.m3u8 + └── programs/ + └── {ACCESS_TOKEN}/ + └── episodes/ + └── {EPISODE_ID}.json + :param episodes: List of episode data, matches output of fetch_episodes_for_access_code + :param destination: Filepath to save to, defaults to current directory + :return: None + """ + + destination_fpath = Path(destination) + destination_fpath.joinpath() + for episode in episodes: + episode_id = str(episode["id"]) + program = episode["program"] + exercises = [] + episode_fpath = destination_fpath.joinpath( + "programs", episode["token"], "episodes" + ) + episode_fpath.mkdir(parents=True, exist_ok=True) + + for exercise in program["program_exercises"]: + exercises.append(exercise) + exercise_id = str(exercise["id"]) + exercise_fpath = destination_fpath.joinpath("exercises", exercise_id) + thumbnails_fpath = exercise_fpath.joinpath("thumbnails") + thumbnails_fpath.mkdir(parents=True, exist_ok=True) + for i, image in enumerate(exercise["exercise"]["thumbnails"]): + r = requests.get(image["image_filepath"]) + fname = image["image_filepath"].split("/")[-1] + with thumbnails_fpath.joinpath(fname).open("wb") as f: + f.write(r.content) + + json_fpath = exercise_fpath.joinpath(exercise_id).with_suffix(".json") + with json_fpath.open("w") as f: + json.dump(exercise, f) + + txt_fpath = exercise_fpath.joinpath(exercise_id).with_suffix(".txt") + with txt_fpath.open("w") as f: + f.write(exercise["name"]) + + if exercise["exercise"]["video_file"]: + video_fpath = exercise_fpath.joinpath("video_file").with_suffix( + ".m3u8" + ) + with video_fpath.open("wb") as f: + r = requests.get("http:" + exercise["exercise"]["video_file"]) + f.write(r.content) + + description_fpath = exercise_fpath.joinpath("description").with_suffix( + ".html" + ) + with description_fpath.open("w") as f: + f.write(exercise["exercise"]["description"]) + + episode_json_fpath = episode_fpath.joinpath(episode_id).with_suffix(".json") + with episode_json_fpath.open("w") as f: + json.dump(episode, fp=f) + + return + + +if __name__ == "__main__": + pass diff --git a/athleticotogo/athletico.py b/athleticotogo/athletico.py deleted file mode 100644 index d97ac32..0000000 --- a/athleticotogo/athletico.py +++ /dev/null @@ -1,111 +0,0 @@ -import requests -from requests import RequestException -import json -import jinja2 -from pathlib import Path - -class Athletico(): - def fetch_episodes_for_access_code(self, access_code:str) -> dict: - ''' - Given a home access code, fetch exercise information - :param access_code: the access code to a home excercise program from Athletico - :return: a dict containing the exercise information - ''' - session = requests.Session() - home_url = "https://athleticopt.medbridgego.com" - fetch_episodes_url = "https://athleticopt.medbridgego.com/api/v4/lite/episodes/" - - session.get(home_url) - csrf_token = session.cookies.get('csrf_cookie_name') - - session.post( - url="https://athleticopt.medbridgego.com/register_token", - data = { - "token": access_code, - "X-CSRF-Token": csrf_token, - "verify_access_code": "Verify+Access+Code" - } - ) - - response = session.get(fetch_episodes_url) - - if response.ok and response.json() and response.json().get('status'): - return response.json().get('episodes') - else: - raise RequestException(f'Request Failed, {response.status_code}: {response.reason}') - - def render(self, template: str, output: str, context: dict) -> str: - ''' - Renders an HTML page. - :param template: Path to jinja template - :param output: Destination for .html file - :param context: The context to be passed to the render - :return: Path to html file - ''' - - output_filepath = Path(output) - template_filepath = Path(output) - environment = jinja2.Environment(loader=jinja2.FileSystemLoader()) - template = environment.get_template(template_filepath) - page = template.render(**context) - - output_filepath.mkdir(parents=True, exist_ok=True) - with output_filepath.open('w') as f: - f.write(page) - - return str(output_filepath) - - def save_episodes(self, episodes: list, destination: str= ".") -> None: - ''' - Outputs episodes to disk in a folder structure - :param episodes: List of episode data, matches output of fetch_episodes_for_access_code - :param destination: Filepath to save to, defaults to current directory - :return: None - ''' - - destination_filepath = Path(destination) - destination_filepath.joinpath() - for episode in episodes: - episode_id = str(episode['id']) - program = episode['program'] - exercises = [] - episode_filepath = destination_filepath.joinpath('programs',episode["token"],'episodes') - episode_filepath.mkdir(parents=True, exist_ok=True) - - for exercise in program['program_exercises']: - exercises.append(exercise) - exercise_id = str(exercise['id']) - exercise_filepath = destination_filepath.joinpath(destination,'exercises', exercise_id) - thumbnails_filepath = exercise_filepath.joinpath('thumbnails') - thumbnails_filepath.mkdir(parents=True, exist_ok=True) - for i, image in enumerate(exercise['exercise']['thumbnails']): - r = requests.get(image['image_filepath']) - fname = image['image_filepath'].split('/')[-1] - with thumbnails_filepath.joinpath(fname).open('wb') as f: - f.write(r.content) - - with exercise_filepath.joinpath(exercise_id).with_suffix('.json').open('w') as f: - json.dump(exercise, f) - - with exercise_filepath.joinpath(exercise_id).with_suffix('.txt').open('w') as f: - f.write(exercise['name']) - - if exercise["exercise"]["video_file"]: - with exercise_filepath.joinpath('video_file').with_suffix('.m3u8').open('wb') as f: - try: - r = requests.get("http:"+exercise["exercise"]["video_file"]) - except: - pass - f.write(r.content) - - with exercise_filepath.joinpath('description').with_suffix('.html').open('w') as f: - f.write(exercise["exercise"]["description"]) - - - with episode_filepath.joinpath(episode_id).with_suffix('.json').open('w') as f: - json.dump(episode, fp=f) - - return - -if __name__ == "__main__": - pass \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index 2e17a8b..e9f3d16 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,2 +1,3 @@ requests==2.28.1 -Jinja2==3.1.2 \ No newline at end of file +Jinja2==3.1.2 +vcrpy==4.2.1 \ No newline at end of file diff --git a/setup.py b/setup.py index 718d7ad..db9711f 100644 --- a/setup.py +++ b/setup.py @@ -1,10 +1,11 @@ from distutils.core import setup -setup(name='AthleticoTogo', +setup(name='Athletico', version='0.1', - packages=['athleticotogo',], + packages=['athletico',], description='Download exercises from Athletico programs.', install_requires = [ + 'Jinja2==3.1.2', 'requests==2.28.1', - 'Jinja2==3.1.2' + 'vcrpy==4.2.1' ] ) \ No newline at end of file diff --git a/tests/test_athletico.py b/tests/test_athletico.py index e7ccc84..2fa51a8 100644 --- a/tests/test_athletico.py +++ b/tests/test_athletico.py @@ -1,37 +1,35 @@ -from athleticotogo.athletico import Athletico -import unittest -import vcr -import os import json +import os +import unittest + +import vcr + +from athletico.athletico import Athletico class TestAthletico(unittest.TestCase): - @classmethod def setUpClass(cls) -> None: - with open('tests/data/input/sample_data.json') as f: - cls.tokens = json.load(f)['tokens'] - - def setUp(self): - os.chdir("..") - pass + with open("tests/data/input/sample_data.json") as f: + cls.tokens = json.load(f)["tokens"] @vcr.use_cassette(cassette_library_dir="tests/fixtures") def test_extract(self): a = Athletico() episodes = a.fetch_episodes_for_access_code(self.tokens[0]) - self.assertIn('id', episodes[0]) - self.assertIn('program', episodes[0]) - self.assertIn('program_exercises', episodes[0]['program']) + self.assertIn("id", episodes[0]) + self.assertIn("program", episodes[0]) + self.assertIn("program_exercises", episodes[0]["program"]) @vcr.use_cassette(cassette_library_dir="tests/fixtures") def test_save_extraction_to_local(self): a = Athletico() + print(os.path.abspath(os.path.curdir)) for token in self.tokens: episodes = a.fetch_episodes_for_access_code(token) - a.save_episodes(episodes, destination='tests/data/output') + a.save_episodes(episodes, destination="tests/data/output") pass -if __name__ == '__main__': - unittest.main() \ No newline at end of file +if __name__ == "__main__": + unittest.main() diff --git a/tox.ini b/tox.ini index d71b998..690ba4b 100644 --- a/tox.ini +++ b/tox.ini @@ -4,12 +4,30 @@ # and then run "tox" from this directory. [tox] -envlist = py310 +envlist = py310, linters + +[flake8] +max-line-length = 120 + +[isort] +profile=black +line_length = 120 [testenv] deps = - requests==2.28.1 Jinja2==3.1.2 - vcrpy + requests==2.28.1 + vcrpy==4.2.1 commands = python -m unittest discover tests + +[testenv:linters] +deps = + black + flake8 + flake8-black +commands = + black --check --diff athletico + black --check --diff tests + flake8 athletico + flake8 tests \ No newline at end of file