Best Coworking Spaces Toggl: Our Experience

#!/usr/bin/env python3
"""
toggl_collector.py — Pull time entries from Toggl Track API,
enrich with workspace tags, and persist to SQLite.

Requirements: requests>=2.31, python-dotenv>=1.0
Set TOGGL_API_TOKEN in .env or environment.
"""

import os
import sys
import sqlite3
import logging
import requests
from datetime import datetime, timedelta
from dotenv import load_dotenv
from typing import Optional

load_dotenv()

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s"
)
logger = logging.getLogger(__name__)

TOGGL_API_TOKEN = os.getenv("TOGGL_API_TOKEN")
WORKSPACE_ID = int(os.getenv("TOGGL_WORKSPACE_ID", "0"))
DB_PATH = os.getenv("OUTPUT_DB", "coworking_data.db")

if not TOGGL_API_TOKEN:
    logger.error("TOGGL_API_TOKEN not set. Create a .env file.")
    sys.exit(1)

AUTH = (TOGGL_API_TOKEN, "api_token")
BASE_URL = "https://api.track.toggl.com/api/v9"


def create_database(db_path: str) -> sqlite3.Connection:
    """Initialize SQLite schema for time-entry storage."""
    conn = sqlite3.connect(db_path)
    conn.execute("PRAGMA journal_mode=WAL")
    conn.execute("""
        CREATE TABLE IF NOT EXISTS time_entries (
            id INTEGER PRIMARY KEY,
            tid INTEGER UNIQUE,
            description TEXT,
            project_id INTEGER,
            project_name TEXT,
            workspace_id INTEGER,
            tag_workspace TEXT,
            start TEXT,
            stop TEXT,
            duration_seconds INTEGER,
            focus_rating INTEGER,
            raw_json TEXT
        )
    """)
    conn.execute("""
        CREATE INDEX IF NOT EXISTS idx_start
        ON time_entries(start)
    """)
    logger.info("Database ready at %s", db_path)
    return conn


def fetch_time_entries(
    session: requests.Session,
    since: str,
    until: str,
    page: int = 1
) -> Optional[list]:
    """
    Fetch a single page of time entries.
    Toggl paginates with ?page=N; returns None on HTTP error.
    """
    url = f"{BASE_URL}/workspace/{WORKSPACE_ID}/time_entries"
    params = {
        "start_date": since,
        "end_date": until,
        "page": page,
        "per_page": 50,
        "order_field": "start",
        "order_dir": "ASC",
    }
    try:
        resp = session.get(url, auth=AUTH, params=params, timeout=30)
        resp.raise_for_status()
        data = resp.json()
        logger.info("Page %d: %d entries", page, len(data))
        return data
    except requests.RequestException as exc:
        logger.error("Page %d request failed: %s", page, exc)
        return None


