优化get-app,优先使用SSH,git全局用户名和密钥Token,支持私有库安装和公共库安装

This commit is contained in:
jingrow 2025-08-21 22:11:25 +08:00
parent 3919eb31df
commit 4940177a5b
2 changed files with 110 additions and 14 deletions

View File

@ -35,6 +35,10 @@ from bench.utils import (
log,
run_jingrow_cmd,
get_file_md5,
GIT_SSH_USER,
GIT_SSH_HOST,
GIT_SSH_PORT,
GIT_DOMAIN,
)
from bench.utils.bench import build_assets, install_python_dev_dependencies
from bench.utils.render import step
@ -64,10 +68,10 @@ class AppMeta:
dependencies = [{"jingrow/jerp": "~13.17.0"}]
"""
self.name = name.rstrip("/")
self.remote_server = "git.jingrow.com"
self.remote_server = GIT_DOMAIN
self.to_clone = to_clone
self.on_disk = False
self.use_ssh = False
self.use_ssh = True
self.from_apps = False
self.is_url = False
self.branch = branch
@ -160,7 +164,7 @@ class AppMeta:
return f"http://{self.remote_server}/{self.org}/{self.repo}.git"
def get_ssh_url(self):
return f"git@{self.remote_server}:{self.org}/{self.repo}.git"
return f"ssh://{GIT_SSH_USER}@{GIT_SSH_HOST}:{GIT_SSH_PORT}/{self.org}/{self.repo}.git"
@lru_cache(maxsize=None)

View File

@ -6,6 +6,8 @@ import re
import subprocess
import sys
import hashlib
import time
import requests
from functools import lru_cache
from glob import glob
from pathlib import Path
@ -29,6 +31,12 @@ from typing import TYPE_CHECKING
if TYPE_CHECKING:
from typing import Optional
# Git 服务器配置
GIT_DOMAIN = 'git.jingrow.com'
GIT_SSH_USER = 'gitea'
GIT_SSH_HOST = 'git01.jingrow.com'
GIT_SSH_PORT = 198
logger = logging.getLogger(PROJECT_NAME)
paths_in_app = ("hooks.py", "modules.txt", "patches.txt")
@ -430,29 +438,113 @@ def get_env_jingrow_commands(bench_path=".") -> List:
def find_org(org_repo, using_cached: bool = False):
import requests
org_repo = org_repo[0]
# 1. 检测 SSH 配置
ssh_available = _check_ssh_availability()
print(f"SSH detection result: {ssh_available}")
# 2. SSH 优先尝试
if ssh_available:
print("Trying SSH method...")
for org in ["jingrow", "jerp"]:
if _try_ssh_access(org, org_repo):
print(f"SSH successfully found: {org}/{org_repo}")
return org, org_repo
# 3. HTTP 回退
print("Falling back to HTTP method...")
username, password = _get_http_credentials()
for org in ["jingrow", "jerp"]:
url = f"http://git.jingrow.com/api/v1/repos/{org}/{org_repo}"
res = requests.get(url)
if res.status_code in (400, 403):
url2 = f"http://git.jingrow.com/{org}/{org_repo}"
res = requests.get(url2)
if res.ok:
if _try_http_access(org, org_repo, username, password):
print(f"HTTP successfully found: {org}/{org_repo}")
return org, org_repo
# 4. 失败处理
if using_cached:
return "", org_repo
raise InvalidRemoteException(
f"{org_repo} not found under jingrow or jerp Git accounts"
)
def _check_ssh_availability():
"""检测 SSH 是否可用"""
try:
# 直接测试 SSH 连接(使用配置的端口)
ssh_cmd = ['ssh', '-p', str(GIT_SSH_PORT), '-T', f'{GIT_SSH_USER}@{GIT_SSH_HOST}']
print(f"SSH connection test command: {' '.join(ssh_cmd)}")
result = subprocess.run(
ssh_cmd,
capture_output=True,
text=True,
timeout=10
)
print(f"SSH connection test: returncode={result.returncode}, stdout='{result.stdout}', stderr='{result.stderr}'")
# SSH 成功或认证失败都说明 SSH 可用returncode 0 或 1
ssh_available = result.returncode in [0, 1]
print(f"SSH availability: {ssh_available}")
return ssh_available
except Exception as e:
print(f"SSH detection exception: {e}")
return False
def _try_ssh_access(org, repo):
"""尝试 SSH 访问"""
try:
ssh_url = f"ssh://{GIT_SSH_USER}@{GIT_SSH_HOST}:{GIT_SSH_PORT}/{org}/{repo}.git"
result = subprocess.run(
['git', 'ls-remote', ssh_url],
capture_output=True,
text=True,
timeout=15
)
return result.returncode == 0
except:
return False
def _get_http_credentials():
"""获取 HTTP 凭据"""
username = os.getenv('GIT_USERNAME')
password = os.getenv('GIT_PASSWORD')
if not username or not password:
try:
git_config = subprocess.check_output(
['git', 'config', '--get', f'url.http://{GIT_DOMAIN}/.insteadOf'],
stderr=subprocess.DEVNULL
).decode('utf-8').strip()
if '@' in git_config:
auth_part = git_config.split('@')[0].split('//')[-1]
if ':' in auth_part:
username, password = auth_part.split(':', 1)
except:
pass
return username, password
def _try_http_access(org, repo, username, password):
"""尝试 HTTP 访问"""
try:
url = f"http://{GIT_DOMAIN}/api/v1/repos/{org}/{repo}"
res = requests.get(url, auth=(username, password) if username and password else None, timeout=10)
if res.status_code in (400, 403):
url2 = f"http://{GIT_DOMAIN}/{org}/{repo}"
res = requests.get(url2, auth=(username, password) if username and password else None, timeout=10)
return res.ok
except:
return False
def fetch_details_from_tag(
_tag: str, using_cached: bool = False
) -> Tuple[str, str, str]: