flask|flask web 实现 harbor 早期版本的镜像 tag 批量可视化删除 flaskweb实现harbor早期版本的镜像

功能展示

代码展示

app.py

#! /usr/bin/env python3 # -*- coding:utf-8 -*- from flask import Flask, request, render_template, redirect, url_for, make_response ,escape, session, flash, g, current_app, abort, jsonify, send_file import requests #from werkzeug.utils import secure_filename import os,time, platform from flask_wtf import Form #from wtforms import TextField #from shell import os_scrpict, harbor #import sqlite3 #from pathlib import Path from flask_sqlalchemy import SQLAlchemyapp = Flask(__name__)# 本脚本只是harbor页面tags清理 # cd /usr/local/src/harbor && docker-compose stop#关闭harbor # 物理清理：docker run -it --name gc --rm --volumes-from registry goharbor/registry-photon:v2.6.2-v1.6.3 garbage-collect /etc/registry/config.yml#docker-compose up -d#启动class RequestClient(object):def __init__(self, login_url, username, password): self.username = username self.password = password self.login_url = login_url self.session = requests.Session() self.login()def login(self): requests.packages.urllib3.disable_warnings() self.session.post(self.login_url, params={"principal": self.username, "password": self.password}, verify=False)class ClearHarbor(object):def __init__(self, harbor_domain, username, password, num, schema="http", ): self.num = num self.schema = schema self.harbor_domain = harbor_domain self.harbor_url = self.schema + "://" + self.harbor_domain self.login_url = self.harbor_url + "/login" self.api_url = self.harbor_url + "/api" self.pro_url = self.api_url + "/projects" self.repos_url = self.api_url + "/repositories" self.username = username self.password = password self.client = RequestClient(self.login_url, self.username, self.password)def __fetch_pros_obj(self): # 获取所有项目名称 self.pros_obj = self.client.session.get(self.pro_url).json() # for n in self.pros_obj: #print("分组：",n.get("name")) print(self.pros_obj) return self.pros_objdef get_pros_id(self): # 获取所有项目ID self.pros_id = [] pro_res = self.__fetch_pros_obj() for i in pro_res: self.pros_id.append(i['project_id']) # print("所有项目ID：",self.pros_id) return self.pros_iddef get_del_repos_name(self, pro_id): # 镜像tag数量大于self.num的镜像仓库名称 self.del_repos_name = [] repos_res = self.client.session.get(self.repos_url, params={"project_id": pro_id}) # print("项目信息：",repos_res.json()) for repo in repos_res.json(): if repo["tags_count"] > self.num: # print("镜像仓库名称：",repo['name']) self.del_repos_name.append(repo['name']) return self.del_repos_namedef fetch_del_repos(self, repo_name): # 删除镜像仓库tag self.del_res = [] tag_url = self.repos_url + "/" + repo_name + "/tags" # 项目镜像仓库的所有tags,按创建时间排序 tags = self.client.session.get(tag_url).json() tags_sort = sorted(tags, key=lambda a: a["created"]) # print(len(tags_sort),tags_sort) # 除了最新的self.num个，其他的tag都添加到待删除列表del_tags del_tags = tags_sort[0:len(tags_sort) - self.num] # print(del_tags) for tag in del_tags: del_repo_tag_url = tag_url + "/" + tag['name'] # print(del_repo_tag_url) del_res = self.client.session.delete(del_repo_tag_url) self.del_res.append("镜像: %s 删除状态: %s" % (del_repo_tag_url,del_res)) return self.del_resdef work(self): # 遍历project id for i in self.get_pros_id(): # 获取所有tag超过self.num的repos repos = self.get_del_repos_name(i) if repos: for repo in repos: del_repos = self.fetch_del_repos(repo) print(del_repos) print(repo)def list_repositories(self, pro_id): # 镜像tag数量大于self.num的镜像仓库名称 self.del_repos_name1 = [] repos_res = self.client.session.get(self.repos_url, params={"project_id": pro_id}) # print("项目信息：",repos_res.json()) for repo in repos_res.json(): #if repo["tags_count"] > self.num: #print(repo,"镜像仓库名称：",repo['name']) #self.del_repos_name1.append({"name": repo["name"], "tags_count": repo["tags_count"]}) self.del_repos_name1.append({"name": repo["name"], "tags_count": repo["tags_count"],"project_id": repo["project_id"]}) #print(self.del_repos_name1) return self.del_repos_name1def list_pros(self): # 获取所有项目ID self.pros_idname = [] pro_res = self.__fetch_pros_obj() for i in pro_res: self.pros_idname.append({"project_id": i["project_id"], "name": i["name"], "repo_count": i["repo_count"] }) #print(self.pros_idname) #print(self.pros_idname[0]['name']) return self.pros_idnamedef work_list(self, pros_id_list): # 遍历project id for i in pros_id_list: # 获取所有tag超过self.num的repos repos = self.get_del_repos_name(i) if repos: for repo in repos: # del_repos = self.fetch_del_repos(repo) # print(del_repos) print(repo)@app.route('/harbor/', methods = ['GET', 'POST']) def harbor_list(): del_images_tag = ClearHarbor(harbor_domain="172.16.167.11", username="admin", password="Harbor12345", num=10) harbor_project = del_images_tag.list_pros() print("###########",harbor_project)return render_template('harbor.html', harbor_project1 = harbor_project)@app.route('/del_harbor_project_all/') def del_harbor_project_all1(): del_images_tag = ClearHarbor(harbor_domain="172.16.167.11", username="admin", password="Harbor12345", num=10) del_images_tag.work()return redirect('/harbor/')@app.route('/del_harbor_project/') def del_harbor_project(project_id): del_images_tag = ClearHarbor(harbor_domain="172.16.167.11", username="admin", password="Harbor12345", num=10) del_reposname = del_images_tag.get_del_repos_name(project_id) print(del_reposname,"#####################",project_id) del_reposname_status = del_images_tag.fetch_del_repos(del_reposname) print(del_reposname_status)return redirect('/harbor/')@app.route('/del_harbor_project_list/', methods = ['GET', 'POST']) def del_harbor_project_list(): if request.method == 'POST': project_id_list = request.form.getlist('vals') #print(project_id_list)vals_list2 = [] for i in project_id_list: if i == '': print("i is none") else: vals_list2.append(i)del_images_tag = ClearHarbor(harbor_domain="172.16.167.11", username="admin", password="Harbor12345", num=10) del_images_tag.work_list(vals_list2)return redirect('/harbor/')@app.route('/harbor_repositories/') def harbor_repositoriest(project_id): del_images_tag = ClearHarbor(harbor_domain="192.168.1.17", username="admin", password="xxxxxx", num=10) harbor_project = del_images_tag.list_repositories(project_id) print("###########",harbor_project)return render_template('harbor_repositories.html', harbor_project1 = harbor_project)@app.route('/del_repositories_url//') def del_repositories_url(project_id,project_name): del_images_tag = ClearHarbor(harbor_domain="192.168.1.17", username="admin", password="xxxxxx", num=10) del_reposname_status = del_images_tag.fetch_del_repos(project_name) print(del_reposname_status)return redirect('/harbor_repositories/{tproject_id}'.format(tproject_id=project_id))if __name__ == '__main__': app.run() #app.run(host="172.16.117.33",port=5888)

