How to use json_load method in Behave

Best Python code snippet using behave

users_metadata.py

Source:users_metadata.py Github

copy

Full Screen

1# coding=utf-82import tweepy3import json4import psycopg25import io6import time7from http.client import IncompleteRead8import sys9import csv10from math import ceil11import time12import oauth2 as oauth13from tweepy import TweepError14from tweepy import RateLimitError15conn_string = "host='localhost' dbname='referendum_catalonia_db' user='postgres' password='########'"16CONSUMER_KEY = 'kHvNMDJhGKPkhi66P3XtqhkHI'17CONSUMER_SECRET = 'SZdo7R8JQ2KXQP5exwOdspNqN5GAgvraxj3ZQUutDQ6sbuaS1X'18ACCESS_TOKEN = '785229010531979264-OZjOEpdk3ecsEkyXekVGZJ24nVwOkj4'19ACCESS_TOKEN_SECRET = 'pIRRqtGsDlTGj2T8r7DOXQhd0cxpLroY4DhZ8vuhaxPIr'20CONSUMER_KEY2 = 'Wh3kQCeravcDqJ0i3IP4Q81dX'21CONSUMER_SECRET2 = 'zfWPaIWwPrhCNHvCBayXzbQVOQ4hIiEv9LGzZU2RxgHj6U0NQp'22ACCESS_TOKEN2 = '785229010531979264-7NAa3Jq90Zlzgqvxcmth69TBuW3MT49'23ACCESS_TOKEN_SECRET2 = 'x8JVhBnURwdj5qvkI64AlzlNsgSn92jBGItJ9s8q0F0k7'24CONSUMER_KEY3 = 'yMoy4FQU4m4Cmzte66imeI5f3'25CONSUMER_SECRET3 = 'ovesXl7vadgYEnN31iZMAvtkRRuUiFsbC9cHN8xlvkFQk1Si36'26ACCESS_TOKEN3 = '785229010531979264-dsRAp13EO2Vm5EJTeDdmhCF3aEXvYAM'27ACCESS_TOKEN_SECRET3 = 'vhk8UcNruuxlr3jUdFIlWt16pZ2NhynTlJaGkUVJtg58T'28auth = tweepy.OAuthHandler(CONSUMER_KEY, CONSUMER_SECRET)29auth.set_access_token(ACCESS_TOKEN, ACCESS_TOKEN_SECRET)30auth2 = tweepy.OAuthHandler(CONSUMER_KEY2, CONSUMER_SECRET2)31auth2.set_access_token(ACCESS_TOKEN2, ACCESS_TOKEN_SECRET2)32auth3 = tweepy.OAuthHandler(CONSUMER_KEY3, CONSUMER_SECRET3)33auth3.set_access_token(ACCESS_TOKEN3, ACCESS_TOKEN_SECRET3)34apis = []35apis_len = 336apis.append(tweepy.API(auth, wait_on_rate_limit=True, wait_on_rate_limit_notify=True))37apis.append(tweepy.API(auth2, wait_on_rate_limit=True, wait_on_rate_limit_notify=True))38apis.append(tweepy.API(auth3, wait_on_rate_limit=True, wait_on_rate_limit_notify=True))39with open('/data/project/catalonia/userids_new.csv', encoding='utf8', mode='r') as userids_csv:40 reader = csv.reader(userids_csv)41 userids_list = list(reader)42userids = [int(userid[0]) for userid in userids_list]43iterations = len(userids)44c = 045apikey_iter = 3746for i in range(c, iterations): 47 try: 48 user = apis[apikey_iter].get_user(userids[c]) 49 c += 150 print(str(c))51 if c % 300 == 0:52 apikey_iter = (apikey_iter + 1) % apis_len 53 54 userid = -155 utc_offset = -156 profile_image_url = None57 url = None58 protected = None59 name = None60 screen_name = None61 location = None 62 description = None63 verified = None 64 followers_count = -165 friends_count = -166 listed_count = -167 favourites_count = -168 statuses_count = -169 created_at = None 70 time_zone = None71 geo_enabled = None 72 lang = None73 74 json_string = json.dumps(user._json)75 json_load = json.loads(json_string)76 if("id" in json_load):77 if(json_load["id"] != None):78 userid = json_load["id"] 79 if("utc_offset" in json_load):80 if(json_load["utc_offset"] != None):81 utc_offset = json_load["utc_offset"]82 if("profile_image_url" in json_load):83 if(json_load["profile_image_url"] != None):84 profile_image_url = json_load["profile_image_url"]85 if("url" in json_load):86 if(json_load["url"] != None):87 url = json_load["url"]88 if("protected" in json_load):89 if(json_load["protected"] != None):90 protected = json_load["protected"]91 if("name" in json_load):92 if(json_load["name"] != None):93 name = json_load["name"]94 if("screen_name" in json_load):95 if(json_load["screen_name"] != None):96 screen_name = json_load["screen_name"] 97 if("location" in json_load):98 if(json_load["location"] != None):99 location = json_load["location"]100 if("description" in json_load):101 if(json_load["description"] != None):102 description = json_load["description"]103 if("verified" in json_load):104 if(json_load["verified"] != None):105 verified = json_load["verified"] 106 if("followers_count" in json_load):107 if(json_load["followers_count"] != None):108 followers_count = json_load["followers_count"]109 if("friends_count" in json_load):110 if(json_load["friends_count"] != None):111 friends_count = json_load["friends_count"]112 if("listed_count" in json_load):113 if(json_load["listed_count"] != None):114 listed_count = json_load["listed_count"]115 if("favourites_count" in json_load):116 if(json_load["favourites_count"] != None):117 favourites_count = json_load["favourites_count"]118 if("statuses_count" in json_load):119 if(json_load["statuses_count"] != None):120 statuses_count = json_load["statuses_count"] 121 if("created_at" in json_load):122 if(json_load["created_at"] != None):123 created_at = json_load["created_at"]124 if("time_zone" in json_load):125 if(json_load["time_zone"] != None):126 time_zone = json_load["time_zone"]127 if("geo_enabled" in json_load):128 if(json_load["geo_enabled"] != None):129 geo_enabled = json_load["geo_enabled"]130 if("lang" in json_load):131 if(json_load["lang"] != None):132 lang = json_load["lang"]133 134 try: 135 conn = psycopg2.connect(conn_string)136 conn.autocommit = True137 cur = conn.cursor() 138 cur.execute("INSERT INTO users_metadata(userid, name, screen_name, location, description, verified, followers_count, friends_count, listed_count, favourites_count, statuses_count, created_at, time_zone, geo_enabled, lang, utc_offset, profile_image_url, url, protected) VALUES(%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)", (userid, name, screen_name, location, description, verified, followers_count, friends_count, listed_count, favourites_count, statuses_count, created_at, time_zone, geo_enabled, lang, utc_offset, profile_image_url, url, protected, ))139 cur.close() 140 except Exception as e:141 print("Twitter User insertion failed! " + str(e))142 error_type = sys.exc_info()[0]143 error_value = sys.exc_info()[1]144 print('ERROR:', error_type, error_value)145 pass 146 finally:147 conn.close()148 except TweepError as e:149 print('Error string: ' + e.response.text) 150 pass151 except RateLimitError:152 apikey_iter = (apikey_iter + 1) % clients_len...

