232 lines
6.9 KiB
Python
232 lines
6.9 KiB
Python
import csv
|
|
import io
|
|
import logging
|
|
from datetime import date, datetime
|
|
from typing import Dict, List
|
|
|
|
from openpyxl import load_workbook
|
|
|
|
from config import Config
|
|
|
|
|
|
def validate_date(date_text: str) -> bool:
|
|
"""
|
|
Validate if date string is in ISO format (YYYY-MM-DD).
|
|
|
|
Args:
|
|
date_text: Date string to validate
|
|
|
|
Returns:
|
|
True if valid, False otherwise
|
|
"""
|
|
try:
|
|
datetime.fromisoformat(date_text)
|
|
return True
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
def parse_csv(file_content: str) -> List[List[str]]:
|
|
"""
|
|
Parse CSV content into rows with automatic delimiter detection.
|
|
|
|
Args:
|
|
file_content: CSV file content as string
|
|
|
|
Returns:
|
|
List of rows, where each row is a list of values
|
|
"""
|
|
# Use csv.Sniffer to auto-detect delimiter
|
|
try:
|
|
sample = file_content[:1024] # Use first 1KB for detection
|
|
sniffer = csv.Sniffer()
|
|
delimiter = sniffer.sniff(sample).delimiter
|
|
logging.info(f"Detected CSV delimiter: '{delimiter}'")
|
|
except Exception as e:
|
|
# Fallback to comma if detection fails
|
|
logging.warning(f"Could not detect delimiter, using comma: {e}")
|
|
delimiter = ','
|
|
|
|
reader = csv.reader(io.StringIO(file_content), delimiter=delimiter)
|
|
return list(reader)
|
|
|
|
|
|
def parse_xlsx(file_bytes: bytes) -> List[List[str]]:
|
|
"""
|
|
Parse XLSX content into rows from the configured worksheet.
|
|
|
|
Args:
|
|
file_bytes: XLSX file content as bytes
|
|
|
|
Returns:
|
|
List of rows, where each row is a list of values
|
|
|
|
Raises:
|
|
ValueError: If the worksheet is missing or the file cannot be read
|
|
"""
|
|
try:
|
|
workbook = load_workbook(io.BytesIO(file_bytes), read_only=True, data_only=True)
|
|
except Exception as exc: # pragma: no cover - relies on openpyxl internals
|
|
raise ValueError(f"Failed to read XLSX file: {exc}") from exc
|
|
|
|
if Config.XLSX_SHEET_NAME not in workbook.sheetnames:
|
|
raise ValueError(f"Worksheet '{Config.XLSX_SHEET_NAME}' not found in XLSX file")
|
|
|
|
sheet = workbook[Config.XLSX_SHEET_NAME]
|
|
rows: List[List[str]] = []
|
|
|
|
for row in sheet.iter_rows(values_only=True):
|
|
cleaned_row = []
|
|
for cell in row:
|
|
if cell is None:
|
|
cleaned_row.append("")
|
|
elif isinstance(cell, datetime):
|
|
cleaned_row.append(cell.date().isoformat())
|
|
elif isinstance(cell, date):
|
|
cleaned_row.append(cell.isoformat())
|
|
else:
|
|
cleaned_row.append(str(cell).strip())
|
|
if any(value for value in cleaned_row):
|
|
rows.append(cleaned_row)
|
|
|
|
return rows
|
|
|
|
|
|
def extract_work_time(rows: List[List[str]]) -> Dict[str, float]:
|
|
"""
|
|
Extract and aggregate work time from CSV rows.
|
|
|
|
Args:
|
|
rows: A list of lists representing the rows of a CSV file
|
|
|
|
Returns:
|
|
A dictionary mapping dates (in 'YYYY-MM-DD' format) to total hours worked
|
|
"""
|
|
work_time: Dict[str, float] = {}
|
|
|
|
# Try to detect column indices from header
|
|
date_col_idx = Config.DATE_COL_INDEX
|
|
hours_col_idx = Config.HOURS_COL_INDEX
|
|
|
|
if rows and len(rows) > 0:
|
|
header = [str(col).strip().lower() for col in rows[0]]
|
|
# Look for date-related columns
|
|
for idx, col in enumerate(header):
|
|
if 'date' in col or 'vouch' in col:
|
|
date_col_idx = idx
|
|
logging.info(f"Detected date column at index {idx}: '{rows[0][idx]}'")
|
|
break
|
|
# Look for hours-related columns
|
|
for idx, col in enumerate(header):
|
|
if 'hour' in col or 'time' in col and 'code' not in col:
|
|
hours_col_idx = idx
|
|
logging.info(f"Detected hours column at index {idx}: '{rows[0][idx]}'")
|
|
break
|
|
|
|
for i, row in enumerate(rows):
|
|
# Skip header row
|
|
if i == 0:
|
|
continue
|
|
|
|
# Skip short rows
|
|
if len(row) <= max(date_col_idx, hours_col_idx):
|
|
logging.debug(
|
|
f"Skipping row {i+1} as it has fewer than "
|
|
f"{max(date_col_idx, hours_col_idx)+1} columns."
|
|
)
|
|
continue
|
|
|
|
date_str = row[date_col_idx].strip() if date_col_idx < len(row) else ""
|
|
hours_str = row[hours_col_idx].strip() if hours_col_idx < len(row) else ""
|
|
|
|
if not date_str or not hours_str:
|
|
logging.debug(f"Skipping row {i+1} due to missing date or hours.")
|
|
continue
|
|
|
|
try:
|
|
# Parse date, trying API format first, then CSV format
|
|
try:
|
|
parsed_date = datetime.strptime(date_str, Config.API_DATE_FORMAT)
|
|
except ValueError:
|
|
parsed_date = datetime.strptime(date_str, Config.CSV_DATE_FORMAT)
|
|
|
|
api_formatted_date = parsed_date.strftime(Config.API_DATE_FORMAT)
|
|
|
|
# Parse hours, allowing for comma as decimal separator and whitespace
|
|
hours = float(hours_str.replace(',', '.').strip())
|
|
|
|
if hours > Config.MIN_HOURS_THRESHOLD:
|
|
work_time[api_formatted_date] = work_time.get(api_formatted_date, 0.0) + hours
|
|
else:
|
|
logging.debug(
|
|
f"Skipping row {i+1} with hours below threshold. Hours: {hours}"
|
|
)
|
|
|
|
except (ValueError, IndexError) as e:
|
|
logging.debug(f"Skipping row {i+1} due to parsing error: {e}")
|
|
|
|
logging.info(f"Extracted work time for {len(work_time)} unique dates.")
|
|
return work_time
|
|
|
|
|
|
def create_time_event(
|
|
user_id: int,
|
|
project_id: int,
|
|
activity: Dict,
|
|
date: str,
|
|
hours: float
|
|
) -> Dict:
|
|
"""
|
|
Create a time event dictionary for the Kleer API.
|
|
|
|
Args:
|
|
user_id: User ID
|
|
project_id: Project ID
|
|
activity: Activity dictionary
|
|
date: Date in YYYY-MM-DD format
|
|
hours: Number of hours
|
|
|
|
Returns:
|
|
Time event dictionary
|
|
"""
|
|
return {
|
|
"foreign-id": "",
|
|
"user": {"id": user_id},
|
|
"client-project": project_id,
|
|
"activity": activity,
|
|
"date": date,
|
|
"hours": hours,
|
|
"comment": "",
|
|
"internal-comment": ""
|
|
}
|
|
|
|
|
|
def get_user_by_name(users_data: Dict, username: str) -> int:
|
|
"""
|
|
Get user ID by username from users data.
|
|
|
|
Args:
|
|
users_data: User data from API
|
|
username: Username to search for
|
|
|
|
Returns:
|
|
User ID
|
|
|
|
Raises:
|
|
ValueError: If user not found
|
|
"""
|
|
if not users_data or "users" not in users_data:
|
|
raise ValueError("Invalid user data received from API")
|
|
|
|
users = {
|
|
user.get('name'): user.get('id')
|
|
for user in users_data.get("users", [])
|
|
if isinstance(user, dict)
|
|
}
|
|
|
|
user_id = users.get(username)
|
|
if not user_id:
|
|
raise ValueError(f"User '{username}' not found")
|
|
|
|
return user_id
|