#!/usr/bin/env python3
import csv
import os
from typing import List, Optional
import numpy as np
import pandas as pd
import logging
import openpyxl
from ...Utils.Exceptions import CsvReadError, ExcelReadError
from LOGS.Auxiliary.Exceptions import LOGSException
from LOGS.Entities import Person, PersonRequestParameter, Role, RoleRequestParameter
from LOGS.LOGS import LOGS
logging.basicConfig(level=logging.INFO)
[docs]
class PersonManager:
"""This class enables the creation of persons in a LOGS instance using a
CSV file, or the export of persons from a LOGS instance into a CSV file."""
def __init__(
self,
logs: LOGS,
source_path: Optional[str] = None,
target_path: Optional[str] = None,
export_format: Optional[str] = ".csv",
) -> None:
"""Initialization.
:param logs: LOGS object to access the LOGS web API
:param source_path: Source path for exporting persons in logs
instance, defaults to None
:param target_path: target path for extracting persons of a logs
instance in csv file, defaults to None
"""
self.__logs = logs
self.__source_path = source_path
self.__target_path = target_path
if self.__target_path is not None:
if self.__target_path.suffix == "":
self.__target_path = os.path.join(
self.__target_path, f"persons_export{export_format}"
)
self.__export_format = export_format
self.__source_format = self.__source_path.suffix if self.__source_path else None
[docs]
def check_role(self, role: str, role_list: List) -> Role:
"""Retrieves the LOGS role object that matches the given role name, if
it exists.
:param role: The name of the role to match.
:param role_list: List of all roles in the LOGS instance.
:return: The LOGS Role object corresponding to the specified
role name.
"""
for r in role_list:
if str(r.name) == str(role):
return r
logging.warning(
"The role '%s' does not exist in the LOGS instance. It will be skipped.",
role,
)
return -1
def _ensure_header(self, df: pd.DataFrame) -> None:
"""Ensure that the DataFrame has the correct header."""
required_columns = [
"Last Name",
"First Name",
"Login",
"E-Mail",
"Roles",
"Password",
]
cols = [str(c).strip() for c in df.columns]
if cols != required_columns:
raise ValueError(
f"Header does not match.\nExpected: {required_columns}\nFound: {cols}"
)
def _split_list_cell(self, v: object) -> Optional[List[str]]:
"""Parse '2, 1' -> [2,1]. Empty/None/NA -> None.
:param v: Cell value to parse.
:return: List of strings or None.
"""
if v is None or pd.isna(v):
return None
s = str(v).strip()
if s == "" or s.lower() in {"none", "nan", "<na>"}:
return None
parts = [p.strip() for p in s.split(",")]
parts = [p for p in parts if p != ""]
return parts if parts else None
[docs]
def post_process_data(self, personal_data: pd.DataFrame) -> pd.DataFrame:
"""Normalize empty strings, parse roles, drop empty rows; strict header.
:param personal_data: DataFrame with personal data.
:return: Processed DataFrame.
"""
self._ensure_header(personal_data)
df = personal_data.copy()
df = df.replace(r"^\s*$", pd.NA, regex=True)
df["Roles"] = df["Roles"].map(self._split_list_cell)
df = df.dropna(how="all")
logging.debug("Personal_Data after post processing:\n%s", df)
return df
[docs]
def read_file(self) -> pd.DataFrame:
"""Reads the persons from the given source file.
:return: DataFrame containing the person data.
"""
if self.__source_format == ".csv":
try:
personal_data = pd.read_csv(
self.__source_path,
delimiter=";",
dtype=str,
keep_default_na=False,
quotechar='"',
skip_blank_lines=True,
)
except Exception as e:
message = f"Error reading CSV file with the persons: {e}"
logging.exception(message)
raise CsvReadError(message) from e
elif self.__source_format == ".xlsx":
try:
personal_data = pd.read_excel(
self.__source_path,
dtype=str,
keep_default_na=False,
engine="openpyxl",
)
except Exception as e:
message = f"Error reading Excel file with the persons: {e}"
logging.exception(message)
raise ExcelReadError(message) from e
else:
raise ValueError(
f"Unsupported source format: {self.__source_format}. Supported formats are: .csv, .xlsx"
)
return self.post_process_data(personal_data)
[docs]
def create_persons(self) -> None:
"""Creates a person by the given csv-file."""
personal_data = self.read_file()
# Get all roles of the LOGS instance
role_list: List[Role] = list(self.__logs.roles(RoleRequestParameter()))
for line_num, (_, person) in enumerate(personal_data.iterrows(), start=1):
# Create person and set attributes excluding roles
log_person = self.__logs.newPerson()
if pd.notna(person["Last Name"]):
log_person.lastName = person["Last Name"].strip()
if pd.notna(person["First Name"]):
log_person.firstName = person["First Name"].strip()
if pd.notna(person["Login"]):
log_person.login = person["Login"].strip()
if pd.notna(person["E-Mail"]):
log_person.email = person["E-Mail"].strip()
if pd.notna(person["Password"]):
log_person.password = person["Password"].strip()
# Set roles of the person
log_per_roles: List[Role] = []
roles = person["Roles"] or []
for role_name in roles:
log_role = self.check_role(role_name, role_list)
if log_role is not None:
log_per_roles.append(log_role)
if log_person.login:
log_person.roles = log_per_roles
try:
self.__logs.create(log_person)
except LOGSException as e:
logging.error(
"The person in line %s could not be created. %s", line_num, e
)
[docs]
def export_persons_csv(self) -> None:
"""Export Persons from LOGS."""
heading = [
"Last Name",
"First Name",
"Login",
"E-Mail",
"Roles",
]
with open(self.__target_path, "w", newline="", encoding="utf-8") as file:
writer = csv.writer(file, delimiter=";")
writer.writerow(heading)
for person in self.__logs.persons(PersonRequestParameter()):
roles_str = ""
if person.roles is not None:
roles_str = ",".join([role.name for role in person.roles])
person_data = [
person.lastName,
person.firstName,
person.login,
person.email,
roles_str,
]
writer.writerow(person_data)
[docs]
def export_persons_excel(self) -> None:
"""Export Persons from LOGS to an Excel file."""
heading = [
"Last Name",
"First Name",
"Login",
"E-Mail",
"Roles",
]
wb = openpyxl.Workbook()
ws = wb.active
ws.append(heading)
for person in self.__logs.persons(PersonRequestParameter()):
if person.roles is not None:
roles_str = ",".join([role.name for role in person.roles])
else:
roles_str = ""
person_data = [
person.lastName,
person.firstName,
person.login,
person.email,
roles_str,
]
ws.append(person_data)
wb.save(self.__target_path)
[docs]
def export_persons(self) -> None:
"""Export persons based on the specified export format."""
if self.__export_format == ".csv":
self.export_persons_csv()
elif self.__export_format == ".xlsx":
self.export_persons_excel()
else:
raise ValueError(
f"Unsupported export format: {self.__export_format}. Supported formats are: .csv, .xlsx"
)