Full Screen

Full Screen

main.py

Source:main.py Github

copy

Full Screen

1from flask import Flask, request, abort2import datetime3import datetime as dt4import json5import time6import asyncio7 8from linebot import (9 LineBotApi, WebhookHandler10)11from linebot.exceptions import (12 InvalidSignatureError13)14from linebot.models import (15 MessageEvent, TextMessage, TextSendMessage,16)17import os18 19app = Flask(__name__)20 21#環境変数取得22YOUR_CHANNEL_ACCESS_TOKEN = os.environ["YOUR_CHANNEL_ACCESS_TOKEN"]23YOUR_CHANNEL_SECRET = os.environ["YOUR_CHANNEL_SECRET"]24line_bot_api=LineBotApi(YOUR_CHANNEL_ACCESS_TOKEN)25handler=WebhookHandler(YOUR_CHANNEL_SECRET)26 27 28## 1 ##29#Webhookからのリクエストをチェックします。30@app.route("/callback", methods=['POST'])31def callback():32 # リクエストヘッダーから署名検証のための値を取得します。33 signature = request.headers['X-Line-Signature']34 35 # リクエストボディを取得します。36 body = request.get_data(as_text=True)37 app.logger.info("Request body: " + body)38 39 # handle webhook body40 # 署名を検証し、問題なければhandleに定義されている関数を呼び出す。41 try:42 handler.handle(body, signature)43 # 署名検証で失敗した場合、例外を出す。44 except InvalidSignatureError:45 abort(400)46 #handleの処理を終えればOK47 return 'OK'48 49 50@handler.add(MessageEvent, message=TextMessage)51def handle_message(event):52 if "授業" in event.message.text:53 content = "明日の授業は..."54 json_open = open('week2.json', 'r',encoding="utf-8")55 json_load = json.load(json_open)56 week = dt.datetime.now().strftime("%A")57 indx = len(json_load[week]["class"])58 if indx == 1:59 buffa = json_load[week]["week"] + json_load[week]["class"][indx - 1]60 content = buffa61 if week == "Monday":62 buffa = json_load[week]["class"][indx - 2] + json_load[week]["time"][indx - 2] + json_load[week]["class"][indx - 1] + json_load[week]["time"][indx - 1]63 content - buffa64 if week == "Wednesday":65 buffa = json_load[week]["class"][indx - 3] + json_load[week]["time"][indx - 3] + json_load[week]["class"][indx - 2] + json_load[week]["time"][indx - 2] + json_load[week]["class"][indx - 1] + json_load[week]["time"][indx - 1]66 content = buffa67 if week == "Thursday":68 buffa = json_load[week]["class"][indx - 4]69 + json_load[week]["time"][indx - 4]70 + json_load[week]["class"][indx - 3]71 + json_load[week]["time"][indx - 3]72 + json_load[week]["class"][indx - 2]73 + json_load[week]["time"][indx - 2]74 + json_load[week]["class"][indx - 1]75 + json_load[week]["time"][indx - 1]76 content = buffa77 if week == "Tuesday":78 buffa = json_load[week]["class"][indx - 5]79 + json_load[week]["time"][indx - 5]80 + json_load[week]["class"][indx - 4]81 + json_load[week]["time"][indx - 4]82 + json_load[week]["class"][indx - 3]83 + json_load[week]["time"][indx - 3]84 + json_load[week]["class"][indx - 2]85 + json_load[week]["time"][indx - 2]86 + json_load[week]["class"][indx - 1]87 + json_load[week]["time"][indx - 1]88 content = buffa89 if week == "Friday":90 buffa = json_load[week]["class"][indx - 2] + json_load[week]["time"][indx - 2] + json_load[week]["class"][indx - 1] + json_load[week]["time"][indx - 1]91 content = buffa92 line_bot_api.reply_message(93 event.reply_token,94 TextSendMessage(text=content))95# ポート番号の設定96if __name__ == "__main__":97# app.run()98 port = int(os.getenv("PORT", 5000))...

