How to use organize method in dbt-osmosis

Best Python code snippet using dbt-osmosis_python

submit_tool.py

Source:submit_tool.py Github

copy

Full Screen

1import requests2import json3import time4from cn.code_predict import base64_api5from pic_code import get_pic_code6header = {7 "User-Agent": "Mozilla/​5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/​537.36 (KHTML, like Gecko) Chrome/​90.0.4430.212 Safari/​537.36",8 'content-type': "application/​json"9}10token_url = 'https:/​/​fsservice.wjj.foshan.gov.cn/​fw2/​foying/​wechatpublic/​wx/​user/​getTokenByCh5'11submit_url = 'https:/​/​fsservice.wjj.foshan.gov.cn/​fw2/​foying/​wechatpublic/​wx/​VaccineOrder/​addVaccineOrderByCh5'12def get_token():13 body = {14 "orderChannel": "ch5"15 }16 res = requests.post(token_url, headers=header, data=json.dumps(body))17 data = res.text18 # data = '{"ResCode": "100","ResMsg": "成功","entity": {"allowOrderUserType": "au1","orderChannel": "ch5","token": "token-adda3382c5ebfc8c84bd28f5d4c0087a065121622552785493"}}'19 dataObj = json.loads(data)20 token = dataObj["entity"]["token"]21 return token22today = time.strftime("%Y-%m-%d", time.localtime())23def submit(token, platformScheduleId, base_organize_name, base_organize_ID, base_organize_phone,24 base_organize_address, ):25 needleTimes = "1" # 针次26 img_path = 'pic_code.jpeg'27 result = base64_api(uname='bozi0822', pwd='111111qq', img=img_path, typeid=11)28 body = {29 "token": token,30 "bookType": "personal", # 个人预约31 "baseOrganizeId": base_organize_ID, # 接种单位32 "platformScheduleId": platformScheduleId, # 排班id33 "baseOrganizeName": base_organize_name, #34 "baseOrganizeAddress": base_organize_address,35 "baseOrganizePhone": base_organize_phone,36 "scheduleDateStr": today, # 接种预约时间 2021-06-0237 "workAddress": workAddress,38 "homeAddress": homeAddress,39 "peopleCode": "pc1", # 职业 不用变40 "needleTimes": needleTimes, # 针次41 "realName": realName, # 姓名42 "idCardNo": idCardNo, # 身份证43 "telephone": telephone, # 电话44 "sex": "1", # 性别 1男45 "idCardNoType": "01", # 证件类型 01身份证46 "picCode": result, # 验证码 #################################47 "birthday": ""48 }49 print('submit body=>' + str(body))50 res = requests.post(token_url, headers=header, data=json.dumps(body))51 data = res.text52 print('submit=>' + data)53all = [54 {"groupArea": "禅城区",55 "groupStreets": {56 "石湾街道", "张槎街道", "祖庙街道", "南庄镇"57 }},58 {"groupArea": "南海区",59 "groupStreets": {60 "桂城街道", "丹灶镇", "狮山镇", "大沥镇", "里水镇"61 }},62 {"groupArea": "顺德区",63 "groupStreets": {64 "大良街道", "伦教街道", "北滘镇", "陈村镇"65 }},66]67url2 = 'https:/​/​fsservice.wjj.foshan.gov.cn/​fw2/​foying/​wechatpublic/​wx/​userBooking/​getScheduleByDate'68url1 = 'https:/​/​fsservice.wjj.foshan.gov.cn/​fw2/​foying/​wechatpublic/​wx/​userBooking/​getOrganizeByGroupArea'69def find_info(groupArea, groupStreet, base_organize_name):70 body = {71 "groupArea": groupArea,72 "groupStreet": groupStreet73 }74 res = requests.post(url1, headers=header, data=json.dumps(body))75 # print(res.text)76 data = res.text77 resList = []78 #################################################################################################################################79 # data = '{"ResCode":"100","ResMsg":"获取机构列表成功","entityList":[{"address":"北滘镇人昌路10号附近","allowOrderUserType":"au1","id":"08a99f34553c471bad32da56fa93c1651621942656007","organizeEnCode":"4406060530","organizeName":"北滘文化中心新冠疫苗大型临时接种点","showFlag":"1","telephone":"0757-26656485"},{"address":"佛山市禅城区绿景西路10号50号铺","allowOrderUserType":"au1","id":"2d1fc3a0991a4c7ca75a2a8d954730061617154824217","organizeEnCode":"4406040434","organizeName":"石湾镇街道番村社区卫生服务站临时接种点","showFlag":"0","telephone":""},{"address":"佛山市禅城区奇槎村奇槎新基42号","allowOrderUserType":"au1","id":"424c9471da194d53b77d36006f7435621618800142086","organizeEnCode":"4406040405","organizeName":"禅城区人民医院奇槎新冠临时接种点","showFlag":"0","telephone":""},{"address":"佛山市第一人民医院健康管理中心3号楼1楼大堂","allowOrderUserType":"au1,au2","id":"501977ac24ab436f8b2c4bcc6cf2b9601617249299578","organizeEnCode":"4406040130","organizeName":"佛山市第一人民医院新冠疫苗接种点","showFlag":"0","telephone":""},{"address":"石湾镇街道澜石前进路88号澜石码头","allowOrderUserType":"au1","id":"56c1e81068ae4227817a0ff4574e2e3c1617155817357","organizeEnCode":"4406040433","organizeName":"佛山市禅城区石湾镇街道大型接种点","showFlag":"0","telephone":""},{"address":"禅城区石湾镇街道汾江南路9号一楼","allowOrderUserType":"au1","id":"6bce5464beda48cbb4a691a1d25f4cad1616920840275","organizeEnCode":"4406040401","organizeName":"禅城区中心医院里水社区卫生服务站新冠接种门诊","showFlag":"0","telephone":"83385330"},{"address":"佛山市禅城区文华中路72号","allowOrderUserType":"au1","id":"8d833ddbd4304f1ba83a3ee3a6c3fa861620958775430","organizeEnCode":"4406040403","organizeName":"汉德（佛山）骨科医院新冠疫苗临时接种点","showFlag":"0","telephone":""},{"address":"禅城区石湾镇街道三友南路3号6号楼儿保园","allowOrderUserType":"au1","id":"8f12e1de035a4ee18e645815d2f1c3151616920568069","organizeEnCode":"4406040430","organizeName":"禅城区中心医院预防接种门诊","showFlag":"1","telephone":"82778663"},{"address":"石湾镇街道前进路1号澜石小学旁（六楼）","allowOrderUserType":"au1","id":"a660b4b013394f1d958495c664374e871616920000674","organizeEnCode":"4406040436","organizeName":"石湾镇街道社区卫生服务中心新冠固定接种点","showFlag":"0","telephone":""},{"address":"禅城区石湾镇街道前进路1号澜石小学旁（三楼）","allowOrderUserType":"au1","id":"bd66900acd2348378be06fd13f16bc731616919497529","organizeEnCode":"4406040432","organizeName":"石湾镇街道社区卫生服务中心预防接种门诊","showFlag":"0","telephone":""},{"address":"佛山市禅城区深村大道深村社区卫生服务站旁","allowOrderUserType":"au1","id":"e836b9aeb52e4acca41160ad0ce0130c1617156391382","organizeEnCode":"4406040445","organizeName":"石湾镇街道深村社区卫生服务站怡康楼","showFlag":"0","telephone":""}]}'80 info_dic = json.loads(data)81 # print("\n")82 # print("==========", groupStreet, "===============")83 # print(info_dic["entityList"])84 global base_organize_ID85 global base_organize_phone86 global base_organize_address87 for entity in info_dic["entityList"]:88 if base_organize_name == entity["organizeName"]:89 base_organize_address = entity["address"]90 base_organize_ID = entity["id"]91 base_organize_phone = entity["telephone"]92 # print("=======================================================")93 # print(entity["address"])94 # print(entity["id"])95 # print(entity["organizeName"])96 # print(entity["telephone"])97 body2 = {98 "scheduleDate": today,99 "baseOrganizeID": base_organize_ID100 }101 res = requests.post(url2, headers=header, data=json.dumps(body2))102 # print(res.text)103 data = res.text104 # data = '{"ResCode":"100","ResMsg":"获取排班信息成功","entityList":[{"beginTimeStr":"14:30:00","count":300,"endTimeStr":"15:30:00","scheduleID":"9086ee28830d43d5866362d4f98349df1622176400468","vaccineProducer":"北京生物（含长春、兰州和成都）"},{"beginTimeStr":"15:30:00","count":0,"endTimeStr":"16:30:00","scheduleID":"e68e3605c8314a20a22792c77851130a1622176430593","vaccineProducer":"北京生物（含长春、兰州和成都）"}]}'105 schedule_ID = ""106 schedule_dic = json.loads(data)107 for item in schedule_dic["entityList"]:108 count = item["count"]109 vaccineProducer = item["vaccineProducer"]110 countInt = int(count)111 if countInt > 0:112 todayTime = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())113 timeName = time.strftime("%Y%m%d", time.localtime())114 res_content = vaccineProducer + ' ==剩余数量：' + str(count) + '【' + todayTime + '】'115 print(res_content)116 with open("log" + timeName + ".txt", "a", newline='') as f:117 f.write(res_content + '\n')118 f.close()119 schedule_ID = item["scheduleID"]120 return base_organize_ID, base_organize_phone, base_organize_address, schedule_ID121#################################自己的信息####################################122workAddress = "" # 公司地址123homeAddress = "" # 住址124realName = "" # 姓名125idCardNo = "" # 身份证126telephone = "" # 电话127groupArea = "顺德区",128groupStreet = "北滘镇",129base_organize_name = "北滘文化中心新冠疫苗大型临时接种点"130###############################################################################131if __name__ == '__main__':132 get_pic_code()133 base_organize_ID, base_organize_phone, base_organize_address, schedule_ID = find_info(groupArea, groupStreet,134 base_organize_name)135 token = get_token()136 submit(token=token, platformScheduleId=schedule_ID, base_organize_name=base_organize_name,137 base_organize_ID=base_organize_ID, base_organize_phone=base_organize_phone,...

