148 lines
4.8 KiB
Python
148 lines
4.8 KiB
Python
import logging
|
|
import pathlib
|
|
import sys
|
|
import traceback
|
|
from typing import Any, Dict, Optional
|
|
|
|
from api_client import KleerAPI
|
|
from config import Config
|
|
from services import (
|
|
create_time_event,
|
|
extract_work_time,
|
|
get_user_by_name,
|
|
parse_csv,
|
|
parse_xlsx,
|
|
)
|
|
|
|
|
|
def run_cli():
|
|
"""Run the CLI interface for time reporting."""
|
|
# Validate configuration
|
|
try:
|
|
Config.validate_config()
|
|
except ValueError as e:
|
|
logging.error(str(e))
|
|
sys.exit(1)
|
|
|
|
# Find CSV and XLSX files
|
|
current_directory = pathlib.Path.cwd()
|
|
data_files = list(current_directory.glob('*.csv')) + list(current_directory.glob('*.xlsx'))
|
|
|
|
if not data_files:
|
|
logging.error("No CSV or XLSX files found in the current directory.")
|
|
sys.exit(1)
|
|
|
|
logging.info(f"Data files found: {', '.join(f.name for f in data_files)}")
|
|
|
|
# Initialize API client
|
|
kleer_api = KleerAPI()
|
|
|
|
# Process each CSV file
|
|
for data_file in data_files:
|
|
logging.info(f"Processing file: {data_file.name}")
|
|
|
|
try:
|
|
# Get user information
|
|
users_data = kleer_api.get_user_info()
|
|
username = Config.get_username()
|
|
|
|
try:
|
|
user_id = get_user_by_name(users_data, username)
|
|
except ValueError as e:
|
|
logging.error(str(e))
|
|
continue
|
|
|
|
logging.info(f"Using user ID: {user_id} for events.")
|
|
|
|
# Get projects
|
|
projects = kleer_api.get_projects(user_id=user_id)
|
|
if not projects:
|
|
logging.warning("No projects found for user.")
|
|
continue
|
|
|
|
logging.info("Available projects from Kleer API:")
|
|
for name, value in projects.items():
|
|
logging.info(f"- {name}: {value}")
|
|
|
|
# Parse CSV file
|
|
if data_file.suffix.lower() == '.xlsx':
|
|
content = data_file.read_bytes()
|
|
rows = parse_xlsx(content)
|
|
else:
|
|
with data_file.open(mode='r', newline='', encoding='utf-8') as f:
|
|
content = f.read()
|
|
rows = parse_csv(content)
|
|
|
|
work_time = extract_work_time(rows)
|
|
|
|
if not work_time:
|
|
logging.info(f"No valid work time found in {data_file.name}.")
|
|
continue
|
|
|
|
# Let user select project
|
|
project_info = _select_project(projects)
|
|
if not project_info:
|
|
logging.warning("No project selected, skipping file.")
|
|
continue
|
|
|
|
# Display work time summary
|
|
logging.info("Total work time by date:")
|
|
total_work_time = 0.0
|
|
|
|
for date, total_time in sorted(work_time.items()):
|
|
logging.info(f"{date}: {total_time:.2f} hours")
|
|
total_work_time += total_time
|
|
|
|
# Create and submit event
|
|
event = create_time_event(
|
|
user_id=user_id,
|
|
project_id=project_info["id"],
|
|
activity=project_info["activity"],
|
|
date=date,
|
|
hours=total_time
|
|
)
|
|
kleer_api.put_event(event)
|
|
|
|
logging.info(f"Total work time for all dates: {total_work_time:.2f} hours")
|
|
|
|
except Exception as e:
|
|
logging.error(f"Error processing {csv_file.name}: {e}")
|
|
logging.error(traceback.format_exc())
|
|
|
|
|
|
def _select_project(projects: Dict[str, Dict[str, Any]]) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Let user select a project from available projects.
|
|
|
|
Args:
|
|
projects: Dictionary of available projects
|
|
|
|
Returns:
|
|
Selected project info or None if cancelled
|
|
"""
|
|
while True:
|
|
print("\nAvailable projects:")
|
|
project_list = list(projects.items())
|
|
|
|
for idx, (name, value) in enumerate(project_list, start=1):
|
|
print(f"{idx}. {name} (ID: {value['id']}, Activity: {value['activity']})")
|
|
|
|
project_choice = input("\nSelect a project by number (or 'q' to quit): ")
|
|
|
|
if project_choice.lower() == 'q':
|
|
return None
|
|
|
|
try:
|
|
project_index = int(project_choice) - 1
|
|
if 0 <= project_index < len(project_list):
|
|
_, project_info = project_list[project_index]
|
|
return project_info
|
|
else:
|
|
logging.error("Invalid project selection. Please try again.")
|
|
except ValueError:
|
|
logging.error("Invalid input. Please enter a number or 'q' to quit.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
run_cli()
|