waybar_time_track/time_track.py

#!/usr/bin/env python3
"""Time tracker for Waybar.

Usage:
    time_track.py toggle [task]          -- start or stop a session
    time_track.py status                 -- emit Waybar JSON
    time_track.py log [n]                -- show last n entries (default 10)
    time_track.py timesheet [week|month] -- generate a Markdown timesheet
    time_track.py current                -- show the active session, if any
    time_track.py stop                   -- stop the active session
    time_track.py sync                   -- upload finished sessions to a server
"""

import argparse
import json
import os
import shutil
import subprocess
import sys
import uuid
from datetime import datetime, timedelta
from pathlib import Path
from urllib.error import HTTPError, URLError
from urllib.parse import urlsplit
from urllib.request import Request, urlopen

DATA_DIR = Path(
    os.environ.get("TIME_TRACK_DIR", Path.home() / ".local" / "share" / "time_track")
)
DATA_FILE = DATA_DIR / "data.json"
DEFAULT_TASK = ""
SYNC_TIMEOUT_SECONDS = 15


def load_data():
    DATA_DIR.mkdir(parents=True, exist_ok=True)
    if DATA_FILE.exists():
        try:
            with DATA_FILE.open("r", encoding="utf-8") as f:
                data = json.load(f)
            migrated = False
            entry_ids = set()
            for entry in data.get("entries", []):
                try:
                    entry_id = str(uuid.UUID(entry.get("id", "")))
                except (AttributeError, TypeError, ValueError):
                    entry_id = ""
                if not entry_id or entry_id in entry_ids:
                    entry_id = str(uuid.uuid4())
                    entry["id"] = entry_id
                    migrated = True
                entry_ids.add(entry_id)
            if migrated:
                save_data(data)
            return data
        except (json.JSONDecodeError, OSError):
            pass
    return {"active": False, "current": None, "entries": []}


def save_data(data):
    DATA_DIR.mkdir(parents=True, exist_ok=True)
    with DATA_FILE.open("w", encoding="utf-8") as f:
        json.dump(data, f, indent=2)


def now_iso():
    return datetime.now().isoformat(timespec="seconds")


def parse_iso(value):
    return datetime.fromisoformat(value)


def format_duration(seconds):
    sign = "-" if seconds < 0 else ""
    seconds = abs(int(seconds))
    hours, remainder = divmod(seconds, 3600)
    minutes, secs = divmod(remainder, 60)
    parts = []
    if hours:
        parts.append(f"{hours}h")
    if minutes:
        parts.append(f"{minutes}m")
    if not hours and not minutes:
        parts.append(f"{secs}s")
    return sign + " ".join(parts)


def format_time(dt):
    return dt.strftime("%H:%M")


def format_date(dt):
    return dt.strftime("%Y-%m-%d")


def previous_tasks(data):
    """Return a sorted list of unique task names from finished entries."""
    tasks = {e.get("task", "") for e in data["entries"]}
    return sorted(t for t in tasks if t)


def prompt_for_task(data):
    """Prompt the user for a task name.

    Returns the selected/typed task string. An empty string is allowed.
    Returns None if the user cancelled the prompt.
    """
    tasks = previous_tasks(data)
    input_text = "\n".join(tasks)

    launchers = [
        (
            "wofi",
            lambda: ["wofi", "--dmenu", "--prompt", "Enter task", "--insensitive"],
        ),
        (
            "rofi",
            lambda: ["rofi", "-dmenu", "-p", "Enter task", "-i"],
        ),
        (
            "bemenu",
            lambda: ["bemenu", "-p", "Enter task", "-i"],
        ),
        (
            "dmenu",
            lambda: ["dmenu", "-p", "Enter task", "-i"],
        ),
    ]

    for binary, cmd in launchers:
        if shutil.which(binary):
            try:
                result = subprocess.run(
                    cmd(),
                    input=input_text,
                    text=True,
                    capture_output=True,
                    check=False,
                )
                if result.returncode != 0:
                    return None
                return result.stdout.strip()
            except (OSError, subprocess.SubprocessError):
                continue

    # Fallback to zenity (plain entry, no dropdown).
    if shutil.which("zenity"):
        try:
            result = subprocess.run(
                [
                    "zenity",
                    "--entry",
                    "--title=Time Tracker",
                    "--text=Enter task (leave empty for none):",
                ],
                text=True,
                capture_output=True,
                check=False,
            )
            if result.returncode != 0:
                return None
            return result.stdout.strip()
        except (OSError, subprocess.SubprocessError):
            pass

    # Terminal fallback.
    print("Enter task (empty for none):", file=sys.stderr)
    if tasks:
        print("Previous tasks:", ", ".join(tasks), file=sys.stderr)
    try:
        return input("Task: ").strip()
    except EOFError:
        return None


