58 lines
1.9 KiB
Python
58 lines
1.9 KiB
Python
from typing import Generator, Any, List
|
|
from collections import defaultdict
|
|
|
|
import pandas as pd
|
|
|
|
from .exceptions import InvalidNameOrTypeHeaderException
|
|
from ..students.models import Student
|
|
|
|
|
|
def check_columns(df: pd.DataFrame) -> bool:
|
|
headers = set(df.keys().values)
|
|
columns = ['NAZWISKO', 'IMIE', 'INDEKS', 'PESEL', 'EMAIL']
|
|
|
|
if len(headers - set(columns)) != 0:
|
|
return False
|
|
|
|
flag = True
|
|
col_types = ['object', 'object', 'int', 'float64', 'object']
|
|
|
|
for name, col_type in zip(columns, col_types):
|
|
if not str(df.dtypes[name]).startswith(col_type):
|
|
flag = False
|
|
break
|
|
|
|
return flag
|
|
|
|
|
|
def parse_csv(file, mode) -> Generator[Student, Any, None]:
|
|
df = pd.read_csv(file)
|
|
|
|
if not check_columns(df):
|
|
raise InvalidNameOrTypeHeaderException
|
|
|
|
students = (Student(last_name=dict(item.items())['NAZWISKO'],
|
|
first_name=dict(item.items())['IMIE'],
|
|
index=dict(item.items())['INDEKS'],
|
|
pesel=str(int(dict(item.items())['PESEL'])) if not pd.isna(
|
|
dict(item.items())['PESEL']) else None,
|
|
email=dict(item.items())['EMAIL'],
|
|
mode=mode)
|
|
for _, item in df.iterrows())
|
|
|
|
return students
|
|
|
|
|
|
def generate_csv(students: List[Student]) -> str:
|
|
headers = ['PESEL', 'INDEKS', 'IMIE', 'NAZWISKO', 'EMAIL', 'CDYD_KOD', 'PRZ_KOD', 'TZAJ_KOD', 'GR_NR', 'PRG_KOD']
|
|
data = [(student.pesel, student.index, student.first_name, student.last_name, student.email,
|
|
student.group.cdyd_kod, student.group.prz_kod, student.group.tzaj_kod, student.group.project_supervisor_id,
|
|
None) for student in students]
|
|
dataframe = defaultdict(list)
|
|
for row in data:
|
|
for idx, item in enumerate(row):
|
|
dataframe[headers[idx]].append(item)
|
|
|
|
df = pd.DataFrame(dataframe)
|
|
return df.to_csv(index=False)
|