commit d279fa416a341c2db47fd7285ed832c337f329c9 Author: asc Date: Fri Oct 28 13:26:32 2022 -0500 initial commit diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..f90fba8 --- /dev/null +++ b/.gitignore @@ -0,0 +1,11 @@ +tests/data/output +tests/data/input/sample_data.json +tests/fixtures/ + +.tox/ +venv/ + +.idea/ +.DS_Store + +*.egg* \ No newline at end of file diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..e69de29 diff --git a/README.md b/README.md new file mode 100644 index 0000000..e69de29 diff --git a/athleticotogo/__init__.py b/athleticotogo/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/athleticotogo/athletico.py b/athleticotogo/athletico.py new file mode 100644 index 0000000..d97ac32 --- /dev/null +++ b/athleticotogo/athletico.py @@ -0,0 +1,111 @@ +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 new file mode 100644 index 0000000..2e17a8b --- /dev/null +++ b/requirements.txt @@ -0,0 +1,2 @@ +requests==2.28.1 +Jinja2==3.1.2 \ No newline at end of file diff --git a/setup.py b/setup.py new file mode 100644 index 0000000..718d7ad --- /dev/null +++ b/setup.py @@ -0,0 +1,10 @@ +from distutils.core import setup +setup(name='AthleticoTogo', + version='0.1', + packages=['athleticotogo',], + description='Download exercises from Athletico programs.', + install_requires = [ + 'requests==2.28.1', + 'Jinja2==3.1.2' + ] + ) \ No newline at end of file diff --git a/templates/episode.html b/templates/episode.html new file mode 100644 index 0000000..c2b93c6 --- /dev/null +++ b/templates/episode.html @@ -0,0 +1,28 @@ + + + + + Title + + + + + \ No newline at end of file diff --git a/templates/extraction_index.html b/templates/extraction_index.html new file mode 100644 index 0000000..739a095 --- /dev/null +++ b/templates/extraction_index.html @@ -0,0 +1,16 @@ + + + + + Title + + +{% for episode_id in extractions %} + +{% endfor %} + + \ No newline at end of file diff --git a/tests/data/input/sample_data_example.json b/tests/data/input/sample_data_example.json new file mode 100644 index 0000000..e1505b7 --- /dev/null +++ b/tests/data/input/sample_data_example.json @@ -0,0 +1,4 @@ +{ + "tokens": [ + ] +} \ No newline at end of file diff --git a/tests/test_athletico.py b/tests/test_athletico.py new file mode 100644 index 0000000..e7ccc84 --- /dev/null +++ b/tests/test_athletico.py @@ -0,0 +1,37 @@ +from athleticotogo.athletico import Athletico +import unittest +import vcr +import os +import json + + +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 + + @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']) + + @vcr.use_cassette(cassette_library_dir="tests/fixtures") + def test_save_extraction_to_local(self): + a = Athletico() + for token in self.tokens: + episodes = a.fetch_episodes_for_access_code(token) + a.save_episodes(episodes, destination='tests/data/output') + pass + + +if __name__ == '__main__': + unittest.main() \ No newline at end of file diff --git a/tox.ini b/tox.ini new file mode 100644 index 0000000..d71b998 --- /dev/null +++ b/tox.ini @@ -0,0 +1,15 @@ +# tox (https://tox.readthedocs.io/) is a tool for running tests +# in multiple virtualenvs. This configuration file will run the +# test suite on all supported python versions. To use it, "pip install tox" +# and then run "tox" from this directory. + +[tox] +envlist = py310 + +[testenv] +deps = + requests==2.28.1 + Jinja2==3.1.2 + vcrpy +commands = + python -m unittest discover tests