139 lines
5.9 KiB
Python
139 lines
5.9 KiB
Python
import logging
|
||
import os
|
||
from typing import Dict, List, NamedTuple, Optional
|
||
|
||
from app.core.dto.training import ApproachDTO, ExerciseDTO, TrainingDTO
|
||
from app.core.utils.date_refactor import parse_training_date
|
||
|
||
TABLE_HEADER_CELLS = {"Упражнение", "Вес", "Подходы"}
|
||
|
||
|
||
class ParseError(ValueError):
|
||
"""Raised when a line looks like training data but can't be parsed."""
|
||
|
||
|
||
class TrainingHeader(NamedTuple):
|
||
date: str
|
||
trainer: str
|
||
year_counter: Optional[str] = None
|
||
|
||
|
||
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 {}
|
||
# Lines skipped by the last parse() call
|
||
self.errors: List[str] = []
|
||
|
||
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", encoding="utf-8") as f:
|
||
content = f.read()
|
||
return content
|
||
|
||
def serialize_exercise(self, reps: str, weight: str, name: str) -> ExerciseDTO:
|
||
"""Convert raw exercise data into Exercise object with approaches."""
|
||
try:
|
||
reps_list: List[int] = [int(rep) for rep in reps.split("-")]
|
||
# "15х2" (cyrillic х) and "15x2" mean the same: 15 kg per hand
|
||
weight_tokens: List[str] = weight.replace("х", "x").split("-") if weight else []
|
||
# "0-12х2-16х2": the first approach may have no weight, so check all of them
|
||
weight_splitted: bool = any("x" in token for token in weight_tokens)
|
||
weight_list: List[float] = [float(w.split("x")[0]) for w in weight_tokens]
|
||
except ValueError as e:
|
||
raise ParseError(f"name={name!r}, weight={weight!r}, reps={reps!r}") from e
|
||
|
||
# Last weight is reused for the remaining approaches
|
||
approaches = [
|
||
ApproachDTO(
|
||
weight=weight_list[min(index, len(weight_list) - 1)] if weight_list else 0.0,
|
||
reps=rep,
|
||
)
|
||
for index, rep in enumerate(reps_list)
|
||
]
|
||
return ExerciseDTO(
|
||
name=name, approaches=approaches, splitted_weight=weight_splitted
|
||
)
|
||
|
||
def parse_exercise_row(self, exercise_line: str) -> Optional[ExerciseDTO]:
|
||
"""Parse exercise data from a table row.
|
||
|
||
Returns None for lines that are not exercise rows (headings, separators, text)
|
||
and raises ParseError for rows that are exercises but are malformed.
|
||
"""
|
||
if "|" not in exercise_line:
|
||
return None
|
||
cells: List[str] = [entry.strip() for entry in exercise_line.split("|")][1:-1]
|
||
if not cells or "---" in cells[0] or TABLE_HEADER_CELLS & set(cells):
|
||
return None
|
||
if len(cells) != 3:
|
||
raise ParseError(f"expected 3 columns, got {len(cells)}")
|
||
name, weight, reps = cells
|
||
return self.serialize_exercise(name=name, weight=weight, reps=reps)
|
||
|
||
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) -> Optional[TrainingHeader]:
|
||
"""Parse training header. Override in subclasses for specific formats."""
|
||
raise NotImplementedError("Subclasses must implement parse_training_header")
|
||
|
||
def parse(self, data: str, map_names: bool = True) -> List[TrainingDTO]:
|
||
"""Parse trainings from a string with notes."""
|
||
self.errors = []
|
||
lines = self.filter_training_data(data).splitlines()
|
||
trainings: List[TrainingDTO] = []
|
||
current_training: Optional[TrainingDTO] = None
|
||
|
||
for line_number, line in enumerate(lines, start=1):
|
||
header = self.parse_training_header(line)
|
||
if header:
|
||
try:
|
||
current_training = TrainingDTO(
|
||
date=parse_training_date(header.date), trainer=header.trainer
|
||
)
|
||
except ValueError as e:
|
||
# Exercises below belong to an unknown training, skip them
|
||
current_training = None
|
||
self._report_error(line_number, line, e)
|
||
continue
|
||
trainings.append(current_training)
|
||
continue
|
||
|
||
if current_training is None:
|
||
continue
|
||
try:
|
||
exercise = self.parse_exercise_row(line)
|
||
except ParseError as e:
|
||
self._report_error(line_number, line, e)
|
||
continue
|
||
if exercise:
|
||
current_training.exercises.append(exercise)
|
||
|
||
return self.apply_exercise_mapping(trainings) if map_names else trainings
|
||
|
||
def parse_file(self, map_names: bool = True) -> List[TrainingDTO]:
|
||
"""Parse trainings from the parser's data file."""
|
||
return self.parse(self.read_data_file(self.data_file_name), map_names=map_names)
|
||
|
||
def _report_error(self, line_number: int, line: str, error: Exception) -> None:
|
||
message = f"line {line_number}: {error} ({line.strip()!r})"
|
||
self.errors.append(message)
|
||
logging.warning(f"{type(self).__name__} skipped {message}")
|
||
|
||
def apply_exercise_mapping(self, trainings: List[TrainingDTO]) -> List[TrainingDTO]:
|
||
"""Apply exercise name mapping to all trainings."""
|
||
for training in trainings:
|
||
for exercise in training.exercises or []:
|
||
exercise.name = self.exercise_mapper.get(exercise.name, exercise.name)
|
||
return trainings
|