fixed tox, update to included linters
This commit is contained in:
138
athletico/athletico.py
Normal file
138
athletico/athletico.py
Normal file
@@ -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
|
||||
@@ -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
|
||||
@@ -1,2 +1,3 @@
|
||||
requests==2.28.1
|
||||
Jinja2==3.1.2
|
||||
Jinja2==3.1.2
|
||||
vcrpy==4.2.1
|
||||
7
setup.py
7
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'
|
||||
]
|
||||
)
|
||||
@@ -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()
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
24
tox.ini
24
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
|
||||
Reference in New Issue
Block a user