File size: 3,480 Bytes
8499730 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 |
import os
import argparse
import requests
from colorama import init, Fore, Style
# from dataflow.utils.paths import BencoPath
from dataflow.cli_funcs import cli_env, cli_init
import importlib.metadata
PYPI_API_URL = 'https://pypi.org/pypi/open-dataflow/json'
from dataflow.version import __version__
def version_and_check_for_updates():
# print a bar by the length of the shell width
print(Fore.BLUE + "=" * os.get_terminal_size().columns + Style.RESET_ALL)
print(f'open-dataflow codebase version: {__version__}')
try:
response = requests.get(PYPI_API_URL, timeout=5)
response.raise_for_status() # 如果响应码不是200，则抛出异常
pypi_data = response.json()
cloud_version = pypi_data['info']['version'] # 获取最新版本号
# cloud_version = '0.1.21' # for debug & test
print("\tChecking for updates...")
print("\tLocal version: ", __version__)
print("\tPyPI newest version: ", cloud_version)
local_version = __version__ # 通过 importlib.metadata 获取当前安装版本
if cloud_version != local_version:
print(Fore.YELLOW + f"New version available: {cloud_version}. Your version: {local_version}." + Style.RESET_ALL)
print("Run 'pip install --upgrade open-dataflow' to upgrade.")
else:
print(Fore.GREEN + f"You are using the latest version: {local_version}." + Style.RESET_ALL)
except requests.exceptions.RequestException as e:
print(Fore.RED + "Failed to check for updates from PyPI. Please check your internet connection." + Style.RESET_ALL)
print(f"Error: {e}")
print(Fore.BLUE + "=" * os.get_terminal_size().columns + Style.RESET_ALL)
def main():
parser = argparse.ArgumentParser(description='Command line interface for DataFlow, with codebase version: ' + __version__)
# Add version argument with update check only when user requests version
parser.add_argument('-v', '--version', action='store_true', help="Show the version of the tool")
subparsers = parser.add_subparsers(dest='command', required=False)
# init command
parser_init = subparsers.add_parser('init', help='Initialize the scripts and configs in a directory')
init_subparsers = parser_init.add_subparsers(dest='subcommand', required=False)
# init all
parser_init_all = init_subparsers.add_parser('all', help='Initialize all components')
parser_init_all.set_defaults(subcommand='all')
# init reasoning
parser_init_reasoning = init_subparsers.add_parser('reasoning', help='Initialize reasoning components')
parser_init_reasoning.set_defaults(subcommand='reasoning')
# env command
parser_env = subparsers.add_parser('env', help='Show environment information')
# parser.add_argument('--config', type=str, help='Path to the configuration file')
args = parser.parse_args()
if args.version:
version_and_check_for_updates()
if args.command == 'init':
if args.subcommand is None:
args.subcommand = 'base'
cli_init(subcommand=args.subcommand)
# print("TODO Calling cli_init with subcommand:", args.subcommand)
from dataflow.cli_funcs.paths import DataFlowPath
# print(DataFlowPath.get_dataflow_dir())
# print(DataFlowPath.get_dataflow_scripts_dir())
elif args.command == 'env':
cli_env()
if __name__ == '__main__':
main()
|