def toggle(task=None):
    data = load_data()

    if data["active"]:
        start = parse_iso(data["current"]["start"])
        end = datetime.now()
        duration = int((end - start).total_seconds())
        entry = {
            "id": str(uuid.uuid4()),
            "task": data["current"].get("task", ""),
            "start": data["current"]["start"],
            "end": end.isoformat(timespec="seconds"),
            "duration": duration,
        }
        data["entries"].append(entry)
        data["active"] = False
        data["current"] = None
        save_data(data)
        label = entry["task"] or "(no task)"
        print(
            f"Stopped: {label} @ {format_time(end)} "
            f"({format_duration(duration)})",
            file=sys.stderr,
        )
        return

    # Starting a new session.
    if task is None:
        task = prompt_for_task(data)
        if task is None:
            print("Start cancelled.", file=sys.stderr)
            return

    data["active"] = True
    data["current"] = {
        "task": task,
        "start": now_iso(),
    }
    save_data(data)
    label = task or "(no task)"
    print(
        f"Started: {label} @ {format_time(datetime.now())}",
        file=sys.stderr,
    )


def stop():
    data = load_data()
    if not data["active"]:
        print("No active session.", file=sys.stderr)
        return
    toggle()


def status():
    data = load_data()
    if data["active"]:
        start = parse_iso(data["current"]["start"])
        elapsed = int((datetime.now() - start).total_seconds())
        elapsed_text = format_duration(elapsed)
        task = data["current"].get("task", "") or "(no task)"
        waybar = {
            "text": f"⏱ {elapsed_text}",
            "tooltip": (
                f"Task: {task}\n"
                f"Started: {format_time(start)} ({format_date(start)})\n"
                f"Elapsed: {elapsed_text}"
            ),
            "class": "running",
        }
    else:
        last_entry = data["entries"][-1] if data["entries"] else None
        if last_entry:
            label = last_entry.get("task", "") or "(no task)"
            tooltip = (
                f"Last: {label} "
                f"@ {format_time(parse_iso(last_entry['end']))} "
                f"({format_duration(last_entry['duration'])}). Click to start."
            )
        else:
            tooltip = "Click to start tracking."
        waybar = {
            "text": "⏱ --:--",
            "tooltip": tooltip,
            "class": "idle",
        }
    print(json.dumps(waybar))


def sync():
    """Upload all finished entries using environment-only configuration."""
    server_url = os.environ.get("TIME_TRACK_SERVER_URL", "").rstrip("/")
    token = os.environ.get("TIME_TRACK_SERVER_TOKEN", "")
    device_id = os.environ.get("TIME_TRACK_DEVICE_ID", "")
    missing = [
        name
        for name, value in (
            ("TIME_TRACK_SERVER_URL", server_url),
            ("TIME_TRACK_SERVER_TOKEN", token),
            ("TIME_TRACK_DEVICE_ID", device_id),
        )
        if not value
    ]
    if missing:
        print(
            "Sync is not configured. Set " + ", ".join(missing) + ".",
            file=sys.stderr,
        )
        return 1

    parsed_url = urlsplit(server_url)
    if parsed_url.scheme not in {"http", "https"} or not parsed_url.netloc:
        print(
            "Sync is not configured: TIME_TRACK_SERVER_URL must be an absolute HTTP(S) URL.",
            file=sys.stderr,
        )
        return 1

    data = load_data()
    entries = [
        {
            "id": entry["id"],
            "task": entry.get("task", ""),
            "start": entry["start"],
            "end": entry["end"],
            "duration": entry["duration"],
        }
        for entry in data["entries"]
    ]
    payload = json.dumps({"device_id": device_id, "entries": entries}).encode("utf-8")
    request = Request(
        f"{server_url}/api/v1/sync/time-track",
        data=payload,
        headers={
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json",
            "Accept": "application/json",
        },
        method="POST",
    )

    try:
        with urlopen(request, timeout=SYNC_TIMEOUT_SECONDS) as response:
            status = response.status
            response.read()
    except HTTPError as error:
        print(f"Sync failed: server returned HTTP {error.code}.", file=sys.stderr)
        return 1
    except URLError as error:
        reason = error.reason
        print(f"Sync failed: could not reach server ({reason}).", file=sys.stderr)
        return 1
    except (OSError, ValueError) as error:
        print(f"Sync failed: request error ({error}).", file=sys.stderr)
        return 1

    print(f"Synced {len(entries)} finished entr{'y' if len(entries) == 1 else 'ies'} (HTTP {status}).")
    return 0


