Spaces:
Running
on
Zero
Running
on
Zero
File size: 2,092 Bytes
4998dac 601ec3d f8ce262 4998dac 5a1d661 5ea6611 4998dac f606abc 4998dac 6b4bf17 4998dac |
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 |
import os
import subprocess
from file_utils import validate_file_types, get_file_summary, read_file_content
SUPPORTED_FILE_TYPES = ["txt", "shell", "python", "markdown", "yaml", "json", "csv", "tsv", "xml", "html", "ini", "jsonl", "ipynb"]
def validate_url(url):
return url.startswith('https://')
def clone_repo(url, repo_dir, hf_token, hf_user):
env = os.environ.copy()
env['GIT_LFS_SKIP_SMUDGE'] = '1'
token_url = url.replace('https://', f'https://{hf_user}:{hf_token}@')
result = subprocess.run(["git", "clone", token_url, repo_dir], env=env, capture_output=True, text=True)
if result.returncode != 0:
return False, result.stderr
return True, None
def extract_repo_content(url, hf_token, hf_user):
if not validate_url(url):
return [{"header": {"name": "Error", "type": "error", "size": 0}, "content": "Invalid URL"}]
repo_dir = "./temp_repo"
if os.path.exists(repo_dir):
subprocess.run(["rm", "-rf", repo_dir])
success, error = clone_repo(url, repo_dir, hf_token, hf_user)
if not success:
return [{"header": {"name": "Error", "type": "error", "size": 0}, "content": f"Failed to clone repository: {error}"}]
file_types = validate_file_types(repo_dir)
extracted_content = []
for file_path, file_type in file_types.items():
file_summary = get_file_summary(file_path, file_type)
# Remove temp_repo/ prefix from the file path
file_summary["name"] = file_summary["name"].replace("temp_repo/", "")
content = {"header": file_summary}
if file_type in SUPPORTED_FILE_TYPES and file_summary["size"] <= 32 * 1024:
try:
content["content"] = read_file_content(file_path)
except Exception as e:
content["content"] = f"Failed to read file content: {str(e)}"
else:
content["content"] = "File too large or binary, content not captured."
extracted_content.append(content)
subprocess.run(["rm", "-rf", repo_dir])
return extracted_content
|