How to use logme method in Playwright Python

Best Python code snippet using playwright-python

Run Playwright Python automation tests on LambdaTest cloud grid

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

output.py

Source: output.py Github

copy
1from datetime import datetime
2
3from . import format, get
4from .tweet import Tweet
5from .user import User
6from .storage import db, elasticsearch, write, panda
7
8import logging as logme
9
10follows_list = []
11tweets_list = []
12users_list = []
13
14author_list = {''}
15author_list.pop()
16
17# used by Pandas
18_follows_object = {}
19
20def _clean_follow_list():
21    logme.debug(__name__+':clean_follow_list')
22    global _follows_object
23    _follows_object = {}
24
25def clean_lists():
26    logme.debug(__name__+':clean_lists')
27    global follows_list
28    global tweets_list
29    global users_list
30    follows_list = []
31    tweets_list = []
32    users_list = []
33
34def datecheck(datetimestamp, config):
35    logme.debug(__name__+':datecheck')
36    if config.Since:
37        logme.debug(__name__+':datecheck:SinceTrue')
38        d = int(datetime.strptime(datetimestamp, "%Y-%m-%d %H:%M:%S").timestamp())
39        s = int(datetime.strptime(config.Since, "%Y-%m-%d %H:%M:%S").timestamp())
40        if d < s:
41           return False
42    if config.Until:
43        logme.debug(__name__+':datecheck:UntilTrue')
44        d = int(datetime.strptime(datetimestamp, "%Y-%m-%d %H:%M:%S").timestamp())
45        s = int(datetime.strptime(config.Until, "%Y-%m-%d %H:%M:%S").timestamp())
46        if d > s:
47           return False
48    logme.debug(__name__+':datecheck:dateRangeFalse')
49    return True
50
51def is_tweet(tw):
52    try:
53        tw["data-item-id"]
54        logme.debug(__name__+':is_tweet:True')
55        return True
56    except:
57        logme.critical(__name__+':is_tweet:False')
58        return False
59
60def _output(obj, output, config, **extra):
61    logme.debug(__name__+':_output')
62    if config.Lowercase:
63        if isinstance(obj, str):
64            logme.debug(__name__+':_output:Lowercase:username')
65            obj = obj.lower()
66        elif obj.__class__.__name__ == "user":
67            logme.debug(__name__+':_output:Lowercase:user')
68            pass
69        elif obj.__class__.__name__ == "tweet":
70            logme.debug(__name__+':_output:Lowercase:tweet')
71            obj.username = obj.username.lower()
72            author_list.update({obj.username})
73            for i in range(len(obj.mentions)):
74                obj.mentions[i] = obj.mentions[i].lower()
75            for i in range(len(obj.hashtags)):
76                obj.hashtags[i] = obj.hashtags[i].lower()
77            for i in range(len(obj.cashtags)):
78                obj.cashtags[i] = obj.cashtags[i].lower()
79        else:
80            logme.info('_output:Lowercase:hiddenTweetFound')
81            print("[x] Hidden tweet found, account suspended due to violation of TOS")
82            return
83    if config.Output != None:
84        if config.Store_csv:
85            try:
86                write.Csv(obj, config)
87                logme.debug(__name__+':_output:CSV')
88            except Exception as e:
89                logme.critical(__name__+':_output:CSV:Error:' + str(e))
90                print(str(e) + " [x] output._output")
91        elif config.Store_json:
92            write.Json(obj, config)
93            logme.debug(__name__+':_output:JSON')
94        else:
95            write.Text(output, config.Output)
96            logme.debug(__name__+':_output:Text')
97
98    if config.Elasticsearch:
99        logme.debug(__name__+':_output:Elasticsearch')
100        print("", end=".", flush=True)
101    else:
102        if not config.Hide_output:
103            try:
104                print(output.replace('\n', ' '))
105            except UnicodeEncodeError:
106                logme.critical(__name__+':_output:UnicodeEncodeError')
107                print("unicode error [x] output._output")
108
109async def checkData(tweet, config, conn):
110    logme.debug(__name__+':checkData')
111    copyright = tweet.find("div", "StreamItemContent--withheld")
112    if copyright is None and is_tweet(tweet):
113        tweet = Tweet(tweet, config)
114
115        if not tweet.datestamp:
116            logme.critical(__name__+':checkData:hiddenTweetFound')
117            print("[x] Hidden tweet found, account suspended due to violation of TOS")
118            return
119
120        if datecheck(tweet.datestamp + " " + tweet.timestamp, config):
121            output = format.Tweet(config, tweet)
122
123            if config.Database:
124                logme.debug(__name__+':checkData:Database')
125                db.tweets(conn, tweet, config)
126
127            if config.Pandas:
128                logme.debug(__name__+':checkData:Pandas')
129                panda.update(tweet, config)
130
131            if config.Store_object:
132                logme.debug(__name__+':checkData:Store_object')
133                if hasattr(config.Store_object_tweets_list, 'append'):
134                    config.Store_object_tweets_list.append(tweet)
135                else:
136                    tweets_list.append(tweet)
137
138            if config.Elasticsearch:
139                logme.debug(__name__+':checkData:Elasticsearch')
140                elasticsearch.Tweet(tweet, config)
141
142            _output(tweet, output, config)
143    else:
144        logme.critical(__name__+':checkData:copyrightedTweet')
145
146async def Tweets(tweets, config, conn, url=''):
147    logme.debug(__name__+':Tweets')
148    if config.Favorites or config.Profile_full or config.Location:
149        logme.debug(__name__+':Tweets:fav+full+loc')
150        for tw in tweets:
151            if tw['data-item-id'] == url.split('?')[0].split('/')[-1]:
152                await checkData(tw, config, conn)
153    elif config.TwitterSearch:
154        logme.debug(__name__+':Tweets:TwitterSearch')
155        await checkData(tweets, config, conn)
156    else:
157        logme.debug(__name__+':Tweets:else')
158        if int(tweets["data-user-id"]) == config.User_id or config.Retweets:
159            await checkData(tweets, config, conn)
160
161async def Users(u, config, conn):
162    logme.debug(__name__+':User')
163    global users_list
164
165    user = User(u)
166    output = format.User(config.Format, user)
167
168    if config.Database:
169        logme.debug(__name__+':User:Database')
170        db.user(conn, config, user)
171
172    if config.Elasticsearch:
173        logme.debug(__name__+':User:Elasticsearch')
174        _save_date = user.join_date
175        _save_time = user.join_time
176        user.join_date = str(datetime.strptime(user.join_date, "%d %b %Y")).split()[0]
177        user.join_time = str(datetime.strptime(user.join_time, "%I:%M %p")).split()[1]
178        elasticsearch.UserProfile(user, config)
179        user.join_date = _save_date
180        user.join_time = _save_time
181
182    if config.Store_object:
183        logme.debug(__name__+':User:Store_object')
184
185        if config.Followers or config.Following:
186            if hasattr(config.Store_object_follow_list, 'append'):
187                config.Store_object_follow_list.append(user)
188            else:
189                users_list.append(user) # twint.user.user
190
191    if config.Pandas:
192        logme.debug(__name__+':User:Pandas+user')
193        panda.update(user, config)
194
195    _output(user, output, config)
196
197async def Username(username, config, conn):
198    logme.debug(__name__+':Username')
199    global _follows_object
200    global follows_list
201    follow_var = config.Following*"following" + config.Followers*"followers"
202
203    if config.Database:
204        logme.debug(__name__+':Username:Database')
205        db.follow(conn, config.Username, config.Followers, username)
206
207    if config.Elasticsearch:
208        logme.debug(__name__+':Username:Elasticsearch')
209        elasticsearch.Follow(username, config)
210
211    if config.Store_object:
212        if hasattr(config.Store_object_follow_list, 'append'):
213            config.Store_object_follow_list.append(username)
214        else:
215            follows_list.append(username) # twint.user.user
216
217    if config.Pandas:
218        logme.debug(__name__+':Username:object+pandas')
219        try:
220            _ = _follows_object[config.Username][follow_var]
221        except KeyError:
222            _follows_object.update({config.Username: {follow_var: []}})
223        _follows_object[config.Username][follow_var].append(username)
224        if config.Pandas_au:
225            logme.debug(__name__+':Username:object+pandas+au')
226            panda.update(_follows_object[config.Username], config)
227    _output(username, username, config)
228
Full Screen

