import requests import questionary from rich.console import Console from rich.table import Table import json import os import re from datetime import datetime, timedelta from dotenv import load_dotenv # Load environment variables from .env if it exists load_dotenv() console = Console() BASE_URL = "https://api.tamigo.com" def parse_tamigo_date(date_str): """ Parses Tamigo date formats: - /Date(1600898400000+0200)/ - 2023-10-27T08:00:00 """ if not date_str: return None # Handle /Date(1600898400000+0200)/ match = re.search(r"/Date\((\d+)([+-]\d+)?\)/", date_str) if match: ms = int(match.group(1)) # Convert ms to seconds return datetime.fromtimestamp(ms / 1000.0) # Handle ISO format try: # datetime.fromisoformat handles T and optional Z/+offset in Python 3.7+ return datetime.fromisoformat(date_str.replace("Z", "+00:00")) except: return None class TamigoClient: def __init__(self): self.session_token = None self.user_info = None self.employee_id = None def login(self, email, password): # Try different URL patterns based on docs and common patterns urls = [ f"{BASE_URL}/login/application", f"{BASE_URL}/Login", f"{BASE_URL}/Login/", f"{BASE_URL}/login", f"{BASE_URL}/login/" ] last_error = "" for url in urls: payload = { "Email": email, "Password": password, "Name": "TamigoCLI", # For /login/application "Key": password # For /login/application } try: headers = { "Content-Type": "application/json", "Accept": "application/json" } response = requests.post(url, json=payload, headers=headers, timeout=15) if response.status_code == 200: try: data = response.json() # Token can be in different fields self.session_token = data.get("SessionToken") or data.get("securitytoken") or data.get("Token") if self.session_token: self.user_info = data self.employee_id = data.get("EmployeeId") return True except json.JSONDecodeError: text = response.text.strip().strip('"') if text and len(text) > 20: self.session_token = text self.user_info = {"Email": email} return True last_error = f"URL: {url}, Status: {response.status_code}" except Exception as e: last_error = f"URL: {url}, Error: {str(e)}" console.print(f"[red]Login failed.[/red]") console.print(f"[dim]Debug: {last_error}[/dim]") return False def get_employee_id(self): if self.employee_id: return self.employee_id headers = { "x-tamigo-token": self.session_token, "securitytoken": self.session_token, "Accept": "application/json" } # Method A: User Info from Login if self.user_info and self.user_info.get("EmployeeId"): self.employee_id = self.user_info.get("EmployeeId") return self.employee_id # Method B: My Overview url = f"{BASE_URL}/shifts/myoverview" try: response = requests.get(url, headers=headers) if response.status_code == 200: shifts = response.json() if shifts and len(shifts) > 0: self.employee_id = shifts[0].get("EmployeeId") return self.employee_id except: pass return None def get_employee_actual_shifts(self, start_date_dt, end_date_dt): """ Fetches actual worked shifts using employee-accessible endpoints. Tamigo's 'past' endpoint often limits to 60 days, so we fetch in chunks. """ if not self.session_token: return None headers = { "x-tamigo-token": self.session_token, "securitytoken": self.session_token, "Accept": "application/json" } all_shifts = [] days_diff = (end_date_dt - start_date_dt).days # Fetch in 60-day chunks moving backwards from end_date for i in range(0, days_diff + 1, 60): target_date = (end_date_dt - timedelta(days=i)).strftime("%Y-%m-%d") # Stop if we've moved past the start date current_dt = end_date_dt - timedelta(days=i) if current_dt < start_date_dt - timedelta(days=60): break url = f"{BASE_URL}/actualshifts/past/{target_date}" try: response = requests.get(url, headers=headers) if response.status_code == 200: data = response.json() if isinstance(data, list): all_shifts.extend(data) elif response.status_code == 401: response = requests.get(url, params={"securitytoken": self.session_token}) if response.status_code == 200: all_shifts.extend(response.json()) except Exception as e: console.print(f"[dim]Failed to fetch chunk at {target_date}: {e}[/dim]") # Supplement with /shifts/period/ start_str = start_date_dt.strftime("%Y-%m-%d") end_str = end_date_dt.strftime("%Y-%m-%d") url_period = f"{BASE_URL}/shifts/period/{start_str}/{end_str}" try: response = requests.get(url_period, headers=headers) if response.status_code == 200: data = response.json() if isinstance(data, list): all_shifts.extend(data) except: pass return all_shifts def calculate_checkins(client): range_choice = questionary.select( "Select period:", choices=[ "Last 365 days", "Last 30 days", "This Month", "This Year", "Custom range..." ] ).ask() end_date_dt = datetime.now() if range_choice == "Last 365 days": start_date_dt = end_date_dt - timedelta(days=365) elif range_choice == "Last 30 days": start_date_dt = end_date_dt - timedelta(days=30) elif range_choice == "This Month": start_date_dt = end_date_dt.replace(day=1) elif range_choice == "This Year": start_date_dt = end_date_dt.replace(month=1, day=1) else: # Custom range while True: start_str = questionary.text("Start date (YYYY-MM-DD):", default=(datetime.now() - timedelta(days=30)).strftime("%Y-%m-%d")).ask() end_str = questionary.text("End date (YYYY-MM-DD):", default=datetime.now().strftime("%Y-%m-%d")).ask() try: start_date_dt = datetime.strptime(start_str, "%Y-%m-%d") end_date_dt = datetime.strptime(end_str, "%Y-%m-%d") if start_date_dt > end_date_dt: console.print("[red]Start date must be before end date![/red]") continue break except ValueError: console.print("[red]Invalid format. Please use YYYY-MM-DD.[/red]") with console.status(f"[bold green]Fetching work history from {start_date_dt.strftime('%Y-%m-%d')} to {end_date_dt.strftime('%Y-%m-%d')}..."): data = client.get_employee_actual_shifts(start_date_dt, end_date_dt) if data: work_days = {} # date_str -> {hours, text} # Filter data to ensure it's strictly within our requested range # (Since the API chunks might return slightly more) requested_start = start_date_dt.date() requested_end = end_date_dt.date() for item in data: raw_date = item.get("Date") or item.get("StartTime") or item.get("CheckInTime") dt = parse_tamigo_date(raw_date) if dt: item_date = dt.date() if not (requested_start <= item_date <= requested_end): continue date_str = dt.strftime("%Y-%m-%d") is_absent = item.get("IsAbsent", False) hours = (item.get("ActualShiftHours") or item.get("CheckInOutShiftHours") or item.get("Hours") or 0) if hours == 0 and item.get("StartTime") and item.get("EndTime"): st = parse_tamigo_date(item.get("StartTime")) et = parse_tamigo_date(item.get("EndTime")) if st and et: hours = (et - st).total_seconds() / 3600.0 has_actual = False if hours > 0 or item.get("CheckInTime") or item.get("ActualStartTime"): has_actual = True if item.get("ActualShift") and item["ActualShift"].get("Shift", 0) > 0: has_actual = True hours = item["ActualShift"]["Shift"] if item.get("StartTime") and not is_absent: has_actual = True if has_actual and not is_absent: if date_str not in work_days: work_days[date_str] = { "hours": float(hours), "text": item.get("ActualShiftText") or item.get("ActivityName") or item.get("DepartmentName") or "Worked" } else: work_days[date_str]["hours"] += float(hours) if not work_days: console.print("[yellow]No work records found for this period.[/yellow]") return all_dates = sorted(work_days.keys(), reverse=True) table = Table(title="Worked Days in Period", show_header=True, header_style="bold magenta") table.add_column("Date", style="cyan") table.add_column("Hours", justify="right") table.add_column("Details", style="dim") for day in all_dates: info = work_days[day] table.add_row(day, f"{info['hours']:.2f}", str(info['text'])) console.print(table) console.print(f"\n[bold]Work Statistics:[/bold]") console.print(f" - Period: [cyan]{start_date_dt.strftime('%Y-%m-%d')} to {end_date_dt.strftime('%Y-%m-%d')}[/cyan]") console.print(f" - Days Worked: [bold green]{len(all_dates)}[/bold green]") total_hours = sum(d['hours'] for d in work_days.values()) console.print(f" - Total Hours: [bold green]{total_hours:.2f}[/bold green]") else: console.print("[yellow]Could not retrieve any shift data for this period.[/yellow]") def show_profile(client): if client.user_info: console.print_json(data=client.user_info) else: console.print("[yellow]No profile info available. Are you logged in?[/yellow]") def main(): client = TamigoClient() console.print("[bold blue]Welcome to Tamigo CLI[/bold blue]") email = os.getenv("TAMIGO_EMAIL") if not email: email = questionary.text("Email:").ask() password = os.getenv("TAMIGO_PASSWORD") if not password: password = questionary.password("Password:").ask() if not email or not password: return with console.status("[bold green]Logging in..."): success = client.login(email, password) if success: console.print("[bold green]Login successful![/bold green]") menu(client) else: console.print("[bold red]Login failed. Please check your credentials.[/bold red]") def menu(client): while True: choice = questionary.select( "What would you like to do?", choices=[ "Calculate actual work days", "Show profile info", "Logout and Exit" ] ).ask() if choice == "Calculate actual work days": calculate_checkins(client) elif choice == "Show profile info": show_profile(client) else: break if __name__ == "__main__": main()