影刀RPA 配置中心设计流程参数集中管理署名林焱什么情况用你的RPA流程里有20个参数数据库地址、账号密码、网页URL、超时时间、文件路径……散落在各个组件里。改一个数据库地址要翻遍整个流程找哪里用了旧地址。更惨的是5个流程共用同一套配置改了一个忘了改其他四个。配置中心就是把这些参数集中到一个地方管理——改一处所有流程生效。就像家里的电箱所有开关集中在一个面板上不用满屋子找开关。典型场景多个RPA流程共用同一套数据库配置开发/测试/生产环境切换不同数据库、不同URL密码等敏感信息统一管理不散落在代码里非开发人员需要修改参数如采集频率、收件人邮箱怎么做1. JSON配置文件店群矩阵自动化突破运营极限最简单的配置中心——一个JSON文件搞定importjsonimportos# 创建配置文件config{database:{host:192.168.1.100,port:3306,user:rpa_user,password:RPA2024#secure,database:erp_system},web:{target_url:https://www.example.com,timeout:30,retry_times:3,headless:False},excel:{output_dir:C:\\RPAData\\导出,template_dir:C:\\RPAData\\模板,max_rows_per_sheet:50000},email:{smtp_server:smtp.qq.com,smtp_port:465,sender:rpa_botqq.com,password:email_auth_code,recipients:[managercompany.com,bosscompany.com]},schedule:{crawl_interval_minutes:60,report_send_hour:9,cleanup_days:30},environment:production}config_pathrC:\RPAData\config.jsonwithopen(config_path,w,encodingutf-8)asf:json.dump(config,f,indent2,ensure_asciiFalse)print(f配置文件已创建{config_path})2. 配置读取封装importjsonimportosfromfunctoolsimportlru_cacheclassConfigManager:配置管理器def__init__(self,config_pathNone):ifconfig_pathisNone:# 默认配置文件路径config_pathos.path.join(os.environ.get(USERPROFILE,),RPAData,config.json)self.config_pathconfig_path self._configNonedefload(self):加载配置ifnotos.path.exists(self.config_path):raiseFileNotFoundError(f配置文件不存在{self.config_path})withopen(self.config_path,r,encodingutf-8)asf:self._configjson.load(f)returnself._configdefget(self,key_path,defaultNone): 用点号路径获取配置值 如config.get(database.host) ifself._configisNone:self.load()keyskey_path.split(.)valueself._configforkeyinkeys:ifisinstance(value,dict)andkeyinvalue:valuevalue[key]else:returndefaultreturnvaluedefset(self,key_path,value):设置配置值ifself._configisNone:self.load()keyskey_path.split(.)configself._configforkeyinkeys[:-1]:ifkeynotinconfig:config[key]{}configconfig[key]config[keys[-1]]valuedefsave(self):保存配置withopen(self.config_path,w,encodingutf-8)asf:json.dump(self._config,f,indent2,ensure_asciiFalse)defreload(self):重新加载配置self._configNonereturnself.load()# 使用示例configConfigManager(rC:\RPAData\config.json)# 读取配置db_hostconfig.get(database.host)db_portconfig.get(database.port)target_urlconfig.get(web.target_url)recipientsconfig.get(email.recipients,[])print(f数据库{db_host}:{db_port})print(f目标URL{target_url})print(f收件人{recipients})# 修改配置config.set(web.timeout,60)config.set(email.recipients,[new_managercompany.com])config.save()print(配置已更新)3. 多环境配置importjsonimportosclassEnvironmentConfig:多环境配置管理def__init__(self,config_dirNone):self.config_dirconfig_dirorrC:\RPAData\configself.envself._detect_environment()self.configself._load_config()def_detect_environment(self):自动检测当前环境# 方式1环境变量envos.environ.get(RPA_ENV,)ifenv:returnenv# 方式2机器名判断hostnameos.environ.get(COMPUTERNAME,).lower()ifprodinhostnameorserverinhostname:returnproductioneliftestinhostnameordevinhostname:returntestelse:returndevelopmentdef_load_config(self):加载当前环境的配置# 先加载基础配置base_pathos.path.join(self.config_dir,base.json)withopen(base_path,r,encodingutf-8)asf:configjson.load(f)# 再加载环境特定配置覆盖基础配置env_pathos.path.join(self.config_dir,f{self.env}.json)ifos.path.exists(env_path):withopen(env_path,r,encodingutf-8)asf:env_configjson.load(f)configself._deep_merge(config,env_config)config[_environment]self.envreturnconfigdef_deep_merge(self,base,override):深度合并两个字典resultbase.copy()forkey,valueinoverride.items():ifkeyinresultandisinstance(result[key],dict)andisinstance(value,dict):result[key]self._deep_merge(result[key],value)else:result[key]valuereturnresultdefget(self,key_path,defaultNone):keyskey_path.split(.)valueself.configforkeyinkeys:ifisinstance(value,dict)andkeyinvalue:valuevalue[key]else:returndefaultreturnvalue# 配置文件结构# C:\RPAData\config\# ├── base.json # 基础配置所有环境共用# ├── development.json # 开发环境覆盖配置# ├── test.json # 测试环境覆盖配置# └── production.json # 生产环境覆盖配置# base.json# {# app_name: RPA采集系统,# version: 2.0,# database: {port: 3306, charset: utf8mb4},# web: {timeout: 30, retry: 3}# }# development.json# {# database: {host: localhost, user: dev, password: dev123},# web: {target_url: https://dev.example.com}# }# production.json# {# database: {host: 192.168.1.100, user: rpa_prod, password: Prod2024},# web: {target_url: https://www.example.com, timeout: 60}# }# 使用env_configEnvironmentConfig(rC:\RPAData\config)print(f当前环境{env_config.get(_environment)})print(f数据库地址{env_config.get(database.host)})print(f目标URL{env_config.get(web.target_url)})4. 配置热更新importjsonimportosimporttimeclassHotReloadConfig:支持热更新的配置管理器def__init__(self,config_path):self.config_pathconfig_path self._configNoneself._last_mtime0self._load()def_load(self):加载配置并记录修改时间withopen(self.config_path,r,encodingutf-8)asf:self._configjson.load(f)self._last_mtimeos.path.getmtime(self.config_path)def_check_reload(self):检查配置文件是否被修改是则重新加载try:current_mtimeos.path.getmtime(self.config_path)ifcurrent_mtimeself._last_mtime:print(检测到配置文件变更重新加载...)self._load()exceptFileNotFoundError:passdefget(self,key_path,defaultNone):每次获取值时检查是否需要重新加载self._check_reload()keyskey_path.split(.)valueself._configforkeyinkeys:ifisinstance(value,dict)andkeyinvalue:valuevalue[key]else:returndefaultreturnvalue# 使用修改config.json后不需要重启RPA流程下次get自动读取最新值configHotReloadConfig(rC:\RPAData\config.json)# 模拟流程持续运行foriinrange(100):timeoutconfig.get(web.timeout,30)# 如果config.json被修改这里自动获取新值urlconfig.get(web.target_url)print(f[{i}] URL{url}, timeout{timeout})time.sleep(10)5. 配置验证importjsonimportosdefvalidate_config(config):验证配置完整性errors[]warnings[]# 检查必填项required_keys[(database.host,数据库地址),(database.port,数据库端口),(database.user,数据库用户名),(database.password,数据库密码),(web.target_url,目标URL),(email.smtp_server,邮件服务器),]forkey_path,descinrequired_keys:value_get_nested(config,key_path)ifvalueisNoneorvalue:errors.append(f缺少必填配置{desc}{key_path})# 检查类型ifisinstance(_get_nested(config,database.port),str):warnings.append(database.port 应为整数当前为字符串)# 检查URL格式url_get_nested(config,web.target_url)ifurlandnoturl.startswith((http://,https://)):errors.append(fweb.target_url 格式错误{url})# 检查路径是否存在output_dir_get_nested(config,excel.output_dir)ifoutput_dirandnotos.path.exists(output_dir):warnings.append(f输出目录不存在{output_dir}将自动创建)# 检查密码强度password_get_nested(config,database.password)ifpasswordandlen(password)8:warnings.append(数据库密码长度不足8位建议加强)returnerrors,warningsdef_get_nested(d,key_path):keyskey_path.split(.)valuedforkeyinkeys:ifisinstance(value,dict)andkeyinvalue:valuevalue[key]else:returnNonereturnvalue# 使用config_pathrC:\RPAData\config.jsonwithopen(config_path,r,encodingutf-8)asf:configjson.load(f)errors,warningsvalidate_config(config)iferrors:print(❌ 配置错误)foreinerrors:print(f -{e})raiseValueError(配置校验失败请修正后重试)ifwarnings:print(⚠ 配置警告)forwinwarnings:print(f -{w})print(✓ 配置校验通过)有什么坑坑1密码明文存储在配置文件中config.json里的密码是明文任何能访问文件的人都能看到。生产环境中必须加密importosfromcryptography.fernetimportFernet# 生成密钥只需一次妥善保管keyFernet.generate_key()# 把key保存到安全位置如Windows凭据管理器# 加密密码cipherFernet(key)encrypted_passwordcipher.encrypt(bRPA2024#secure)# 把encrypted_password存入配置文件# 解密decryptedcipher.decrypt(encrypted_password)或者更简单用环境变量存密码配置文件只存非敏感信息importos# 配置文件里不存密码# config.json: {database: {host: ..., user: ...}}# 密码从环境变量读取db_passwordos.environ.get(RPA_DB_PASSWORD)ifnotdb_password:raiseEnvironmentError(请设置环境变量 RPA_DB_PASSWORD)坑2配置文件被多人同时修改temu店群自动化报活动案例A改了timeout从30到60B同时改了retry从3到5A先保存B后保存——B的保存覆盖了A的修改timeout又变回30。简单解决配置文件只允许一个人改或者用数据库存储配置数据库有行锁。如果必须用文件加文件锁importfcntl# Linux only# Windows用msvcrt或pywin32的LockFile# 或简单地用配置修改记录表# 每次修改前先读取最新版本号修改后版本号1# 如果版本号不匹配说明有人抢先修改了需要重新读取坑3配置项过多导致JSON文件巨大配置项从20个增长到200个JSON文件几百行找一项要翻半天。解决按模块拆分C:\RPAData\config\ ├── database.json # 数据库相关 ├── web.json # 网页相关 ├── excel.json # Excel相关 ├── email.json # 邮件相关 └── schedule.json # 调度相关classMultiFileConfig:def__init__(self,config_dir):self.config_dirconfig_dir self._cache{}defget(self,module,key,defaultNone):如config.get(database, host)ifmodulenotinself._cache:pathos.path.join(self.config_dir,f{module}.json)ifos.path.exists(path):withopen(path,r,encodingutf-8)asf:self._cache[module]json.load(f)else:returndefaultreturnself._cache[module].get(key,default)坑4配置文件路径写死配置文件路径硬编码为C:\RPAData\config.json换台电脑就找不到。解决用相对路径或环境变量# 方案1环境变量指定配置文件位置config_pathos.environ.get(RPA_CONFIG_PATH,os.path.join(os.path.dirname(__file__),config.json))# 方案2从用户目录读取config_pathos.path.join(os.path.expanduser(~),.rpa,config.json)# 方案3影刀全局变量# 在影刀客户端设置中配置全局变量CONFIG_PATH流程中用【获取全局变量】读取坑5缺少配置导致流程中途崩溃流程跑到第50步才发现某个配置项没设置直接报错退出。前面50步白跑了。# 解决方案流程启动时先做配置完整性检查configConfigManager(config_path).load()errors,warningsvalidate_config(config)iferrors:# 配置不完整不启动流程print(配置不完整无法启动)foreinerrors:print(f{e})# 发送通知给管理员send_alert(RPA流程启动失败,f配置错误\n\n.join(errors))sys.exit(1)