import os from ruamel import yaml from common.db.db import dbOP import datetime curpath = os.path.dirname(os.path.realpath(__file__)) rootPath = os.path.split(curpath)[0] # 数据读入和写入文件 class FileUtils(object): def w_info(self, info,keyname): module=info[2] dict = {} value = {} value['username'] = info[0] value['goodsname'] = info[1] key=keyname dict[key] = value w_path=rootPath+os.sep+'data'+os.sep+module # print(w_path) yamlpath = os.path.join(w_path, "message") # 写入到yaml文件 with open(yamlpath, "w", encoding="utf-8") as f: yaml.dump(dict, f, Dumper=yaml.RoundTripDumper,allow_unicode=True) def r_info(self, module,keyname): w_path = rootPath + os.sep + 'data' + os.sep + module yamlpath = os.path.join(w_path, "message") file_value = open(yamlpath, 'r',encoding='utf-8') result = yaml.load(file_value.read(), Loader=yaml.Loader) if result is not None: key = keyname if key in result: return result[key] else: return None else: return None #供货关系信息写入文件 def w_info1(self, info,module,keyname): module=module dict = {} value = {} value['e_corpId'] = info[0] value['h_corpId'] = info[1] value['e_username'] = info[2] value['h_name'] = info[3] key=keyname dict[key] = value w_path=rootPath+os.sep+'data'+os.sep+module # print(w_path) yamlpath = os.path.join(w_path, "message") # 写入到yaml文件 with open(yamlpath, "w", encoding="utf-8") as f: yaml.dump(dict, f, Dumper=yaml.RoundTripDumper,allow_unicode=True) #写入院区数据,新函数,存在已有内容,替换掉旧内容 def w_info2_new(self, newinfo, module, keyname, oldinfo): module = module dict = {} value = {} value['branch_id'] = newinfo[0] value['branch_name'] = newinfo[1] key = keyname dict[key] = value w_path = rootPath + os.sep + 'data' + os.sep + module # print(w_path) yamlpath = os.path.join(w_path, "message") if (self.r_info(module, keyname) != None): # 替换文件内容 file = open(yamlpath, "r+", encoding="utf-8") content = file.read() for i in range(len(newinfo)): new_content = content.replace(oldinfo[i], newinfo[i]) file.seek(0) file.write(new_content) file.close() else: # 写入到yaml文件 with open(yamlpath, "a", encoding="utf-8") as f: yaml.dump(dict, f, Dumper=yaml.RoundTripDumper, allow_unicode=True) #写入院区数据 def w_info2(self, info,module,keyname): module=module dict = {} value = {} value['branch_id'] = info[0] value['branch_name'] = info[1] key=keyname dict[key] = value w_path=rootPath+os.sep+'data'+os.sep+module # print(w_path) yamlpath = os.path.join(w_path, "message") if (self.r_info(module,keyname)!=None): # 写入到yaml文件 with open(yamlpath, "w", encoding="utf-8") as f: yaml.dump(dict, f, Dumper=yaml.RoundTripDumper,allow_unicode=True) else: # 写入到yaml文件 with open(yamlpath, "a", encoding="utf-8") as f: yaml.dump(dict, f, Dumper=yaml.RoundTripDumper, allow_unicode=True) #写入公告id和公告标题 def w_info3(self, info,module,keyname): module=module dict = {} value = {} value['notices_id'] = info[0] value['notices_name'] = info[1] key=keyname dict[key] = value w_path=rootPath+os.sep+'data'+os.sep+module # print(w_path) yamlpath = os.path.join(w_path, "message1") # 写入到yaml文件 with open(yamlpath, "w", encoding="utf-8") as f: yaml.dump(dict, f, Dumper=yaml.RoundTripDumper, allow_unicode=True) #读取公告id和标题 def r_info3(self, module,keyname): w_path = rootPath + os.sep + 'data' + os.sep + module yamlpath = os.path.join(w_path, "message1") file_value = open(yamlpath, 'r',encoding='utf-8') result = yaml.load(file_value.read(), Loader=yaml.Loader) if result is not None: key = keyname if key in result: return result[key] else: return None else: return None #写入科室信息 def w_info4(self, info,module,keyname,titlename): module=module dict = {} value = {} for i in range(len(titlename)): value[titlename[i]] = info[i] # value[titlename[0]] = info[0] # value[titlename[1]] = info[1] key=keyname dict[key] = value w_path=rootPath+os.sep+'data'+os.sep+module # print(w_path) yamlpath = os.path.join(w_path, "message") if (self.r_info(module,keyname)!=None): print('已有相同信息，不写入') # 写入到yaml文件 # with open(yamlpath, "w", encoding="utf-8") as f: # yaml.dump(dict, f, Dumper=yaml.RoundTripDumper,allow_unicode=True) else: # 写入到yaml文件 with open(yamlpath, "a", encoding="utf-8") as f: yaml.dump(dict, f, Dumper=yaml.RoundTripDumper, allow_unicode=True) #写入医院信息 def w_info5(self, info,module,keyname,titlename): module=module dict = {} value = {} for i in range(len(titlename)): value[titlename[i]] = info[i] # value[titlename[0]] = info[0] # value[titlename[1]] = info[1] key=keyname dict[key] = value w_path=rootPath+os.sep+'data'+os.sep+module # print(w_path) yamlpath = os.path.join(w_path, "message") if (self.r_info(module,keyname)!=None): print('已有相同信息，覆盖写入') # 写入到yaml文件 with open(yamlpath, "w", encoding="utf-8") as f: yaml.dump(dict, f, Dumper=yaml.RoundTripDumper,allow_unicode=True) else: # 写入到yaml文件 with open(yamlpath, "a", encoding="utf-8") as f: yaml.dump(dict, f, Dumper=yaml.RoundTripDumper, allow_unicode=True) #写入产品审核信息 def w_info6(self, info,module,keyname,titlename,message_no): module=module dict = {} value = {} for i in range(len(titlename)): value[titlename[i]] = info[i] # value[titlename[0]] = info[0] # value[titlename[1]] = info[1] key=keyname dict[key] = value w_path=rootPath+os.sep+'data'+os.sep+module # print(w_path) yamlpath = os.path.join(w_path, message_no) if (self.r_info(module,keyname)!=None): print('已有相同信息，覆盖写入') # 写入到yaml文件 with open(yamlpath, "w", encoding="utf-8") as f: yaml.dump(dict, f, Dumper=yaml.RoundTripDumper,allow_unicode=True) else: # 写入到yaml文件 with open(yamlpath, "a", encoding="utf-8") as f: yaml.dump(dict, f, Dumper=yaml.RoundTripDumper, allow_unicode=True) def r_info6(self, module,keyname,message_no): w_path = rootPath + os.sep + 'data' + os.sep + module yamlpath = os.path.join(w_path, message_no) file_value = open(yamlpath, 'r',encoding='utf-8') result = yaml.load(file_value.read(), Loader=yaml.Loader) if result is not None: key = keyname if key in result: return result[key] else: return None else: return None if __name__ == '__main__': # info=("aaaa","bbbbbb","mdm3-pim") # FileUtils().w_info(info,"产品新增") aa=FileUtils().r_info("mdm3-pim","产品新增") print(aa['username'])