harbor.html

Titleharbor 批量清除选中项目images 清除all project >10 tag images {##} {#text： {{ conflook.conflook }} #} {##} {% for ele in harbor_project1 %} {% endfor %} ID 项目镜像仓库数操作 {{ ele.project_id }} {{ ele.name }} {{ ele.repo_count }} 清除

	ID	项目	镜像仓库数	操作
	{{ ele.project_id }}	{{ ele.name }}	{{ ele.repo_count }}	清除

harbor_repositories.html

Titleharbor{##} {#text： {{ conflook.conflook }} #} {##} {% for ele in harbor_project1 %} {% endfor %} 项目标签数操作 {{ ele.name }} {{ ele.tags_count }} 清除

	项目	标签数	操作
	{{ ele.name }}	{{ ele.tags_count }}	清除

python 脚本

harbor_del_tag.py

#! /usr/bin/env python3 # -*- coding:utf-8 -*- # 本脚本只是harbor页面tags清理 # 物理清理：docker run -it --name gc --rm --volumes-from registry goharbor/registry-photon:v2.6.2-v1.6.3 garbage-collect /etc/registry/config.yml import requestsclass RequestClient(object):def __init__(self, login_url, username, password): self.username = username self.password = password self.login_url = login_url self.session = requests.Session() self.login()def login(self): requests.packages.urllib3.disable_warnings() self.session.post(self.login_url, params={"principal": self.username, "password": self.password}, verify=False)class ClearHarbor(object):def __init__(self, harbor_domain, username, password, num, schema="http", ): self.num = num self.schema = schema self.harbor_domain = harbor_domain self.harbor_url = self.schema + "://" + self.harbor_domain self.login_url = self.harbor_url + "/login" self.api_url = self.harbor_url + "/api" self.pro_url = self.api_url + "/projects" self.repos_url = self.api_url + "/repositories" self.username = username self.password = password self.client = RequestClient(self.login_url, self.username, self.password)def __fetch_pros_obj(self): # 获取所有项目名称 self.pros_obj = self.client.session.get(self.pro_url).json() # for n in self.pros_obj: #print("分组：",n.get("name")) print(self.pros_obj) return self.pros_objdef get_pros_id(self): # 获取所有项目ID self.pros_id = [] pro_res = self.__fetch_pros_obj() for i in pro_res: self.pros_id.append(i['project_id']) # print("所有项目ID：",self.pros_id) return self.pros_iddef get_del_repos_name(self, pro_id): # 镜像tag数量大于self.num的镜像仓库名称 self.del_repos_name = [] repos_res = self.client.session.get(self.repos_url, params={"project_id": pro_id}) # print("项目信息：",repos_res.json()) for repo in repos_res.json(): if repo["tags_count"] > self.num: # print("镜像仓库名称：",repo['name']) self.del_repos_name.append(repo['name']) return self.del_repos_namedef fetch_del_repos(self, repo_name): # 删除镜像仓库tag self.del_res = [] tag_url = self.repos_url + "/" + repo_name + "/tags" # 项目镜像仓库的所有tags,按创建时间排序 tags = self.client.session.get(tag_url).json() tags_sort = sorted(tags, key=lambda a: a["created"]) # print(len(tags_sort),tags_sort) # 除了最新的self.num个，其他的tag都添加到待删除列表del_tags del_tags = tags_sort[0:len(tags_sort) - self.num] # print(del_tags) for tag in del_tags: del_repo_tag_url = tag_url + "/" + tag['name'] # print(del_repo_tag_url) del_res = self.client.session.delete(del_repo_tag_url) self.del_res.append("镜像: %s 删除状态: %s" % (del_repo_tag_url,del_res)) return self.del_resdef work(self): # 遍历project id for i in self.get_pros_id(): # 获取所有tag超过self.num的repos repos = self.get_del_repos_name(i) if repos: for repo in repos: del_repos = self.fetch_del_repos(repo) print(del_repos) print(repo)def list_repositories(self, pro_id): # 镜像tag数量大于self.num的镜像仓库名称 self.del_repos_name1 = [] repos_res = self.client.session.get(self.repos_url, params={"project_id": pro_id}) # print("项目信息：",repos_res.json()) for repo in repos_res.json(): #if repo["tags_count"] > self.num: #print(repo,"镜像仓库名称：",repo['name']) #self.del_repos_name1.append({"name": repo["name"], "tags_count": repo["tags_count"]}) self.del_repos_name1.append({"name": repo["name"], "tags_count": repo["tags_count"],"project_id": repo["project_id"]}) #print(self.del_repos_name1) return self.del_repos_name1def list_pros(self): # 获取所有项目ID self.pros_idname = [] pro_res = self.__fetch_pros_obj() for i in pro_res: self.pros_idname.append({"project_id": i["project_id"], "name": i["name"], "repo_count": i["repo_count"] }) #print(self.pros_idname) #print(self.pros_idname[0]['name']) return self.pros_idnamedef work_list(self, pros_id_list): # 遍历project id for i in pros_id_list: # 获取所有tag超过self.num的repos repos = self.get_del_repos_name(i) if repos: for repo in repos: # del_repos = self.fetch_del_repos(repo) # print(del_repos) print(repo)if __name__ == "__main__": clean = ClearHarbor(harbor_domain="192.168.167.17", username="admin", password="xxxxxxxx", num=6) clean.work()

harbor_del_tag.sh

#!/bin/sh ############################################################# # File Name: harbor_del_tag.sh # Author: Sean_Li # Created Time: 20210427for crontab use #==================================================================#PATH='/root/.sdkman/candidates/groovy/current/bin:/usr/local/jdk825x64/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/root/bin' echo "1.清除 tag" python3 /root/python/harbor_del_tag.pysleep 3echo "2. stop harbor 物理删除 tag"ansible 192.168.167.17 -m raw -a "cd /usr/local/src/harbor && docker-compose stop"sleep 3ansible 192.168.167.17 -m raw -a "docker run -it --name gc --rm --volumes-from registry harbor.reg/vmware/registry-photon:v2.6.2-v1.5.0 garbage-collect /etc/registry/config.yml"sleep 3echo "3. 启动 harbor"ansible 192.168.167.17 -m raw -a "cd /usr/local/src/harbor && docker-compose up -d"

【flask|flask web 实现 harbor 早期版本的镜像 tag 批量可视化删除】