Full Screen

Full Screen

manager.py

Source:manager.py Github

copy

Full Screen

...23 self.interval = interval24 def run(self):25 while True:26 time.sleep(self.interval)27 self.manager.flush_organize()28class Manager(Loggable):29 # NOTE : this massive class is a source of many problems of mm and30 # is in dire need of breaking up and refactoring.31 """ An abstraction over media monitors core pyinotify functions.32 These include adding watched,store, organize directories, etc.33 Basically composes over WatchManager from pyinotify """34 def __init__(self):35 self.wm = pyinotify.WatchManager()36 # These two instance variables are assumed to be constant37 self.watch_channel = getsig('watch')38 self.organize_channel = getsig('organize')39 self.watch_listener = StoreWatchListener(signal = self.watch_channel)40 self.__timeout_thread = ManagerTimeout(self)41 self.__timeout_thread.daemon = True42 self.__timeout_thread.start()43 self.organize = {44 'organize_path' : None,45 'imported_path' : None,46 'recorded_path' : None,47 'problem_files_path' : None,48 'organizer' : None,49 'problem_handler' : None,50 'organize_listener' : OrganizeListener(signal=51 self.organize_channel),52 }53 def dummy(sender, event): self.watch_move( event.path, sender=sender )54 dispatcher.connect(dummy, signal=getsig('watch_move'), 55 sender=dispatcher.Any, weak=False)56 def subwatch_add(sender, directory):57 self.__add_watch(directory, self.watch_listener)58 dispatcher.connect(subwatch_add, signal=getsig('add_subwatch'),59 sender=dispatcher.Any, weak=False)60 # A private mapping path => watch_descriptor61 # we use the same dictionary for organize, watch, store wd events.62 # this is a little hacky because we are unable to have multiple wd's63 # on the same path.64 self.__wd_path = {}65 # The following set isn't really necessary anymore. Should be66 # removed...67 self.watched_directories = set([])68 # This is the only event that we are unable to process "normally". I.e.69 # through dedicated handler objects. Because we must have access to a70 # manager instance. Hence we must slightly break encapsulation.71 def watch_move(self, watch_dir, sender=None):72 """ handle 'watch move' events directly sent from listener """73 self.logger.info("Watch dir '%s' has been renamed (hence removed)" %74 watch_dir)75 self.remove_watch_directory(normpath(watch_dir))76 def watch_signal(self):77 """ Return the signal string our watch_listener is reading78 events from """79 return getsig(self.watch_listener.signal)80 def __remove_watch(self,path):81 """ Remove path from being watched (first will check if 'path'82 is watched) """83 # only delete if dir is actually being watched84 if path in self.__wd_path:85 wd = self.__wd_path[path]86 self.wm.rm_watch(wd, rec=True)87 del(self.__wd_path[path])88 def __add_watch(self,path,listener):89 """ Start watching 'path' using 'listener'. First will check if90 directory is being watched before adding another watch """91 self.logger.info("Attempting to add listener to path '%s'" % path)92 self.logger.info( 'Listener: %s' % str(listener) )93 if not self.has_watch(path):94 wd = self.wm.add_watch(path, pyinotify.ALL_EVENTS, rec=True,95 auto_add=True, proc_fun=listener)96 if wd: self.__wd_path[path] = wd.values()[0]97 def __create_organizer(self, target_path, recorded_path):98 """ creates an organizer at new destination path or modifies the99 old one """100 # TODO : find a proper fix for the following hack101 # We avoid creating new instances of organize because of the way102 # it interacts with pydispatch. We must be careful to never have103 # more than one instance of OrganizeListener but this is not so104 # easy. (The singleton hack in Organizer) doesn't work. This is105 # the only thing that seems to work.106 if self.organize['organizer']:107 o = self.organize['organizer']108 o.channel = self.organize_channel109 o.target_path = target_path110 o.recorded_path = recorded_path111 else:112 self.organize['organizer'] = Organizer(channel=113 self.organize_channel, target_path=target_path,114 recorded_path=recorded_path)115 def get_problem_files_path(self):116 """ returns the path where problem files should go """117 return self.organize['problem_files_path']118 def set_problem_files_path(self, new_path):119 """ Set the path where problem files should go """120 self.organize['problem_files_path'] = new_path121 self.organize['problem_handler'] = \122 ProblemFileHandler( PathChannel(signal=getsig('badfile'),123 path=new_path) )124 def get_recorded_path(self):125 """ returns the path of the recorded directory """126 return self.organize['recorded_path']127 def set_recorded_path(self, new_path):128 self.__remove_watch(self.organize['recorded_path'])129 self.organize['recorded_path'] = new_path130 self.__create_organizer( self.organize['imported_path'], new_path)131 self.__add_watch(new_path, self.watch_listener)132 def get_organize_path(self):133 """ returns the current path that is being watched for134 organization """135 return self.organize['organize_path']136 def set_organize_path(self, new_path):137 """ sets the organize path to be new_path. Under the current138 scheme there is only one organize path but there is no reason139 why more cannot be supported """140 # if we are already organizing a particular directory we remove the141 # watch from it first before organizing another directory142 self.__remove_watch(self.organize['organize_path'])143 self.organize['organize_path'] = new_path144 # the OrganizeListener instance will walk path and dispatch an organize145 # event for every file in that directory146 self.organize['organize_listener'].flush_events(new_path)147 #self.__add_watch(new_path, self.organize['organize_listener'])148 def flush_organize(self):149 path = self.organize['organize_path']150 self.organize['organize_listener'].flush_events(path)151 def get_imported_path(self):152 return self.organize['imported_path']153 def set_imported_path(self,new_path):154 """ set the directory where organized files go to. """155 self.__remove_watch(self.organize['imported_path'])156 self.organize['imported_path'] = new_path157 self.__create_organizer( new_path, self.organize['recorded_path'])158 self.__add_watch(new_path, self.watch_listener)159 def change_storage_root(self, store):160 """ hooks up all the directories for you. Problem, recorded,161 imported, organize. """162 store_paths = mmp.expand_storage(store)...

