fs_add_robot已经放在~/.bashrc 文件中,在本地可以执行fs_add_robot 192.168.101.36,我需要在py中执行,下面是sh代码 #!/bin/bash . scripts/helper.sh fs_add_robot () { if [ $# -eq 0 ]; then echo "Aborting: Please provide an ip address !" return 1 fi fs_query_full_stack_path if ! test -f "$full_stack_path/credentials.env"; then echo "Aborting: credentials.env not present! Please provide ./credentials.env" return 1 fi export $(grep -v '^#' $full_stack_path/credentials.env | xargs) ip=$1 ip_underscore=$(echo $ip | sed 's/\./_/g') if [ ! "$(docker ps | grep mysql)" ]; then docker compose up -d db fs_wait_for_container_to_be_healthy mysql fi docker exec -i mysql bash -c "mariadb -uroot -p\"$MYSQL_ROOT_PASSWORD\"" <<< $(sed "s/IPADDRESS/$ip/g" $full_stack_path/db_dumps/helper_snippets/add_robot.sql | sed "s/IP_ADDRESS/$ip_underscore/") echo Added robot with IP: $ip to database }
时间: 2025-07-05 14:46:16 浏览: 2
<think>好的,我需要帮助用户将他们的fs_add_robot bash脚本转换成Python代码。首先,我得仔细分析原脚本的功能和流程。
用户提到fs_add_robot已经放在.bashrc里,本地可以执行。现在他们想在Python中实现同样的功能。我需要理解原脚本每一步的作用。
首先,脚本source了helper.sh,可能载入了一些函数或变量。然后定义了fs_add_robot函数。函数检查参数是否存在,没有IP地址就报错。接着调用fs_query_full_stack_path获取路径,确认credentials.env存在。然后导出环境变量,处理IP地址中的点替换为下划线。检查MySQL容器是否运行,如果没有就启动并等待健康状态。最后用docker exec执行SQL脚本,替换其中的IP占位符。
现在要转换成Python,我需要考虑如何模拟这些步骤。首先,参数检查,Python可以用sys.argv获取参数。检查文件是否存在用os.path.exists。处理环境变量可能需要使用dotenv库或者手动解析。执行docker命令可能需要subprocess.run,需要注意命令的拼接和安全性。等待容器健康可以用循环检查docker inspect的结果。
需要注意的是,原脚本用了sed替换SQL文件中的IPADDRESS和IP_ADDRESS,Python中可以用字符串替换或模板处理。还要确保数据库连接的安全性,比如密码传递时避免泄露。
另外,原脚本中的fs_query_full_stack_path和fs_wait_for_container_to_be_healthy这两个函数需要确认在Python中如何实现。可能涉及到获取项目路径和轮询检查容器状态。
可能的难点在于正确模拟shell脚本中的环境变量导出和命令执行,尤其是docker compose和docker exec的参数处理。还要处理子进程的输出和错误,确保异常能被捕获。
需要逐步将这些步骤翻译成Python代码,确保逻辑一致,并考虑异常处理,比如文件不存在、命令执行失败等情况。同时,保持代码结构清晰,方便用户理解和维护。</think>我将逐步指导你将这个bash脚本转换为Python实现:
1. **参数检查**:
```python
import sys
if len(sys.argv) < 2:
print("Aborting: Please provide an ip address!")
sys.exit(1)
ip = sys.argv[1]
```
2. **环境变量处理**:
```python
from dotenv import load_dotenv
import os
# 获取项目路径(需要根据实际情况实现)
full_stack_path = "/path/to/your/project"
env_path = os.path.join(full_stack_path, "credentials.env")
if not os.path.exists(env_path):
print("Aborting: credentials.env not present!")
sys.exit(1)
load_dotenv(env_path)
```
3. **处理IP地址**:
```python
ip_underscore = ip.replace(".", "_")
```
4. **Docker容器检查与启动**:
```python
import subprocess
import time
def check_container_healthy(container_name):
cmd = f"docker inspect --format='{{{{.State.Health.Status}}}}' {container_name}"
result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
return result.stdout.strip() == "healthy"
# 检查MySQL容器
if "mysql" not in subprocess.getoutput("docker ps"):
subprocess.run("docker compose up -d db", shell=True)
# 等待容器健康
while not check_container_healthy("mysql"):
time.sleep(1)
```
5. **执行SQL脚本**:
```python
# 读取并处理SQL模板
sql_path = os.path.join(full_stack_path, "db_dumps/helper_snippets/add_robot.sql")
with open(sql_path) as f:
sql = f.read()
sql = sql.replace("IPADDRESS", ip).replace("IP_ADD
阅读全文
相关推荐













