Add daily coding summary tool for Obsidian
Integrates with Zed editor history and Git commits to track daily coding activity. Writes idempotent `## Coding Work` sections into Obsidian daily notes with project breakdowns and commit stats.
This commit is contained in:
@@ -0,0 +1,43 @@
|
||||
# Daily Coding Summary
|
||||
|
||||
Track your coding activity from **Zed** editors and **Git** commits, then write a summary into your Obsidian daily note.
|
||||
|
||||
## Features
|
||||
|
||||
- **Zed Integration** – reads Zed's SQLite database to find which files you edited today and groups them by project.
|
||||
- **Git Tracking** – scans `~/Code` (or custom directories) for git repositories and collects today's commits with stats (insertions, deletions, files changed).
|
||||
- **Obsidian Daily Notes** – writes or appends a `## Coding Work` section to your daily note (`YYYY-MM-DD.md`).
|
||||
- **Idempotent** – running multiple times on the same day replaces the existing `## Coding Work` section rather than duplicating it.
|
||||
|
||||
## Usage
|
||||
|
||||
```bash
|
||||
# Use default vault (from OBSIDIAN_VAULT env var or ~/Obsidian)
|
||||
python main.py
|
||||
|
||||
# Specify vault and code directories
|
||||
python main.py --vault ~/Documents/obsidian_vault --code-dir ~/Code --code-dir ~/Work
|
||||
|
||||
# Dry-run to preview without writing
|
||||
python main.py --dry-run
|
||||
|
||||
# Summarize a specific day
|
||||
python main.py --date 2026-05-20
|
||||
|
||||
# Skip Zed or Git tracking
|
||||
python main.py --no-zed
|
||||
python main.py --no-git
|
||||
```
|
||||
|
||||
## Configuration
|
||||
|
||||
| Environment Variable | Description | Default |
|
||||
|---------------------|-------------|---------|
|
||||
| `OBSIDIAN_VAULT` | Path to your Obsidian vault | auto-detected from `~/Obsidian` |
|
||||
| `OBSIDIAN_DAILY_FOLDER` | Folder inside the vault for daily notes | `Daily Notes` |
|
||||
|
||||
## Requirements
|
||||
|
||||
- Python 3.9+
|
||||
- Git (for commit tracking)
|
||||
- Zed editor (optional, for editor tracking)
|
||||
Binary file not shown.
Executable
+7
@@ -0,0 +1,7 @@
|
||||
#!/usr/bin/env bash
|
||||
set -euo pipefail
|
||||
|
||||
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
|
||||
|
||||
cd "$SCRIPT_DIR"
|
||||
python3 main.py "$@"
|
||||
@@ -0,0 +1,141 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Daily Coding Summary for Obsidian
|
||||
|
||||
Track coding activity from Zed editors and Git commits, then write
|
||||
a summary into your Obsidian daily note.
|
||||
|
||||
Usage:
|
||||
python main.py
|
||||
python main.py --vault ~/Obsidian/MyVault
|
||||
python main.py --code-dir ~/Code --code-dir ~/Work
|
||||
python main.py --date 2026-05-20
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import sys
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
from utils.config import date_range, get_vault_path
|
||||
from utils.git_tracker import fetch_git_activity
|
||||
from utils.obsidian_writer import write_daily_summary
|
||||
from utils.zed_tracker import fetch_zed_activity
|
||||
|
||||
|
||||
def parse_date(date_str: str) -> datetime:
|
||||
"""Parse YYYY-MM-DD string into a datetime object."""
|
||||
try:
|
||||
dt = datetime.strptime(date_str, "%Y-%m-%d")
|
||||
return dt.replace(tzinfo=timezone.utc).astimezone()
|
||||
except ValueError:
|
||||
raise argparse.ArgumentTypeError(
|
||||
f"Invalid date format: '{date_str}'. Expected YYYY-MM-DD."
|
||||
)
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Track daily coding activity and write it to Obsidian."
|
||||
)
|
||||
parser.add_argument(
|
||||
"--vault",
|
||||
dest="vault",
|
||||
default=None,
|
||||
help="Path to the Obsidian vault. Overrides OBSIDIAN_VAULT env var.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--code-dir",
|
||||
dest="code_dirs",
|
||||
action="append",
|
||||
default=None,
|
||||
help="Directory to scan for git repos (default: ~/Code). Can be given multiple times.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--daily-folder",
|
||||
dest="daily_folder",
|
||||
default=None,
|
||||
help="Folder inside the vault for daily notes (default: Daily Notes). Overrides OBSIDIAN_DAILY_FOLDER env var.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--date",
|
||||
dest="date",
|
||||
type=parse_date,
|
||||
default=None,
|
||||
help="Date to summarize (YYYY-MM-DD). Defaults to today.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--no-zed",
|
||||
dest="no_zed",
|
||||
action="store_true",
|
||||
help="Skip Zed editor tracking.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--no-git",
|
||||
dest="no_git",
|
||||
action="store_true",
|
||||
help="Skip Git commit tracking.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--dry-run",
|
||||
dest="dry_run",
|
||||
action="store_true",
|
||||
help="Print the summary instead of writing to Obsidian.",
|
||||
)
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
try:
|
||||
vault_path = get_vault_path(args.vault)
|
||||
except Exception as e:
|
||||
print(f"Error: {e}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
if args.daily_folder:
|
||||
daily_folder = args.daily_folder
|
||||
else:
|
||||
import os
|
||||
|
||||
daily_folder = os.environ.get("OBSIDIAN_DAILY_FOLDER", "Daily Notes")
|
||||
|
||||
code_dirs = args.code_dirs
|
||||
if code_dirs:
|
||||
code_dirs = [Path(d).expanduser().resolve() for d in code_dirs]
|
||||
else:
|
||||
code_dirs = [Path.home() / "Code"]
|
||||
|
||||
target_date = args.date or datetime.now(timezone.utc).astimezone()
|
||||
start_ts, end_ts = date_range(target_date)
|
||||
|
||||
zed_projects = []
|
||||
git_repos = []
|
||||
|
||||
if not args.no_zed:
|
||||
print("🔍 Reading Zed editor history…")
|
||||
zed_projects = fetch_zed_activity(start_ts, end_ts, code_dirs=code_dirs)
|
||||
print(f" Found {len(zed_projects)} project(s) with editor activity.")
|
||||
|
||||
if not args.no_git:
|
||||
print("🔍 Scanning git repositories…")
|
||||
git_repos = fetch_git_activity(start_ts, end_ts, code_dirs=code_dirs)
|
||||
print(f" Found {len(git_repos)} repo(s) with commits today.")
|
||||
|
||||
if args.dry_run:
|
||||
from utils.obsidian_writer import generate_markdown
|
||||
|
||||
print("\n--- Daily Summary ---\n")
|
||||
print(generate_markdown(zed_projects, git_repos, target_date))
|
||||
print("\n---------------------")
|
||||
return
|
||||
|
||||
note_path = write_daily_summary(
|
||||
vault_path=vault_path,
|
||||
zed_projects=zed_projects,
|
||||
git_repos=git_repos,
|
||||
daily_folder=daily_folder,
|
||||
date=target_date,
|
||||
)
|
||||
print(f"✅ Daily note updated: {note_path}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
@@ -0,0 +1,61 @@
|
||||
"""Configuration and common utilities for daily summary."""
|
||||
|
||||
import os
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
|
||||
def get_vault_path(cli_vault: Optional[str] = None) -> Path:
|
||||
"""Resolve the Obsidian vault path from CLI arg, env var, or default locations."""
|
||||
if cli_vault:
|
||||
vault = Path(cli_vault).expanduser().resolve()
|
||||
if not vault.exists():
|
||||
raise FileNotFoundError(f"Vault path does not exist: {vault}")
|
||||
return vault
|
||||
|
||||
env_vault = os.environ.get("OBSIDIAN_VAULT")
|
||||
if env_vault:
|
||||
vault = Path(env_vault).expanduser().resolve()
|
||||
if vault.exists():
|
||||
return vault
|
||||
|
||||
obsidian_dir = Path.home() / "Obsidian"
|
||||
if obsidian_dir.exists():
|
||||
vaults = [d for d in obsidian_dir.iterdir() if d.is_dir()]
|
||||
if len(vaults) == 1:
|
||||
return vaults[0]
|
||||
elif len(vaults) > 1:
|
||||
raise RuntimeError(
|
||||
f"Multiple vaults found in {obsidian_dir}. "
|
||||
"Set OBSIDIAN_VAULT or use --vault."
|
||||
)
|
||||
|
||||
raise RuntimeError(
|
||||
"Could not determine Obsidian vault path. "
|
||||
"Use --vault or set OBSIDIAN_VAULT environment variable."
|
||||
)
|
||||
|
||||
|
||||
def get_daily_notes_folder(vault_path: Path) -> Path:
|
||||
"""Return the daily notes folder, defaulting to 'Daily Notes'."""
|
||||
folder = os.environ.get("OBSIDIAN_DAILY_FOLDER", "Daily Notes")
|
||||
return vault_path / folder
|
||||
|
||||
|
||||
def date_range(dt: datetime) -> tuple[int, int]:
|
||||
"""Return Unix timestamp range for the given datetime's date (start, end)."""
|
||||
start = datetime(dt.year, dt.month, dt.day, tzinfo=dt.tzinfo or timezone.utc)
|
||||
end = start + timedelta(days=1)
|
||||
return int(start.timestamp()), int(end.timestamp())
|
||||
|
||||
|
||||
def format_duration(seconds: int) -> str:
|
||||
"""Format seconds as 'Xh Ym' or 'Ym'."""
|
||||
if seconds < 60:
|
||||
return f"{seconds}s"
|
||||
hours, remainder = divmod(seconds, 3600)
|
||||
minutes = remainder // 60
|
||||
if hours:
|
||||
return f"{hours}h {minutes}m"
|
||||
return f"{minutes}m"
|
||||
@@ -0,0 +1,225 @@
|
||||
"""Scan git repositories for today's commits."""
|
||||
|
||||
import subprocess
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import List, Optional
|
||||
|
||||
|
||||
@dataclass
|
||||
class CommitInfo:
|
||||
hash: str
|
||||
message: str
|
||||
author: str
|
||||
timestamp: int
|
||||
files_changed: int
|
||||
insertions: int
|
||||
deletions: int
|
||||
|
||||
|
||||
@dataclass
|
||||
class RepoSummary:
|
||||
name: str
|
||||
path: Path
|
||||
commits: List[CommitInfo]
|
||||
total_insertions: int
|
||||
total_deletions: int
|
||||
total_files: int
|
||||
|
||||
|
||||
def _run_git(args: List[str], cwd: Path, check: bool = True) -> str:
|
||||
"""Run a git command and return stdout."""
|
||||
result = subprocess.run(
|
||||
["git"] + args,
|
||||
cwd=cwd,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=check,
|
||||
)
|
||||
if result.returncode != 0 and check:
|
||||
raise subprocess.CalledProcessError(
|
||||
result.returncode,
|
||||
["git"] + args,
|
||||
output=result.stdout,
|
||||
stderr=result.stderr,
|
||||
)
|
||||
return result.stdout
|
||||
|
||||
|
||||
def _parse_git_date(date_str: str) -> int:
|
||||
"""Parse a git date string into Unix timestamp.
|
||||
|
||||
Handles both ISO ('2026-05-26 14:30:00 +0200') and
|
||||
RFC ('Tue, 26 May 2026 14:30:00 +0200') formats.
|
||||
"""
|
||||
date_str = date_str.strip()
|
||||
# Try ISO format first
|
||||
for fmt in ("%Y-%m-%d %H:%M:%S %z", "%a, %d %b %Y %H:%M:%S %z"):
|
||||
try:
|
||||
dt = datetime.strptime(date_str, fmt)
|
||||
return int(dt.timestamp())
|
||||
except ValueError:
|
||||
continue
|
||||
return 0
|
||||
|
||||
|
||||
def _get_commits_for_date(
|
||||
repo_path: Path, start_ts: int, end_ts: int
|
||||
) -> List[CommitInfo]:
|
||||
"""Get commits in the given timestamp range."""
|
||||
# Use --since and --until with ISO format which git understands reliably
|
||||
since = datetime.fromtimestamp(start_ts, tz=timezone.utc).isoformat()
|
||||
until = datetime.fromtimestamp(end_ts, tz=timezone.utc).isoformat()
|
||||
|
||||
log_format = "%H%x00%s%x00%an%x00%ad%x00"
|
||||
try:
|
||||
output = _run_git(
|
||||
[
|
||||
"log",
|
||||
f"--since={since}",
|
||||
f"--until={until}",
|
||||
f"--format={log_format}",
|
||||
"--date=iso",
|
||||
"--no-merges",
|
||||
],
|
||||
cwd=repo_path,
|
||||
check=False,
|
||||
)
|
||||
except subprocess.CalledProcessError:
|
||||
return []
|
||||
|
||||
if not output.strip():
|
||||
return []
|
||||
|
||||
commits = []
|
||||
entries = output.strip().split("\n")
|
||||
for entry in entries:
|
||||
parts = entry.split("\x00")
|
||||
if len(parts) < 4:
|
||||
continue
|
||||
commit_hash, message, author, date_str = parts[0], parts[1], parts[2], parts[3]
|
||||
ts = _parse_git_date(date_str)
|
||||
|
||||
# Get stats for this commit
|
||||
try:
|
||||
stat_output = _run_git(
|
||||
["show", "--stat", "--format=", commit_hash],
|
||||
cwd=repo_path,
|
||||
check=False,
|
||||
)
|
||||
except subprocess.CalledProcessError:
|
||||
stat_output = ""
|
||||
|
||||
files_changed, insertions, deletions = 0, 0, 0
|
||||
for line in stat_output.splitlines():
|
||||
line = line.strip()
|
||||
if "file changed" in line or "files changed" in line:
|
||||
# Parse line like "3 files changed, 10 insertions(+), 2 deletions(-)"
|
||||
parts_stat = line.split(",")
|
||||
for part in parts_stat:
|
||||
part = part.strip()
|
||||
if "changed" in part:
|
||||
try:
|
||||
files_changed = int(part.split()[0])
|
||||
except ValueError:
|
||||
pass
|
||||
elif "insertion" in part:
|
||||
try:
|
||||
insertions = int(part.split()[0])
|
||||
except ValueError:
|
||||
pass
|
||||
elif "deletion" in part:
|
||||
try:
|
||||
deletions = int(part.split()[0])
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
commits.append(
|
||||
CommitInfo(
|
||||
hash=commit_hash[:8],
|
||||
message=message.strip(),
|
||||
author=author.strip(),
|
||||
timestamp=ts,
|
||||
files_changed=files_changed,
|
||||
insertions=insertions,
|
||||
deletions=deletions,
|
||||
)
|
||||
)
|
||||
|
||||
return commits
|
||||
|
||||
|
||||
def _is_git_repo(path: Path) -> bool:
|
||||
"""Check if a directory is a git repository."""
|
||||
return (path / ".git").is_dir()
|
||||
|
||||
|
||||
def find_git_repos(base_dir: Path, max_depth: int = 3) -> List[Path]:
|
||||
"""Find git repositories under base_dir up to max_depth levels."""
|
||||
repos = []
|
||||
base_dir = base_dir.expanduser().resolve()
|
||||
if not base_dir.exists():
|
||||
return repos
|
||||
|
||||
for depth in range(max_depth + 1):
|
||||
pattern = "/".join(["*"] * depth) if depth > 0 else "."
|
||||
if depth == 0:
|
||||
if _is_git_repo(base_dir):
|
||||
repos.append(base_dir)
|
||||
continue
|
||||
|
||||
for candidate in base_dir.glob(pattern):
|
||||
if candidate.is_dir() and _is_git_repo(candidate):
|
||||
repos.append(candidate)
|
||||
|
||||
return repos
|
||||
|
||||
|
||||
def fetch_git_activity(
|
||||
start_ts: int,
|
||||
end_ts: int,
|
||||
code_dirs: Optional[List[Path]] = None,
|
||||
max_depth: int = 3,
|
||||
) -> List[RepoSummary]:
|
||||
"""Scan code directories and return git activity summaries."""
|
||||
if code_dirs is None:
|
||||
code_dirs = [Path.home() / "Code"]
|
||||
|
||||
all_repos = []
|
||||
for d in code_dirs:
|
||||
all_repos.extend(find_git_repos(d, max_depth=max_depth))
|
||||
|
||||
# Deduplicate
|
||||
seen = set()
|
||||
unique_repos = []
|
||||
for r in all_repos:
|
||||
resolved = r.resolve()
|
||||
if resolved not in seen:
|
||||
seen.add(resolved)
|
||||
unique_repos.append(resolved)
|
||||
|
||||
summaries = []
|
||||
for repo_path in unique_repos:
|
||||
commits = _get_commits_for_date(repo_path, start_ts, end_ts)
|
||||
if not commits:
|
||||
continue
|
||||
|
||||
total_insertions = sum(c.insertions for c in commits)
|
||||
total_deletions = sum(c.deletions for c in commits)
|
||||
total_files = sum(c.files_changed for c in commits)
|
||||
|
||||
summaries.append(
|
||||
RepoSummary(
|
||||
name=repo_path.name,
|
||||
path=repo_path,
|
||||
commits=commits,
|
||||
total_insertions=total_insertions,
|
||||
total_deletions=total_deletions,
|
||||
total_files=total_files,
|
||||
)
|
||||
)
|
||||
|
||||
# Sort by number of commits descending
|
||||
summaries.sort(key=lambda r: -len(r.commits))
|
||||
return summaries
|
||||
@@ -0,0 +1,112 @@
|
||||
"""Write or append daily summaries to Obsidian notes."""
|
||||
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import List
|
||||
|
||||
from .git_tracker import RepoSummary
|
||||
from .zed_tracker import ProjectSummary
|
||||
|
||||
DAILY_NOTE_FORMAT = "%Y-%m-%d"
|
||||
SECTION_HEADER = "## Coding Work"
|
||||
|
||||
|
||||
def _format_time(ts: int) -> str:
|
||||
"""Format a Unix timestamp as HH:MM."""
|
||||
dt = datetime.fromtimestamp(ts, tz=timezone.utc).astimezone()
|
||||
return dt.strftime("%H:%M")
|
||||
|
||||
|
||||
def generate_markdown(
|
||||
zed_projects: List[ProjectSummary],
|
||||
git_repos: List[RepoSummary],
|
||||
date: datetime,
|
||||
) -> str:
|
||||
"""Generate the markdown content for the daily summary."""
|
||||
lines = []
|
||||
lines.append(SECTION_HEADER)
|
||||
lines.append("")
|
||||
|
||||
if not zed_projects and not git_repos:
|
||||
lines.append("No coding activity tracked today. 🌴")
|
||||
return "\n".join(lines)
|
||||
|
||||
if zed_projects:
|
||||
lines.append("### Zed Editors")
|
||||
lines.append("")
|
||||
for project in zed_projects:
|
||||
langs = (
|
||||
f" ({', '.join(sorted(project.languages))})"
|
||||
if project.languages
|
||||
else ""
|
||||
)
|
||||
lines.append(
|
||||
f"- **{project.project}**{langs} — {len(project.files)} file(s)"
|
||||
)
|
||||
# Show up to 5 most recent files
|
||||
for path, mtime in project.files[:5]:
|
||||
time_str = _format_time(mtime)
|
||||
lines.append(f" - `{Path(path).name}` at {time_str}")
|
||||
if len(project.files) > 5:
|
||||
lines.append(f" - *…and {len(project.files) - 5} more*")
|
||||
lines.append("")
|
||||
|
||||
if git_repos:
|
||||
lines.append("### Git Commits")
|
||||
lines.append("")
|
||||
for repo in git_repos:
|
||||
lines.append(
|
||||
f"- **{repo.name}** — {len(repo.commits)} commit(s), "
|
||||
f"+{repo.total_insertions}/-{repo.total_deletions} lines, "
|
||||
f"{repo.total_files} file(s)"
|
||||
)
|
||||
for commit in repo.commits:
|
||||
time_str = _format_time(commit.timestamp)
|
||||
lines.append(f" - `{commit.hash}` {commit.message} — {time_str}")
|
||||
lines.append("")
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
# Backward compatibility alias
|
||||
_generate_markdown = generate_markdown
|
||||
|
||||
|
||||
def write_daily_summary(
|
||||
vault_path: Path,
|
||||
zed_projects: List[ProjectSummary],
|
||||
git_repos: List[RepoSummary],
|
||||
daily_folder: str = "Daily Notes",
|
||||
date: datetime = None,
|
||||
) -> Path:
|
||||
"""Write or append a coding summary to the Obsidian daily note."""
|
||||
if date is None:
|
||||
date = datetime.now(timezone.utc).astimezone()
|
||||
|
||||
note_dir = vault_path / daily_folder
|
||||
note_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
note_name = date.strftime(DAILY_NOTE_FORMAT)
|
||||
note_path = note_dir / f"{note_name}.md"
|
||||
|
||||
content = _generate_markdown(zed_projects, git_repos, date)
|
||||
|
||||
if note_path.exists():
|
||||
existing = note_path.read_text(encoding="utf-8")
|
||||
# If section already exists, replace it; otherwise append
|
||||
if SECTION_HEADER in existing:
|
||||
before, _, after = existing.partition(SECTION_HEADER)
|
||||
# Find the next section header or end of file
|
||||
next_section_idx = after.find("\n## ")
|
||||
if next_section_idx != -1:
|
||||
after = after[next_section_idx:]
|
||||
else:
|
||||
after = ""
|
||||
new_content = before.rstrip() + "\n\n" + content + "\n" + after.lstrip()
|
||||
else:
|
||||
new_content = existing.rstrip() + "\n\n" + content + "\n"
|
||||
else:
|
||||
new_content = f"# {note_name}\n\n" + content + "\n"
|
||||
|
||||
note_path.write_text(new_content, encoding="utf-8")
|
||||
return note_path
|
||||
@@ -0,0 +1,140 @@
|
||||
"""Read recent editor activity from Zed's SQLite database."""
|
||||
|
||||
import sqlite3
|
||||
from collections import defaultdict
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Dict, List, Optional, Tuple
|
||||
|
||||
|
||||
@dataclass
|
||||
class EditorSession:
|
||||
path: str
|
||||
mtime: int
|
||||
language: Optional[str]
|
||||
|
||||
|
||||
@dataclass
|
||||
class ProjectSummary:
|
||||
project: str
|
||||
files: List[Tuple[str, int]] # (relative_path, mtime)
|
||||
languages: set
|
||||
|
||||
|
||||
def _find_zed_db() -> Optional[Path]:
|
||||
"""Locate the most recent Zed stable database."""
|
||||
zed_db_dir = Path.home() / ".local" / "share" / "zed" / "db"
|
||||
if not zed_db_dir.exists():
|
||||
return None
|
||||
|
||||
candidates = []
|
||||
for sub in zed_db_dir.iterdir():
|
||||
db_file = sub / "db.sqlite"
|
||||
if db_file.exists():
|
||||
# Prefer stable, then preview, then anything else
|
||||
priority = 0
|
||||
if "stable" in sub.name:
|
||||
priority = 2
|
||||
elif "preview" in sub.name:
|
||||
priority = 1
|
||||
candidates.append((priority, db_file.stat().st_mtime, db_file))
|
||||
|
||||
if not candidates:
|
||||
return None
|
||||
|
||||
candidates.sort(key=lambda x: (x[0], x[1]), reverse=True)
|
||||
return candidates[0][2]
|
||||
|
||||
|
||||
def _resolve_project(file_path: Path, code_dirs: List[Path]) -> Optional[str]:
|
||||
"""Return the project name for a file path based on code directories."""
|
||||
for code_dir in code_dirs:
|
||||
try:
|
||||
rel = file_path.relative_to(code_dir)
|
||||
parts = rel.parts
|
||||
if parts:
|
||||
return parts[0]
|
||||
except ValueError:
|
||||
continue
|
||||
# Fallback: use parent directory name
|
||||
return file_path.parent.name if file_path.parent != Path("/") else None
|
||||
|
||||
|
||||
def fetch_zed_activity(
|
||||
start_ts: int,
|
||||
end_ts: int,
|
||||
code_dirs: Optional[List[Path]] = None,
|
||||
) -> List[ProjectSummary]:
|
||||
"""Fetch editor activity from Zed for the given time range."""
|
||||
db_path = _find_zed_db()
|
||||
if not db_path:
|
||||
return []
|
||||
|
||||
if code_dirs is None:
|
||||
code_dirs = [Path.home() / "Code"]
|
||||
code_dirs = [d.expanduser().resolve() for d in code_dirs if d.exists()]
|
||||
|
||||
sessions: List[EditorSession] = []
|
||||
try:
|
||||
conn = sqlite3.connect(str(db_path))
|
||||
cursor = conn.cursor()
|
||||
cursor.execute(
|
||||
"""
|
||||
SELECT path, mtime_seconds, language
|
||||
FROM editors
|
||||
WHERE mtime_seconds >= ? AND mtime_seconds < ?
|
||||
ORDER BY mtime_seconds DESC
|
||||
""",
|
||||
(start_ts, end_ts),
|
||||
)
|
||||
for row in cursor.fetchall():
|
||||
path_blob, mtime, language = row
|
||||
if not path_blob:
|
||||
continue
|
||||
try:
|
||||
path_str = (
|
||||
path_blob.decode("utf-8")
|
||||
if isinstance(path_blob, bytes)
|
||||
else str(path_blob)
|
||||
)
|
||||
except (UnicodeDecodeError, AttributeError):
|
||||
continue
|
||||
sessions.append(
|
||||
EditorSession(path=path_str, mtime=mtime, language=language)
|
||||
)
|
||||
except sqlite3.Error as e:
|
||||
print(f"Warning: could not read Zed database: {e}")
|
||||
return []
|
||||
|
||||
# Group by project
|
||||
by_project: Dict[str, List[EditorSession]] = defaultdict(list)
|
||||
for s in sessions:
|
||||
fp = Path(s.path)
|
||||
project = _resolve_project(fp, code_dirs)
|
||||
if not project:
|
||||
project = fp.parent.name
|
||||
by_project[project].append(s)
|
||||
|
||||
summaries = []
|
||||
for project, items in by_project.items():
|
||||
files = []
|
||||
seen = set()
|
||||
languages = set()
|
||||
for item in items:
|
||||
if item.language:
|
||||
languages.add(item.language)
|
||||
rel = item.path
|
||||
# Store unique files with their latest mtime
|
||||
key = rel
|
||||
if key not in seen:
|
||||
seen.add(key)
|
||||
files.append((rel, item.mtime))
|
||||
# Sort by mtime descending
|
||||
files.sort(key=lambda x: x[1], reverse=True)
|
||||
summaries.append(
|
||||
ProjectSummary(project=project, files=files, languages=languages)
|
||||
)
|
||||
|
||||
# Sort projects by number of files, then by latest mtime
|
||||
summaries.sort(key=lambda s: (-len(s.files), -s.files[0][1] if s.files else 0))
|
||||
return summaries
|
||||
Reference in New Issue
Block a user