|
3 | 3 | import argparse |
4 | 4 | import os |
5 | 5 |
|
6 | | -from typing import Dict, Optional |
7 | | -from github import Github, Repository, GithubException |
8 | | - |
9 | | -class IndexGenerator: |
10 | | - def __init__(self, agency: str, verison: str, token: Optional[str] = None,): |
11 | | - self.github = Github(token) if token else Github() |
12 | | - |
13 | | - # user can change agency and version depending on paramters |
14 | | - self.index = { |
15 | | - "agency": agency, |
16 | | - "version": verison, |
17 | | - "measurementType": { |
18 | | - "method": "projects" |
19 | | - }, |
20 | | - "releases": [] |
21 | | - } |
22 | | - |
23 | | - def get_code_json(self, repo: Repository) -> Optional[Dict]: |
24 | | - try: |
25 | | - content = repo.get_contents("code.json", ref = "main") |
26 | | - except GithubException as e: |
27 | | - print(f"GitHub Error: {e.data.get('message', 'No message available')}") |
28 | | - return None |
29 | | - |
30 | | - try: |
31 | | - decoded_content = base64.b64decode(content.content) |
32 | | - return json.loads(decoded_content) |
33 | | - except (json.JSONDecodeError, ValueError) as e: |
34 | | - print(f"JSON Error: {str(e)}") |
35 | | - return None |
36 | | - |
37 | | - def update_index(self, index: Dict, code_json: Dict, org_name: str, repo_name: str) -> None: |
38 | | - baseline = { |
39 | | - 'organization': org_name, |
40 | | - 'name': repo_name |
41 | | - } |
42 | | - |
43 | | - baseline.update(code_json) |
44 | | - |
45 | | - index['releases'].append(baseline) |
46 | | - |
47 | | - def process_organization(self, org_name: str) -> None: |
48 | | - try: |
49 | | - org = self.github.get_organization(org_name) |
50 | | - print(f"\nProcessing organization: {org_name}") |
51 | | - |
52 | | - total_repos = org.public_repos |
53 | | - print(f"Found {total_repos} public repositories") |
54 | | - |
55 | | - for id, repo in enumerate(org.get_repos(type='public'), 1): |
56 | | - print(f"\nChecking {repo.name} [{id}/{total_repos}]") |
57 | | - |
58 | | - code_json = self.get_code_json(repo) |
59 | | - if code_json: |
60 | | - print(f"✅ Found code.json in {repo.name}") |
61 | | - self.update_index(self.index, code_json, org_name, repo.name) |
62 | | - else: |
63 | | - print(f"❌ No code.json found in {repo.name}") |
64 | | - |
65 | | - except GithubException as e: |
66 | | - print(f"Error processing organization {org_name}: {str(e)}") |
67 | | - |
68 | | - def save_index(self, output_path: str) -> None: |
69 | | - # sorts index by organizaiton then by name |
70 | | - self.index['releases'].sort(key=lambda x: (x.get('organization', ''), x.get('name', ''))) |
71 | | - |
72 | | - with open(output_path, 'w') as f: |
73 | | - json.dump(self.index, f, indent=2) |
| 6 | +from codejson_index_generator import IndexGenerator |
74 | 7 |
|
75 | 8 | def main(): |
76 | 9 | parser = argparse.ArgumentParser( |
|
0 commit comments