Coverage for little_loops / cli / messages.py: 7%
82 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-03-18 16:18 -0500
1"""ll-messages: Extract user messages from Claude Code session logs."""
3from __future__ import annotations
5import argparse
6from pathlib import Path
8from little_loops.logger import Logger
11def main_messages() -> int:
12 """Entry point for ll-messages command.
14 Extract user messages from Claude Code session logs.
16 Returns:
17 Exit code (0 = success)
18 """
19 import json
20 from datetime import datetime
22 from little_loops.user_messages import (
23 CommandRecord,
24 UserMessage,
25 extract_commands,
26 extract_user_messages,
27 get_project_folder,
28 )
30 parser = argparse.ArgumentParser(
31 description="Extract user messages from Claude Code logs",
32 formatter_class=argparse.RawDescriptionHelpFormatter,
33 epilog="""
34Examples:
35 %(prog)s # Last 100 messages to file
36 %(prog)s -n 50 # Last 50 messages
37 %(prog)s --since 2026-01-01 # Messages since date
38 %(prog)s -o output.jsonl # Custom output path
39 %(prog)s --stdout # Print to terminal
40 %(prog)s --include-response-context # Include response metadata
41 %(prog)s --skip-cli # Exclude CLI commands from output
42 %(prog)s --commands-only # Extract only CLI commands
43""",
44 )
45 parser.add_argument(
46 "-n",
47 "--limit",
48 type=int,
49 default=100,
50 help="Maximum number of messages to extract (default: 100)",
51 )
52 parser.add_argument(
53 "--since",
54 type=str,
55 help="Only include messages after this date (YYYY-MM-DD or ISO format)",
56 )
57 parser.add_argument(
58 "-o",
59 "--output",
60 type=Path,
61 help="Output file path (default: .claude/user-messages-{timestamp}.jsonl)",
62 )
63 parser.add_argument(
64 "--cwd",
65 type=Path,
66 help="Working directory to use (default: current directory)",
67 )
68 parser.add_argument(
69 "--exclude-agents",
70 action="store_true",
71 help="Exclude agent session files (agent-*.jsonl)",
72 )
73 parser.add_argument(
74 "--stdout",
75 action="store_true",
76 help="Print messages to stdout instead of writing to file",
77 )
78 parser.add_argument(
79 "-v",
80 "--verbose",
81 action="store_true",
82 help="Print verbose progress information",
83 )
84 parser.add_argument(
85 "--include-response-context",
86 action="store_true",
87 help="Include metadata from assistant responses (tools used, files modified)",
88 )
89 parser.add_argument(
90 "--skip-cli",
91 action="store_true",
92 help="Exclude CLI commands from output (included by default)",
93 )
94 parser.add_argument(
95 "--commands-only",
96 action="store_true",
97 help="Extract only CLI commands, no user messages",
98 )
99 parser.add_argument(
100 "--tools",
101 type=str,
102 default="Bash",
103 help="Comma-separated list of tools to extract commands from (default: Bash)",
104 )
106 args = parser.parse_args()
108 logger = Logger(verbose=args.verbose)
110 # Parse since date if provided
111 since = None
112 if args.since:
113 try:
114 # Try ISO format first
115 since = datetime.fromisoformat(args.since.replace("Z", "+00:00"))
116 except ValueError:
117 try:
118 # Try YYYY-MM-DD format
119 since = datetime.strptime(args.since, "%Y-%m-%d")
120 except ValueError:
121 logger.error(f"Invalid date format: {args.since}")
122 logger.error("Use YYYY-MM-DD or ISO format")
123 return 1
125 # Get project folder
126 cwd = args.cwd or Path.cwd()
127 project_folder = get_project_folder(cwd)
129 if project_folder is None:
130 logger.error(f"No Claude project folder found for: {cwd}")
131 logger.error(f"Expected: ~/.claude/projects/{str(cwd).replace('/', '-')}")
132 return 1
134 logger.info(f"Project folder: {project_folder}")
135 logger.info(f"Limit: {args.limit}")
136 if since:
137 logger.info(f"Since: {since}")
139 # Parse tools list
140 tools_list = [t.strip() for t in args.tools.split(",")]
142 # Extract data based on flags
143 messages: list[UserMessage] = []
144 commands: list[CommandRecord] = []
146 if not args.commands_only:
147 messages = extract_user_messages(
148 project_folder=project_folder,
149 limit=None, # Apply limit after merging
150 since=since,
151 include_agent_sessions=not args.exclude_agents,
152 include_response_context=args.include_response_context,
153 )
155 if not args.skip_cli or args.commands_only:
156 commands = extract_commands(
157 project_folder=project_folder,
158 limit=None, # Apply limit after merging
159 since=since,
160 include_agent_sessions=not args.exclude_agents,
161 tools=tools_list,
162 )
164 if not messages and not commands:
165 logger.warning("No user messages or commands found")
166 return 0
168 # Merge and sort by timestamp
169 combined: list[UserMessage | CommandRecord] = []
170 combined.extend(messages)
171 combined.extend(commands)
172 combined.sort(key=lambda x: x.timestamp, reverse=True)
174 # Apply limit
175 if args.limit is not None:
176 combined = combined[: args.limit]
178 msg_count = len([x for x in combined if isinstance(x, UserMessage)])
179 cmd_count = len([x for x in combined if isinstance(x, CommandRecord)])
180 logger.info(f"Found {msg_count} messages, {cmd_count} commands")
182 # Output
183 if args.stdout:
184 for item in combined:
185 print(json.dumps(item.to_dict()))
186 else:
187 output_path = _save_combined(combined, args.output)
188 logger.success(f"Saved {len(combined)} records to: {output_path}")
190 return 0
193def _save_combined(
194 items: list,
195 output_path: Path | None = None,
196) -> Path:
197 """Save combined messages and commands to JSONL file."""
198 import json
199 from datetime import datetime
201 if output_path is None:
202 timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
203 output_dir = Path.cwd() / ".claude"
204 output_dir.mkdir(parents=True, exist_ok=True)
205 output_path = output_dir / f"user-messages-{timestamp}.jsonl"
207 output_path = Path(output_path)
208 output_path.parent.mkdir(parents=True, exist_ok=True)
210 with open(output_path, "w", encoding="utf-8") as f:
211 for item in items:
212 f.write(json.dumps(item.to_dict()) + "\n")
214 return output_path