Full Screen

Full Screen

app.py

Source:app.py Github

copy

Full Screen

1import os2from flask import render_template, request, json3from tweepy import OAuthHandler4from tweepy import Stream5from tweepy.streaming import StreamListener6from Tweets import create_app7from collections import Counter8from textblob import TextBlob9import operator10config_name = os.getenv('FLASK_CONFIG', 'development')11app = create_app(config_name)12ckey = "xDHW87silQGmPgtEgE2M1wWdp"13csecret = "lTt3WgG9dHj6IdIadGDCB1SHGe3KtDzhe0xT0yeWkhDgntTojE"14atoken = "1342146790976487431-fHbYzoQHHybPuIT3XUwqyJlIIJda21"15asecret = "dngthM03Sh1b4t7bi999pt83DM9HEKAE1wxSWCw2AfXzy"16@app.route('/', methods=['GET', 'POST'])17def twt_search():18 if request.method == 'GET':19 return render_template('index.html')20 elif request.method == 'POST':21 tweets = []22 lst_tweets = []23 class Listener(StreamListener):24 def on_data(self, data):25 json_load = json.loads(data)26 dict_tweets = dict()27 for i in json_load:28 analysis = TextBlob(json_load["text"])29 print(analysis.sentiment)30 if analysis.sentiment[0] > 0:31 dict_tweets['sentiment'] = "Positive"32 elif analysis.sentiment[0] < 0:33 dict_tweets['sentiment'] = "Negative"34 else:35 dict_tweets['sentiment'] = "Neutral"36 dict_tweets["created_at"] = json_load["created_at"]37 dict_tweets["text"] = json_load["text"]38 dict_tweets["favourites_count"] = json_load["user"]["favourites_count"]39 dict_tweets["followers_count"] = json_load["user"]["followers_count"]40 dict_tweets["friends_count"] = json_load["user"]["friends_count"]41 dict_tweets["profile_image_url_https"] = json_load["user"]["profile_image_url_https"]42 dict_tweets["retweet_count"] = json_load.get('retweeted_status', {}).get('retweet_count',43 json_load.get('quoted_status',44 {}).get(45 'retweet_count'))46 lst_tweets.append(dict_tweets)47 tweets.append(json_load['text'])48 if len(tweets) == 10:49 return False50 return True51 keyword = request.form.get('keyword')52 auth = OAuthHandler(ckey, csecret)53 auth.set_access_token(atoken, asecret)54 twitterstream = Stream(auth, Listener())55 twitterstream.filter(languages=['en'], track=[keyword])56 lst_words = []57 for i in tweets:58 lst_words += i.split()59 dict_twt_word = Counter(lst_words)60 del dict_twt_word["RT"]61 word = max(dict_twt_word.items(), key=operator.itemgetter(1))[0]62 return render_template('tweet.html', tweets=lst_tweets, word=word)63if __name__ == '__main__':...

