#!/usr/bin/env python3
import csv
import json
import logging
import os
import sys
from pathlib import Path
from typing import List, Optional, Set
import openpyxl
import pandas as pd
from LOGS.Auxiliary.Exceptions import LOGSException
from LOGS.Entities import (
CustomTypeRequestParameter,
InventoryItemRequestParameter,
ProjectRequestParameter,
)
from LOGS.LOGS import LOGS
from ...Utils.Exceptions import CsvReadError, ExcelReadError
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
[docs]
class InventoryManager:
"""This class enables the creation of inventories in a LOGS instance using a
CSV file, or the export of inventories 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",
log_path: Optional[str] = None,
) -> None:
"""Initialization.
:param logs: LOGS object to access the LOGS web API
:param source_path: Source path for exporting inventories in logs
instance, defaults to None
:param target_path: target path for extracting inventories of a logs
instance in csv file, defaults to None
:param export_format: Should be set to ".xlsx" if the export format should
be an Excel table instead of a CSV file; default: ".csv"
:param log_path: Directory where the log file will be saved,
defaults to None (no file logging)
"""
self.__logs = logs
self.__source_path = source_path
self.__target_path = target_path
# Configure file logging
if log_path is None:
log_path = Path.cwd() / "InventoryManager.log"
log_path = Path(log_path)
# Remove any previous file handler from this logger
for h in logger.handlers[:]:
if isinstance(h, logging.FileHandler):
logger.removeHandler(h)
h.close()
file_handler = logging.FileHandler(log_path)
file_handler.setLevel(logging.INFO)
file_handler.setFormatter(
logging.Formatter("%(asctime)s:%(levelname)s:%(name)s:%(message)s")
)
logger.addHandler(file_handler)
if self.__target_path is not None:
if self.__target_path.suffix == "":
self.__target_path = os.path.join(
self.__target_path, f"inventories_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.
:raises ValueError: If the header does not match the expected format.
:return: None
"""
required_columns = [
"Custom Type ID",
"Name",
"Projects",
]
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_int_list_cell(self, v: object) -> Optional[List[int]]:
"""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 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(",")]
out: List[int] = []
for p in parts:
if not p:
continue
try:
out.append(int(p))
except ValueError:
continue
return out if out else None
[docs]
def check_customtypes(self, inventories_data) -> pd.DataFrame:
"""Removes rows whose Custom Type ID does not exist in the LOGS instance.
:param inventories_data: DataFrame containing inventory data.
:return: DataFrame with invalid custom types removed.
"""
logs_customtype_ids = {
customtype.id
for customtype in self.__logs.customTypes(CustomTypeRequestParameter())
}
invalid_types = (
set(inventories_data["Custom Type ID"].dropna().unique())
- logs_customtype_ids
)
for customtype in invalid_types:
if customtype != "":
rows = inventories_data.index[
inventories_data["Custom Type ID"] == customtype
].tolist()
csv_lines = [r + 2 for r in rows]
logger.warning(
"The inventory in line(s) %s has a custom type %s that does not exist in this LOGS instance and will be skipped.",
csv_lines,
customtype,
)
inventories_data = inventories_data[
inventories_data["Custom Type ID"].isin(logs_customtype_ids)
]
return inventories_data
[docs]
def check_projects(self, inventories_data) -> pd.DataFrame:
"""Removes rows whose project IDs do not exist in the LOGS instance.
:param inventories_data: DataFrame containing inventory data.
:return: DataFrame with invalid projects removed.
"""
logs_project_ids = {
str(project.id)
for project in self.__logs.projects(ProjectRequestParameter())
}
def row_is_valid(row):
project_list = row["Projects"]
if project_list is None:
return True
if isinstance(project_list, float) and pd.isna(project_list):
return True
csv_line = row.name + 2
for project in project_list:
if str(project) not in logs_project_ids:
logger.warning(
"The inventory in line %s has a project %s that does not exist in this LOGS instance and will be skipped.",
csv_line,
project,
)
return False
return True
inventories_data = inventories_data[
inventories_data.apply(row_is_valid, axis=1)
]
return inventories_data
[docs]
def post_process_data(self, inventories_data: pd.DataFrame) -> pd.DataFrame:
"""Normalize empties, parse Projects list, parse Custom Type; drop bad rows.
:param inventories_data: DataFrame read from source file.
:return: Post-processed DataFrame.
"""
logger.debug("Inventories_Data before post processing:\n%s", inventories_data)
self._ensure_header(inventories_data)
df = inventories_data.copy()
df = df.replace(r"^\s*$", pd.NA, regex=True)
df["Custom Type ID"] = pd.to_numeric(
df["Custom Type ID"], errors="coerce"
).astype("Int64")
df["Projects"] = df["Projects"].map(self._split_int_list_cell)
df = df.dropna(subset=["Custom Type ID", "Name"], how="any")
df = df.dropna(how="all")
logger.debug("Inventories_Data after post processing:\n%s", df)
return df
[docs]
def read_file(self) -> pd.DataFrame:
"""Reads the inventories from the given source file.
:return: DataFrame with inventories data.
"""
logger.info("Reading inventory data from file: %s", self.__source_path)
if self.__source_format == ".csv":
try:
inventories_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 inventories: {e}"
logger.exception(message)
raise CsvReadError(message) from e
elif self.__source_format in [".xlsx"]:
try:
inventories_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 inventories: {e}"
logger.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(inventories_data)
[docs]
def create_inventories(self) -> None:
"""Creates inventories in the LOGS instance based on the data from
the source CSV or Excel file.
:return: None
"""
inventories_data = self.read_file()
logger.info("Starting inventory creation process.")
inventories_data = self.check_projects(inventories_data)
inventories_data = self.check_customtypes(inventories_data)
inventory_count = 0
for idx, inventory_item in inventories_data.iterrows():
inventory_count += 1
csv_line = idx + 2
projects = inventory_item["Projects"]
inventory_customtype = (
self.__logs.customTypes(
CustomTypeRequestParameter(
ids=[int(inventory_item["Custom Type ID"])]
)
).first()
if not pd.isna(inventory_item["Custom Type ID"])
else None
)
if inventory_customtype.isHierarchyRoot:
logger.warning(
"The customtype %s in line %s is hierarchical and will be skipped.",
inventory_customtype,
csv_line,
)
continue
logger.info(
"Custom Type Name: %s",
inventory_customtype.name if inventory_customtype else "None",
)
log_inventory = self.__logs.newInventoryItem(
customTypeOrId=inventory_customtype
)
log_inventory.name = inventory_item["Name"]
log_inventory.projects = projects
try:
self.__logs.create(log_inventory)
logger.info("The inventory item in line %s has been created.", csv_line)
except LOGSException:
logger.exception(
"The inventory item in line %s could not be created.",
csv_line,
)
[docs]
def export_inventories_json(self) -> None:
"""Exports inventories from the LOGS instance to JSON files.
:return: None
"""
target_dir = os.path.dirname(self.__target_path)
for inventory_item in self.__logs.inventoryItems(
InventoryItemRequestParameter()
):
inventory_json = inventory_item.toJson()
json_filename = f"inventory_{inventory_item.id}.json"
json_path = os.path.join(target_dir, json_filename)
with open(json_path, "w", encoding="utf-8") as json_file:
json.dump(inventory_json, json_file, ensure_ascii=False, indent=2)
[docs]
def export_inventories_csv(self) -> None:
"""Exports inventories from the LOGS instance to a CSV file.
:return: None
"""
heading = [
"Custom Type ID",
"Name",
"Projects",
]
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 inventory_item in self.__logs.inventoryItems(
InventoryItemRequestParameter()
):
if (
inventory_item.customType
and inventory_item.customType.isHierarchyRoot
):
continue
projects_str = ""
if inventory_item.projects is not None:
projects_str = ",".join(
str(project.id) for project in inventory_item.projects
)
inventory_data = [
inventory_item.customType.id if inventory_item.customType else "",
inventory_item.name,
projects_str,
]
writer.writerow(inventory_data)
[docs]
def export_inventories_excel(self) -> None:
"""Exports inventories from the LOGS instance to an Excel file.
:return: None
"""
heading = [
"Custom Type ID",
"Name",
"Projects",
]
wb = openpyxl.Workbook()
ws = wb.active
ws.append(heading)
for inventory_item in self.__logs.inventoryItems(
InventoryItemRequestParameter()
):
if inventory_item.customType and inventory_item.customType.isHierarchyRoot:
continue
projects_str = ""
if inventory_item.projects is not None:
projects_str = ",".join(
str(project.id) for project in inventory_item.projects
)
inventory_data = [
inventory_item.customType.id if inventory_item.customType else "",
inventory_item.name,
projects_str,
]
ws.append(inventory_data)
wb.save(self.__target_path)
[docs]
def export_inventories(self) -> None:
"""Exports inventories from the LOGS instance to a CSV or Excel file
at the target path.
:return: None
"""
logger.info("Starting inventory export process.")
if self.__export_format == ".csv":
self.export_inventories_csv()
elif self.__export_format == ".xlsx":
self.export_inventories_excel()
else:
raise ValueError(
f"Invalid export format: {self.__export_format}. Supported formats are: .csv, .xlsx"
)
self.export_inventories_json()