258 lines
7.4 KiB
Python
258 lines
7.4 KiB
Python
"""
|
|
Export service for inventory snapshot and audit trail exports.
|
|
Supports CSV and Excel (.xlsx) formats.
|
|
"""
|
|
|
|
import csv
|
|
import io
|
|
from datetime import datetime
|
|
from typing import List, Optional
|
|
from openpyxl import Workbook
|
|
from openpyxl.styles import Font, PatternFill, Alignment
|
|
from openpyxl.utils import get_column_letter
|
|
|
|
|
|
class InventorySnapshotExporter:
|
|
"""Export inventory items to CSV or Excel format."""
|
|
|
|
HEADERS = [
|
|
"ID",
|
|
"Name",
|
|
"Part Number",
|
|
"Barcode",
|
|
"Category",
|
|
"Type",
|
|
"Quantity",
|
|
"Min Quantity",
|
|
"Description",
|
|
"Color",
|
|
"Size",
|
|
"Connector",
|
|
"Box Label",
|
|
"Created",
|
|
"Modified",
|
|
]
|
|
|
|
@staticmethod
|
|
def _item_to_row(item) -> list:
|
|
"""Convert Item object to row data."""
|
|
return [
|
|
item.id,
|
|
item.name or "",
|
|
item.part_number or "",
|
|
item.barcode or "",
|
|
item.category or "",
|
|
item.type or "",
|
|
str(item.quantity or 0),
|
|
str(item.min_quantity or 1),
|
|
item.description or "",
|
|
item.color or "",
|
|
item.size or "",
|
|
item.connector or "",
|
|
item.box_label or "",
|
|
item.created_at.isoformat() if hasattr(item, "created_at") and item.created_at else "",
|
|
item.updated_at.isoformat() if hasattr(item, "updated_at") and item.updated_at else "",
|
|
]
|
|
|
|
@classmethod
|
|
def to_csv(cls, items: List, timestamp: str) -> str:
|
|
"""
|
|
Export items to CSV format.
|
|
|
|
Args:
|
|
items: List of Item objects
|
|
timestamp: ISO 8601 date string (YYYY-MM-DD)
|
|
|
|
Returns:
|
|
CSV string content
|
|
"""
|
|
output = io.StringIO()
|
|
writer = csv.writer(output, quoting=csv.QUOTE_MINIMAL)
|
|
|
|
# Write header
|
|
writer.writerow(cls.HEADERS)
|
|
|
|
# Write data rows
|
|
for item in items:
|
|
writer.writerow(cls._item_to_row(item))
|
|
|
|
return output.getvalue()
|
|
|
|
@classmethod
|
|
def to_excel(cls, items: List, timestamp: str) -> bytes:
|
|
"""
|
|
Export items to Excel (.xlsx) format.
|
|
|
|
Args:
|
|
items: List of Item objects
|
|
timestamp: ISO 8601 date string (YYYY-MM-DD)
|
|
|
|
Returns:
|
|
Bytes containing Excel file content
|
|
"""
|
|
wb = Workbook()
|
|
ws = wb.active
|
|
ws.title = "Inventory Snapshot"
|
|
|
|
# Add title row
|
|
ws.append([f"Inventory Snapshot - {timestamp}"])
|
|
title_cell = ws["A1"]
|
|
title_cell.font = Font(bold=False, size=12)
|
|
title_cell.fill = PatternFill(start_color="E8F4F8", end_color="E8F4F8", fill_type="solid")
|
|
ws.merge_cells("A1:O1")
|
|
|
|
# Add headers
|
|
ws.append(cls.HEADERS)
|
|
header_fill = PatternFill(start_color="D3D3D3", end_color="D3D3D3", fill_type="solid")
|
|
header_font = Font(bold=False)
|
|
for cell in ws[2]:
|
|
cell.fill = header_fill
|
|
cell.font = header_font
|
|
cell.alignment = Alignment(horizontal="center", vertical="center")
|
|
|
|
# Add data rows
|
|
for item in items:
|
|
ws.append(cls._item_to_row(item))
|
|
|
|
# Adjust column widths
|
|
column_widths = [8, 20, 15, 15, 15, 12, 10, 12, 20, 12, 10, 15, 15, 20, 20]
|
|
for i, width in enumerate(column_widths, 1):
|
|
ws.column_dimensions[get_column_letter(i)].width = width
|
|
|
|
# Center align quantity columns
|
|
for row in ws.iter_rows(min_row=3, max_row=ws.max_row, min_col=7, max_col=8):
|
|
for cell in row:
|
|
cell.alignment = Alignment(horizontal="right")
|
|
|
|
# Convert to bytes
|
|
output = io.BytesIO()
|
|
wb.save(output)
|
|
output.seek(0)
|
|
return output.getvalue()
|
|
|
|
|
|
class AuditTrailExporter:
|
|
"""Export audit logs to CSV or Excel format."""
|
|
|
|
HEADERS = [
|
|
"ID",
|
|
"Timestamp",
|
|
"User",
|
|
"Action",
|
|
"Item ID",
|
|
"Item Name",
|
|
"Item Part Number",
|
|
"Item Barcode",
|
|
"Quantity Change",
|
|
"Details",
|
|
]
|
|
|
|
@staticmethod
|
|
def _log_to_row(log) -> list:
|
|
"""Convert AuditLog object to row data."""
|
|
user_name = log.user.username if (hasattr(log, "user") and log.user) else ""
|
|
return [
|
|
log.id,
|
|
log.timestamp.isoformat() if log.timestamp else "",
|
|
user_name,
|
|
log.action or "",
|
|
log.target_item_id or "",
|
|
log.target_item_name or "",
|
|
log.target_item_pn or "",
|
|
log.target_item_barcode or "",
|
|
str(log.quantity_change or ""),
|
|
log.details or "",
|
|
]
|
|
|
|
@classmethod
|
|
def to_csv(cls, logs: List, timestamp: str) -> str:
|
|
"""
|
|
Export audit logs to CSV format.
|
|
|
|
Args:
|
|
logs: List of AuditLog objects
|
|
timestamp: ISO 8601 date string (YYYY-MM-DD)
|
|
|
|
Returns:
|
|
CSV string content
|
|
"""
|
|
output = io.StringIO()
|
|
writer = csv.writer(output, quoting=csv.QUOTE_MINIMAL)
|
|
|
|
# Write header
|
|
writer.writerow(cls.HEADERS)
|
|
|
|
# Write data rows
|
|
for log in logs:
|
|
writer.writerow(cls._log_to_row(log))
|
|
|
|
return output.getvalue()
|
|
|
|
@classmethod
|
|
def to_excel(cls, logs: List, timestamp: str) -> bytes:
|
|
"""
|
|
Export audit logs to Excel (.xlsx) format.
|
|
|
|
Args:
|
|
logs: List of AuditLog objects
|
|
timestamp: ISO 8601 date string (YYYY-MM-DD)
|
|
|
|
Returns:
|
|
Bytes containing Excel file content
|
|
"""
|
|
wb = Workbook()
|
|
ws = wb.active
|
|
ws.title = "Audit Trail"
|
|
|
|
# Add title row
|
|
ws.append([f"Audit Trail - {timestamp}"])
|
|
title_cell = ws["A1"]
|
|
title_cell.font = Font(bold=False, size=12)
|
|
title_cell.fill = PatternFill(start_color="E8F4F8", end_color="E8F4F8", fill_type="solid")
|
|
ws.merge_cells("A1:J1")
|
|
|
|
# Add headers
|
|
ws.append(cls.HEADERS)
|
|
header_fill = PatternFill(start_color="D3D3D3", end_color="D3D3D3", fill_type="solid")
|
|
header_font = Font(bold=False)
|
|
for cell in ws[2]:
|
|
cell.fill = header_fill
|
|
cell.font = header_font
|
|
cell.alignment = Alignment(horizontal="center", vertical="center")
|
|
|
|
# Add data rows
|
|
for log in logs:
|
|
ws.append(cls._log_to_row(log))
|
|
|
|
# Adjust column widths
|
|
column_widths = [8, 25, 15, 15, 10, 20, 18, 15, 15, 30]
|
|
for i, width in enumerate(column_widths, 1):
|
|
ws.column_dimensions[get_column_letter(i)].width = width
|
|
|
|
# Right-align quantity change
|
|
for row in ws.iter_rows(min_row=3, max_row=ws.max_row, min_col=9, max_col=9):
|
|
for cell in row:
|
|
cell.alignment = Alignment(horizontal="right")
|
|
|
|
# Convert to bytes
|
|
output = io.BytesIO()
|
|
wb.save(output)
|
|
output.seek(0)
|
|
return output.getvalue()
|
|
|
|
|
|
def get_export_filename(report_type: str, format_type: str, timestamp: str) -> str:
|
|
"""
|
|
Generate export filename with timestamp.
|
|
|
|
Args:
|
|
report_type: 'inventory_snapshot' or 'audit_trail'
|
|
format_type: 'csv' or 'xlsx'
|
|
timestamp: ISO 8601 date string (YYYY-MM-DD)
|
|
|
|
Returns:
|
|
Filename string (e.g., 'inventory_snapshot_2026-04-22.csv')
|
|
"""
|
|
extension = "csv" if format_type == "csv" else "xlsx"
|
|
return f"{report_type}_{timestamp}.{extension}"
|