Full Screen

Full Screen

test_crop_coordinates.py

Source:test_crop_coordinates.py Github

copy

Full Screen

...19 my_case = [[1, 2, 3], [4, 5, 6]]20 with make_case(my_case) as (input, output):21 main(["--input", input,22 "--output", output])23 result = self.json_load(output)24 self.assertSequenceEqual(my_case, result)25 def json_load(self, output):26 with open(output) as fd:27 return json.load(fd)28 def test_x0(self):29 my_case = [[1, 2, 3], [4, 5, 6]]30 with make_case(my_case) as (input, output):31 main(["--input", input,32 "--output", output,33 "--x0", "4"])34 result = self.json_load(output)35 self.assertSequenceEqual(my_case[1:], result)36 def test_x1(self):37 my_case = [[1, 2, 3], [4, 5, 6]]38 with make_case(my_case) as (input, output):39 main(["--input", input,40 "--output", output,41 "--x1", "4"])42 result = self.json_load(output)43 self.assertSequenceEqual(my_case[:1], result)44 def test_y0(self):45 my_case = [[1, 2, 3], [4, 5, 6]]46 with make_case(my_case) as (input, output):47 main(["--input", input,48 "--output", output,49 "--y0", "5"])50 result = self.json_load(output)51 self.assertSequenceEqual(my_case[1:], result)52 def test_y1(self):53 my_case = [[1, 2, 3], [4, 5, 6]]54 with make_case(my_case) as (input, output):55 main(["--input", input,56 "--output", output,57 "--y1", "5"])58 result = self.json_load(output)59 self.assertSequenceEqual(my_case[:1], result)60 def test_z0(self):61 my_case = [[1, 2, 3], [4, 5, 6]]62 with make_case(my_case) as (input, output):63 main(["--input", input,64 "--output", output,65 "--z0", "6"])66 result = self.json_load(output)67 self.assertSequenceEqual(my_case[1:], result)68 def test_z1(self):69 my_case = [[1, 2, 3], [4, 5, 6]]70 with make_case(my_case) as (input, output):71 main(["--input", input,72 "--output", output,73 "--z1", "6"])74 result = self.json_load(output)75 self.assertSequenceEqual(my_case[:1], result)76if __name__ == '__main__':...

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 Behave automation tests on LambdaTest cloud grid

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

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful