mirror of
https://github.com/The-Pocket/PocketFlow-Tutorial-Codebase-Knowledge.git
synced 2026-08-29 08:34:31 +08:00
fix(utils/crawl_github_files): branch names that includes "/"
To fix this problem, my solution was getting branch names from Github api and checking for matching. If there is no match, this time check is a tree like this exists.
This commit is contained in:
+65
-11
@@ -135,21 +135,75 @@ def crawl_github_files(
|
||||
owner = path_parts[0]
|
||||
repo = path_parts[1]
|
||||
|
||||
# Check if URL contains a specific branch/commit
|
||||
if 'tree' in path_parts:
|
||||
tree_index = path_parts.index('tree')
|
||||
ref = path_parts[tree_index + 1]
|
||||
# Combine all parts after the ref as the path
|
||||
path_start = tree_index + 2
|
||||
specific_path = '/'.join(path_parts[path_start:]) if path_start < len(path_parts) else ""
|
||||
else:
|
||||
ref = "main" # Default branch
|
||||
specific_path = ""
|
||||
|
||||
# Setup for GitHub API
|
||||
headers = {"Accept": "application/vnd.github.v3+json"}
|
||||
if token:
|
||||
headers["Authorization"] = f"token {token}"
|
||||
|
||||
def fetch_branches(owner: str, repo: str):
|
||||
"""Get brancshes of the repository"""
|
||||
|
||||
url = f"https://api.github.com/repos/{owner}/{repo}/branches"
|
||||
response = requests.get(url, headers=headers)
|
||||
|
||||
if response.status_code == 404:
|
||||
if not token:
|
||||
print(f"Error 404: Repository not found or is private.\n"
|
||||
f"If this is a private repository, please provide a valid GitHub token via the 'token' argument or set the GITHUB_TOKEN environment variable.")
|
||||
else:
|
||||
print(f"Error 404: Path '{path}' not found in repository or insufficient permissions with the provided token.\n"
|
||||
f"Please verify the token has access to this repository and the path exists.")
|
||||
return []
|
||||
|
||||
if response.status_code != 200:
|
||||
print(f"Error fetching {path}: {response.status_code} - {response.text}")
|
||||
return []
|
||||
|
||||
return response.json()
|
||||
|
||||
def check_tree(owner: str, repo: str, tree: str):
|
||||
"""Check the repository has the given tree"""
|
||||
|
||||
url = f"https://api.github.com/repos/{owner}/{repo}/git/trees/{tree}"
|
||||
response = requests.get(url, headers=headers)
|
||||
|
||||
return True if response.status_code == 200 else False
|
||||
|
||||
# Check if URL contains a specific branch/commit
|
||||
if len(path_parts) > 2 and 'tree' == path_parts[2]:
|
||||
join_parts = lambda i: '/'.join(path_parts[i:])
|
||||
|
||||
branches = fetch_branches(owner, repo)
|
||||
branch_names = map(lambda branch: branch.get("name"), branches)
|
||||
|
||||
# Fetching branches is not successfully
|
||||
if len(branches) == 0:
|
||||
return
|
||||
|
||||
# To check branch name
|
||||
relevant_path = join_parts(3)
|
||||
|
||||
# Find a match with relevant path and get the branch name
|
||||
filter_gen = (name for name in branch_names if relevant_path.startswith(name))
|
||||
ref = next(filter_gen, None)
|
||||
|
||||
# If match is not found, check for is it a tree
|
||||
if ref == None:
|
||||
tree = path_parts[3]
|
||||
ref = tree if check_tree(owner, repo, tree) else None
|
||||
|
||||
# If it is neither a tree nor a branch name
|
||||
if ref == None:
|
||||
print(f"The given path does not match with any branch and any tree in the repository.\n"
|
||||
f"Please verify the path is exists.")
|
||||
return
|
||||
|
||||
# Combine all parts after the ref as the path
|
||||
part_index = 5 if '/' in ref else 4
|
||||
specific_path = join_parts(part_index) if part_index < len(path_parts) else ""
|
||||
else:
|
||||
ref = "main" # Default branch
|
||||
specific_path = ""
|
||||
|
||||
# Dictionary to store path -> content mapping
|
||||
files = {}
|
||||
|
||||
Reference in New Issue
Block a user