Spaces:
Sleeping
Sleeping
File size: 5,982 Bytes
a724f0f |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 |
from googlesearch import search as google_search
from smolagents import tool, Tool
from curl_scraper import BasicScraper
from huggingface_hub import hf_hub_download, upload_folder
import os
class FileReader(Tool):
name = "read_file"
description = "Reads a file and returns the contents as a string."
inputs = {"file": {"type": "string", "description": "The absolute path to the file to be read."}}
output_type = "string"
def __init__(self, space_id, folder_path, **kwargs):
super().__init__()
self.space_id = space_id
self.folder_path = folder_path
def forward(self, file: str) -> str:
file_path = os.path.join(self.folder_path, file)
if not os.path.exists(file_path):
hf_hub_download(repo_id=self.space_id, filename=file, repo_type="space", local_dir=self.folder_path, local_dir_use_symlinks=False)
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
return content
class FileWriter(Tool):
name = "write_file"
description = "Overwrite or append content to a file. Use for creating new files, appending content, or modifying existing files."
inputs = {
"file": {"type": "string", "description": "The absolute path to the file to be written."},
"content": {"type": "string", "description": "The content to be written to the file."},
"append": {"type": "boolean", "description": "Whether to append to the file or overwrite it.", "nullable": True},
}
output_type = "string"
def __init__(self, space_id, folder_path, **kwargs):
super().__init__()
self.space_id = space_id
self.folder_path = folder_path
def forward(self, file: str, content: str, append: bool = False) -> str:
file_path = os.path.join(self.folder_path, file)
mode = 'a' if append else 'w'
with open(file_path, mode, encoding='utf-8') as f:
f.write(content if append else content.strip())
return f"Successfully {'appended to' if append else 'wrote to'} file at {file_path}"
@tool
def search(query: str) -> list:
"""
Search for web pages and scrape their content.
Args:
query (str): The search query.
Returns:
list: List of top web results including url, title, description from the search results.
"""
# Get search results
results = []
for result in google_search(query, advanced=True, unique=True):
results.append(result)
return results
@tool
def scraper(url: str, response_type: str = 'plain_text') -> str:
"""
Scrapes the content of a web page. Important: Don't use regex to extract information from the output of a this scraper tool, Think more and find answer yourself.
Args:
url (str): The URL of the web page to scrape.
response_type (str): The type of response to return. Valid options are 'markdown' and 'plain_text'. Default is 'plain_text'.
Returns:
str: The content of the web page in the specified format.
"""
response = BasicScraper(timeout=5).get(url)
if response_type == 'markdown':
return response.markdown
elif response_type == 'plain_text':
return response.plain_text
else:
raise ValueError("Invalid response_type. Use 'markdown' or 'plain_text'.")
class CommitChanges(Tool):
name = "commit_changes"
description = "Commits changes to a repository."
inputs = {
"commit_message": {"type": "string", "description": "The commit message."},
"commit_description": {"type": "string", "description": "The commit description."},
}
output_type = "string"
def __init__(self, space_id, folder_path, **kwargs):
super().__init__()
self.space_id = space_id
self.folder_path = folder_path
def forward(self, commit_message: str, commit_description: str) -> str:
try:
upload_folder(
folder_path=self.folder_path,
repo_id=self.space_id,
repo_type="space",
create_pr=True,
commit_message=commit_message,
commit_description=commit_description,
)
return "Changes committed successfully."
except Exception as e:
return f"Error committing changes: {str(e)}"
from huggingface_hub import HfApi
def get_repository_structure(space_id: str) -> str:
api = HfApi()
space_info = api.space_info(space_id)
printed_dirs = set()
output_lines = []
sorted_siblings = sorted(space_info.siblings, key=lambda x: x.rfilename)
for sibling in sorted_siblings:
rfilename = sibling.rfilename
path_parts = rfilename.split('/')
current_dir_path_parts = []
for i in range(len(path_parts) - 1):
dir_name = path_parts[i]
current_dir_path_parts.append(dir_name)
dir_path = '/'.join(current_dir_path_parts)
if dir_path not in printed_dirs:
depth = i
indent = ' ' * depth
output_lines.append(f"{indent}{dir_name}")
printed_dirs.add(dir_path)
file_name = path_parts[-1]
file_indent_depth = len(path_parts) - 1
file_indent = ' ' * file_indent_depth
if file_indent_depth == 0:
output_lines.append(file_name)
else:
is_last_file = sibling == sorted_siblings[-1] or not any(
s.rfilename.startswith('/'.join(path_parts[:-1]) + '/')
for s in sorted_siblings[sorted_siblings.index(sibling)+1:]
)
file_prefix = '└─ ' if is_last_file else '├─ '
output_lines.append(f"{file_indent}{file_prefix}{file_name}")
return '\n'.join(output_lines) |