claude-proxy / src /free_claude_code /api /command_utils.py
dheraingoud's picture
feat: synchronize proxy with upstream commits up to 5305bd7 and integrate NIM key rotator
0a54372
Raw
History Blame Contribute Delete
4.94 kB
"""Command parsing utilities for API optimizations."""
import re
import shlex
_ENV_ASSIGNMENT_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*=.*$")
def _is_env_assignment(part: str) -> bool:
"""Return True when a token is a shell-style env assignment."""
return bool(_ENV_ASSIGNMENT_RE.match(part))
def _strip_env_assignments(parts: list[str]) -> list[str]:
"""Return command parts after leading shell-style env assignments."""
cmd_start = 0
for i, part in enumerate(parts):
if _is_env_assignment(part):
cmd_start = i + 1
else:
break
return parts[cmd_start:]
def extract_command_prefix(command: str) -> str:
"""Extract the command prefix for fast prefix detection.
Parses a shell command safely, handling environment variables and
command injection attempts. Returns the command prefix suitable
for quick identification.
Returns:
Command prefix (e.g., "git", "git commit", "npm install")
or "none" if no valid command found
"""
if "`" in command or "$(" in command:
return "command_injection_detected"
try:
parts = shlex.split(command, posix=False)
if not parts:
return "none"
env_prefix = []
cmd_start = 0
for i, part in enumerate(parts):
if _is_env_assignment(part):
env_prefix.append(part)
cmd_start = i + 1
else:
break
if cmd_start >= len(parts):
return "none"
cmd_parts = parts[cmd_start:]
if not cmd_parts:
return "none"
first_word = cmd_parts[0]
two_word_commands = {
"git",
"npm",
"docker",
"kubectl",
"cargo",
"go",
"pip",
"yarn",
}
if first_word in two_word_commands and len(cmd_parts) > 1:
second_word = cmd_parts[1]
if not second_word.startswith("-"):
return f"{first_word} {second_word}"
return first_word
return first_word if not env_prefix else " ".join(env_prefix) + " " + first_word
except ValueError:
parts = command.split()
if not parts:
return "none"
cmd_parts = _strip_env_assignments(parts)
return cmd_parts[0] if cmd_parts else "none"
def extract_filepaths_from_command(command: str, output: str) -> str:
"""Extract file paths from a command locally without API call.
Determines if the command reads file contents and extracts paths accordingly.
Commands like ls/dir/find just list files, so return empty.
Commands like cat/head/tail actually read contents, so extract the file path.
Returns:
Filepath extraction result in <filepaths> format
"""
listing_commands = {
"ls",
"dir",
"find",
"tree",
"pwd",
"cd",
"mkdir",
"rmdir",
"rm",
}
reading_commands = {"cat", "head", "tail", "less", "more", "bat", "type"}
try:
parts = shlex.split(command, posix=False)
if not parts:
return "<filepaths>\n</filepaths>"
cmd_parts = _strip_env_assignments(parts)
if not cmd_parts:
return "<filepaths>\n</filepaths>"
base_cmd = cmd_parts[0].split("/")[-1].split("\\")[-1].lower()
if base_cmd in listing_commands:
return "<filepaths>\n</filepaths>"
if base_cmd in reading_commands:
filepaths = []
for part in cmd_parts[1:]:
if part.startswith("-"):
continue
filepaths.append(part)
if filepaths:
paths_str = "\n".join(filepaths)
return f"<filepaths>\n{paths_str}\n</filepaths>"
return "<filepaths>\n</filepaths>"
if base_cmd == "grep":
flags_with_args = {"-e", "-f", "-m", "-A", "-B", "-C"}
pattern_provided_via_flag = False
positional = []
skip_next = False
for part in cmd_parts[1:]:
if skip_next:
skip_next = False
continue
if part.startswith("-"):
if part in flags_with_args:
if part in {"-e", "-f"}:
pattern_provided_via_flag = True
skip_next = True
continue
positional.append(part)
filepaths = positional if pattern_provided_via_flag else positional[1:]
if filepaths:
paths_str = "\n".join(filepaths)
return f"<filepaths>\n{paths_str}\n</filepaths>"
return "<filepaths>\n</filepaths>"
return "<filepaths>\n</filepaths>"
except ValueError:
return "<filepaths>\n</filepaths>"