140 lines
5.6 KiB
Python
140 lines
5.6 KiB
Python
import os
|
||
from typing import List, Tuple, Dict, Optional
|
||
|
||
from app.core.models.training import Approach, Exercise, Training
|
||
from app.core.utils.date_refactor import parse_training_date
|
||
|
||
|
||
class BaseNotesParser:
|
||
"""Base class for parsing training data from different note formats."""
|
||
|
||
def __init__(self, data_file_name: str, exercise_mapper: Optional[Dict[str, str]] = None):
|
||
self.data_file_name = data_file_name
|
||
self.project_root = os.getcwd()
|
||
self.exercise_mapper = exercise_mapper or {}
|
||
|
||
def get_data_path(self) -> str:
|
||
return os.path.join(self.project_root, "data")
|
||
|
||
def get_data_file_path(self, file_name: str) -> str:
|
||
return os.path.join(self.get_data_path(), file_name)
|
||
|
||
def read_data_file(self, file_name: str) -> str:
|
||
path_to_file = self.get_data_file_path(file_name)
|
||
with open(path_to_file, "r") as f:
|
||
content = f.read()
|
||
return content
|
||
|
||
def serialize_exercise(self, reps: str, weight: str, name: str) -> Exercise:
|
||
"""Convert raw exercise data into Exercise object with approaches."""
|
||
reps_list: List[int] = [int(rep) for rep in reps.split("-")]
|
||
weight_splitted: bool = False
|
||
weight_list: List[float] = []
|
||
|
||
if weight:
|
||
weight_str_list: List[str] = [weight for weight in weight.split("-")]
|
||
if any(split_anchor in weight_str_list[0] for split_anchor in ["x", "х"]):
|
||
weight_splitted = True
|
||
splitter = "x" if "x" in weight_str_list[0] else "х"
|
||
weight_list = [
|
||
float(xweight.split(splitter)[0]) for xweight in weight_str_list
|
||
]
|
||
else:
|
||
weight_list = [float(w) for w in weight_str_list]
|
||
|
||
approaches = []
|
||
if not weight:
|
||
for rep_index in range(0, len(reps_list)):
|
||
approach = Approach(weight=0.0, reps=reps_list[rep_index])
|
||
approaches.append(approach)
|
||
else:
|
||
weight_pointer = 0
|
||
for rep_index in range(0, len(reps_list)):
|
||
approach = Approach(
|
||
weight=weight_list[weight_pointer], reps=reps_list[rep_index]
|
||
)
|
||
if rep_index < len(weight_list) - 1:
|
||
weight_pointer += 1
|
||
approaches.append(approach)
|
||
|
||
exercise = Exercise(
|
||
name=name, approaches=approaches, splitted_weight=weight_splitted
|
||
)
|
||
return exercise
|
||
|
||
def parse_training_exercises(self, exercise_line: str) -> Exercise:
|
||
"""Parse exercise data from a table row."""
|
||
stripped: List[str] = [entry.strip() for entry in exercise_line.split("|")][
|
||
1:-1
|
||
]
|
||
for entry in stripped:
|
||
if entry in ["Упражнение", "Вес", "Подходы"]:
|
||
raise ValueError
|
||
if stripped:
|
||
if "---" in stripped[0]:
|
||
raise ValueError
|
||
if len(stripped) != 3:
|
||
raise ValueError
|
||
return self.serialize_exercise(
|
||
name=stripped[0], weight=stripped[1], reps=stripped[2]
|
||
)
|
||
raise ValueError("No valid exercise data found")
|
||
|
||
def filter_training_data(self, training_data: str) -> str:
|
||
"""Filter and clean training data. Override in subclasses for specific formats."""
|
||
return training_data
|
||
|
||
def parse_training_header(
|
||
self, training_data_line: str
|
||
) -> Tuple[bool, str, str, str]:
|
||
"""Parse training header. Override in subclasses for specific formats."""
|
||
raise NotImplementedError("Subclasses must implement parse_training_header")
|
||
|
||
def create_training_from_date(self, date_str: str) -> Training:
|
||
"""Create Training object from date string using utility function."""
|
||
return Training(date=parse_training_date(date_str), exercises=[])
|
||
|
||
def parse_training_data(self) -> List[Training]:
|
||
"""Main parsing method. Override for specific parsing logic."""
|
||
training_data = self.filter_training_data(
|
||
self.read_data_file(self.data_file_name)
|
||
)
|
||
lines = training_data.splitlines()
|
||
current_training = None
|
||
trains = []
|
||
|
||
for index, line in enumerate(lines):
|
||
header_parsed, date, _, _ = self.parse_training_header(line)
|
||
if index == len(lines) - 1:
|
||
trains.append(current_training)
|
||
if header_parsed:
|
||
trains.append(current_training)
|
||
current_training = self.create_training_from_date(date)
|
||
continue
|
||
try:
|
||
exr = self.parse_training_exercises(line)
|
||
if current_training:
|
||
current_training.exercises.append(exr)
|
||
except ValueError:
|
||
pass
|
||
|
||
return [train for train in trains if train is not None]
|
||
|
||
def apply_exercise_mapping(self, trainings: List[Training]) -> List[Training]:
|
||
"""Apply exercise name mapping to all trainings."""
|
||
for training in trainings:
|
||
if not training or not training.exercises:
|
||
continue
|
||
for exercise in training.exercises:
|
||
if not exercise:
|
||
continue
|
||
mapped_name = self.exercise_mapper.get(exercise.name)
|
||
if mapped_name is not None:
|
||
exercise.name = mapped_name
|
||
return trainings
|
||
|
||
def parse_and_map_training_data(self) -> List[Training]:
|
||
"""Parse training data and apply exercise mapping."""
|
||
trainings = self.parse_training_data()
|
||
return self.apply_exercise_mapping(trainings)
|
||
|