def current():
    data = load_data()
    if data["active"]:
        start = parse_iso(data["current"]["start"])
        elapsed = int((datetime.now() - start).total_seconds())
        task = data["current"].get("task", "") or "(no task)"
        print(
            f"Task: {task}\n"
            f"Started: {start.isoformat(' ', timespec='seconds')}\n"
            f"Elapsed: {format_duration(elapsed)}"
        )
    else:
        print("No active session.")


def log_cmd(count=10):
    data = load_data()
    entries = data["entries"][-count:]
    if not entries:
        print("No logged entries yet.")
        return
    print(f"{'Start':20} {'End':20} {'Task':12} {'Duration':10}")
    print("-" * 65)
    for e in entries:
        start = parse_iso(e["start"])
        end = parse_iso(e["end"])
        label = e.get("task", "") or "(no task)"
        print(
            f"{start.isoformat(' ', timespec='seconds'):20} "
            f"{end.isoformat(' ', timespec='seconds'):20} "
            f"{label:12} {format_duration(e['duration']):>10}"
        )


def timesheet(period="week"):
    data = load_data()
    today = datetime.now().date()

    if period == "week":
        start_date = today - timedelta(days=today.weekday())
        end_date = start_date + timedelta(days=6)
    elif period == "month":
        start_date = today.replace(day=1)
        next_month = (today.replace(day=28) + timedelta(days=4)).replace(day=1)
        end_date = (next_month - timedelta(days=1))
    else:
        start_date = today - timedelta(days=7)
        end_date = today

    entries = [
        e for e in data["entries"]
        if start_date <= parse_iso(e["start"]).date() <= end_date
    ]

    if not entries:
        print(f"No entries for {period} ({start_date} → {end_date}).")
        return

    by_task: dict[str, int] = {}
    by_date: dict[str, int] = {}
    rows = []

    for e in entries:
        start = parse_iso(e["start"])
        end = parse_iso(e["end"])
        duration = e["duration"]
        date_key = format_date(start)
        label = e.get("task", "") or "(no task)"
        by_task[label] = by_task.get(label, 0) + duration
        by_date[date_key] = by_date.get(date_key, 0) + duration
        rows.append(
            f"| {format_date(start)} | {format_time(start)} | "
            f"{format_time(end)} | {label} | "
            f"{format_duration(duration)} |"
        )

    total = sum(by_task.values())

    print(f"# Timesheet: {period} ({start_date} → {end_date})")
    print()
    print("| Date | Start | End | Task | Duration |")
    print("|------|-------|-----|------|----------|")
    for row in rows:
        print(row)
    print()
    print("## Summary by task")
    print()
    print("| Task | Total |")
    print("|------|-------|")
    for task, seconds in sorted(by_task.items()):
        print(f"| {task} | {format_duration(seconds)} |")
    print(f"| **Total** | **{format_duration(total)}** |")
    print()
    print("## Summary by day")
    print()
    print("| Day | Total |")
    print("|-----|-------|")
    for date_key in sorted(by_date):
        print(f"| {date_key} | {format_duration(by_date[date_key])} |")


def main():
    parser = argparse.ArgumentParser(description="Waybar time tracker")
    subparsers = parser.add_subparsers(dest="command", required=True)

    p_toggle = subparsers.add_parser("toggle", help="start or stop a session")
    p_toggle.add_argument(
        "task",
        nargs="?",
        default=None,
        help="task name (prompts if omitted and not stopping)",
    )

    subparsers.add_parser("status", help="emit Waybar JSON")
    subparsers.add_parser("current", help="show the active session")
    subparsers.add_parser("stop", help="stop the active session")
    subparsers.add_parser("sync", help="upload finished sessions to the configured server")

    p_log = subparsers.add_parser("log", help="show recent entries")
    p_log.add_argument("count", nargs="?", type=int, default=10)

    p_sheet = subparsers.add_parser("timesheet", help="generate a Markdown timesheet")
    p_sheet.add_argument(
        "period",
        nargs="?",
        choices=["week", "month"],
        default="week",
        help="time period to summarize",
    )

    args = parser.parse_args()

    if args.command == "toggle":
        toggle(args.task)
    elif args.command == "status":
        status()
    elif args.command == "current":
        current()
    elif args.command == "stop":
        stop()
    elif args.command == "sync":
        return sync()
    elif args.command == "log":
        log_cmd(args.count)
    elif args.command == "timesheet":
        timesheet(args.period)


if __name__ == "__main__":
    sys.exit(main() or 0)