| 20 | |
| 21 | |
| 22 | class FetchRepo(Node): |
| 23 | def prep(self, shared): |
| 24 | repo_url = shared.get("repo_url") |
| 25 | local_dir = shared.get("local_dir") |
| 26 | project_name = shared.get("project_name") |
| 27 | |
| 28 | if not project_name: |
| 29 | # Basic name derivation from URL or directory |
| 30 | if repo_url: |
| 31 | project_name = repo_url.split("/")[-1].replace(".git", "") |
| 32 | else: |
| 33 | project_name = os.path.basename(os.path.abspath(local_dir)) |
| 34 | shared["project_name"] = project_name |
| 35 | |
| 36 | # Get file patterns directly from shared |
| 37 | include_patterns = shared["include_patterns"] |
| 38 | exclude_patterns = shared["exclude_patterns"] |
| 39 | max_file_size = shared["max_file_size"] |
| 40 | |
| 41 | return { |
| 42 | "repo_url": repo_url, |
| 43 | "local_dir": local_dir, |
| 44 | "token": shared.get("github_token"), |
| 45 | "include_patterns": include_patterns, |
| 46 | "exclude_patterns": exclude_patterns, |
| 47 | "max_file_size": max_file_size, |
| 48 | "use_relative_paths": True, |
| 49 | } |
| 50 | |
| 51 | def exec(self, prep_res): |
| 52 | if prep_res["repo_url"]: |
| 53 | print(f"Crawling repository: {prep_res['repo_url']}...") |
| 54 | result = crawl_github_files( |
| 55 | repo_url=prep_res["repo_url"], |
| 56 | token=prep_res["token"], |
| 57 | include_patterns=prep_res["include_patterns"], |
| 58 | exclude_patterns=prep_res["exclude_patterns"], |
| 59 | max_file_size=prep_res["max_file_size"], |
| 60 | use_relative_paths=prep_res["use_relative_paths"], |
| 61 | ) |
| 62 | else: |
| 63 | print(f"Crawling directory: {prep_res['local_dir']}...") |
| 64 | |
| 65 | result = crawl_local_files( |
| 66 | directory=prep_res["local_dir"], |
| 67 | include_patterns=prep_res["include_patterns"], |
| 68 | exclude_patterns=prep_res["exclude_patterns"], |
| 69 | max_file_size=prep_res["max_file_size"], |
| 70 | use_relative_paths=prep_res["use_relative_paths"] |
| 71 | ) |
| 72 | |
| 73 | # Convert dict to list of tuples: [(path, content), ...] |
| 74 | files_list = list(result.get("files", {}).items()) |
| 75 | if len(files_list) == 0: |
| 76 | raise (ValueError("Failed to fetch files")) |
| 77 | print(f"Fetched {len(files_list)} files.") |
| 78 | return files_list |
| 79 |
no outgoing calls
no test coverage detected