run.py

Source: run.py Github

copy
1import sys, os, time
2from asyncio import get_event_loop, TimeoutError, ensure_future, new_event_loop, set_event_loop
3
4from . import datelock, feed, get, output, verbose, storage
5from .storage import db
6
7import logging as logme
8
9class Twint:
10    def __init__(self, config):
11        logme.debug(__name__+':Twint:__init__')
12        if config.Resume is not None and (config.TwitterSearch or config.Followers or config.Following):
13            logme.debug(__name__+':Twint:__init__:Resume')
14            self.init = self.get_resume(config.Resume)
15        else:
16            self.init = '-1'
17
18        self.feed = [-1]
19        self.count = 0
20        self.user_agent = ""
21        self.config = config
22        self.conn = db.Conn(config.Database)
23        self.d = datelock.Set(self.config.Until, self.config.Since)
24        verbose.Elastic(config.Elasticsearch)
25
26        if self.config.Store_object:
27            logme.debug(__name__+':Twint:__init__:clean_follow_list')
28            output._clean_follow_list()
29
30        if self.config.Pandas_clean:
31            logme.debug(__name__+':Twint:__init__:pandas_clean')
32            storage.panda.clean()
33
34    def get_resume(self, resumeFile):
35        if not os.path.exists(resumeFile):
36            return '-1'
37        with open(resumeFile, 'r') as rFile:
38            _init = rFile.readlines()[-1].strip('\n')
39            return _init
40
41    async def Feed(self):
42        logme.debug(__name__+':Twint:Feed')
43        consecutive_errors_count = 0
44        while True:
45            response = await get.RequestUrl(self.config, self.init, headers=[("User-Agent", self.user_agent)])
46            if self.config.Debug:
47                print(response, file=open("twint-last-request.log", "w", encoding="utf-8"))
48
49            self.feed = []
50            try:
51                if self.config.Favorites:
52                    self.feed, self.init = feed.Mobile(response)
53                    if not self.count % 40:
54                        time.sleep(5)
55                elif self.config.Followers or self.config.Following:
56                    self.feed, self.init = feed.Follow(response)
57                    if not self.count % 40:
58                        time.sleep(5)
59                elif self.config.Profile:
60                    if self.config.Profile_full:
61                        self.feed, self.init = feed.Mobile(response)
62                    else:
63                        self.feed, self.init = feed.profile(response)
64                elif self.config.TwitterSearch:
65                    self.feed, self.init = feed.Json(response)
66                break
67            except TimeoutError as e:
68                if self.config.Proxy_host.lower() == "tor":
69                    print("[?] Timed out, changing Tor identity...")
70                    if self.config.Tor_control_password is None:
71                        logme.critical(__name__+':Twint:Feed:tor-password')
72                        sys.stderr.write("Error: config.Tor_control_password must be set for proxy autorotation!\r\n")
73                        sys.stderr.write("Info: What is it? See https://stem.torproject.org/faq.html#can-i-interact-with-tors-controller-interface-directly\r\n")
74                        break
75                    else:
76                        get.ForceNewTorIdentity(self.config)
77                        continue
78                else:
79                    logme.critical(__name__+':Twint:Feed:' + str(e))
80                    print(str(e))
81                    break
82            except Exception as e:
83                if self.config.Profile or self.config.Favorites:
84                    print("[!] Twitter does not return more data, scrape stops here.")
85                    break
86                logme.critical(__name__+':Twint:Feed:noData' + str(e))
87                # Sometimes Twitter says there is no data. But it's a lie.
88                consecutive_errors_count += 1
89                if consecutive_errors_count < self.config.Retries_count:
90                    # skip to the next iteration if wait time does not satisfy limit constraints
91                    delay = round(consecutive_errors_count ** self.config.Backoff_exponent, 1)
92
93                    # if the delay is less than users set min wait time then replace delay
94                    if self.config.Min_wait_time > delay:
95                        delay = self.config.Min_wait_time
96
97                    sys.stderr.write('sleeping for {} secs\n'.format(delay))
98                    time.sleep(delay)
99                    self.user_agent = await get.RandomUserAgent(wa=True)
100                    continue
101                logme.critical(__name__+':Twint:Feed:Tweets_known_error:' + str(e))
102                sys.stderr.write(str(e) + " [x] run.Feed")
103                sys.stderr.write("[!] if get this error but you know for sure that more tweets exist, please open an issue and we will investigate it!")
104                break
105        if self.config.Resume:
106            print(self.init, file=open(self.config.Resume, "a", encoding="utf-8"))
107
108    async def follow(self):
109        await self.Feed()
110        if self.config.User_full:
111            logme.debug(__name__+':Twint:follow:userFull')
112            self.count += await get.Multi(self.feed, self.config, self.conn)
113        else:
114            logme.debug(__name__+':Twint:follow:notUserFull')
115            for user in self.feed:
116                self.count += 1
117                username = user.find("a")["name"]
118                await output.Username(username, self.config, self.conn)
119
120    async def favorite(self):
121        logme.debug(__name__+':Twint:favorite')
122        await self.Feed()
123        self.count += await get.Multi(self.feed, self.config, self.conn)
124
125    async def profile(self):
126        await self.Feed()
127        if self.config.Profile_full:
128            logme.debug(__name__+':Twint:profileFull')
129            self.count += await get.Multi(self.feed, self.config, self.conn)
130        else:
131            logme.debug(__name__+':Twint:notProfileFull')
132            for tweet in self.feed:
133                self.count += 1
134                await output.Tweets(tweet, self.config, self.conn)
135
136    async def tweets(self):
137        await self.Feed()
138        if self.config.Location:
139            logme.debug(__name__+':Twint:tweets:location')
140            self.count += await get.Multi(self.feed, self.config, self.conn)
141        else:
142            logme.debug(__name__+':Twint:tweets:notLocation')
143            for tweet in self.feed:
144                self.count += 1
145                await output.Tweets(tweet, self.config, self.conn)
146
147    async def main(self, callback=None):
148
149        task = ensure_future(self.run())  # Might be changed to create_task in 3.7+.
150
151        if callback:
152            task.add_done_callback(callback)
153
154        await task
155
156    async def run(self):
157        if self.config.TwitterSearch:
158            self.user_agent = await get.RandomUserAgent(wa=True)
159        else:
160            self.user_agent = await get.RandomUserAgent()
161
162        if self.config.User_id is not None:
163            logme.debug(__name__+':Twint:main:user_id')
164            self.config.Username = await get.Username(self.config.User_id)
165
166        if self.config.Username is not None:
167            logme.debug(__name__+':Twint:main:username')
168            url = f"https://twitter.com/{self.config.Username}?lang=en"
169            self.config.User_id = await get.User(url, self.config, self.conn, True)
170
171        if self.config.TwitterSearch and self.config.Since and self.config.Until:
172            logme.debug(__name__+':Twint:main:search+since+until')
173            while self.d._since < self.d._until:
174                self.config.Since = str(self.d._since)
175                self.config.Until = str(self.d._until)
176                if len(self.feed) > 0:
177                    await self.tweets()
178                else:
179                    logme.debug(__name__+':Twint:main:gettingNewTweets')
180                    break
181
182                if get.Limit(self.config.Limit, self.count):
183                    break
184        else:
185            logme.debug(__name__+':Twint:main:not-search+since+until')
186            while True:
187                if len(self.feed) > 0:
188                    if self.config.Followers or self.config.Following:
189                        logme.debug(__name__+':Twint:main:follow')
190                        await self.follow()
191                    elif self.config.Favorites:
192                        logme.debug(__name__+':Twint:main:favorites')
193                        await self.favorite()
194                    elif self.config.Profile:
195                        logme.debug(__name__+':Twint:main:profile')
196                        await self.profile()
197                    elif self.config.TwitterSearch:
198                        logme.debug(__name__+':Twint:main:twitter-search')
199                        await self.tweets()
200                else:
201                    logme.debug(__name__+':Twint:main:no-more-tweets')
202                    break
203
204                #logging.info("[<] " + str(datetime.now()) + ':: run+Twint+main+CallingGetLimit2')
205                if get.Limit(self.config.Limit, self.count):
206                    logme.debug(__name__+':Twint:main:reachedLimit')
207                    break
208
209        if self.config.Count:
210            verbose.Count(self.count, self.config)
211
212def run(config, callback=None):
213    logme.debug(__name__+':run')
214    try:
215        get_event_loop()
216    except RuntimeError as e:
217        if "no current event loop" in str(e):
218            set_event_loop(new_event_loop())
219        else:
220            logme.exception(__name__+':Lookup:Unexpected exception while handling an expected RuntimeError.')
221            raise
222    except Exception as e:
223        logme.exception(__name__+':Lookup:Unexpected exception occured while attempting to get or create a new event loop.')
224        raise
225
226    get_event_loop().run_until_complete(Twint(config).main(callback))
227
228def Favorites(config):
229    logme.debug(__name__+':Favorites')
230    config.Favorites = True
231    config.Following = False
232    config.Followers = False
233    config.Profile = False
234    config.Profile_full = False
235    config.TwitterSearch = False
236    run(config)
237    if config.Pandas_au:
238        storage.panda._autoget("tweet")
239
240def Followers(config):
241    logme.debug(__name__+':Followers')
242    config.Followers = True
243    config.Following = False
244    config.Profile = False
245    config.Profile_full = False
246    config.Favorites = False
247    config.TwitterSearch = False
248    run(config)
249    if config.Pandas_au:
250        storage.panda._autoget("followers")
251        if config.User_full:
252            storage.panda._autoget("user")
253    if config.Pandas_clean and not config.Store_object:
254        #storage.panda.clean()
255        output._clean_follow_list()
256
257def Following(config):
258    logme.debug(__name__+':Following')
259    config.Following = True
260    config.Followers = False
261    config.Profile = False
262    config.Profile_full = False
263    config.Favorites = False
264    config.TwitterSearch = False
265    run(config)
266    if config.Pandas_au:
267        storage.panda._autoget("following")
268        if config.User_full:
269            storage.panda._autoget("user")
270    if config.Pandas_clean and not config.Store_object:
271        #storage.panda.clean()
272        output._clean_follow_list()
273
274def Lookup(config):
275    logme.debug(__name__+':Lookup')
276
277    try:
278        get_event_loop()
279    except RuntimeError as e:
280        if "no current event loop" in str(e):
281            set_event_loop(new_event_loop())
282        else:
283            logme.exception(__name__+':Lookup:Unexpected exception while handling an expected RuntimeError.')
284            raise
285    except Exception as e:
286        logme.exception(__name__+':Lookup:Unexpected exception occured while attempting to get or create a new event loop.')
287        raise
288
289    try:
290        if config.User_id is not None:
291            logme.debug(__name__+':Twint:Lookup:user_id')
292            config.Username = get_event_loop().run_until_complete(get.Username(config.User_id))
293
294        url = f"https://twitter.com/{config.Username}?lang=en"
295        get_event_loop().run_until_complete(get.User(url, config, db.Conn(config.Database)))
296
297        if config.Pandas_au:
298            storage.panda._autoget("user")
299    except RuntimeError as e:
300        if "no current event loop" in str(e):
301            logme.exception(__name__+':Lookup:Previous attempt to to create an event loop failed.')
302
303        raise
304    except Exception as e:
305        logme.exception(__name__+':Lookup:Unexpected exception occured.')
306        raise
307
308def Profile(config):
309    logme.debug(__name__+':Profile')
310    config.Profile = True
311    config.Favorites = False
312    config.Following = False
313    config.Followers = False
314    config.TwitterSearch = False
315    run(config)
316    if config.Pandas_au:
317        storage.panda._autoget("tweet")
318
319def Search(config, callback=None):
320    logme.debug(__name__+':Search')
321    config.TwitterSearch = True
322    config.Favorites = False
323    config.Following = False
324    config.Followers = False
325    config.Profile = False
326    config.Profile_full = False
327    run(config, callback)
328    if config.Pandas_au:
329        storage.panda._autoget("tweet")
330
Full Screen

Accelerate Your Automation Test Cycles With LambdaTest

Leverage LambdaTest’s cloud-based platform to execute your automation tests in parallel and trim down your test execution time significantly. Your first 100 automation testing minutes are on us.

Try LambdaTest

Run Python Tests on LambdaTest Cloud Grid

Execute automation tests with Playwright Python on a cloud-based Grid of 3000+ real browsers and operating systems for both web and mobile applications.

Test now for Free
LambdaTestX

We use cookies to give you the best experience. Cookies help to provide a more personalized experience and relevant advertising for you, and web analytics for us. Learn More in our Cookies policy, Privacy & Terms of service

Allow Cookie
Sarah

I hope you find the best code examples for your project.

If you want to accelerate automated browser testing, try LambdaTest. Your first 100 automation testing minutes are FREE.

Sarah Elson (Product & Growth Lead)