装完新的环境发现官方的同步方案会有各种坑跑不起来,索性直接用gitea自带的api来做。
gitea自带接口可以在swagger页面查看
import requests
import threading
dest_host = "https://x.com"
dest_token = "xx"
source_host = "https://xxxx.com"
source_token = "yy"
source_username = "u"
source_password = "p"
def job(r):
print(r['name'])
json = {}
json["auth_username"] = source_username
json["auth_password"] = source_password
json["access_token"] = source_token
json["clone_addr"] = r['clone_addr']
json["repo_name"] = r["name"]
json["mirror"] = False
json["private"] = True
json["uid"] = 1
migrate_url = "%s/api/v1/repos/migrate?access_token=%s" % (dest_host, dest_token)
resp = requests.post(url=migrate_url, json=json)
print(resp.status_code, resp.text)
for i in range(5):
repo_list_page = "%s/api/v1/repos/search?access_token=%s&limit%d&page=%d" % (source_host, source_token, 50, i)
resp = requests.get(url=repo_list_page)
if resp.status_code != 200:
break
repos = [{'id': s['id'], 'clone_addr': s['clone_url'], 'name': s['name']} for s in resp.json()['data']]
if len(repos) == 0 :
break
print(len(repos))
threads = [threading.Thread(target=job, args=[r]) for r in repos]
for t in threads:
t.start()
for t in threads:
t.join()