9
9
10
10
def get_hyphenated_cwd () -> str :
11
11
"""Convert current working directory to hyphenated format for Claude log path.
12
-
12
+
13
13
Returns:
14
14
Hyphenated directory name (e.g., "/Users/john/project" -> "users-john-project")
15
15
"""
16
16
cwd = os .getcwd ()
17
17
# Remove leading slash and replace slashes and spaces with hyphens
18
- hyphenated = cwd .replace ('/' , '-' ).replace (' ' , '-' )
18
+ hyphenated = cwd .replace ("/" , "-" ).replace (" " , "-" ). replace ( "_" , "-" )
19
19
# Remove any double hyphens
20
- hyphenated = re .sub (r'-+' , '-' , hyphenated )
20
+ hyphenated = re .sub (r"-+" , "-" , hyphenated )
21
21
return hyphenated
22
22
23
23
24
24
def get_claude_session_log_path (session_id : str ) -> Path :
25
25
"""Get the path to the Claude session log file.
26
-
26
+
27
27
Args:
28
28
session_id: The Claude session ID
29
-
29
+
30
30
Returns:
31
31
Path to the session log file
32
32
"""
33
33
claude_dir = Path .home () / ".claude"
34
34
projects_dir = claude_dir / "projects"
35
35
hyphenated_cwd = get_hyphenated_cwd ()
36
36
project_dir = projects_dir / hyphenated_cwd
37
-
37
+
38
38
log_file = project_dir / f"{ session_id } .jsonl"
39
39
return log_file
40
40
41
41
42
42
def parse_jsonl_line (line : str ) -> Optional [Dict [str , Any ]]:
43
43
"""Parse a single line from a JSONL file.
44
-
44
+
45
45
Args:
46
46
line: Raw line from JSONL file
47
-
47
+
48
48
Returns:
49
49
Parsed JSON object or None if parsing fails
50
50
"""
51
51
line = line .strip ()
52
52
if not line :
53
53
return None
54
-
54
+
55
55
try :
56
56
return json .loads (line )
57
57
except json .JSONDecodeError :
@@ -60,10 +60,10 @@ def parse_jsonl_line(line: str) -> Optional[Dict[str, Any]]:
60
60
61
61
def ensure_log_directory (session_id : str ) -> Path :
62
62
"""Ensure the log directory exists and return the log file path.
63
-
63
+
64
64
Args:
65
65
session_id: The Claude session ID
66
-
66
+
67
67
Returns:
68
68
Path to the session log file
69
69
"""
@@ -74,53 +74,53 @@ def ensure_log_directory(session_id: str) -> Path:
74
74
75
75
def read_existing_log_lines (log_path : Path ) -> int :
76
76
"""Count existing lines in a log file.
77
-
77
+
78
78
Args:
79
79
log_path: Path to the log file
80
-
80
+
81
81
Returns:
82
82
Number of existing lines
83
83
"""
84
84
if not log_path .exists ():
85
85
return 0
86
-
86
+
87
87
try :
88
- with open (log_path , 'r' , encoding = ' utf-8' ) as f :
88
+ with open (log_path , "r" , encoding = " utf-8" ) as f :
89
89
return sum (1 for _ in f )
90
90
except (OSError , UnicodeDecodeError ):
91
91
return 0
92
92
93
93
94
94
def validate_log_entry (log_entry : Dict [str , Any ]) -> bool :
95
95
"""Validate a log entry before sending to API.
96
-
96
+
97
97
Args:
98
98
log_entry: The log entry to validate
99
-
99
+
100
100
Returns:
101
101
True if valid, False otherwise
102
102
"""
103
103
if not isinstance (log_entry , dict ):
104
104
return False
105
-
105
+
106
106
# Basic validation - ensure it has some content
107
107
if not log_entry :
108
108
return False
109
-
109
+
110
110
# Optionally validate specific fields that Claude Code uses
111
111
# This can be expanded based on actual Claude log format
112
112
return True
113
113
114
114
115
115
def format_log_for_api (log_entry : Dict [str , Any ]) -> Dict [str , Any ]:
116
116
"""Format a log entry for sending to the API.
117
-
117
+
118
118
Args:
119
119
log_entry: Raw log entry from Claude
120
-
120
+
121
121
Returns:
122
122
Formatted log entry ready for API
123
123
"""
124
124
# For now, pass through as-is since API expects dict[str, Any]
125
125
# This can be enhanced to transform or filter fields as needed
126
- return log_entry
126
+ return log_entry
0 commit comments