filtrar repos por owner usando endpoints dedicados

This commit is contained in:
2026-05-18 10:26:13 +02:00
parent b5935f2511
commit d14c246bcb
+38 -32
View File
@@ -118,39 +118,45 @@ def gitea_get(cfg: Config, path: str, params: dict[str, str] | None = None) -> t
raise RuntimeError(f"no se pudo contactar con {cfg.url}: {e.reason}") from e raise RuntimeError(f"no se pudo contactar con {cfg.url}: {e.reason}") from e
def fetch_remote_repos(cfg: Config, owner: str) -> list[RemoteRepo]: def _parse_repo_list(body: bytes) -> list[RemoteRepo]:
repos: list[RemoteRepo] = [] items = json.loads(body or b"[]")
page = 1 return [
while True: RemoteRepo(
status, body, _ = gitea_get( name=item.get("name", ""),
cfg, full_name=item.get("full_name", ""),
"/repos/search", clone_url=item.get("clone_url", ""),
{"owner": owner, "limit": "50", "page": str(page)}, ssh_url=item.get("ssh_url", ""),
description=item.get("description") or "",
private=bool(item.get("private", False)),
default_branch=item.get("default_branch") or "main",
) )
if status == 401: for item in items
raise RuntimeError("401 no autorizado — revisa server.token en la config") ]
if status >= 400:
raise RuntimeError(f"Gitea devolvió HTTP {status}: {body[:200].decode('utf-8', 'replace')}")
data = json.loads(body or b"{}") def fetch_remote_repos(cfg: Config, owner: str) -> list[RemoteRepo]:
items = data.get("data") or [] """Lista los repos de un owner. Prueba primero como organización, luego como usuario."""
if not items: for endpoint in (f"/orgs/{owner}/repos", f"/users/{owner}/repos"):
break repos: list[RemoteRepo] = []
for item in items: page = 1
repos.append( found = False
RemoteRepo( while True:
name=item.get("name", ""), status, body, _ = gitea_get(cfg, endpoint, {"limit": "50", "page": str(page)})
full_name=item.get("full_name", ""), if status == 401:
clone_url=item.get("clone_url", ""), raise RuntimeError("401 no autorizado — revisa server.token en la config")
ssh_url=item.get("ssh_url", ""), if status == 404:
description=item.get("description") or "", break # no es este tipo, probamos el siguiente
private=bool(item.get("private", False)), if status >= 400:
default_branch=item.get("default_branch") or "main", raise RuntimeError(f"Gitea devolvió HTTP {status}: {body[:200].decode('utf-8', 'replace')}")
) found = True
) items = _parse_repo_list(body)
if len(items) < 50: repos.extend(items)
break if len(items) < 50:
page += 1 break
return repos page += 1
if found:
return repos
raise RuntimeError(f"'{owner}' no existe como usuario ni como organización en el servidor")
# --- escaneo local ---------------------------------------------------------- # --- escaneo local ----------------------------------------------------------