Last active
May 9, 2025 07:44
-
-
Save zipus/89ed5a751a23512a8ec3ce9feaf54cd5 to your computer and use it in GitHub Desktop.
Tryton Chronos CLI
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| requests | |
| prompt_toolkit |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| from pathlib import Path | |
| import json | |
| import datetime | |
| import os | |
| import sys | |
| import requests | |
| from prompt_toolkit import PromptSession | |
| from prompt_toolkit.key_binding import KeyBindings | |
| from prompt_toolkit.completion import WordCompleter | |
| # Platform-specific keypress detection | |
| if sys.platform.startswith("win"): | |
| import msvcrt | |
| def get_keypress(): | |
| return msvcrt.getch().decode('utf-8').lower() | |
| else: | |
| import termios | |
| import tty | |
| def get_keypress(): | |
| fd = sys.stdin.fileno() | |
| old_settings = termios.tcgetattr(fd) | |
| try: | |
| tty.setraw(sys.stdin.fileno()) | |
| ch = sys.stdin.read(1).lower() | |
| finally: | |
| termios.tcsetattr(fd, termios.TCSADRAIN, old_settings) | |
| return ch | |
| # Paths and config | |
| today = datetime.date.today().isoformat() | |
| data_dir = Path("tracker_data") | |
| data_dir.mkdir(exist_ok=True) | |
| file_path = data_dir / f"{today}.json" | |
| config_path = Path("config.json") | |
| # Load data | |
| if file_path.exists(): | |
| with open(file_path, "r") as f: | |
| data = json.load(f) | |
| else: | |
| data = {"events": [], "blocks": []} | |
| # Load config | |
| if not config_path.exists(): | |
| print("β Missing config.json with url, db, key and employee.") | |
| sys.exit(1) | |
| with open(config_path) as f: | |
| config = json.load(f) | |
| # Utilities | |
| def time_to_minutes(t): | |
| h, m = map(int, t.split(":")) | |
| return h * 60 + m | |
| def minutes_to_time(m): | |
| h = m // 60 | |
| m = m % 60 | |
| return f"{h}:{m:02d}" | |
| def formatDuration(duration): | |
| h = duration // 3600 | |
| m = (duration % 3600) // 60 | |
| return f"{h}:{m:02d}" | |
| def save_data(): | |
| with open(file_path, "w") as f: | |
| json.dump(data, f, indent=2) | |
| def show_report(): | |
| print("\n--- DAILY SUMMARY ---") | |
| total_tracked = 0 | |
| total_assigned = 0 | |
| now_minutes = time_to_minutes(datetime.datetime.now().strftime("%H:%M")) | |
| last_open_in = None | |
| for event in data["events"]: | |
| if event["type"] == "in" and "out" in event: | |
| total_tracked += time_to_minutes(event["out"]) - time_to_minutes(event["time"]) | |
| elif event["type"] == "in": | |
| last_open_in = event["time"] | |
| if last_open_in: | |
| total_tracked += now_minutes - time_to_minutes(last_open_in) | |
| for block in data["blocks"]: | |
| total_assigned += time_to_minutes(block["duration"]) | |
| print(f"β³ Tracked (in/out): {minutes_to_time(total_tracked)}") | |
| print(f"π§© Assigned (blocks): {minutes_to_time(total_assigned)}") | |
| diff = total_tracked - total_assigned | |
| print(f"β Unassigned time: {minutes_to_time(abs(diff))}" + (" (over-assigned)" if diff < 0 else "")) | |
| if data["blocks"]: | |
| print("\nπ§Ύ Assigned blocks:") | |
| for block in data["blocks"]: | |
| synced = "β " if block.get("synced") else "β" | |
| print(f" {synced} {block['label']} ({block['duration']}) - {block.get('work_name', 'No work')}") | |
| else: | |
| print("\nπ No blocks assigned.") | |
| def register_in(): | |
| now = datetime.datetime.now().strftime("%H:%M") | |
| data["events"].append({"type": "in", "time": now}) | |
| print(f"π΅ Check-in at {now}") | |
| save_data() | |
| def register_out(): | |
| now = datetime.datetime.now().strftime("%H:%M") | |
| for block in reversed(data["events"]): | |
| if block["type"] == "in" and "out" not in block: | |
| block["out"] = now | |
| print(f"π΄ Check-out at {now}") | |
| save_data() | |
| return | |
| print("β οΈ No open check-in found.") | |
| def get_works_from_api(): | |
| url = f"{config['url'].rstrip('/')}/{config['db']}/timesheet/employee/{config['employee']}/works" | |
| headers = {"Authorization": f"bearer {config['key']}"} | |
| r = requests.get(url, headers=headers) | |
| return r.json() if r.ok else [] | |
| def choose_work(): | |
| works = get_works_from_api() | |
| if not works: | |
| print("β οΈ Work list is empty.") | |
| return None, None | |
| current = datetime.datetime.now().replace(hour=0, minute=0, second=0, microsecond=0) | |
| filtered_works = [] | |
| for w in works: | |
| start = datetime.datetime.fromisoformat(w['start']) if w.get('start') else None | |
| end = datetime.datetime.fromisoformat(w['end']) if w.get('end') else None | |
| if (not start or start <= current) and (not end or end >= current): | |
| filtered_works.append(w) | |
| if not filtered_works: | |
| print("β οΈ No works available for the current date.") | |
| return None, None | |
| names = [w["name"] for w in filtered_works] | |
| completer = WordCompleter(names, ignore_case=True) | |
| session = PromptSession() | |
| selected_name = session.prompt("π Select work: ", completer=completer) | |
| for w in filtered_works: | |
| if w["name"].lower() == selected_name.lower(): | |
| return w["id"], w["name"] | |
| print("β οΈ Work not found.") | |
| return None, None | |
| def push_block_to_chronos(date_str, duration_str, label, uuid=None, work_id=None, work_name=None): | |
| parts = list(map(int, duration_str.split(":"))) | |
| duration = parts[0] * 3600 + parts[1] * 60 | |
| values = { | |
| "duration": duration, | |
| "work": work_id, | |
| "work.name": work_name, | |
| "description": label, | |
| "uuid": uuid or os.urandom(4).hex(), | |
| "employee": config["employee"], | |
| "date": date_str | |
| } | |
| url = f"{config['url'].rstrip('/')}/{config['db']}/timesheet/line" | |
| headers = { | |
| "Authorization": f"bearer {config['key']}", | |
| "Content-Type": "application/json" | |
| } | |
| r = requests.post(url, headers=headers, json=values) | |
| if r.ok: | |
| print(f"β Line sent: {label} - {duration_str}") | |
| return True | |
| else: | |
| print(f"β Error sending: {r.status_code} - {r.text}") | |
| return False | |
| def add_block(): | |
| session = PromptSession() | |
| kb = KeyBindings() | |
| duration_minutes = 0 | |
| @kb.add('up') | |
| def _(event): | |
| nonlocal duration_minutes | |
| duration_minutes = min(duration_minutes + 15, 23*60 + 59) | |
| event.app.current_buffer.text = minutes_to_time(duration_minutes) | |
| @kb.add('down') | |
| def _(event): | |
| nonlocal duration_minutes | |
| duration_minutes = max(duration_minutes - 15, 0) | |
| event.app.current_buffer.text = minutes_to_time(duration_minutes) | |
| duration = session.prompt("Duration (HH:MM): ", key_bindings=kb) | |
| work_id, work_name = choose_work() | |
| label = input("Label: ") | |
| print("β/β to adjust duration. Enter to confirm.") | |
| if not work_id: | |
| return | |
| block = { | |
| "label": label, | |
| "duration": duration, | |
| "work_id": work_id, | |
| "work_name": work_name, | |
| "synced": False, | |
| "uuid": os.urandom(4).hex() | |
| } | |
| data["blocks"].append(block) | |
| save_data() | |
| if push_block_to_chronos(today, duration, label, block["uuid"], work_id, work_name): | |
| block["synced"] = True | |
| save_data() | |
| def retry_unsynced(): | |
| for block in data["blocks"]: | |
| if not block.get("synced", False): | |
| success = push_block_to_chronos(today, block["duration"], block["label"], block.get("uuid"), block.get("work_id"), block.get("work_name")) | |
| block["synced"] = success | |
| save_data() | |
| print("π Retry completed.") | |
| def list_remote_lines(store=True): | |
| url = f"{config['url'].rstrip('/')}/{config['db']}/timesheet/employee/{config['employee']}/lines/{today}" | |
| headers = {"Authorization": f"bearer {config['key']}"} | |
| r = requests.get(url, headers=headers) | |
| if not r.ok: | |
| print("β Error fetching lines:", r.status_code) | |
| return | |
| lines = r.json() | |
| print(f"\nπ€ Remote lines in Chronos for {today}:") | |
| if store: | |
| data["blocks"] = [b for b in data["blocks"] if not b.get("synced")] | |
| for line in lines: | |
| dur = int(line["duration"]) | |
| label = line.get("description", "") | |
| work_name = line.get("work.name", "Unknown") | |
| work_id = line.get("work") | |
| print(f" - {work_name}: {formatDuration(dur)} | {label}") | |
| if store: | |
| block = { | |
| "label": label, | |
| "duration": formatDuration(dur), | |
| "work_id": work_id, | |
| "work_name": work_name, | |
| "synced": True, | |
| "remote_only": True, | |
| "uuid": line.get("uuid") or os.urandom(4).hex() | |
| } | |
| data["blocks"].append(block) | |
| if store: | |
| save_data() | |
| print("π Remote lines added to blocks.") | |
| def open_in_sublime(): | |
| os.system(f"subl -n {file_path}") | |
| # Main loop | |
| show_report() | |
| while True: | |
| print("\nOptions: [i=in, o=out, a=add, u=unsynced, l=remote, r=report, s=sublime, e=exit]") | |
| key = get_keypress() | |
| if key == 'i': | |
| register_in() | |
| elif key == 'o': | |
| register_out() | |
| elif key == 'a': | |
| add_block() | |
| elif key == 'u': | |
| retry_unsynced() | |
| elif key == 'l': | |
| list_remote_lines() | |
| elif key == 'r': | |
| show_report() | |
| elif key == 's': | |
| open_in_sublime() | |
| elif key == 'e': | |
| print("π Exiting...") | |
| break | |
| else: | |
| print(f"Key '{key}' not recognized.") |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| { | |
| "url": "https://tryton.example.com/", | |
| "db": "erp", | |
| "key": "APPLICATION_KEY", | |
| "employee": 1 | |
| } |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment