feat(fgit): 添加下载 ZIP 功能并优化日志输出

- 新增 download-zip 命令，支持下载仓库的 ZIP 文件
- 优化日志输出，增加详细信息和错误处理
- 更新 README 文档

Author: NaivG

.github/workflows/build.yml

@@ -74,6 +74,6 @@ jobs:
     - name: Upload Artifact
       uses: actions/upload-artifact@v4.6.2
       with:
-        name: my_script-${{ matrix.os }}
+        name: fgit-${{ matrix.os }}
         path: dist/fgit_*
         if-no-files-found: error

README.md

@@ -29,7 +29,9 @@
 **完全无感化**，和常规的git命令行相同，支持各种命令行参数，也就是说，平时git命令行怎么用，fgit就怎么用，区别只是git换成了fgit。
 
 - **镜像加速**  
-  自动测试多个Git镜像源延迟，选择最快的源进行克隆/拉取（支持`clone`/`pull`/`push`/`fetch`）。
+  自动测试多个Git镜像源延迟，选择最快的源进行克隆/拉取(支持`clone`/`pull`/`push`/`fetch`)。
+- **下载文件**
+  支持直接下载仓库压缩包(TODO: 支持下载release文件)。
 - **代理支持**  
   可通过命令行参数或配置文件设置HTTP/HTTPS代理。
 - **智能缓存**  
@@ -45,7 +47,7 @@
 
 ### 直接使用
 
-1. 下载编译后的可执行文件：[releases](https://github.com/NaivG/fgit/releases)
+1. 下载编译后的可执行文件：[releases](https://github.com/NaivG/fastgit/releases)(稳定版)  / [GitHub Actions](https://github.com/NaivG/fastgit/actions)(最新版)
 2. 将可执行文件加入系统路径环境变量，或是移动到PATH目录下，如`C:\Windows\System32`
 3. 打开命令行，输入`fgit`命令，即可使用加速版git命令。
 
@@ -82,6 +84,10 @@ fgit clone <user>/<repo> # fgit的特色方式，会自动转换为https://githu
 # 拉取仓库（在失败时自动选择镜像源）
 fgit pull
 
+# 下载仓库压缩包（自动选择镜像源）
+fgit download-zip <仓库URL>
+fgit download-zip <user>/<repo>
+
 # 启用代理
 fgit --use-proxy http://127.0.0.1:7890 clone <仓库URL>
 fgit --use-proxy http://127.0.0.1:7890 push
@@ -118,4 +124,4 @@ fgit --verbose clone <仓库URL>
 
 Star History：
 
-[![Star History Chart](https://api.star-history.com/svg?repos=NaivG/fgit&type=Date)](https://star-history.com/#NaivG/fgit&Date)
+[![Star History Chart](https://api.star-history.com/svg?repos=NaivG/fastgit&type=Date)](https://star-history.com/#NaivG/fastgit&Date)

config.py

@@ -37,6 +37,15 @@ def save_proxy(self, proxy):
             self.config.set('proxy', k, v)
         self._save()
 
+    def get_downloader_config(self):
+        if self.config.has_section('downloader'):
+            return dict(self.config.items('downloader'))
+        self.config.add_section('downloader')
+        self.config.set('downloader', 'chunk_size', '1024')
+        self.config.set('downloader', 'MIN_FILE_SIZE', '1')
+        self._save()
+        return dict(self.config.items('downloader'))
+
     def _save(self):
         with open(self.config_file, 'w') as f:
             self.config.write(f)

downloader.py

@@ -0,0 +1,22 @@
+from tqdm import tqdm
+import requests
+from loguru import logger
+
+def download_file(url: str, file_path: str, chunk_size: int = 1024, MIN_FILE_SIZE: int = 1) -> bool:
+    try:
+        response = requests.get(url, stream=True)
+        total_size = int(response.headers.get('content-length', 0))
+        logger.info(f"下载文件: {file_path} ({total_size/1024:.1f}KB)")
+        if total_size < MIN_FILE_SIZE:
+            logger.debug(f"镜像源错误: 文件大小小于{MIN_FILE_SIZE}字节")
+            return False
+
+        with open(file_path, 'wb') as f:
+            with tqdm(total=total_size, unit='B', unit_scale=True, desc="下载文件") as pbar:
+                for data in response.iter_content(chunk_size=chunk_size):
+                    f.write(data)
+                    pbar.update(len(data))
+        return True
+    except Exception as e:
+        logger.debug(f"下载失败: {str(e)}")
+        return False

fgit.py

@@ -4,20 +4,27 @@
 import argparse
 import os
 import json
-import traceback
 import threading
-from urllib.parse import urlparse
+from loguru import logger
 from urllib.request import urlopen, Request
 from urllib.error import HTTPError
 from colorama import Fore, Style, init
-from tempfile import NamedTemporaryFile
-import configparser
 from config import ConfigHandler
-from mirrors import test_latency, select_mirror, MIRRORS, convert_url
+from mirrors import select_mirror, convert_url
+from downloader import download_file
 from proxy import ProxyHandler
 
 init(autoreset=True)
 
+parser = argparse.ArgumentParser(description='Git加速工具，支持镜像源和代理')
+parser.add_argument('command', type=str, help='git命令, 或是fgit命令')
+parser.add_argument('--use-proxy', type=str, help='设置HTTP代理（格式: http://[user:pass@]host:port）')
+parser.add_argument('--verbose', action='store_true', help='显示详细输出')
+args, unknown_args = parser.parse_known_args()
+
+logger.remove()
+logger.add(sys.stderr, level='DEBUG', colorize=True, format='{time:HH:mm:ss} | {level} | {message}') if args.verbose else logger.add(sys.stderr, level='INFO', colorize=True, format='{time:HH:mm:ss} | {level} | {message}')
+
 GIT_COMMANDS_NEED_MIRROR = {'clone', 'pull', 'push', 'fetch'}
 
 headers = {'User-Agent': 'Mozilla/5.0',
@@ -27,22 +34,21 @@
 
 def main():
     config = ConfigHandler()
-    parser = argparse.ArgumentParser(description='Git加速工具，支持镜像源和代理')
-    parser.add_argument('command', type=str, help='git命令')
-    parser.add_argument('--use-proxy', type=str, help='设置HTTP代理（格式: http://[user:pass@]host:port）')
-    parser.add_argument('--verbose', action='store_true', help='显示详细输出')
-    args, unknown_args = parser.parse_known_args()
 
     proxy = ProxyHandler(args.use_proxy, config, args.verbose)
     env = proxy.setup_proxy_env()
 
-    if args.verbose:
-        if proxy.proxy_url:
-            print(Fore.CYAN + f"🔧 运行于代理模式" + Style.RESET_ALL)
-        else:
-            print(Fore.CYAN + f"🔧 运行于镜像模式" + Style.RESET_ALL)
-        print(Fore.CYAN + f"命令参数: {' '.join(sys.argv)}" + Style.RESET_ALL)
+    if proxy.proxy_url:
+        logger.debug(Fore.CYAN + "🔧 运行于代理模式" + Style.RESET_ALL)
+    else:
+        logger.debug(Fore.CYAN + "🔧 运行于镜像模式" + Style.RESET_ALL)
+    logger.debug(Fore.CYAN + f"命令参数: {' '.join(sys.argv)}" + Style.RESET_ALL)
 
+
+    if args.command == 'download-zip':
+        handle_download_zip(args, unknown_args, config, env, args.verbose)
+        return
+    
     if args.command not in GIT_COMMANDS_NEED_MIRROR:
         subprocess.run(['git'] + sys.argv[1:], env=env)
         return
@@ -55,6 +61,41 @@ def main():
     finally:
         proxy.restore_proxy_settings()
 
+def handle_download_zip(args, unknown_args, config, env, verbose):
+    downloader_config = config.get_downloader_config()
+    if not downloader_config:
+        logger.warning(Fore.YELLOW + "🧐 下载配置不存在, 使用默认配置" + Style.RESET_ALL)
+    chunk_size = downloader_config.get('chunk_size', 1024)
+    MIN_FILE_SIZE = downloader_config.get('min_file_size', 1)
+    
+    original_url = unknown_args[0]
+    if '://' not in original_url and '/' in original_url:
+        if '@' in original_url:  # SSH格式
+            original_url = f"https://{original_url.split('@')[1].replace(':', '/', 1)}"
+        else:  # 简写格式
+            original_url = f"https://github.com/{original_url}"
+
+    original_url = original_url.split('.git')[0]
+    repo_name = original_url.split('/')[-1].split('.git')[0]
+    if os.path.exists(os.path.join(os.getcwd(), repo_name + '.zip')):
+        logger.warning(Fore.YELLOW + f"😪 压缩包 {repo_name}.zip 已存在" + Style.RESET_ALL)
+        return
+    
+    repo_status = get_repo(original_url)
+    if repo_status is None:
+        logger.warning(Fore.YELLOW + "🧐 无法获取到仓库信息, 尝试下载" + Style.RESET_ALL)
+    elif repo_status is False and not input_with_timeout(Fore.YELLOW + "🧐 仓库可能不存在，5秒内按任意键忽略..." + Style.RESET_ALL, 5):
+        return
+        
+    mirror_list = select_mirror(config, args.verbose)
+    for mirror in mirror_list:
+        new_url = convert_url(original_url, mirror) + '/archive/refs/heads/main.zip'
+        logger.info(Fore.GREEN + f"🔄 尝试镜像源 {mirror} [{mirror_list.index(mirror) + 1}/{len(mirror_list)}]: {new_url}" + Style.RESET_ALL)
+        if download_file(new_url, os.path.join(os.getcwd(), repo_name + '.zip'), chunk_size=chunk_size, MIN_FILE_SIZE=MIN_FILE_SIZE):
+            return
+    logger.error(Fore.RED + "❌ 所有镜像源尝试失败" + Style.RESET_ALL)
+    return
+
 def handle_clone(args, unknown_args, config, env, verbose, proxy):
     original_url = unknown_args[0]
     if '://' not in original_url and '/' in original_url:
@@ -65,12 +106,12 @@ def handle_clone(args, unknown_args, config, env, verbose, proxy):
 
     repo_name = original_url.split('/')[-1].split('.git')[0]
     if os.path.exists(os.path.join(os.getcwd(), repo_name)):
-        print(Fore.YELLOW + f"😪 仓库 {repo_name} 已存在" + Style.RESET_ALL)
+        logger.warning(Fore.YELLOW + f"😪 仓库 {repo_name} 已存在" + Style.RESET_ALL)
         return
 
-    repo_status = get_repo(original_url, verbose)
+    repo_status = get_repo(original_url)
     if repo_status is None:
-        print(Fore.YELLOW + "🧐 无法获取到仓库信息, 尝试克隆" + Style.RESET_ALL)
+        logger.warning(Fore.YELLOW + "🧐 无法获取到仓库信息, 尝试克隆" + Style.RESET_ALL)
     elif repo_status is False and not input_with_timeout(Fore.YELLOW + "🧐 仓库可能不存在，5秒内按任意键忽略..." + Style.RESET_ALL, 5):
         return
 
@@ -80,29 +121,29 @@ def handle_clone(args, unknown_args, config, env, verbose, proxy):
         if result.returncode == 0:
             return
         else:
-            print(Fore.RED + "❌ 在代理模式下克隆失败, 尝试使用镜像模式..." + Style.RESET_ALL)
+            logger.error(Fore.RED + "❌ 在代理模式下克隆失败, 尝试使用镜像模式..." + Style.RESET_ALL)
     mirror_list = select_mirror(config, verbose)
     for mirror in mirror_list:
         new_url = convert_url(original_url, mirror)
-        print(Fore.GREEN + f"🔄 尝试镜像源 {mirror} [{mirror_list.index(mirror) + 1}/{len(mirror_list)}]: {new_url}" + Style.RESET_ALL)
+        logger.info(Fore.GREEN + f"🔄 尝试镜像源 {mirror} [{mirror_list.index(mirror) + 1}/{len(mirror_list)}]: {new_url}" + Style.RESET_ALL)
         cmd = ['git', 'clone', new_url] + unknown_args[1:]
         result = subprocess.run(cmd, env=env, check=False)
         if result.returncode == 0:
             repo_path = os.path.join(os.getcwd(), repo_name)
             subprocess.run(['git', '-C', repo_path, 'remote', 'set-url', 'origin', original_url], check=True) # 还原原始 URL
             return
-    print(Fore.RED + "❌ 所有镜像源尝试失败" + Style.RESET_ALL)
+    logger.error(Fore.RED + "❌ 所有镜像源尝试失败" + Style.RESET_ALL)
 
 def handle_other_commands(args, unknown_args, config, env, verbose, proxy):
     if not os.path.exists(os.path.join(os.getcwd(), '.git')):
-        print(Fore.YELLOW + "❌ 当前目录不是有效的 Git 仓库" + Style.RESET_ALL)
+        logger.warning(Fore.YELLOW + "❌ 当前目录不是有效的 Git 仓库" + Style.RESET_ALL)
         return
     git_args = [args.command] + unknown_args
     result = subprocess.run(['git'] + git_args, env=env, check=False)
     if result.returncode == 0:
         return
     elif proxy.proxy_url:
-        print(Fore.RED + "❌ 在代理模式下运行失败, 尝试使用镜像模式..." + Style.RESET_ALL)
+        logger.error(Fore.RED + "❌ 在代理模式下运行失败, 尝试使用镜像模式..." + Style.RESET_ALL)
 
     mirror_list = select_mirror(config, verbose)
     for mirror in mirror_list:
@@ -113,43 +154,38 @@ def handle_other_commands(args, unknown_args, config, env, verbose, proxy):
                 return
         finally:
             restore_git_config()
-    print(Fore.RED + "❌ 所有镜像源尝试失败" + Style.RESET_ALL)
+    logger.error(Fore.RED + "❌ 所有镜像源尝试失败" + Style.RESET_ALL)
 
-def get_repo(repo: str, verbose: bool = False) -> bool | None:
+def get_repo(repo: str) -> bool | None:
     if repo.endswith('.git'):
         repo = repo[:-4]
     if '://' in repo and '/' in repo:
         repo = repo.split('/')[-2] + '/' + repo.split('/')[-1]  # 处理 URL 形式的仓库名
-    if verbose:
-        print(Fore.CYAN + f"🔍 正在获取仓库: {repo}" + Style.RESET_ALL)
+    logger.debug(Fore.CYAN + f"🔍 正在获取仓库: {repo}" + Style.RESET_ALL)
     url = f"https://api.github.com/repos/{repo}"
     req = Request(url, headers=headers)
     try:
         with urlopen(req) as response:
             if response.status == 200:
                 result = json.loads(response.read().decode())
                 # return [result['full_name'], result['id']]
-                print(Fore.GREEN + f"✅ 获取到仓库信息: {result['full_name']}({result['id']})" + Style.RESET_ALL)
+                logger.info(Fore.GREEN + f"✅ 获取到仓库信息: {result['full_name']}({result['id']})" + Style.RESET_ALL)
                 return True
             elif response.status == 404:
-                print(Fore.RED + f"❌ 获取仓库信息失败，该仓库可能不存在或未公开" + Style.RESET_ALL)
+                logger.warning(Fore.RED + "❌ 获取仓库信息失败，该仓库可能不存在或未公开" + Style.RESET_ALL)
                 return False
             else:
-                if verbose:
-                    print(Fore.RED + f"❌ 获取仓库信息失败: {response.status} {response.reason}" + Style.RESET_ALL)
+                logger.debug(Fore.RED + f"❌ 获取仓库信息失败: {response.status} {response.reason}" + Style.RESET_ALL)
                 return None
     except HTTPError as e:
         if e.code == 404:
-            print(Fore.RED + f"❌ 获取仓库信息失败，该仓库可能不存在或未公开" + Style.RESET_ALL)
+            logger.warning(Fore.RED + "❌ 获取仓库信息失败，该仓库可能不存在或未公开" + Style.RESET_ALL)
             return False
         else:
-            if verbose:
-                print(Fore.RED + f"❌ 获取仓库信息失败: {e}" + Style.RESET_ALL)
+            logger.debug(Fore.RED + f"❌ 获取仓库信息失败: {e}" + Style.RESET_ALL)
             return None
     except Exception as e:
-        if verbose:
-            print(Fore.RED + f"❌ 获取仓库信息失败: {e}" + Style.RESET_ALL)
-            traceback.print_exc()
+        logger.debug(Fore.RED + f"❌ 获取仓库信息失败: {e}" + Style.RESET_ALL)
         return None
 
 def modify_git_config(mirror):
@@ -159,7 +195,7 @@ def restore_git_config():
     subprocess.run(['git', 'config', '--local', '--unset', 'url.https://github.com/.insteadOf'])
 
 def input_with_timeout(prompt, timeout):
-    print(prompt)
+    logger.info(prompt)
     result = []
     thread = threading.Thread(target=lambda: result.append(sys.stdin.read(1)))
     thread.daemon = True
@@ -169,13 +205,12 @@ def input_with_timeout(prompt, timeout):
 
 if __name__ == '__main__':
     try:
-        print(Fore.GREEN + f"fastgit🚀 by NaivG" + Style.RESET_ALL)
+        print(Fore.GREEN + "fastgit🚀 by NaivG" + Style.RESET_ALL)
         main()
     except KeyboardInterrupt:
-        print(Fore.YELLOW + "❗ 操作已取消" + Style.RESET_ALL)
+        logger.warning(Fore.YELLOW + "❗ 操作已取消" + Style.RESET_ALL)
     except Exception as e:
-        print(Fore.RED + f"❌ 错误: {str(e)}" + Style.RESET_ALL)
-        traceback.print_exc()
+        logger.exception(Fore.RED + f"❌ 错误: {str(e)}" + Style.RESET_ALL)
         sys.exit(1)
     finally:
         sys.exit(0)

mirrors.py

@@ -1,3 +1,4 @@
+from loguru import logger
 from ping3 import ping
 from concurrent.futures import ThreadPoolExecutor
 from prettytable import PrettyTable
@@ -19,24 +20,27 @@
     'gh-deno': 'https://gh-deno.mocn.top/https://github.com'
 }
 
+RAWCONTENT_MIRRORS = {
+    'github': 'https://raw.githubusercontent.com',
+    'ghproxy.net': 'https://ghproxy.net/https://raw.githubusercontent.com',
+}
+
 def test_latency(verbose=False):
-    if verbose:
-        print(Fore.CYAN + "🔎 测试镜像源延迟..." + Style.RESET_ALL)
-        table = PrettyTable()
-        table.field_names = ['Git 镜像源', 'Latency 延迟']
+    logger.debug(Fore.CYAN + "🔎 测试镜像源延迟..." + Style.RESET_ALL)
+    table = PrettyTable()
+    table.field_names = ['Git 镜像源', 'Latency 延迟']
 
     results = {}
     with ThreadPoolExecutor() as executor:
         futures = {executor.submit(test_single, name, url): name for name, url in MIRRORS.items()}
         for future in futures:
             name, latency = future.result()
             results[name] = latency
-            if verbose:
-                if latency is not None and latency != 0.0:
-                    color = Fore.GREEN if latency < 200 else Fore.YELLOW if latency < 500 else Fore.RED
-                    table.add_row([name, color + f"{latency:.1f}ms" + Style.RESET_ALL])
-                else:
-                    table.add_row([name, Fore.RED + "超时" + Style.RESET_ALL])
+            if latency is not None and latency != 0.0:
+                color = Fore.GREEN if latency < 200 else Fore.YELLOW if latency < 500 else Fore.RED
+                table.add_row([name, color + f"{latency:.1f}ms" + Style.RESET_ALL])
+            else:
+                table.add_row([name, Fore.RED + "超时" + Style.RESET_ALL])
 
     if verbose:
         print(table)
@@ -53,13 +57,11 @@ def test_single(name, url):
 
 def select_mirror(config, verbose=False):
     if cached := config.get_mirrors():
-        if verbose:
-            print(Fore.CYAN + f"✔️ 已选择 {cached}(缓存) 作为 Git 镜像源" + Style.RESET_ALL)
+        logger.debug(Fore.CYAN + f"✔️ 已选择 {cached}(缓存) 作为 Git 镜像源" + Style.RESET_ALL)
         return cached
     mirrors = test_latency(verbose)
     config.save_mirrors(mirrors)
-    if verbose:
-        print(Fore.CYAN + f"✔️ 已选择 {mirrors} 作为 Git 镜像源" + Style.RESET_ALL)
+    logger.debug(Fore.CYAN + f"✔️ 已选择 {mirrors} 作为 Git 镜像源" + Style.RESET_ALL)
     return mirrors
 
 def convert_url(url, mirror):