DeleteGiteeRepoInBatch.py

DeleteGiteeRepoInBatch.py

import requests
import datetime
import threading

access_token = "xxxx"


def check_latest_commit_date(item_list):
    # 获取当前日期
    current_date = datetime.datetime.now(datetime.timezone.utc)
    # 获取最新提交的日期
    latest_commit_date = datetime.datetime.strptime(
        item_list[0]["commit"]["author"]["date"], "%Y-%m-%dT%H:%M:%S%z"
    )
    # 计算日期差
    delta = current_date - latest_commit_date
    # 如果最新提交超过2年（730天），打印消息
    if delta.days > 730:
        # print("The latest commit is more than 2 years ago.")
        return True
    else:
        return False


# 获取仓库列表
def get_repository_list():
    repositories = []
    for i in range(1, 3):
        headers = {"Authorization": f"Bearer {access_token}"}
        params = {
            "access_token": access_token,
            "page": i,
            "per_page": 100,
        }
        response = requests.get(
            "https://gitee.com/api/v5/user/repos", headers=headers, params=params
        )
        repositories += response.json()
    return repositories


def sort_by_date(item_list):
    return sorted(
        item_list,
        key=lambda item: datetime.datetime.strptime(
            item["commit"]["author"]["date"], "%Y-%m-%dT%H:%M:%S%z"
        ),
        reverse=True,
    )


def delete_repo(repo):
    url = f"https://gitee.com/api/v5/repos/lovelife_lizhe/{repo}"
    headers = {
        "Content-Type": "application/json;charset=UTF-8",
    }
    params = {
        "access_token": access_token,
    }

    response = requests.delete(url, headers=headers, params=params)

    if response.status_code == 204:
        print("Repository deleted successfully.")
    else:
        print(
            f"Failed to delete repository. Status code: {response.status_code}, Response: {response.text}"
        )


def get_repository_commits_date(repo):
    if "setting" in repo.lower():
        return

    url = f"https://gitee.com/api/v5/repos/lovelife_lizhe/{repo}/commits"
    headers = {
        "Content-Type": "application/json;charset=UTF-8",
    }
    params = {
        "access_token": access_token,
        "page": 1,
        "per_page": 20,
    }
    response = requests.get(url, headers=headers, params=params)
    item_list = response.json()
    # print(repo, item_list)
    if "message" in item_list and item_list["message"] == "Repository is empty":
        delete_repo(repo)
    else:
        item_list = sort_by_date(item_list)
        date_list = [item["commit"]["author"]["date"] for item in item_list]
        if check_latest_commit_date(item_list):
            print(f"Repo: {repo}, Dates: {date_list[0]}")
            delete_repo(repo)
        # else:
        # print(f"Repo: {repo} You are still there.")


def main():
    repository_list = get_repository_list()
    print(len(repository_list))
    threads = []
    for repo in repository_list:  # 限制为前20个仓库，以避免请求过多
        repo_name = repo["full_name"].split("/")[-1]
        # get_repository_commits_date(repo_name)
        thread = threading.Thread(target=get_repository_commits_date, args=(repo_name,))
        threads.append(thread)
        thread.start()

    for thread in threads:
        thread.join()


if __name__ == "__main__":
    main()