import base64
import hashlib
import mimetypes
import os
import time
from typing import Any, Optional
from rp.editor import RPEditor
from ..tools.patch import display_content_diff
from ..ui.diff_display import get_diff_stats
from ..ui.edit_feedback import track_edit, tracker
_id = 0
def get_uid():
global _id
_id += 3
return _id
def read_specific_lines(
filepath: str, start_line: int, end_line: Optional[int] = None, db_conn: Optional[Any] = None
) -> dict:
"""
Read specific lines or a range of lines from a file.
This function allows reading a single line or a contiguous range of lines from the specified file.
It supports optional database connection for tracking read operations.
Args:
filepath (str): The path to the file to read from. Supports user home directory expansion (e.g., ~).
start_line (int): The 1-based line number to start reading from.
end_line (Optional[int]): The 1-based line number to end reading at (inclusive). If None, reads only the start_line.
db_conn (Optional[Any]): An optional database connection object for tracking read operations. If provided, marks the file as read in the database.
Returns:
dict: A dictionary containing the status and either the content or an error message.
- On success: {"status": "success", "content": str} where content is the lines joined by newlines.
- On error: {"status": "error", "error": str} with the exception message.
Raises:
None: Exceptions are caught and returned in the response dictionary.
Examples:
# Read line 5 only
result = read_specific_lines("example.txt", 5)
# Read lines 10 to 20
result = read_specific_lines("example.txt", 10, 20)
"""
try:
path = os.path.expanduser(filepath)
with open(path, "r") as file:
lines = file.readlines()
total_lines = len(lines)
if start_line < 1 or start_line > total_lines:
return {
"status": "error",
"error": f"Start line {start_line} is out of range. File has {total_lines} lines.",
}
if end_line is None:
end_line = start_line
if end_line < start_line or end_line > total_lines:
return {
"status": "error",
"error": f"End line {end_line} is out of range. File has {total_lines} lines.",
}
selected_lines = lines[start_line - 1 : end_line]
content = "".join(selected_lines)
if db_conn:
from rp.tools.database import db_set
db_set("read:" + path, "true", db_conn)
return {"status": "success", "content": content}
except Exception as e:
return {"status": "error", "error": str(e)}
def replace_specific_line(
filepath: str,
line_number: int,
new_content: str,
db_conn: Optional[Any] = None,
show_diff: bool = True,
) -> dict:
"""
Replace the content of a specific line in a file.
This function replaces the entire content of a single line with new text. It supports optional database tracking
and diff display. The file must be read first if a database connection is provided.
Args:
filepath (str): The path to the file to modify. Supports user home directory expansion (e.g., ~).
line_number (int): The 1-based line number to replace.
new_content (str): The new content to place on the specified line (should not include trailing newline).
db_conn (Optional[Any]): An optional database connection for tracking. Requires the file to be read first.
show_diff (bool): If True, displays a diff of the changes after replacement.
Returns:
dict: A dictionary with status and message or error.
- On success: {"status": "success", "message": str} describing the operation.
- On error: {"status": "error", "error": str} with the exception or validation message.
Raises:
None: Exceptions are handled internally.
Examples:
# Replace line 3 with new text
result = replace_specific_line("file.txt", 3, "New line content")
"""
try:
path = os.path.expanduser(filepath)
if not os.path.exists(path):
return {"status": "error", "error": "File does not exist"}
if db_conn:
from rp.tools.database import db_get
read_status = db_get("read:" + path, db_conn)
if read_status.get("status") != "success" or read_status.get("value") != "true":
return {
"status": "error",
"error": "File must be read before writing. Please read the file first.",
}
with open(path, "r") as file:
lines = file.readlines()
total_lines = len(lines)
if line_number < 1 or line_number > total_lines:
return {
"status": "error",
"error": f"Line number {line_number} is out of range. File has {total_lines} lines.",
}
old_content = "".join(lines)
lines[line_number - 1] = (
new_content + "\n" if not new_content.endswith("\n") else new_content
)
new_full_content = "".join(lines)
with open(path, "w") as file:
file.writelines(lines)
if show_diff:
diff_result = display_content_diff(old_content, new_full_content, filepath)
if diff_result["status"] == "success":
print(diff_result["visual_diff"])
return {"status": "success", "message": f"Replaced line {line_number} in {path}"}
except Exception as e:
return {"status": "error", "error": str(e)}
def insert_line_at_position(
filepath: str,
line_number: int,
new_content: str,
db_conn: Optional[Any] = None,
show_diff: bool = True,
) -> dict:
"""
Insert a new line at a specific position in a file.
This function inserts new content as a new line before the specified line number. If line_number is beyond the file's length,
it appends to the end. Supports database tracking and diff display.
Args:
filepath (str): The path to the file to modify. Supports user home directory expansion (e.g., ~).
line_number (int): The 1-based line number before which to insert the new line. If greater than total lines, appends.
new_content (str): The content for the new line (trailing newline is added if missing).
db_conn (Optional[Any]): Optional database connection for tracking. File must be read first if provided.
show_diff (bool): If True, displays a diff of the changes after insertion.
Returns:
dict: Status and message or error.
- Success: {"status": "success", "message": str}
- Error: {"status": "error", "error": str}
Examples:
# Insert before line 5
result = insert_line_at_position("file.txt", 5, "Inserted line")
"""
try:
path = os.path.expanduser(filepath)
if not os.path.exists(path):
return {"status": "error", "error": "File does not exist"}
if db_conn:
from rp.tools.database import db_get
read_status = db_get("read:" + path, db_conn)
if read_status.get("status") != "success" or read_status.get("value") != "true":
return {
"status": "error",
"error": "File must be read before writing. Please read the file first.",
}
with open(path, "r") as file:
lines = file.readlines()
old_content = "".join(lines)
insert_index = min(line_number - 1, len(lines))
lines.insert(
insert_index, new_content + "\n" if not new_content.endswith("\n") else new_content
)
new_full_content = "".join(lines)
with open(path, "w") as file:
file.writelines(lines)
if show_diff:
diff_result = display_content_diff(old_content, new_full_content, filepath)
if diff_result["status"] == "success":
print(diff_result["visual_diff"])
return {
"status": "success",
"message": f"Inserted line at position {line_number} in {path}",
}
except Exception as e:
return {"status": "error", "error": str(e)}
def delete_specific_line(
filepath: str, line_number: int, db_conn: Optional[Any] = None, show_diff: bool = True
) -> dict:
"""
Delete a specific line from a file.
This function removes the specified line from the file. Supports database tracking and diff display.
Args:
filepath (str): The path to the file to modify. Supports user home directory expansion (e.g., ~).
line_number (int): The 1-based line number to delete.
db_conn (Optional[Any]): Optional database connection for tracking. File must be read first if provided.
show_diff (bool): If True, displays a diff of the changes after deletion.
Returns:
dict: Status and message or error.
- Success: {"status": "success", "message": str}
- Error: {"status": "error", "error": str}
Examples:
# Delete line 10
result = delete_specific_line("file.txt", 10)
"""
try:
path = os.path.expanduser(filepath)
if not os.path.exists(path):
return {"status": "error", "error": "File does not exist"}
if db_conn:
from rp.tools.database import db_get
read_status = db_get("read:" + path, db_conn)
if read_status.get("status") != "success" or read_status.get("value") != "true":
return {
"status": "error",
"error": "File must be read before writing. Please read the file first.",
}
with open(path, "r") as file:
lines = file.readlines()
total_lines = len(lines)
if line_number < 1 or line_number > total_lines:
return {
"status": "error",
"error": f"Line number {line_number} is out of range. File has {total_lines} lines.",
}
old_content = "".join(lines)
del lines[line_number - 1]
new_full_content = "".join(lines)
with open(path, "w") as file:
file.writelines(lines)
if show_diff:
diff_result = display_content_diff(old_content, new_full_content, filepath)
if diff_result["status"] == "success":
print(diff_result["visual_diff"])
return {"status": "success", "message": f"Deleted line {line_number} from {path}"}
except Exception as e:
return {"status": "error", "error": str(e)}
def read_file(filepath: str, db_conn: Optional[Any] = None) -> dict:
"""
Read the contents of a file.
Args:
filepath: Path to the file to read
db_conn: Optional database connection for tracking
Returns:
dict: Status and content or error
"""
try:
path = os.path.expanduser(filepath)
mime_type, _ = mimetypes.guess_type(str(path))
if mime_type and (
mime_type.startswith("text/") or mime_type in ["application/json", "application/xml"]
):
with open(path, encoding="utf-8", errors="replace") as f:
content = f.read()
else:
with open(path, "rb") as f:
binary_content = f.read()
content = f"data:{mime_type if mime_type else 'application/octet-stream'};base64,{base64.b64encode(binary_content).decode('utf-8')}"
if db_conn:
from rp.tools.database import db_set
db_set("read:" + path, "true", db_conn)
return {"status": "success", "content": content}
except Exception as e:
return {"status": "error", "error": str(e)}
def write_file(
filepath: str, content: str, db_conn: Optional[Any] = None, show_diff: bool = True
) -> dict:
"""
Write content to a file.
Args:
filepath: Path to the file to write
content: Content to write
db_conn: Optional database connection for tracking
show_diff: Whether to show diff of changes
Returns:
dict: Status and message or error
"""
operation = None
try:
path = os.path.expanduser(filepath)
old_content = ""
is_new_file = not os.path.exists(path)
if not is_new_file and db_conn:
from rp.tools.database import db_get
read_status = db_get("read:" + path, db_conn)
if read_status.get("status") != "success" or read_status.get("value") != "true":
return {
"status": "error",
"error": "File must be read before writing. Please read the file first.",
}
write_mode = "w"
write_encoding = "utf-8"
decoded_content = content
if content.startswith("data:"):
parts = content.split(",", 1)
if len(parts) == 2:
header = parts[0]
encoded_data = parts[1]
if ";base64" in header:
try:
decoded_content = base64.b64decode(encoded_data)
write_mode = "wb"
write_encoding = None
except Exception:
pass # Not a valid base64, treat as plain text
if not is_new_file:
if write_mode == "wb":
with open(path, "rb") as f:
old_content = f.read()
else:
with open(path, encoding="utf-8", errors="replace") as f:
old_content = f.read()
operation = track_edit("WRITE", filepath, content=content, old_content=old_content)
tracker.mark_in_progress(operation)
if show_diff and (not is_new_file) and write_mode == "w": # Only show diff for text files
diff_result = display_content_diff(old_content, content, filepath)
if diff_result["status"] == "success":
print(diff_result["visual_diff"])
if write_mode == "wb":
with open(path, write_mode) as f:
f.write(decoded_content)
else:
with open(path, write_mode, encoding=write_encoding) as f:
f.write(decoded_content)
if os.path.exists(path) and db_conn:
try:
cursor = db_conn.cursor()
file_hash = hashlib.md5(
old_content.encode() if isinstance(old_content, str) else old_content
).hexdigest()
cursor.execute(
"SELECT MAX(version) FROM file_versions WHERE filepath = ?", (filepath,)
)
result = cursor.fetchone()
version = result[0] + 1 if result[0] else 1
cursor.execute(
"INSERT INTO file_versions (filepath, content, hash, timestamp, version)\n VALUES (?, ?, ?, ?, ?)",
(
filepath,
(
old_content
if isinstance(old_content, str)
else old_content.decode("utf-8", errors="replace")
),
file_hash,
time.time(),
version,
),
)
db_conn.commit()
except Exception:
pass
tracker.mark_completed(operation)
message = f"File written to {path}"
if show_diff and (not is_new_file) and write_mode == "w":
stats = get_diff_stats(old_content, content)
message += f" ({stats['insertions']}+ {stats['deletions']}-)"
return {"status": "success", "message": message}
except Exception as e:
if operation is not None:
tracker.mark_failed(operation)
return {"status": "error", "error": str(e)}
def list_directory(path=".", recursive=False):
"""List files and directories in the specified path."""
try:
path = os.path.expanduser(path)
items = []
if recursive:
for root, dirs, files in os.walk(path):
for name in files:
item_path = os.path.join(root, name)
items.append(
{"path": item_path, "type": "file", "size": os.path.getsize(item_path)}
)
for name in dirs:
items.append({"path": os.path.join(root, name), "type": "directory"})
else:
for item in os.listdir(path):
item_path = os.path.join(path, item)
items.append(
{
"name": item,
"type": "directory" if os.path.isdir(item_path) else "file",
"size": os.path.getsize(item_path) if os.path.isfile(item_path) else None,
}
)
return {"status": "success", "items": items}
except Exception as e:
return {"status": "error", "error": str(e)}
def mkdir(path):
try:
os.makedirs(os.path.expanduser(path), exist_ok=True)
return {"status": "success", "message": f"Directory created at {path}"}
except Exception as e:
return {"status": "error", "error": str(e)}
def chdir(path):
"""Change the current working directory."""
try:
os.chdir(os.path.expanduser(path))
return {"status": "success", "new_path": os.getcwd()}
except Exception as e:
return {"status": "error", "error": str(e)}
def getpwd():
"""Get the current working directory."""
try:
return {"status": "success", "path": os.getcwd()}
except Exception as e:
return {"status": "error", "error": str(e)}
def index_source_directory(path: str) -> dict:
"""
Index directory recursively and read all source files.
Args:
path: Path to index
Returns:
dict: Status and indexed files or error
"""
extensions = [
".py",
".js",
".ts",
".java",
".cpp",
".c",
".h",
".hpp",
".html",
".css",
".json",
".xml",
".md",
".sh",
".rb",
".go",
]
source_files = []
try:
for root, _, files in os.walk(os.path.expanduser(path)):
for file in files:
if any((file.endswith(ext) for ext in extensions)):
filepath = os.path.join(root, file)
try:
with open(filepath, encoding="utf-8") as f:
content = f.read()
source_files.append({"path": filepath, "content": content})
except Exception:
continue
return {"status": "success", "indexed_files": source_files}
except Exception as e:
return {"status": "error", "error": str(e)}
def search_replace(
filepath: str, old_string: str, new_string: str, db_conn: Optional[Any] = None
) -> dict:
"""
Search and replace text in a file.
Args:
filepath: Path to the file
old_string: String to replace
new_string: Replacement string
db_conn: Optional database connection for tracking
Returns:
dict: Status and message or error
"""
try:
path = os.path.expanduser(filepath)
if not os.path.exists(path):
return {"status": "error", "error": "File does not exist"}
mime_type, _ = mimetypes.guess_type(str(path))
if not (
mime_type
and (
mime_type.startswith("text/")
or mime_type in ["application/json", "application/xml"]
)
):
return {
"status": "error",
"error": f"Cannot perform search and replace on binary file: {filepath}",
}
if db_conn:
from rp.tools.database import db_get
read_status = db_get("read:" + path, db_conn)
if read_status.get("status") != "success" or read_status.get("value") != "true":
return {
"status": "error",
"error": "File must be read before writing. Please read the file first.",
}
with open(path, encoding="utf-8", errors="replace") as f:
content = f.read()
content = content.replace(old_string, new_string)
with open(path, "w") as f:
f.write(content)
return {
"status": "success",
"message": f"Replaced '{old_string}' with '{new_string}' in {path}",
}
except Exception as e:
return {"status": "error", "error": str(e)}
_editors = {}
def get_editor(filepath):
if filepath not in _editors:
_editors[filepath] = RPEditor(filepath)
return _editors[filepath]
def close_editor(filepath):
try:
path = os.path.expanduser(filepath)
editor = get_editor(path)
editor.close()
return {"status": "success", "message": f"Editor closed for {path}"}
except Exception as e:
return {"status": "error", "error": str(e)}
def open_editor(filepath):
try:
path = os.path.expanduser(filepath)
editor = RPEditor(path)
editor.start()
return {"status": "success", "message": f"Editor opened for {path}"}
except Exception as e:
return {"status": "error", "error": str(e)}
def editor_insert_text(filepath, text, line=None, col=None, show_diff=True, db_conn=None):
operation = None
try:
path = os.path.expanduser(filepath)
mime_type, _ = mimetypes.guess_type(str(path))
if not (
mime_type
and (
mime_type.startswith("text/")
or mime_type in ["application/json", "application/xml"]
)
):
return {"status": "error", "error": f"Cannot insert text into binary file: {filepath}"}
if db_conn:
from rp.tools.database import db_get
read_status = db_get("read:" + path, db_conn)
if read_status.get("status") != "success" or read_status.get("value") != "true":
return {
"status": "error",
"error": "File must be read before writing. Please read the file first.",
}
old_content = ""
if os.path.exists(path):
with open(path, encoding="utf-8", errors="replace") as f:
old_content = f.read()
position = (line if line is not None else 0) * 1000 + (col if col is not None else 0)
operation = track_edit("INSERT", filepath, start_pos=position, content=text)
tracker.mark_in_progress(operation)
editor = get_editor(path)
if line is not None and col is not None:
editor.move_cursor_to(line, col)
editor.insert_text(text)
editor.save_file()
if show_diff and old_content:
with open(path) as f:
new_content = f.read()
diff_result = display_content_diff(old_content, new_content, filepath)
if diff_result["status"] == "success":
print(diff_result["visual_diff"])
tracker.mark_completed(operation)
return {"status": "success", "message": f"Inserted text in {path}"}
except Exception as e:
if operation is not None:
tracker.mark_failed(operation)
return {"status": "error", "error": str(e)}
def editor_replace_text(
filepath, start_line, start_col, end_line, end_col, new_text, show_diff=True, db_conn=None
):
try:
operation = None
path = os.path.expanduser(filepath)
mime_type, _ = mimetypes.guess_type(str(path))
if not (
mime_type
and (
mime_type.startswith("text/")
or mime_type in ["application/json", "application/xml"]
)
):
return {"status": "error", "error": f"Cannot replace text in binary file: {filepath}"}
if db_conn:
from rp.tools.database import db_get
read_status = db_get("read:" + path, db_conn)
if read_status.get("status") != "success" or read_status.get("value") != "true":
return {
"status": "error",
"error": "File must be read before writing. Please read the file first.",
}
old_content = ""
if os.path.exists(path):
with open(path, encoding="utf-8", errors="replace") as f:
old_content = f.read()
start_pos = start_line * 1000 + start_col
end_pos = end_line * 1000 + end_col
operation = track_edit(
"REPLACE",
filepath,
start_pos=start_pos,
end_pos=end_pos,
content=new_text,
old_content=old_content,
)
tracker.mark_in_progress(operation)
editor = get_editor(path)
editor.replace_text(start_line, start_col, end_line, end_col, new_text)
editor.save_file()
if show_diff and old_content:
with open(path) as f:
new_content = f.read()
diff_result = display_content_diff(old_content, new_content, filepath)
if diff_result["status"] == "success":
print(diff_result["visual_diff"])
tracker.mark_completed(operation)
return {"status": "success", "message": f"Replaced text in {path}"}
except Exception as e:
if operation is not None:
tracker.mark_failed(operation)
return {"status": "error", "error": str(e)}
def display_edit_summary():
from ..ui.edit_feedback import display_edit_summary
return display_edit_summary()
def display_edit_timeline(show_content=False):
from ..ui.edit_feedback import display_edit_timeline
return display_edit_timeline(show_content)
def clear_edit_tracker():
from ..ui.edit_feedback import clear_tracker
clear_tracker()
return {"status": "success", "message": "Edit tracker cleared"}
Reference in New Issue View Git Blame Copy Permalink