import csv
import os
from typing import Optional, List
import numpy as np
import logging
import pandas as pd
import openpyxl
from ...Utils.Exceptions import CsvReadError, ExcelReadError
from LOGS.Auxiliary.Exceptions import LOGSException
from LOGS.Entities import (
Project,
ProjectRequestParameter,
ProjectPersonPermission,
PersonRequestParameter,
)
from LOGS.LOGS import LOGS
logging.basicConfig(level=logging.INFO)
[docs]
class ProjectManager:
"""This class enables the creation of projects in a LOGS instance using a
CSV file, or the export of projects 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 projects in logs
instance, defaults to None
:param target_path: target path for extracting projects 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"projects_export{export_format}"
)
self.__export_format = export_format
self.__source_format = self.__source_path.suffix if self.__source_path else None
def _ensure_header(self, df: pd.DataFrame) -> None:
"""Ensure that the DataFrame has the correct header.
:param df: DataFrame to check.
"""
required_columns = [
"Project Name",
"Person ID",
"Admin Permission",
"Edit Permission",
"Add Permission",
"Read Permission",
]
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 integers or None.
"""
if v is None:
return None
if isinstance(v, float) and pd.isna(v):
return None
s = str(v).strip()
if s == "" or s.lower() in {"none", "nan"}:
return None
parts = [p.strip() for p in s.split(",")]
parts = [p for p in parts if p != ""]
return parts if parts else None
def _valid_project_name(self, v: object) -> bool:
"""Check if project name is valid (not None, not empty, not just commas).
:param v: Cell value to check.
:return: True if valid, False otherwise.
"""
if v is None:
return False
if pd.isna(v):
return False
s = str(v).strip()
if s == "":
return False
if s.replace(",", "").strip() == "":
return False
return True
[docs]
def post_process_data(self, projects_data: pd.DataFrame) -> pd.DataFrame:
"""Post-processes the project data DataFrame by stripping whitespace from string fields and dropping empty rows.
:param projects_data: DataFrame read from source file.
:return: Post-processed DataFrame.
"""
self._ensure_header(projects_data)
df = projects_data.copy()
df = df.replace(r"^\s*$", pd.NA, regex=True)
df = df[df["Project Name"].apply(self._valid_project_name)]
for col in [
"Person ID",
"Admin Permission",
"Edit Permission",
"Add Permission",
"Read Permission",
]:
df[col] = df[col].map(self._split_list_cell)
df = df.dropna(how="all")
logging.debug("Projects_Data after post processing:\n%s", df)
return df
[docs]
def read_file(self) -> pd.DataFrame:
"""Reads the project data file and returns a DataFrame.
:return: DataFrame containing the project data.
"""
if self.__source_format == ".csv":
try:
projects_data = pd.read_csv(
self.__source_path,
delimiter=";",
dtype=str,
keep_default_na=False,
quotechar='"',
skip_blank_lines=True,
)
projects_data = self.post_process_data(projects_data)
except Exception as e:
message = f"Error reading CSV file with the projects: {e}"
logging.exception(message)
raise CsvReadError(message) from e
elif self.__source_format == ".xlsx":
try:
projects_data = pd.read_excel(
self.__source_path,
keep_default_na=False,
engine="openpyxl",
dtype=str,
)
projects_data = self.post_process_data(projects_data)
except Exception as e:
message = f"Error reading Excel file with the projects: {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 projects_data
[docs]
def create_projects(self) -> None:
"""Creates a new project in the LOGS instance."""
def split_and_strip(val):
if isinstance(val, list):
return [str(v).strip() for v in val]
if pd.isna(val) or val == "":
return []
return [v.strip() for v in str(val).split(",")]
projects_data = self.read_file()
for col in [
"Person ID",
"Admin Permission",
"Edit Permission",
"Add Permission",
"Read Permission",
]:
if col in projects_data.columns:
projects_data[col] = projects_data[col].apply(split_and_strip)
for index, project in projects_data.iterrows():
name_found = False
for proj in self.__logs.projects(ProjectRequestParameter()):
name = project["Project Name"]
if pd.isna(name):
continue
name = str(name).strip()
if name == proj.name:
logging.info(
"Project '%s' already exists. Project will be upgraded.",
project["Project Name"],
)
existing_project = self.__logs.project(proj.id)
name_found = True
if not name_found:
log_project = self.__logs.newProject()
log_project.name = project["Project Name"]
person_ids = project["Person ID"]
admins = project["Admin Permission"]
edits = project["Edit Permission"]
adds = project["Add Permission"]
reads = project["Read Permission"]
persons_count = len(person_ids)
if any(persons_count != len(lst) for lst in (admins, edits, adds, reads)):
logging.error(
"There are too few or too many arguments in permission columns for project '%s'. It will be skipped.",
project["Project Name"],
)
continue
if not name_found:
log_project.projectPersonPermissions = []
for pid, admin, edit, add, read in zip(
person_ids, admins, edits, adds, reads
):
if self.__logs.persons(PersonRequestParameter(ids=[pid])).count == 0:
logging.warning(
"Person with id %s does not exist and will be skipped.", pid
)
continue
if not isinstance(pid, int):
try:
pid = int(pid.strip())
except ValueError:
logging.error(
"Invalid person ID '%s' for project '%s'. Skipping.",
pid,
project["Project Name"],
)
continue
def to_bool_if_valid(s):
if isinstance(s, str):
lower = s.lower()
return lower == "true"
admin = to_bool_if_valid(admin)
edit = to_bool_if_valid(edit)
add = to_bool_if_valid(add)
read = to_bool_if_valid(read)
if not read:
logging.error(
"Read permission for person ID '%s' in project '%s' is required. Person will be skipped.",
pid,
project["Project Name"],
)
continue
if None in [admin, edit, add, read]:
logging.error(
"Invalid permission values for person ID '%s' in project '%s'. Skipping.",
pid,
project["Project Name"],
)
continue
if admin and any(not x for x in (edit, add, read)):
logging.error(
"Admin permission for person ID '%s' in project '%s' requires all other permissions to be true. Person will be skipped.",
pid,
project["Project Name"],
)
continue
if not name_found:
project_permission = ProjectPersonPermission(connection=self.__logs)
project_permission.person = self.__logs.persons(
PersonRequestParameter(ids=[pid])
).first()
project_permission.administer = admin
project_permission.edit = edit
project_permission.add = add
log_project.projectPersonPermissions.append(project_permission)
else:
existing_perm = next(
(
perm
for perm in existing_project.projectPersonPermissions
if perm.person.id == pid
),
None,
)
if existing_perm:
existing_perm.administer = admin
existing_perm.edit = edit
existing_perm.add = add
else:
project_permission = ProjectPersonPermission()
project_permission.person = pid
project_permission.administer = admin
project_permission.edit = edit
project_permission.add = add
existing_project.projectPersonPermissions.append(
project_permission
)
try:
if not name_found:
logging.info("Creating project '%s'.", log_project.name)
self.__logs.create(log_project)
logging.info("Project '%s' created successfully.", log_project.name)
if name_found:
logging.info("Updating project '%s'.", existing_project.name)
self.__logs.update(existing_project)
logging.info(
"Project '%s' updated successfully.", existing_project.name
)
except LOGSException as e:
logging.error(
"Failed to create project '%s'. %s", project["Project Name"], e
)
[docs]
def export_projects_csv(self) -> None:
"""Exports projects from the LOGS instance to a CSV file."""
heading = [
"Project Name",
"Person ID",
"Admin Permission",
"Edit Permission",
"Add Permission",
"Read Permission",
]
with open(self.__target_path, "w", newline="", encoding="utf-8") as file:
writer = csv.writer(
file, delimiter=";", quotechar='"', quoting=csv.QUOTE_ALL
)
writer.writerow(heading)
for project in self.__logs.projects(ProjectRequestParameter()):
project_name = project.name
persons = []
admin = []
edit = []
add = []
read = []
for permission in project.projectPersonPermissions:
persons.append(permission.person.id)
admin.append(permission.administer)
edit.append(permission.edit)
add.append(permission.add)
read.append(permission.read)
writer.writerow(
[
project_name,
",".join(map(str, persons)),
",".join(map(str, admin)),
",".join(map(str, edit)),
",".join(map(str, add)),
",".join(map(str, read)),
]
)
[docs]
def export_projects_excel(self) -> None:
"""Exports projects from the LOGS instance to an Excel file."""
heading = [
"Project Name",
"Person ID",
"Admin Permission",
"Edit Permission",
"Add Permission",
"Read Permission",
]
wb = openpyxl.Workbook()
ws = wb.active
ws.append(heading)
for project in self.__logs.projects(ProjectRequestParameter()):
project_name = project.name
persons = []
admin = []
edit = []
add = []
read = []
for permission in project.projectPersonPermissions:
persons.append(permission.person.id)
admin.append(permission.administer)
edit.append(permission.edit)
add.append(permission.add)
read.append(permission.read)
ws.append(
[
project_name,
",".join(map(str, persons)),
",".join(map(str, admin)),
",".join(map(str, edit)),
",".join(map(str, add)),
",".join(map(str, read)),
]
)
wb.save(self.__target_path)
[docs]
def export_projects(self) -> None:
"""Exports projects from the LOGS instance to a CSV file or Excel
file."""
if self.__export_format == ".csv":
self.export_projects_csv()
elif self.__export_format == ".xlsx":
self.export_projects_excel()
else:
raise ValueError(
f"Unsupported export format: {self.__export_format}. Supported formats are: .csv, .xlsx"
)