影刀RPA 环境变量管理：多环境配置自动切换-育师

影刀RPA 环境变量管理：多环境配置自动切换

作者：林焱 | 分类：影刀RPA新手教程 | 难度：★★

什么情况用

开发、测试、预发布、生产四套环境，每套环境有不同的数据库地址、API域名、密钥。手动切换环境时容易搞混——把测试环境的数据写到了生产数据库，或者用生产的API Key调了测试接口。

用影刀RPA统一管理多环境配置，自动在环境间切换，防止低级配置错误。适合需要频繁在多个环境间切换的测试和运维场景。

怎么做

第一步：设计环境配置表

拼多多店群自动化上架方案

在Excel或JSON中维护各环境的配置：

# 影刀Python节点：加载环境配置importjson environments={"dev":{"name":"开发环境","db":{"host":"192.168.1.10","port":3306,"database":"app_dev","user":"dev_user",},"api":{"base_url":"https://dev-api.example.com","timeout":10,},"redis":{"host":"192.168.1.11","port":6379},"log_level":"DEBUG",},"test":{"name":"测试环境","db":{"host":"192.168.2.10","port":3306,"database":"app_test","user":"test_user",},"api":{"base_url":"https://test-api.example.com","timeout":10,},"redis":{"host":"192.168.2.11","port":6379},"log_level":"INFO",},"staging":{"name":"预发布环境","db":{"host":"10.0.1.10","port":3306,"database":"app_staging","user":"staging_user",},"api":{"base_url":"https://staging-api.example.com","timeout":5,},"redis":{"host":"10.0.1.11","port":6379},"log_level":"WARN",},"production":{"name":"生产环境 ⚠️","db":{"host":"10.0.0.10","port":3306,"database":"app_prod","user":"prod_readonly",# 生产只能用只读账号！},"api":{"base_url":"https://api.example.com","timeout":5,},"redis":{"host":"10.0.0.11","port":6379},"log_level":"ERROR",},}

第二步：安全切换机制

# 影刀Python节点：环境切换安全校验defswitch_environment(current,target):"""切换环境，带安全检查"""WARNINGS={"production":"⚠️ 即将切换到生产环境！所有操作将影响真实数据。","staging":"⚠️ 预发布环境，数据接近生产。",}# 到生产环境的操作需要二次确认iftarget=="production":print("="*50)print("🔴 正在切换到生产环境！")print("🔴 请确认以下事项：")print(" 1. 确认操作内容是安全的（只读/非破坏性）")print(" 2. 确认数据库账号是只读账号")print(" 3. 确认不会影响线上业务")print("="*50)# 这里应该暂停等待人工确认# 影刀中可以用「确认弹窗」或企微确认confirmed=True# 实际需要等待确认ifnotconfirmed:print("操作已取消")returnNoneprint(f"✅ 已切换到：{environments[target]['name']}")returnenvironments[target]current_env=switch_environment("dev","production")

第三步：配置注入到实际操作

# 影刀Python节点：使用当前环境配置操作数据库defexecute_with_current_env(env,operation,**params):"""在指定环境中执行操作"""db=env["db"]api=env["api"]ifoperation=="query_db":conn=pymysql.connect(host=db["host"],port=db["port"],user=db["user"],database=db["database"],)sql=params.get("sql","SELECT 1")# 执行查询withconn.cursor()ascursor:cursor.execute(sql)result=cursor.fetchall()conn.close()![在这里插入图片描述](https://i-blog.csdnimg.cn/direct/84a47320fe454123a4566a53566802c1.png#pic_center)returnresultelifoperation=="call_api":url=api["base_url"]+params.get("endpoint","/")resp=requests.get(url,timeout=api["timeout"])returnresp.json()elifoperation=="clear_cache":redis_client=redis.Redis(host=env["redis"]["host"],port=env["redis"]["port"],)pattern=params.get("pattern","*")keys=redis_client.keys(pattern)ifkeys:redis_client.delete(*keys)returnf"清除了{len(keys)}个缓存键"

第四步：配置漂移检测

# 影刀Python节点：检测环境间配置差异defdetect_config_drift(envs):"""检测各环境之间的配置是否一致（应该一致的配置）"""drift_issues=[]# 获取所有环境的配置键env_names=list(envs.keys())base_env=env_names[0]base_keys=set(envs[base_env].keys())forenv_nameinenv_names[1:]:current_keys=set(envs[env_name].keys())# 检查缺失的配置项missing=base_keys-current_keysifmissing:drift_issues.append(f"{env_name}缺失配置项：{missing}")# 检查多余的配置项extra=current_keys-base_keysifextra:drift_issues.append(f"{env_name}有多余配置项：{extra}")returndrift_issues drifts=detect_config_drift(environments)ifdrifts:print("⚠️ 环境配置不一致：")fordindrifts:print(f"{d}")else:print("✅ 所有环境配置结构一致")

第五步：配置备份与恢复

# 影刀Python节点：环境配置快照defbackup_env_config(env_name,config):"""备份当前环境配置"""timestamp=datetime.now().strftime("%Y%m%d_%H%M%S")backup_file=f"env_backup_{env_name}_{timestamp}.json"withopen(backup_file,"w",encoding="utf-8")asf:json.dump(config,f,ensure_ascii=False,indent=2)print(f"配置已备份：{backup_file}")returnbackup_file ![在这里插入图片描述](https://i-blog.csdnimg.cn/direct/867c01b5556342d3888a1d4e16c2918a.png#pic_center)defrestore_env_config(backup_file):"""从备份恢复配置"""withopen(backup_file,"r",encoding="utf-8")asf:config=json.load(f)print(f"配置已从{backup_file}恢复")returnconfig