scripts/focus_session_logger.py
10.4 KB| 1 | #!/usr/bin/env python3 |
| 2 | """focus_session_logger.py — JSON-backed focus-session log: deep hours this week, streaks, targets. |
| 3 | |
| 4 | You cannot improve a deep-work practice you do not measure. This script keeps a tiny local |
| 5 | ledger of focus sessions and answers two questions: |
| 6 | |
| 7 | status — how many deep hours this week (ISO week, Mon-Sun) vs --target (default 15)? |
| 8 | streak — how many consecutive days (ending today, with a one-day grace) have at least |
| 9 | one logged session? |
| 10 | |
| 11 | Subcommands: |
| 12 | log --minutes N --label X [--date YYYY-MM-DD] append one focus session |
| 13 | status [--target HOURS] [--date YYYY-MM-DD] this week's deep hours vs target |
| 14 | streak [--date YYYY-MM-DD] consecutive-day streak |
| 15 | |
| 16 | State lives in a single JSON file (--state, default ~/.deep-work/sessions.json), created on |
| 17 | demand. Writes are atomic (tmp file + os.replace) so a crash never corrupts the ledger. |
| 18 | The top-level --sample flag prints a canned status report WITHOUT touching disk. |
| 19 | |
| 20 | NO LLM CALLS. Stdlib only. Deterministic given the same state file and --date. |
| 21 | |
| 22 | Exit codes: |
| 23 | 0 success (log appended / report printed) |
| 24 | 1 usage error, bad date, or unreadable/corrupt state file |
| 25 | |
| 26 | Usage: |
| 27 | python focus_session_logger.py log --minutes 90 --label "Write product spec" |
| 28 | python focus_session_logger.py status --target 15 |
| 29 | python focus_session_logger.py streak |
| 30 | python focus_session_logger.py --sample |
| 31 | """ |
| 32 | |
| 33 | import argparse |
| 34 | import datetime as dt |
| 35 | import json |
| 36 | import os |
| 37 | import sys |
| 38 | import tempfile |
| 39 | from typing import Any, Dict, List, Tuple |
| 40 | |
| 41 | DEFAULT_STATE = os.path.join(os.path.expanduser("~"), ".deep-work", "sessions.json") |
| 42 | DEFAULT_TARGET_HOURS = 15.0 |
| 43 | |
| 44 | |
| 45 | def parse_date(value: str, flag: str) -> dt.date: |
| 46 | try: |
| 47 | return dt.date.fromisoformat(value) |
| 48 | except ValueError: |
| 49 | raise ValueError(f"{flag} must be YYYY-MM-DD, got: {value!r}") |
| 50 | |
| 51 | |
| 52 | def load_state(path: str) -> Dict[str, Any]: |
| 53 | if not os.path.exists(path): |
| 54 | return {"sessions": []} |
| 55 | try: |
| 56 | with open(path, "r", encoding="utf-8") as f: |
| 57 | state = json.load(f) |
| 58 | except (OSError, json.JSONDecodeError) as exc: |
| 59 | raise ValueError(f"cannot read state file {path}: {exc}") |
| 60 | if not isinstance(state, dict) or not isinstance(state.get("sessions"), list): |
| 61 | raise ValueError(f"state file {path} is not a valid session ledger") |
| 62 | return state |
| 63 | |
| 64 | |
| 65 | def save_state(path: str, state: Dict[str, Any]) -> None: |
| 66 | """Atomic write: temp file in the same directory, then os.replace.""" |
| 67 | directory = os.path.dirname(os.path.abspath(path)) |
| 68 | os.makedirs(directory, exist_ok=True) |
| 69 | fd, tmp = tempfile.mkstemp(prefix=".sessions-", suffix=".tmp", dir=directory) |
| 70 | try: |
| 71 | with os.fdopen(fd, "w", encoding="utf-8") as f: |
| 72 | json.dump(state, f, indent=2) |
| 73 | f.write("\n") |
| 74 | os.replace(tmp, path) |
| 75 | except BaseException: |
| 76 | if os.path.exists(tmp): |
| 77 | os.unlink(tmp) |
| 78 | raise |
| 79 | |
| 80 | |
| 81 | def week_bounds(day: dt.date) -> Tuple[dt.date, dt.date]: |
| 82 | monday = day - dt.timedelta(days=day.weekday()) |
| 83 | return monday, monday + dt.timedelta(days=6) |
| 84 | |
| 85 | |
| 86 | def cmd_log(state: Dict[str, Any], minutes: int, label: str, day: dt.date) -> Dict[str, Any]: |
| 87 | state["sessions"].append({"date": day.isoformat(), "minutes": minutes, "label": label}) |
| 88 | total_today = sum(s["minutes"] for s in state["sessions"] if s["date"] == day.isoformat()) |
| 89 | return {"logged": {"date": day.isoformat(), "minutes": minutes, "label": label}, |
| 90 | "sessions_total": len(state["sessions"]), |
| 91 | "minutes_today": total_today} |
| 92 | |
| 93 | |
| 94 | def cmd_status(state: Dict[str, Any], target_hours: float, day: dt.date) -> Dict[str, Any]: |
| 95 | monday, sunday = week_bounds(day) |
| 96 | week = [s for s in state["sessions"] |
| 97 | if monday.isoformat() <= s["date"] <= sunday.isoformat()] |
| 98 | minutes = sum(s["minutes"] for s in week) |
| 99 | hours = round(minutes / 60.0, 2) |
| 100 | by_day: Dict[str, int] = {} |
| 101 | for s in week: |
| 102 | by_day[s["date"]] = by_day.get(s["date"], 0) + s["minutes"] |
| 103 | remaining = round(max(0.0, target_hours - hours), 2) |
| 104 | on_track = hours >= target_hours |
| 105 | return { |
| 106 | "week": f"{monday.isoformat()} .. {sunday.isoformat()}", |
| 107 | "as_of": day.isoformat(), |
| 108 | "sessions_this_week": len(week), |
| 109 | "deep_hours_this_week": hours, |
| 110 | "target_hours": target_hours, |
| 111 | "remaining_hours": remaining, |
| 112 | "on_track": on_track, |
| 113 | "by_day_minutes": dict(sorted(by_day.items())), |
| 114 | "headline": (f"{hours}h of deep work this week vs a {target_hours:g}h target — " |
| 115 | + ("target met. Protect the streak." |
| 116 | if on_track else f"{remaining}h still to block.")), |
| 117 | } |
| 118 | |
| 119 | |
| 120 | def cmd_streak(state: Dict[str, Any], day: dt.date) -> Dict[str, Any]: |
| 121 | days = {s["date"] for s in state["sessions"]} |
| 122 | cursor = day |
| 123 | if cursor.isoformat() not in days: # one-day grace: today not logged yet |
| 124 | cursor = cursor - dt.timedelta(days=1) |
| 125 | streak = 0 |
| 126 | while cursor.isoformat() in days: |
| 127 | streak += 1 |
| 128 | cursor = cursor - dt.timedelta(days=1) |
| 129 | return { |
| 130 | "as_of": day.isoformat(), |
| 131 | "streak_days": streak, |
| 132 | "headline": (f"{streak} consecutive day(s) with at least one focus session." |
| 133 | if streak else "No active streak — one 90-minute block today starts it."), |
| 134 | } |
| 135 | |
| 136 | |
| 137 | def render_human(kind: str, r: Dict[str, Any]) -> str: |
| 138 | out: List[str] = [] |
| 139 | if kind == "log": |
| 140 | s = r["logged"] |
| 141 | out.append(f"Logged: {s['minutes']} min — \"{s['label']}\" on {s['date']}") |
| 142 | out.append(f" Today so far: {r['minutes_today']} min · Ledger total: {r['sessions_total']} session(s)") |
| 143 | elif kind == "status": |
| 144 | out.append(f"Deep-Work Status — week {r['week']} (as of {r['as_of']})") |
| 145 | out.append("=" * 64) |
| 146 | tail = "ON TRACK" if r["on_track"] else f"{r['remaining_hours']}h remaining" |
| 147 | out.append(f" Deep hours: {r['deep_hours_this_week']}h / {r['target_hours']:g}h target ({tail})") |
| 148 | for d, m in r["by_day_minutes"].items(): |
| 149 | out.append(f" {d}: {m} min") |
| 150 | out.append(f" {r['headline']}") |
| 151 | else: # streak |
| 152 | out.append(f"Deep-Work Streak (as of {r['as_of']})") |
| 153 | out.append("=" * 64) |
| 154 | out.append(f" {r['headline']}") |
| 155 | return "\n".join(out) |
| 156 | |
| 157 | |
| 158 | SAMPLE_STATUS = { |
| 159 | "kind": "status", |
| 160 | "sample": True, |
| 161 | "week": "2026-07-13 .. 2026-07-19", |
| 162 | "as_of": "2026-07-16", |
| 163 | "deep_hours_this_week": 8.5, |
| 164 | "target_hours": 15.0, |
| 165 | "on_track": False, |
| 166 | "remaining_hours": 6.5, |
| 167 | "by_day_minutes": {"2026-07-13": 120, "2026-07-14": 180, |
| 168 | "2026-07-15": 90, "2026-07-16": 120}, |
| 169 | "streak_days": 4, |
| 170 | "headline": "8.5h of deep work this week vs a 15h target — 6.5h still to block.", |
| 171 | } |
| 172 | |
| 173 | SAMPLE_REPORT = """Deep-Work Status — week 2026-07-13 .. 2026-07-19 (sample, no disk touched) |
| 174 | ================================================================ |
| 175 | Deep hours: 8.5h / 15h target (6.5h remaining) |
| 176 | 2026-07-13: 120 min |
| 177 | 2026-07-14: 180 min |
| 178 | 2026-07-15: 90 min |
| 179 | 2026-07-16: 120 min |
| 180 | 8.5h of deep work this week vs a 15h target — 6.5h still to block. |
| 181 | Streak: 4 consecutive day(s) with at least one focus session.""" |
| 182 | |
| 183 | |
| 184 | def main(argv: List[str]) -> int: |
| 185 | p = argparse.ArgumentParser( |
| 186 | description="JSON-backed focus-session log: deep hours this week, streaks, targets.", |
| 187 | formatter_class=argparse.RawDescriptionHelpFormatter, |
| 188 | epilog=("Exit codes:\n" |
| 189 | " 0 success (log appended / report printed)\n" |
| 190 | " 1 usage error, bad date, or unreadable/corrupt state file"), |
| 191 | ) |
| 192 | p.add_argument("--state", default=DEFAULT_STATE, |
| 193 | help=f"Path to the JSON ledger (default {DEFAULT_STATE})") |
| 194 | p.add_argument("--sample", action="store_true", |
| 195 | help="Print a canned status report without touching disk") |
| 196 | p.add_argument("--json", action="store_true", |
| 197 | help="Emit JSON instead of the report (also honored with --sample)") |
| 198 | common = argparse.ArgumentParser(add_help=False) |
| 199 | common.add_argument("--json", action="store_true", help="Emit JSON instead of the report") |
| 200 | sub = p.add_subparsers(dest="cmd") |
| 201 | |
| 202 | p_log = sub.add_parser("log", help="Append one focus session", parents=[common]) |
| 203 | p_log.add_argument("--minutes", type=int, required=True, help="Session length in minutes") |
| 204 | p_log.add_argument("--label", default="deep work", help="What the session was spent on") |
| 205 | p_log.add_argument("--date", help="Session date YYYY-MM-DD (default today)") |
| 206 | |
| 207 | p_status = sub.add_parser("status", help="This week's deep hours vs target", parents=[common]) |
| 208 | p_status.add_argument("--target", type=float, default=DEFAULT_TARGET_HOURS, |
| 209 | help=f"Weekly deep-hours target (default {DEFAULT_TARGET_HOURS:g})") |
| 210 | p_status.add_argument("--date", help="As-of date YYYY-MM-DD (default today)") |
| 211 | |
| 212 | p_streak = sub.add_parser("streak", help="Consecutive days with at least one session", parents=[common]) |
| 213 | p_streak.add_argument("--date", help="As-of date YYYY-MM-DD (default today)") |
| 214 | |
| 215 | args = p.parse_args(argv) |
| 216 | |
| 217 | if args.sample: |
| 218 | if args.json: |
| 219 | print(json.dumps(SAMPLE_STATUS, indent=2)) |
| 220 | else: |
| 221 | print(SAMPLE_REPORT) |
| 222 | return 0 |
| 223 | if not args.cmd: |
| 224 | p.print_help() |
| 225 | print("\nerror: a subcommand is required (log / status / streak) or --sample", |
| 226 | file=sys.stderr) |
| 227 | return 1 |
| 228 | |
| 229 | try: |
| 230 | day = parse_date(args.date, "--date") if getattr(args, "date", None) else dt.date.today() |
| 231 | state = load_state(args.state) |
| 232 | if args.cmd == "log": |
| 233 | if args.minutes <= 0: |
| 234 | raise ValueError("--minutes must be a positive integer") |
| 235 | result = cmd_log(state, args.minutes, args.label, day) |
| 236 | save_state(args.state, state) |
| 237 | elif args.cmd == "status": |
| 238 | result = cmd_status(state, args.target, day) |
| 239 | else: |
| 240 | result = cmd_streak(state, day) |
| 241 | except ValueError as exc: |
| 242 | print(f"error: {exc}", file=sys.stderr) |
| 243 | return 1 |
| 244 | |
| 245 | if args.json: |
| 246 | print(json.dumps(result, indent=2)) |
| 247 | else: |
| 248 | print(render_human(args.cmd, result)) |
| 249 | return 0 |
| 250 | |
| 251 | |
| 252 | if __name__ == "__main__": |
| 253 | sys.exit(main(sys.argv[1:])) |
文件内容是平台检查时读出的文本，只供查看。平台不会运行包里的任何脚本或程序。