def enrich_and_store(
    conn: sqlite3.Connection,
    entries: list
) -> int:
    """Insert or ignore entries; return count of new rows."""
    inserted = 0
    for entry in entries:
        tag_workspace = None
        for tag in entry.get("tags", []):
            if tag.startswith("ws-"):
                tag_workspace = tag
                break
        try:
            cursor = conn.execute(
                """
                INSERT OR IGNORE INTO time_entries
                (tid, description, project_id, project_name,
                 workspace_id, tag_workspace, start, stop,
                 duration_seconds, focus_rating, raw_json)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    entry["id"],
                    entry.get("description", ""),
                    entry.get("pid"),
                    entry.get("project"),
                    entry.get("wid"),
                    tag_workspace,
                    entry.get("start"),
                    entry.get("stop"),
                    entry.get("duration", 0),
                    None,  # filled by Slack webhook later
                    str(entry),
                ),
            )
            inserted += cursor.rowcount
        except sqlite3.Error as exc:
            logger.warning("Insert failed for tid=%d: %s", entry["id"], exc)
    conn.commit()
    return inserted


def main():
    """Orchestrate full collection for the past 90 days."""
    end = datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ")
    start = (datetime.utcnow() - timedelta(days=90)).strftime(
        "%Y-%m-%dT%H:%M:%SZ"
    )
    logger.info("Collecting entries from %s to %s", start, end)

    conn = create_database(DB_PATH)
    session = requests.Session()

    page = 1
    total_inserted = 0
    while True:
        entries = fetch_time_entries(session, start, end, page)
        if not entries:
            break
        n = enrich_and_store(conn, entries)
        total_inserted += n
        if len(entries) < 50:
            break
        page += 1

    logger.info("Total new entries inserted: %d", total_inserted)
    conn.close()


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
space_analyzer.py — Compute per-coworking-space productivity metrics
from the SQLite database populated by toggl_collector.py.

Outputs:
  1. Console summary table
  2. CSV export for downstream visualization
  3. Statistical significance tests (Mann-Whitney U)

Requirements: pandas>=2.0, scipy>=1.11, tabulate>=0.9
"""

import sqlite3
import csv
import logging
from collections import defaultdict
from datetime import datetime
from typing import Dict, List, Tuple

import pandas as pd
from scipy.stats import mannwhitneyu
from tabulate import tabulate

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

DB_PATH = "coworking_data.db"
OUTPUT_CSV = "space_comparison.csv"

# Minimum session length (seconds) to count as "deep work"
DEEP_WORK_THRESHOLD = 5400  # 90 minutes

# Known workspace metadata (manually maintained)
WORKSPACE_META = {
    "ws-webflow": {"name": "Webflow Hub", "city": "SF", "type": "open"},
    "ws-knotel": {"name": "Knotel FiDi", "city": "NYC", "type": "open"},
    "ws-industrious": {"name": "Industrious Flatiron", "city": "NYC", "type": "pod"},
    "ws-crew": {"name": "Crew Office", "city": "Austin", "type": "pod"},
    "ws-impact": {"name": "Impact Hub", "city": "Austin", "type": "open"},
    "ws-betahaus": {"name": "Betahaus", "city": "Berlin", "type": "hybrid"},
    "ws-st-obert": {"name": "St. Oberholz", "city": "Berlin", "type": "cafe"},
}


def load_entries(db_path: str) -> pd.DataFrame:
    """Load time entries into a DataFrame with parsed timestamps."""
    conn = sqlite3.connect(db_path)
    df = pd.read_sql_query(
        "SELECT * FROM time_entries WHERE duration_seconds > 0", conn
    )
    conn.close()
    df["start_dt"] = pd.to_datetime(df["start"])
    df["hour"] = df["start_dt"].dt.hour
    df["weekday"] = df["start_dt"].dt.day_name()
    logger.info("Loaded %d entries from database", len(df))
    return df


def compute_metrics(df: pd.DataFrame) -> Dict[str, dict]:
    """Compute per-workspace productivity metrics."""
    results = {}
    for tag, meta in WORKSPACE_META.items():
        subset = df[df["tag_workspace"] == tag]
        if subset.empty:
            logger.warning("No entries for workspace %s, skipping", tag)
            continue

        total_hours = subset["duration_seconds"].sum() / 3600
        deep = subset[subset["duration_seconds"] >= DEEP_WORK_THRESHOLD]
        deep_hours = deep["duration_seconds"].sum() / 3600
        deep_ratio = deep_hours / total_hours if total_hours > 0 else 0

        # Average focus rating (entries that have one)
        rated = subset.dropna(subset=["focus_rating"])
        avg_focus = rated["focus_rating"].mean() if len(rated) > 0 else None

        # Peak hour: which hour-of-day has the most tracked time?
        hour_counts = subset.groupby("hour")["duration_seconds"].sum()
        peak_hour = int(hour_counts.idxmax())

        results[tag] = {
            "name": meta["name"],
            "city": meta["city"],
            "type": meta["type"],
            "total_hours": round(total_hours, 1),
            "deep_hours": round(deep_hours, 1),
            "deep_ratio": round(deep_ratio, 3),
            "avg_focus": round(avg_focus, 2) if avg_focus else None,
            "peak_hour": peak_hour,
            "entry_count": len(subset),
        }
    return results


def significance_test(df: pd.DataFrame, tag_a: str, tag_b: str) -> Tuple[float, bool]:
    """
    Mann-Whitney U test on per-day deep-work hours between two spaces.
    Returns (p_value, significant_at_005).
    """
    a = (
        df[df["tag_workspace"] == tag_a]
        .groupby(df["start_dt"].dt.date)["duration_seconds"]
        .sum()
        .apply(lambda s: s / 3600)
    )
    b = (
        df[df["tag_workspace"] == tag_b]
        .groupby(df["start_dt"].dt.date)["duration_seconds"]
        .sum()
        .apply(lambda s: s / 3600)
    )
    # Use Mann-Whitney because daily hours are not normally distributed
    stat, p = mannwhitneyu(a, b, alternative="greater")
    return p, p < 0.05


def export_csv(results: Dict[str, dict], path: str):
    """Write comparison data to CSV."""
    rows = []
    for tag, m in results.items():
        rows.append({"workspace_id": tag, **m})
    pd.DataFrame(rows).to_csv(path, index=False)
    logger.info("CSV exported to %s", path)


def main():
    df = load_entries(DB_PATH)
    metrics = compute_metrics(df)

    # Print comparison table
    table = []
    for tag, m in sorted(metrics.items(), key=lambda x: -x[1]["deep_ratio"]):
        table.append([
            m["name"],
            m["city"],
            m["type"],
            f"{m['total_hours']:.0f}h",
            f"{m['deep_hours']:.0f}h",
            f"{m['deep_ratio']:.0%}",
            f"{m['avg_focus']:.1f}" if m["avg_focus"] else "N/A",
            f"{m['peak_hour']:02d}:00",
        ])

    headers = [
        "Workspace", "City", "Type", "Total Hrs", "Deep Hrs",
        "Deep Ratio", "Avg Focus", "Peak Hour",
    ]
    print("\n" + tabulate(table, headers=headers, tablefmt="github"))

    # Run significance test: best vs worst deep ratio
    sorted_tags = sorted(metrics, key=lambda t: metrics[t]["deep_ratio"])
    best, worst = sorted_tags[-1], sorted_tags[0]
    p_val, sig = significance_test(df, best, worst)
    logger.info(
        "Mann-Whitney U (%s vs %s): p=%.4f, significant=%s",
        best, worst, p_val, sig,
    )

    export_csv(metrics, OUTPUT_CSV)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
focus_webhook.py — Receive Toggl Track webhook events,
detect long unbroken sessions, and post focus prompts to Slack.

Requirements: flask>=3.0, slack-sdk>=3.22
Environment: SLACK_BOT_TOKEN, SLACK_CHANNEL_ID
"""

import os
import json
import logging
import hmac
import hashlib
from datetime import datetime, timezone

from flask import Flask, request, abort
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError

app = Flask(__name__)
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

SLACK_TOKEN = os.environ["SLACK_BOT_TOKEN"]
SLACK_CHANNEL = os.environ["SLACK_CHANNEL_ID"]
TOGGL_WEBHOOK_SECRET = os.environ.get("TOGGL_WEBHOOK_SECRET", "")
LONG_SESSION_THRESHOLD = 5400  # 90 minutes in seconds

slack = WebClient(token=SLACK_TOKEN)


def verify_signature(payload: bytes, signature: str) -> bool:
    """Verify HMAC-SHA256 signature from Toggl webhook headers."""
    if not TOGGL_WEBHOOK_SECRET:
        logger.warning("No webhook secret configured; skipping verification")
        return True
    expected = hmac.new(
        TOGGL_WEBHOOK_SECRET.encode(),
        payload,
        hashlib.sha256,
    ).hexdigest()
    return hmac.compare_digest(f"sha256={expected}", signature)


def format_duration(seconds: int) -> str:
    """Human-readable duration string."""
    hours, remainder = divmod(abs(seconds), 3600)
    minutes = remainder // 60
    return f"{int(hours)}h {int(minutes)}m"


def post_focus_reminder(entry: dict, session_length: int):
    """Send a Slack message nudging the user to take a break."""
    description = entry.get("description", "Untitled entry")
    user = entry.get("user", "Unknown")
    try:
        slack.chat_postMessage(
            channel=SLACK_CHANNEL,
            blocks=[
                {
                    "type": "header",
                    "text": {
                        "type": "plain_text",
                        "text": "🧠 Deep Work Check-In",
                    },
                },
                {
                    "type": "section",
                    "text": {
                        "type": "mrkdwn",
                        "text": (
                            f"*{user}* has been working on \"{description}\" "
                            f"for {format_duration(session_length)} without a break.\n\n"
                            f"Consider a 10-minute pause. Reply with "
                            f"`/focus-rate <1-5>` to log your perceived focus level."
                        ),
                    },
                },
            ],
        )
        logger.info("Focus reminder posted for %s", user)
    except SlackApiError as exc:
        logger.error("Slack API error: %s", exc.response["error"])


@app.route("/webhook/toggl", methods=["POST"])
def handle_toggl_webhook():
    """Main webhook endpoint."""
    signature = request.headers.get("Toggl-Webhook-Signature", "")
    payload = request.get_data()

    if not verify_signature(payload, signature):
        logger.warning("Webhook signature verification failed")
        abort(403)

    try:
        event = json.loads(payload)
    except json.JSONDecodeError:
        logger.error("Invalid JSON payload")
        abort(400)

    # Toggl sends a list of events per webhook batch
    events = event if isinstance(event, list) else [event]
    for ev in events:
        entry = ev.get("data", {})
        duration = entry.get("duration", 0)
        if duration > LONG_SESSION_THRESHOLD:
            post_focus_reminder(entry, duration)

    return "", 200


if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8080)

# Sample YAML config for the auto-tagger
workspaces:
  - ssid: "Webflow_Guest"
    tag: "ws-webflow"
  - ssid: "Industrious_FiDi"
    tag: "ws-industrious"
  - ssid: "StOberholz_5G"
    tag: "ws-st-obert"
  - ssid: "Crew_Austin_Main"
    tag: "ws-crew"

polling_interval_seconds: 30
api_base: "https://api.track.toggl.com/api/v9"

import requests
from weasyprint import HTML

API_TOKEN = "your_api_token"
WORKSPACE_ID = 123456

params = {
    "workspace_id": WORKSPACE_ID,
    "since": "2024-09-01",
    "until": "2024-09-07",
    "grouping": "tags",
    "subgroups": "date",
    "togglables": "yes",
}
headers = {"Content-Type": "application/json"}
resp = requests.get(
    "https://api.track.toggl.com/reports/api/v2/summary",
    auth=(API_TOKEN, "api_token"),
    params=params,
    headers=headers,
    timeout=30,
)
resp.raise_for_status()
data = resp.json()
# Render HTML template with data, convert to PDF
html_string = render_template("weekly_report.html", report=data)
HTML(string=html_string).write_pdf("weekly_focus_report.pdf")

from git import Repo
import requests
from datetime import datetime

def get_focused_entries(api_token, project_id, since, until):
    """Return entries that are >= 45 min with no project switches nearby."""
    resp = requests.get(
        f"https://api.track.toggl.com/api/v9/me/time_entries",
        auth=(api_token, "api_token"),
        params={
            "start_date": since.isoformat(),
            "end_date": until.isoformat(),
            "project_ids": project_id,
        },
        timeout=30,
    )
    resp.raise_for_status()
    entries = resp.json()
    focused = []
    for entry in entries:
        dur = entry.get("duration", 0)
        if dur >= 2700:  # 45 minutes
            focused.append(entry)
    return focused

def get_commit_count(repo_path, since, until, author_email):
    """Count commits by author in date range."""
    repo = Repo(repo_path)
    commits = list(repo.iter_commits(
        all=True,
        since=since.strftime("%Y-%m-%d"),
        until=until.strftime("%Y-%m-%d"),
        author=author_email,
    ))
    return len(commits)

def compute_ratio(api_token, project_id, repo_path, author_email):
    since = datetime(2024, 9, 1)
    until = datetime(2024, 9, 30)
    focused = get_focused_entries(api_token, project_id, since, until)
    total_focused_hours = sum(e["duration"] for e in focused) / 3600
    commits = get_commit_count(repo_path, since, until, author_email)
    if total_focused_hours == 0:
        return 0, commits, 0
    ratio = commits / total_focused_hours
    return round(ratio, 2), commits, round(total_focused_hours, 1)