Best Python code snippet using behave
users_metadata.py
Source:users_metadata.py
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...
main.py
Source:main.py
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))...
app.py
Source:app.py
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__':...
test_crop_coordinates.py
Source:test_crop_coordinates.py
...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__':...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!