Compare commits

base: retoor:8a985d53f10573e4c2dfd0d692926205788f199b
Branches Tags
retoor:main
..
compare: retoor:ed9f39899229538eb3c8d98725471b85a7749cb8
Branches Tags
retoor:main

No commits in common. "8a985d53f10573e4c2dfd0d692926205788f199b" and "ed9f39899229538eb3c8d98725471b85a7749cb8" have entirely different histories.
8a985d53f1 ... ed9f398992

3 changed files with 0 additions and 145 deletions
Show Stats Expand all files Collapse all files

2
.gitignore vendored
View File

@ -1,5 +1,3 @@
service.py
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]

140
gitea_api.py
View File

@ -155,145 +155,6 @@ class GiteaRepoManager:
logger.error(f"Exception in should_delete for repo {repo}: {e}", exc_info=True)
return False
def list_users(self, page: int = 1, limit: int = 50) -> List[Dict[str, Any]]:
"""
List all users in the Gitea instance.
Args:
page (int): Page number for pagination.
limit (int): Number of users per page.
Returns:
List[Dict[str, Any]]: List of user data dictionaries.
"""
url = f"{self.api_url}/admin/users?page={page}&limit={limit}"
all_users = []
try:
response = requests.get(url, headers=self.headers)
if response.status_code == 200:
users = response.json()
if users:
all_users.extend(users)
logger.info(f"Successfully listed {len(users)} users from page {page}.")
# If we got a full page, there might be more users
if len(users) == limit:
all_users.extend(self.list_users(page + 1, limit))
else:
logger.info(f"No more users found on page {page}.")
elif response.status_code == 403:
logger.error("Access denied. Admin privileges required to list users.")
else:
logger.warning(f"Error listing users: {response.status_code} - {response.text}")
except requests.RequestException as e:
logger.error(f"Exception during listing users: {e}", exc_info=True)
return all_users
def delete_user(self, username: str, purge: bool = True, dry_run: bool = True) -> bool:
"""
Delete a user by username.
Args:
username (str): Username to delete.
purge (bool): If True, completely purge user data.
dry_run (bool): If True, only logs the intended deletion without executing.
Returns:
bool: True if deletion was successful, False otherwise.
"""
purge_param = "?purge=true" if purge else ""
url = f"{self.api_url}/admin/users/{username}{purge_param}"
if dry_run:
logger.info(f"Would delete user: {username} (purge: {purge})")
return True
try:
response = requests.delete(url, headers=self.headers)
if response.status_code == 204:
logger.info(f"Successfully deleted user: {username}")
return True
elif response.status_code == 403:
logger.error(f"Access denied when deleting user {username}. Admin privileges required.")
elif response.status_code == 404:
logger.warning(f"User {username} not found.")
else:
logger.warning(f"Error deleting user {username}: {response.status_code} - {response.text}")
except requests.RequestException as e:
logger.error(f"Exception during deleting user {username}: {e}", exc_info=True)
return False
def filter_users_by_id(self, users: List[Dict[str, Any]], min_id: int) -> List[Dict[str, Any]]:
"""
Filter users with ID above the specified threshold.
Args:
users (List[Dict[str, Any]]): List of user dictionaries.
min_id (int): Minimum ID threshold (exclusive).
Returns:
List[Dict[str, Any]]: Filtered list of users with ID > min_id.
"""
filtered_users = [user for user in users if user.get('id', 0) > min_id]
logger.info(f"Filtered {len(filtered_users)} users with ID > {min_id} from {len(users)} total users.")
return filtered_users
def delete_users_above_id(self, min_id: int = 18, purge: bool = False, dry_run: bool = False) -> Dict[str, Any]:
"""
Delete all users with ID above the specified threshold.
Args:
min_id (int): Minimum ID threshold (exclusive). Default is 18.
purge (bool): If True, completely purge user data.
dry_run (bool): If True, only logs actions without executing deletions.
Returns:
Dict[str, Any]: Summary of the operation including counts and failed deletions.
"""
logger.info(f"Starting user deletion process for users with ID > {min_id} (dry_run: {dry_run})")
# Get all users
all_users = self.list_users()
if not all_users:
logger.warning("No users found or error occurred while listing users.")
return {"total_users": 0, "filtered_users": 0, "deleted_count": 0, "failed_deletions": []}
# Filter users by ID
users_to_delete = self.filter_users_by_id(all_users, min_id)
if not users_to_delete:
logger.info(f"No users found with ID > {min_id}")
return {"total_users": len(all_users), "filtered_users": 0, "deleted_count": 0, "failed_deletions": []}
# Delete filtered users
deleted_count = 0
failed_deletions = []
for user in users_to_delete:
username = user.get('login', user.get('username', 'unknown'))
user_id = user.get('id', 'unknown')
logger.info(f"Processing user: {username} (ID: {user_id})")
if self.delete_user(username, purge=purge, dry_run=dry_run):
deleted_count += 1
else:
failed_deletions.append({"username": username, "id": user_id})
result = {
"total_users": len(all_users),
"filtered_users": len(users_to_delete),
"deleted_count": deleted_count,
"failed_deletions": failed_deletions
}
logger.info(f"User deletion process completed. Deleted: {deleted_count}, Failed: {len(failed_deletions)}")
return result
def manage_repositories(self, dry_run: bool = False) -> List[Dict[str, Any]]:
"""
List repositories and delete those that meet deletion criteria.
@ -313,7 +174,6 @@ class GiteaRepoManager:
repo['pending_delete'] = self.should_delete(repo)
if repo['pending_delete']:
self.delete_repository(repo['owner']['username'], repo['name'], dry_run=dry_run)
self.delete_user(repo['owner']['username'], purge=True, dry_run=dry_run)
except Exception as e:
logger.error(f"Exception managing repository {repo.get('name', 'unknown')}: {e}", exc_info=True)
return repos

3
main.py
View File

@ -1,5 +1,3 @@
import functools
print = functools.partial(print,flush=True)
import read_env
import os
import argparse
@ -92,7 +90,6 @@ def main() -> None:
# Manage repositories with dry_run as specified
manager.manage_repositories(dry_run=args.dry_run)
#manager.delete_users_above_id(1337, not args.dry_run, args.dry_run)
if __name__ == '__main__':
main()