fgh/client.py

#!/usr/bin/env python
# -*- encoding: utf-8 -*-
"""
@Contact :   liuyuqi.gov@msn.cn
@Time    :   2024/04/09 12:56:07
@License :   Copyright © 2017-2022 liuyuqi. All Rights Reserved.
@Desc    :   client enter point
"""

import os,sys,re
import argparse
from fgh.utils.frozen_dir import get_app_path
import requests, subprocess, platform
import dotenv

class Client:

    help_str ="""
    fgh(fast github) quick clone github repo tools
        contact: liuyuqi.gov@msn.cn

Usage: fgh <command> [<args>]

# clone repo
fgh clone https://github.com/xx/yy.git
fgh push https://github.com/xx/yy.git

# download file
fgh wget https://ghproxy.org/https://github.com/microsoft/vscode/archive/refs/tags/1.84.2.zip
fgh wget https://ghproxy.org/https://raw.githubusercontent.com/microsoft/vscode/main/README.md

fgh curl -O https://ghproxy.org/https://github.com/microsoft/vscode/archive/refs/tags/1.84.2.zip
fgh curl -O https://ghproxy.org/https://raw.githubusercontent.com/microsoft/vscode/main/README.md


"""

    def __init__(self):
        self.fgit_host = ''
        self.token = ''
        self.sess = requests.Session()
        self.env_file_path = os.path.join(get_app_path(), '.env')
        self.update_host()
        self.read_config()
        self.args = self.parse_args()
        
    def read_config(self):
        """ read config """
        dotenv.load_dotenv(self.env_file_path)
        fgit_host = os.getenv('FGH_HOST')
        print("当前host: " + fgit_host)
        try:
            if fgit_host[-1] == '/':
                self.fgit_host = fgit_host[:-1]
            self.token = os.getenv('FGh_TOKEN')
        except Exception as e:
            print(".env配置错误"+e)

    @staticmethod
    def ping(host: str) -> None|int:
        """ ping host """
        if host.startswith('https://') or host.startswith('http://'):
            host = host.split('://')[1]
        host = host.split(':')[0]
        host = host.split('/')[0]
        try:
            if platform.system().lower() == 'windows':
                command = ['ping', '-n', '1', '-w', '1000', host]
            else:
                command = ['ping', '-c', '1', '-W', '1', host]
        
            output = subprocess.run(command, capture_output=True, text=True)

            # 从输出中提取延迟信息
            if output.returncode == 0:
                # 使用正则表达式从输出中匹配延迟时间
                pattern = r"time=(\d+)ms"
                match = re.search(pattern, output.stdout)
                if match:
                    return int(match.group(1))  # 返回延迟时间（毫秒）
            return None  # 如果 ping 失败或延迟时间无法提取，则返回 None
        except Exception as e:
            print(e)
        return None

    @staticmethod
    def parse_args():
        """ parse args """
        print("解析参数...")
        parser = argparse.ArgumentParser(description='fgit client')
        parser.add_argument('command', type=str, help='fgit command', 
                            choices=['git','wget','curl','clone', 'push', 'pull', 'commit', 'add', 'status', 'log', 'diff', 'branch', 'checkout', 'merge', 'rebase', 'reset', 'tag', 'fetch', 'remote', 'init', 'config', 'help'])
        # all args behind command
        parser.add_argument('args', nargs=argparse.REMAINDER, help='git command args')
        return parser.parse_args()

    def update_host(self):
        """update host"""
        print("正在更新host...")
        res = []
        try:
            res = self.sess.get('https://git.yoqi.me/lyq/fgh/raw/master/docs/fgh_host.json')
            fgh_hosts= res.json()["hosts"]
            fgh_host= self._get_best_host(fgh_hosts)
            dotenv.set_key(self.env_file_path, 'FGH_HOST', fgh_host)
        except Exception as e:
            print(e)
        
    def _get_best_host(self, hosts: list) -> str:
        """ 在所有 hosts 中，ping 延迟最短的返回 """
        print("即将检测网络延迟...")
        res =[]
        for host in hosts:
            res.append({'host': host, 'delay': self.ping(host) if self.ping(host) else 999999})
            print("正在ping "+ host + " 延迟为：" + str(res[-1]['delay']) + "ms")
        min_delay = 999999
        min_host = res[0]['host']
        for item in res:
            if item['delay'] and item['delay'] < min_delay:
                min_delay = item['delay']
                min_host = item['host']
        print("选择最佳host: " + min_host + " 延迟为：" + str(min_delay) + "ms")
        return min_host
    
    def run(self):
        """ run """
        args = self.parse_args()
        command = args.command
        if len(args.args) == 0 and command != 'help':
            print('Usage: fgh <command> [<args>]')
            sys.exit(1)
        if command == "wget":
            if len(args.args) > 0:
                args.args = [re.sub(r'https://github.com', self.fgit_host+'/https://github.com', arg) for arg in args.args]
                # print('wget ' + ' '.join(args.args))
                os.system('wget ' + ' '.join(args.args))
        elif command =="curl":
            if len(args.args) > 0:
                args.args = [re.sub(r'https://github.com', self.fgit_host+'/https://github.com', arg) for arg in args.args]
                print('curl ' + ' '.join(args.args))
                os.system('curl ' + ' '.join(args.args))
        elif command == "help":
            print(Client.help_str)
        else:
            if len(args.args) > 0:
                args.args = [re.sub(r'https://github.com', self.fgit_host+'/https://github.com', arg) for arg in args.args]
                cmd_str = 'git ' + command + ' ' + ' '.join(args.args)
                try:
                    if command == 'git':
                        cmd_str = command + ' ' + ' '.join(args.args)
                    os.system(cmd_str)
                    # print(cmd_str)
                except Exception as e:
                    print(cmd_str + e)

if __name__=='__main__':
    client = Client()
    client.run()