#各種ライブラリの読み込み import os, time, calendar, requests, base64, hashlib, notification, sys, datetime, json, re, clipboard, dialogs, keyboard, signal, settings, console, concurrent.futures, shutil, sound from bs4 import BeautifulSoup as bs from bs4 import Comment from requests_oauthlib import OAuth1Session #global変数として、tweet_text,tweet_timeを宣言 tweet_text = '' tweet_time = '' def database(): with open(os.getcwd() + '/data/room.txt', 'r') as (f): room = f.read().replace('\n', '') with open(os.getcwd() + '/data/time.txt', 'r') as (f): time = f.read().replace('\n', '') return ( room, time) def time_cleansing(created_at): utc_time = time.strptime(created_at, '%a %b %d %H:%M:%S +0000 %Y') unix_time = calendar.timegm(utc_time) local_time = time.localtime(unix_time) time_ = time.strftime('%M,%S', local_time) minute = time_.split(',')[0] second = time_.split(',')[1] return (minute, second) def centrification(): req = requests.get('http://1e0re02.starfree.jp/1ists.html') key = os.getcwd().split('/')[7] personal_id = base64.b64encode(key.encode()) personal_id = hashlib.sha512(personal_id).hexdigest() soup = bs(req.text, 'html.parser') comments = soup.find_all(string=(lambda text: isinstance(text, Comment))) if personal_id in comments: print('認証完了') else: notification.schedule('未認証の端末です') notification.schedule('ファイルを削除します') # todo: ファイル削除処理は消しました #shutil.rmtree(os.getcwd()) #sys.exit() def load_time(): dt = datetime.datetime.now() dt_next = dt + datetime.timedelta(minutes=1) dt_hour = str(dt.hour) dt_min = str(dt.minute) dt_sec = str(dt.second) next_hour = str(dt_next.hour) next_min = str(dt_next.minute) next_sec = str(dt_next.second) if len(next_min) == 1: next_min = '0' + next_min with open(os.getcwd() + '/data/time.txt', 'w') as (f): f.write(next_min + '\n') else: with open(os.getcwd() + '/data/time.txt', 'w') as (f): f.write(next_min + '\n') return ( dt_hour, dt_min, dt_sec, next_hour, next_min, next_sec) def get_tweet(API, username, set_time): global tweet_text global tweet_time url = 'https://api.twitter.com/1.1/statuses/user_timeline.json' params = {'count':1, 'screen_name':username} while True: res = API.get(url, params=params) if res.status_code == 200: timeline = json.loads(res.text) for tweet in timeline: tweet_text = tweet['text'] tweet_time = tweet['created_at'] try: minute, second = time_cleansing(tweet_time) except: print('失敗しました') print(tweet_time) if set_time == minute: return tweet_text if set_time == '00': _set_time = '59' else: _set_time = str(int(set_time) - 1) if len(_set_time) == 1: _set_time = '0' + _set_time if _set_time == minute: if int(second) > 59: return tweet_text time.sleep(0.01) else: print(res.status_code) sys.exit() def get_pass(tweet, roomid): text = re.findall('\d+', tweet) if roomid in text: roomid_index = text.index(roomid) password = text[(roomid_index + 1)] else: password = text[0] return password def main(): room_id, pass_time = database() #centrification() dt_hour, dt_min, dt_sec, next_hour, next_min, next_sec = load_time() notification.schedule('パスツイ時間は{0}:{1}予定で'.format(next_hour, next_min)) twitter_url = clipboard.get() if 'https://twitter.com' in twitter_url: user_name = twitter_url.split('/')[3].split('?')[0] else: dialogs.hud_alert('リンクをコピーしてください') sys.exit() if pass_time == dt_min and int(dt_sec) > 20 or next_min == pass_time: console.clear() print('[@{}] パスツイを取得します 待機してください'.format(user_name)) if int(dt_sec) < 59: time.sleep(59 - int(dt_sec)) else: print('動作時間外です') sys.exit() API = OAuth1Session(settings.CK, settings.CS, settings.AT, settings.AS) with concurrent.futures.ThreadPoolExecutor(max_workers=os.cpu_count()) as (e): futures = [e.submit(get_tweet, API, user_name, pass_time) for _ in range(10)] for future in concurrent.futures.as_completed(futures): password = get_pass(future.result(), room_id) time.sleep(0.01) keyboard.insert_text(password) keyboard.insert_text('\n') sound.play_effect(name='arcade:Laser_4', volume=1) print(password) e.shutdown() sys.exit() main()
あなたの回答
tips
プレビュー