first commit
This commit is contained in:
+231
@@ -0,0 +1,231 @@
|
||||
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
|
||||
Reference in New Issue
Block a user