ccproxy.plugins.command_replay.formatter

ccproxy.plugins.command_replay.formatter

File formatter for command replay output.

CommandFileFormatter

CommandFileFormatter(
    log_dir="/tmp/ccproxy/command_replay",
    enabled=True,
    separate_files_per_command=False,
)

Formats and writes command replay data to files.

Parameters:

Name	Type	Description	Default
log_dir	str	Directory for command replay files	'/tmp/ccproxy/command_replay'
enabled	bool	Enable file writing	True
separate_files_per_command	bool	Create separate files for curl/xh vs combined	False

Source code in ccproxy/plugins/command_replay/formatter.py

def __init__(
    self,
    log_dir: str = "/tmp/ccproxy/command_replay",
    enabled: bool = True,
    separate_files_per_command: bool = False,
) -> None:
    """Initialize with configuration.

    Args:
        log_dir: Directory for command replay files
        enabled: Enable file writing
        separate_files_per_command: Create separate files for curl/xh vs combined
    """
    self.enabled = enabled
    self.log_dir = Path(log_dir)
    self.separate_files_per_command = separate_files_per_command

    if self.enabled:
        # Create log directory if it doesn't exist
        try:
            self.log_dir.mkdir(parents=True, exist_ok=True)
        except OSError as e:
            logger.error(
                "failed_to_create_command_replay_directory",
                log_dir=str(self.log_dir),
                error=str(e),
                exc_info=e,
            )
            # Disable file writing if we can't create the directory
            self.enabled = False

    # Track which files we've already created (for logging purposes only)
    self._created_files: set[str] = set()

should_write_files

should_write_files()

Check if file writing is enabled.

Source code in ccproxy/plugins/command_replay/formatter.py

def should_write_files(self) -> bool:
    """Check if file writing is enabled."""
    return bool(self.enabled)

write_commands async

write_commands(
    request_id,
    curl_command,
    xh_command,
    provider=None,
    timestamp_prefix=None,
    method=None,
    url=None,
    headers=None,
    body=None,
    is_json=False,
)

Write command replay data to files.

Parameters:

Name	Type	Description	Default
request_id	str	Request ID for correlation	required
curl_command	str	Generated curl command	required
xh_command	str	Generated xh command	required
provider	str | None	Provider name (anthropic, openai, etc.)	None
timestamp_prefix	str | None	Optional timestamp prefix from RequestContext	None
method	str | None	HTTP method for shell script generation	None
url	str | None	Request URL for shell script generation	None
headers	dict[str, str] | None	HTTP headers for shell script generation	None
body	Any	Request body for shell script generation	None
is_json	bool	Whether body is JSON for shell script generation	False

Returns:

Type	Description
list[str]	List of file paths that were written

Source code in ccproxy/plugins/command_replay/formatter.py

async def write_commands(
    self,
    request_id: str,
    curl_command: str,
    xh_command: str,
    provider: str | None = None,
    timestamp_prefix: str | None = None,
    method: str | None = None,
    url: str | None = None,
    headers: dict[str, str] | None = None,
    body: Any = None,
    is_json: bool = False,
) -> list[str]:
    """Write command replay data to files.

    Args:
        request_id: Request ID for correlation
        curl_command: Generated curl command
        xh_command: Generated xh command
        provider: Provider name (anthropic, openai, etc.)
        timestamp_prefix: Optional timestamp prefix from RequestContext
        method: HTTP method for shell script generation
        url: Request URL for shell script generation
        headers: HTTP headers for shell script generation
        body: Request body for shell script generation
        is_json: Whether body is JSON for shell script generation

    Returns:
        List of file paths that were written
    """
    if not self.enabled:
        return []

    written_files = []

    # Use provided timestamp prefix or generate our own
    if timestamp_prefix:
        base_id = f"{self._compose_file_id(request_id)}_{timestamp_prefix}"
    else:
        base_id = self._compose_file_id_with_timestamp(request_id)

    # Add provider to filename if available
    if provider:
        base_id = f"{base_id}_{provider}"

    try:
        if self.separate_files_per_command:
            # Write separate files for curl and xh
            if curl_command:
                curl_file = await self._write_single_command_file(
                    base_id, "curl", curl_command, request_id
                )
                if curl_file:
                    written_files.append(curl_file)

            if xh_command:
                xh_file = await self._write_single_command_file(
                    base_id, "xh", xh_command, request_id
                )
                if xh_file:
                    written_files.append(xh_file)
        else:
            # Write combined file with both commands
            combined_file = await self._write_combined_command_file(
                base_id, curl_command, xh_command, request_id, provider
            )
            if combined_file:
                written_files.append(combined_file)

        # Generate executable shell script if we have raw request data
        if method and url:
            shell_script_file = await self._write_shell_script_file(
                base_id, request_id, method, url, headers, body, is_json, provider
            )
            if shell_script_file:
                written_files.append(shell_script_file)

        # Make files executable
        await self._make_files_executable(written_files)

    except Exception as e:
        logger.error(
            "command_replay_file_write_error",
            request_id=request_id,
            error=str(e),
            exc_info=e,
        )

    return written_files

get_log_dir

get_log_dir()

Get the log directory path.

Source code in ccproxy/plugins/command_replay/formatter.py

def get_log_dir(self) -> str:
    """Get the log directory path."""
    return str(self.log_dir)

cleanup

cleanup()

Clean up resources (if any).

Source code in ccproxy/plugins/command_replay/formatter.py

def cleanup(self) -> None:
    """Clean up resources (if any)."""
    self._created_files.clear()