DengQN·一个普通程序员;
Gitea to Gitea 仓库同步脚本
2023-09-17 16:30 88
#gitea#python#维护

装完新的环境发现官方的同步方案会有各种坑跑不起来,索性直接用gitea自带的api来做。

gitea自带接口可以在swagger页面查看

https://host/api/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()