Skip to content

Instantly share code, notes, and snippets.

@zipus
Last active May 9, 2025 07:44
Show Gist options
  • Select an option

  • Save zipus/89ed5a751a23512a8ec3ce9feaf54cd5 to your computer and use it in GitHub Desktop.

Select an option

Save zipus/89ed5a751a23512a8ec3ce9feaf54cd5 to your computer and use it in GitHub Desktop.
Tryton Chronos CLI
requests
prompt_toolkit
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.")
{
"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