Full Screen

Full Screen

myfunc.py

Source:myfunc.py Github

copy

Full Screen

1import os2import shutil3from glob import glob4# unify path name5def unify_path_name(organize_dirctrory_path):6 organize_dirctrory_path = organize_dirctrory_path.replace('\ ', ' ')7 organize_dirctrory_path = organize_dirctrory_path.strip(' ')8 if organize_dirctrory_path[-1] != '/​':9 organize_dirctrory_path = organize_dirctrory_path + '/​'10 11 return organize_dirctrory_path12# get extension to create files every extension13def check_file_extensions(organize_dirctrory_path):14 file_list = glob(organize_dirctrory_path + '*')15 file_extensions = []16 17 for file in file_list:18 if os.path.isfile(file) == True:19 file_extensions.append(file.split('.')[-1])20 21 file_extensions = list(set(file_extensions))22 23 return file_extensions24# create directories to move files25def make_directories(organize_dirctrory_path):26 file_extensions = check_file_extensions(organize_dirctrory_path)27 created_directory = []28 for file_extension in file_extensions:29 mkdir_to_path = organize_dirctrory_path + file_extension30 try:31 os.makedirs(mkdir_to_path)32 except FileExistsError:33 check = ''34 while True:35 print(f'すでに{organize_dirctrory_path}ディレクトリに{file_extension}ディレクトリが存在します。')36 print('このまま続けても良いですか？[y/​n]')37 check = input()38 if check == 'y' or check == 'n':39 break40 else:41 print('yかnを入力してください。')42 43 if check == 'y':44 continue45 elif check == 'n':46 break47 48 49# move files50def move_f(file_extension, move_path):51 file_ls = glob(move_path + '*')52 for file in file_ls:53 move_f = move_path + file[-3:]...

Full Screen

Full Screen

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run dbt-osmosis automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

... View sample repo... View DocumentationSign up Free
›
_

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

...
Sign up with GoogleSign up with Email
Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful