agentready/scripts/query_github_cache.py at main · ambient-code/agentready · GitHub

History

executable file

223 lines (172 loc) · 6.55 KB

Raw

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

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

#!/usr/bin/env python3

"""

Query cached GitHub organization data.

Provides various queries and filters on the cached repository data.

"""

import json

import sys

from datetime import datetime

from pathlib import Path

from typing import Any

def load_cache(cache_file: Path) -> dict[str, Any]:

"""Load cached data from JSON file."""

if not cache_file.exists():

print(f"Error: Cache file not found: {cache_file}", file=sys.stderr)

print("Run analyze_github_org.py first to fetch data", file=sys.stderr)

sys.exit(1)

with cache_file.open() as f:

return json.load(f)

def query_by_language(

repos: list[dict[str, Any]], language: str

) -> list[dict[str, Any]]:

"""Filter repositories by primary language."""

return [

r for r in repos if r.get("primaryLanguageName", "").lower() == language.lower()

]

def query_active_repos(

repos: list[dict[str, Any]], days: int = 30

) -> list[dict[str, Any]]:

"""Filter repositories active within specified days."""

return [r for r in repos if r["daysSinceLastPush"] <= days]

def query_stale_repos(

repos: list[dict[str, Any]], days: int = 180, exclude_archived: bool = True

) -> list[dict[str, Any]]:

"""Filter repositories not updated in specified days."""

result = [r for r in repos if r["daysSinceLastPush"] > days]

if exclude_archived:

result = [r for r in result if not r["isArchived"]]

return result

def query_archived_repos(repos: list[dict[str, Any]]) -> list[dict[str, Any]]:

"""Filter archived repositories."""

return [r for r in repos if r["isArchived"]]

def query_by_stars(

repos: list[dict[str, Any]], min_stars: int = 10

) -> list[dict[str, Any]]:

"""Filter repositories with minimum star count."""

return [r for r in repos if r["stargazerCount"] >= min_stars]

def query_forks_only(repos: list[dict[str, Any]]) -> list[dict[str, Any]]:

"""Filter forked repositories."""

return [r for r in repos if r["isFork"]]

def query_original_only(repos: list[dict[str, Any]]) -> list[dict[str, Any]]:

"""Filter original (non-fork) repositories."""

return [r for r in repos if not r["isFork"]]

def print_repo_list(repos: list[dict[str, Any]], title: str):

"""Print formatted repository list."""

print(f"\n{'=' * 100}")

print(f"{title} ({len(repos)} repositories)")

print(f"{'=' * 100}\n")

if not repos:

print("No repositories found.\n")

return

# Sort by activity

sorted_repos = sorted(repos, key=lambda r: r["daysSinceLastPush"])

# Print header

print(

f"{'Repository':<45} {'Language':<15} {'Stars':<7} {'Forks':<7} {'Last Push':<12} {'Status'}"

)

print("-" * 100)

for repo in sorted_repos:

name = repo["name"]

if len(name) > 43:

name = name[:40] + "..."

language = repo["primaryLanguageName"] or "N/A"

if len(language) > 13:

language = language[:10] + "..."

stars = repo["stargazerCount"]

forks = repo["forkCount"]

days_ago = repo["daysSinceLastPush"]

if days_ago == 0:

last_push = "Today"

elif days_ago == 1:

last_push = "Yesterday"

elif days_ago < 30:

last_push = f"{days_ago}d ago"

elif days_ago < 365:

last_push = f"{days_ago // 30}mo ago"

else:

last_push = f"{days_ago // 365}y ago"

status_parts = []

if repo["isArchived"]:

status_parts.append("ARCHIVED")

if repo["isFork"]:

status_parts.append("fork")

status = " | ".join(status_parts) if status_parts else ""

print(

f"{name:<45} {language:<15} {stars:<7} {forks:<7} {last_push:<12} {status}"

)

print()

def main():

import argparse

parser = argparse.ArgumentParser(

description="Query cached GitHub organization data"

)

parser.add_argument(

"--cache",

type=Path,

default=Path(".cache/github-org-analysis/opendatahub-io-repos.json"),

help="Path to cache file",

)

# Query filters

parser.add_argument("--language", help="Filter by language (e.g., Python, Go)")

parser.add_argument(

"--active", type=int, metavar="DAYS", help="Repos active within N days"

)

parser.add_argument(

"--stale", type=int, metavar="DAYS", help="Repos not updated in N days"

)

parser.add_argument(

"--archived", action="store_true", help="Show only archived repos"

)

parser.add_argument("--min-stars", type=int, metavar="N", help="Minimum star count")

parser.add_argument(

"--forks-only", action="store_true", help="Show only forked repos"

)

parser.add_argument(

"--original-only", action="store_true", help="Show only original repos"

)

# Output options

parser.add_argument("--json", action="store_true", help="Output as JSON")

parser.add_argument("--count", action="store_true", help="Show only count")

args = parser.parse_args()

# Load cache

cache_data = load_cache(args.cache)

repos = cache_data["repositories"]

# Apply filters

filtered = repos

title_parts = ["Repositories"]

if args.language:

filtered = query_by_language(filtered, args.language)

title_parts.append(f"Language: {args.language}")

if args.active is not None:

filtered = query_active_repos(filtered, args.active)

title_parts.append(f"Active within {args.active} days")

if args.stale is not None:

filtered = query_stale_repos(filtered, args.stale)

title_parts.append(f"Stale (>{args.stale} days)")

if args.archived:

filtered = query_archived_repos(filtered)

title_parts.append("Archived")

if args.min_stars is not None:

filtered = query_by_stars(filtered, args.min_stars)

title_parts.append(f"≥{args.min_stars} stars")

if args.forks_only:

filtered = query_forks_only(filtered)

title_parts.append("Forks only")

if args.original_only:

filtered = query_original_only(filtered)

title_parts.append("Original repos only")

# Output

if args.count:

print(len(filtered))

elif args.json:

print(json.dumps(filtered, indent=2, default=str))

else:

title = " - ".join(title_parts)

print_repo_list(filtered, title)

# Print cache info

fetched_at = datetime.fromisoformat(cache_data["fetchedAt"])

print(f"Data fetched: {fetched_at.strftime('%Y-%m-%d %H:%M:%S')}")

print(f"Cache file: {args.cache}")

print()

if __name__ == "__main__":

main()

agentready/scripts/query_github_cache.py at main · ambient-code/agentready · GitHub

Uh oh!

Expand file tree

Breadcrumbs

query_github_cache.py

Latest commit

History

Breadcrumbs

query_github_cache.py

File metadata and controls