Addition of procedure parser in import.

This commit is contained in:
lwark
2025-06-17 15:09:51 -05:00
parent 0233bc3ac2
commit d8c3f3bbb2
31 changed files with 688 additions and 304 deletions

View File

@@ -1,14 +1,15 @@
"""
"""
from __future__ import annotations
import logging, re
from pathlib import Path
from typing import Generator, Tuple
from openpyxl import load_workbook
from typing import Generator, Tuple, TYPE_CHECKING
from pandas import DataFrame
from backend.validators import pydant
from backend.db.models import Procedure
from dataclasses import dataclass
if TYPE_CHECKING:
from backend.db.models import ProcedureType
logger = logging.getLogger(f"submissions.{__name__}")
@@ -30,7 +31,7 @@ class DefaultParser(object):
return instance
def __init__(self, filepath: Path | str, procedure: Procedure|None=None, range_dict: dict | None = None, *args, **kwargs):
def __init__(self, filepath: Path | str, proceduretype: ProcedureType|None=None, range_dict: dict | None = None, *args, **kwargs):
"""
Args:
@@ -40,7 +41,7 @@ class DefaultParser(object):
*args ():
**kwargs ():
"""
self.procedure = procedure
self.proceduretype = proceduretype
try:
self._pyd_object = getattr(pydant, f"Pyd{self.__class__.__name__.replace('Parser', '')}")
except AttributeError:
@@ -58,6 +59,13 @@ class DefaultParser(object):
data['filepath'] = self.filepath
return self._pyd_object(**data)
@classmethod
def correct_procedure_type(cls, proceduretype: str | "ProcedureType"):
from backend.db.models import ProcedureType
if isinstance(proceduretype, str):
proceduretype = ProcedureType.query(name=proceduretype)
return proceduretype
class DefaultKEYVALUEParser(DefaultParser):
@@ -90,7 +98,6 @@ class DefaultTABLEParser(DefaultParser):
default_range_dict = [dict(
header_row=20,
end_row=116,
sheet="Sample List"
)]
@@ -98,15 +105,25 @@ class DefaultTABLEParser(DefaultParser):
def parsed_info(self):
for item in self.range_dict:
list_worksheet = self.workbook[item['sheet']]
list_df = DataFrame([item for item in list_worksheet.values][item['header_row'] - 1:])
if "end_row" in item.keys():
list_df = DataFrame([item for item in list_worksheet.values][item['header_row'] - 1:item['end_row']-1])
else:
list_df = DataFrame([item for item in list_worksheet.values][item['header_row'] - 1:])
list_df.columns = list_df.iloc[0]
list_df = list_df[1:]
list_df = list_df.dropna(axis=1, how='all')
for ii, row in enumerate(list_df.iterrows()):
output = {key.lower().replace(" ", "_"): value for key, value in row[1].to_dict().items()}
output = {}
for key, value in row[1].to_dict().items():
if isinstance(key, str):
key = key.lower().replace(" ", "_")
key = re.sub(r"_(\(.*\)|#)", "", key)
logger.debug(f"Row {ii} values: {key}: {value}")
output[key] = value
yield output
def to_pydantic(self, **kwargs):
return [self._pyd_object(**output) for output in self.parsed_info]
from .submission_parser import *
from .clientsubmission_parser import *
from backend.excel.parsers.results_parsers.pcr_results_parser import *