From dc87da6e61b023ac8171405d0fc527d78881767a Mon Sep 17 00:00:00 2001 From: vunhat_minh Date: Wed, 4 Jun 2014 12:00:18 +0900 Subject: [PATCH] delete build folder --- build/lib/rainbowstream/__init__.py | 0 build/lib/rainbowstream/colors.py | 83 ---- build/lib/rainbowstream/config.py | 15 - build/lib/rainbowstream/consumer.py | 3 - build/lib/rainbowstream/db.py | 43 -- build/lib/rainbowstream/interactive.py | 65 --- build/lib/rainbowstream/rainbow.py | 632 ------------------------- build/lib/rainbowstream/table_def.py | 20 - 8 files changed, 861 deletions(-) delete mode 100644 build/lib/rainbowstream/__init__.py delete mode 100644 build/lib/rainbowstream/colors.py delete mode 100644 build/lib/rainbowstream/config.py delete mode 100644 build/lib/rainbowstream/consumer.py delete mode 100644 build/lib/rainbowstream/db.py delete mode 100644 build/lib/rainbowstream/interactive.py delete mode 100644 build/lib/rainbowstream/rainbow.py delete mode 100644 build/lib/rainbowstream/table_def.py diff --git a/build/lib/rainbowstream/__init__.py b/build/lib/rainbowstream/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/build/lib/rainbowstream/colors.py b/build/lib/rainbowstream/colors.py deleted file mode 100644 index 4b8ab2c..0000000 --- a/build/lib/rainbowstream/colors.py +++ /dev/null @@ -1,83 +0,0 @@ -import random -import itertools -from functools import wraps -from termcolor import * -from pyfiglet import figlet_format - -grey = lambda x: colored(x, 'grey', attrs=['bold']) -red = lambda x: colored(x, 'red', attrs=['bold']) -green = lambda x: colored(x, 'green', attrs=['bold']) -yellow = lambda x: colored(x, 'yellow', attrs=['bold']) -blue = lambda x: colored(x, 'blue', attrs=['bold']) -magenta = lambda x: colored(x, 'magenta', attrs=['bold']) -cyan = lambda x: colored(x, 'cyan', attrs=['bold']) -white = lambda x: colored(x, 'white', attrs=['bold']) - -on_grey = lambda x: colored(x, 'white', 'on_grey', attrs=['bold']) -on_red = lambda x: colored(x, 'white', 'on_red', attrs=['bold']) -on_green = lambda x: colored(x, 'white', 'on_green', attrs=['bold']) -on_yellow = lambda x: colored(x, 'white', 'on_yellow', attrs=['bold']) -on_blue = lambda x: colored(x, 'white', 'on_blue', attrs=['bold']) -on_magenta = lambda x: colored(x, 'white', 'on_magenta', attrs=['bold']) -on_cyan = lambda x: colored(x, 'white', 'on_cyan', attrs=['bold']) -on_white = lambda x: colored(x, 'white', 'on_white', attrs=['bold']) - -colors_shuffle = [grey, red, green, yellow, blue, magenta, cyan] -background_shuffle = [ - on_grey, - on_red, - on_green, - on_yellow, - on_blue, - on_magenta, - on_cyan] -cyc = itertools.cycle(colors_shuffle[1:]) - - -def order_rainbow(s): - """ - Print a string with ordered color with each character - """ - c = [colors_shuffle[i % 7](s[i]) for i in xrange(len(s))] - return reduce(lambda x, y: x + y, c) - - -def random_rainbow(s): - """ - Print a string with random color with each character - """ - c = [random.choice(colors_shuffle)(i) for i in s] - return reduce(lambda x, y: x + y, c) - - -def Memoize(func): - """ - Memoize decorator - """ - cache = {} - - @wraps(func) - def wrapper(*args): - if args not in cache: - cache[args] = func(*args) - return cache[args] - return wrapper - - -@Memoize -def cycle_color(s): - """ - Cycle the colors_shuffle - """ - return next(cyc)(s) - - -def ascii_art(text): - """ - Draw the Ascii Art - """ - fi = figlet_format(text, font='doom') - print('\n'.join( - [next(cyc)(i) for i in fi.split('\n')] - ) - ) diff --git a/build/lib/rainbowstream/config.py b/build/lib/rainbowstream/config.py deleted file mode 100644 index a77c2cb..0000000 --- a/build/lib/rainbowstream/config.py +++ /dev/null @@ -1,15 +0,0 @@ -# Max Search record -SEARCH_MAX_RECORD = 5 -# Default home tweet -HOME_TWEET_NUM = 5 - -# Stream Domain -USER_DOMAIN = 'userstream.twitter.com' -PUBLIC_DOMAIN = 'stream.twitter.com' -SITE_DOMAIN = 'sitestream.twitter.com' -# Actually called -DOMAIN = USER_DOMAIN - -# Filter and Ignore list ex: ['@fat','@mdo'] -ONLY_LIST = [] -IGNORE_LIST = [] diff --git a/build/lib/rainbowstream/consumer.py b/build/lib/rainbowstream/consumer.py deleted file mode 100644 index a0cdc22..0000000 --- a/build/lib/rainbowstream/consumer.py +++ /dev/null @@ -1,3 +0,0 @@ -# This is RainbowStream App info -CONSUMER_KEY = 'Xk1DGhR1FJa4xjg7GbdogzLJw' -CONSUMER_SECRET = 'SpHtDmbSGCSm55AAlIeb2PsD3kGEzxyo1325rJgrND5abeOh2T' diff --git a/build/lib/rainbowstream/db.py b/build/lib/rainbowstream/db.py deleted file mode 100644 index cac9a73..0000000 --- a/build/lib/rainbowstream/db.py +++ /dev/null @@ -1,43 +0,0 @@ -import os -from sqlalchemy import create_engine -from sqlalchemy.orm import sessionmaker -from table_def import Map -from .table_def import * - - -class RainbowDB(): - - engine = None - - def __init__(self): - if not os.path.isfile('rainbow.db'): - init_db() - self.engine = create_engine('sqlite:///rainbow.db', echo=False) - - def store(self, tweet_id): - """ - Store tweet id - """ - Session = sessionmaker(bind=self.engine) - session = Session() - m = Map(tweet_id) - session.add(m) - session.commit() - - def rainbow_query(self, rid): - """ - Query base of rainbow id - """ - Session = sessionmaker(bind=self.engine) - session = Session() - res = session.query(Map).filter_by(rainbow_id=rid).all() - return res - - def tweet_query(self, tid): - """ - Query base of tweet id - """ - Session = sessionmaker(bind=self.engine) - session = Session() - res = session.query(Map).filter_by(tweet_id=tid).all() - return res \ No newline at end of file diff --git a/build/lib/rainbowstream/interactive.py b/build/lib/rainbowstream/interactive.py deleted file mode 100644 index b7fabbc..0000000 --- a/build/lib/rainbowstream/interactive.py +++ /dev/null @@ -1,65 +0,0 @@ -import readline -import rlcompleter - -class RainbowCompleter(object): - - def __init__(self, options): - """ - Init - """ - self.options = options - self.current_candidates = [] - return - - def complete(self, text, state): - """ - Complete - """ - response = None - if state == 0: - origline = readline.get_line_buffer() - begin = readline.get_begidx() - end = readline.get_endidx() - being_completed = origline[begin:end] - words = origline.split() - - if not words: - self.current_candidates = sorted(self.options.keys()) - else: - try: - if begin == 0: - candidates = self.options.keys() - else: - first = words[0] - candidates = self.options[first] - - if being_completed: - self.current_candidates = [ w for w in candidates - if w.startswith(being_completed) ] - else: - self.current_candidates = candidates - - except (KeyError, IndexError), err: - self.current_candidates = [] - - try: - response = self.current_candidates[state] - except IndexError: - response = None - return response - - -def init_interactive_shell(d): - """ - Init the rainbow shell - """ - readline.set_completer(RainbowCompleter(d).complete) - readline.parse_and_bind('set editing-mode vi') - readline.parse_and_bind("set input-meta on") - readline.parse_and_bind("set convert-meta off") - readline.parse_and_bind("set output-meta on") - if 'libedit' in readline.__doc__: - readline.parse_and_bind("bind ^I rl_complete") - else: - readline.parse_and_bind("tab: complete") - diff --git a/build/lib/rainbowstream/rainbow.py b/build/lib/rainbowstream/rainbow.py deleted file mode 100644 index 8b71c34..0000000 --- a/build/lib/rainbowstream/rainbow.py +++ /dev/null @@ -1,632 +0,0 @@ -""" -Colorful user's timeline stream -""" -from multiprocessing import Process -from dateutil import parser - -import os -import os.path -import sys -import signal -import argparse -import time -import datetime - -from twitter.stream import TwitterStream, Timeout, HeartbeatTimeout, Hangup -from twitter.api import * -from twitter.oauth import OAuth, read_token_file -from twitter.oauth_dance import oauth_dance -from twitter.util import printNicely - -from .colors import * -from .config import * -from .consumer import * -from .interactive import * -from .db import * - -g = {} -db = RainbowDB() -cmdset = [ - 'switch', - 'home', - 'view', - 't', - 'rt', - 'fav', - 'rep', - 'del', - 'ufav', - 's', - 'fr', - 'fl', - 'h', - 'c', - 'q' -] - - -def draw(t, keyword=None, fil=[], ig=[]): - """ - Draw the rainbow - """ - - # Retrieve tweet - tid = t['id'] - text = t['text'] - screen_name = t['user']['screen_name'] - name = t['user']['name'] - created_at = t['created_at'] - favorited = t['favorited'] - date = parser.parse(created_at) - date = date - datetime.timedelta(seconds=time.timezone) - clock = date.strftime('%Y/%m/%d %H:%M:%S') - - # Filter and ignore - screen_name = '@' + screen_name - if fil and screen_name not in fil: - return - if ig and screen_name in ig: - return - - res = db.tweet_query(tid) - if not res: - db.store(tid) - res = db.tweet_query(tid) - rid = res[0].rainbow_id - - # Format info - user = cycle_color(name) + grey(' ' + screen_name + ' ') - meta = grey('[' + clock + '] [id=' + str(rid) + '] ') - if favorited: - meta = meta + green(u'\u2605') - tweet = text.split() - # Highlight RT - tweet = map(lambda x: grey(x) if x == 'RT' else x, tweet) - # Highlight screen_name - tweet = map(lambda x: cycle_color(x) if x[0] == '@' else x, tweet) - # Highlight link - tweet = map(lambda x: cyan(x) if x[0:7] == 'http://' else x, tweet) - # Highlight search keyword - if keyword: - tweet = map( - lambda x: on_yellow(x) if - ''.join(c for c in x if c.isalnum()).lower() == keyword.lower() - else x, - tweet - ) - tweet = ' '.join(tweet) - - # Draw rainbow - line1 = u"{u:>{uw}}:".format( - u=user, - uw=len(user) + 2, - ) - line2 = u"{c:>{cw}}".format( - c=meta, - cw=len(meta) + 2, - ) - line3 = ' ' + tweet - - printNicely('') - printNicely(line1) - printNicely(line2) - printNicely(line3) - - -def parse_arguments(): - """ - Parse the arguments - """ - parser = argparse.ArgumentParser(description=__doc__ or "") - parser.add_argument( - '-to', - '--timeout', - help='Timeout for the stream (seconds).') - parser.add_argument( - '-ht', - '--heartbeat-timeout', - help='Set heartbeat timeout.', - default=90) - parser.add_argument( - '-nb', - '--no-block', - action='store_true', - help='Set stream to non-blocking.') - parser.add_argument( - '-tt', - '--track-keywords', - help='Search the stream for specific text.') - parser.add_argument( - '-fil', - '--filter', - help='Filter specific screen_name.') - parser.add_argument( - '-ig', - '--ignore', - help='Ignore specific screen_name.') - return parser.parse_args() - - -def authen(): - """ - Authenticate with Twitter OAuth - """ - # When using rainbow stream you must authorize. - twitter_credential = os.environ.get( - 'HOME', - os.environ.get( - 'USERPROFILE', - '')) + os.sep + '.rainbow_oauth' - if not os.path.exists(twitter_credential): - oauth_dance("Rainbow Stream", - CONSUMER_KEY, - CONSUMER_SECRET, - twitter_credential) - oauth_token, oauth_token_secret = read_token_file(twitter_credential) - return OAuth( - oauth_token, - oauth_token_secret, - CONSUMER_KEY, - CONSUMER_SECRET) - - -def get_decorated_name(): - """ - Beginning of every line - """ - t = Twitter(auth=authen()) - name = '@' + t.account.verify_credentials()['screen_name'] - g['original_name'] = name[1:] - g['decorated_name'] = grey('[') + grey(name) + grey(']: ') - - -def switch(): - """ - Switch stream - """ - try: - target = g['stuff'].split()[0] - - # Filter and ignore - args = parse_arguments() - try: - if g['stuff'].split()[-1] == '-f': - only = raw_input('Only nicks: ') - ignore = raw_input('Ignore nicks: ') - args.filter = filter(None, only.split(',')) - args.ignore = filter(None, ignore.split(',')) - elif g['stuff'].split()[-1] == '-d': - args.filter = ONLY_LIST - args.ignore = IGNORE_LIST - except: - printNicely(red('Sorry, wrong format.')) - return - - # Public stream - if target == 'public': - keyword = g['stuff'].split()[1] - if keyword[0] == '#': - keyword = keyword[1:] - # Kill old process - os.kill(g['stream_pid'], signal.SIGKILL) - args.track_keywords = keyword - # Start new process - p = Process( - target=stream, - args=( - PUBLIC_DOMAIN, - args)) - p.start() - g['stream_pid'] = p.pid - - # Personal stream - elif target == 'mine': - # Kill old process - os.kill(g['stream_pid'], signal.SIGKILL) - # Start new process - p = Process( - target=stream, - args=( - USER_DOMAIN, - args, - g['original_name'])) - p.start() - g['stream_pid'] = p.pid - printNicely('') - printNicely(green('Stream switched.')) - if args.filter: - printNicely(cyan('Only: ' + str(args.filter))) - if args.ignore: - printNicely(red('Ignore: ' + str(args.ignore))) - printNicely('') - except: - printNicely(red('Sorry I can\'t understand.')) - - -def home(): - """ - Home - """ - t = Twitter(auth=authen()) - num = HOME_TWEET_NUM - if g['stuff'].isdigit(): - num = g['stuff'] - for tweet in reversed(t.statuses.home_timeline(count=num)): - draw(t=tweet) - printNicely('') - - -def view(): - """ - Friend view - """ - t = Twitter(auth=authen()) - user = g['stuff'].split()[0] - if user[0] == '@': - try: - num = int(g['stuff'].split()[1]) - except: - num = HOME_TWEET_NUM - for tweet in reversed(t.statuses.user_timeline(count=num, screen_name=user[1:])): - draw(t=tweet) - printNicely('') - else: - printNicely(red('A name should begin with a \'@\'')) - - -def tweet(): - """ - Tweet - """ - t = Twitter(auth=authen()) - t.statuses.update(status=g['stuff']) - - -def retweet(): - """ - ReTweet - """ - t = Twitter(auth=authen()) - try: - id = int(g['stuff'].split()[0]) - tid = db.rainbow_query(id)[0].tweet_id - t.statuses.retweet(id=tid, include_entities=False, trim_user=True) - except: - printNicely(red('Sorry I can\'t retweet for you.')) - - -def favorite(): - """ - Favorite - """ - t = Twitter(auth=authen()) - try: - id = int(g['stuff'].split()[0]) - tid = db.rainbow_query(id)[0].tweet_id - t.favorites.create(_id=tid, include_entities=False) - printNicely(green('Favorited.')) - draw(t.statuses.show(id=tid)) - except: - printNicely(red('Omg some syntax is wrong.')) - - -def reply(): - """ - Reply - """ - t = Twitter(auth=authen()) - try: - id = int(g['stuff'].split()[0]) - tid = db.rainbow_query(id)[0].tweet_id - user = t.statuses.show(id=tid)['user']['screen_name'] - status = ' '.join(g['stuff'].split()[1:]) - status = '@' + user + ' ' + status.decode('utf-8') - t.statuses.update(status=status, in_reply_to_status_id=tid) - except: - printNicely(red('Sorry I can\'t understand.')) - - -def delete(): - """ - Delete - """ - t = Twitter(auth=authen()) - try: - id = int(g['stuff'].split()[0]) - tid = db.rainbow_query(id)[0].tweet_id - t.statuses.destroy(id=tid) - printNicely(green('Okay it\'s gone.')) - except: - printNicely(red('Sorry I can\'t delete this tweet for you.')) - - -def unfavorite(): - """ - Unfavorite - """ - t = Twitter(auth=authen()) - try: - id = int(g['stuff'].split()[0]) - tid = db.rainbow_query(id)[0].tweet_id - t.favorites.destroy(_id=tid) - printNicely(green('Okay it\'s unfavorited.')) - draw(t.statuses.show(id=tid)) - except: - printNicely(red('Sorry I can\'t unfavorite this tweet for you.')) - - -def search(): - """ - Search - """ - t = Twitter(auth=authen()) - try: - if g['stuff'][0] == '#': - rel = t.search.tweets(q=g['stuff'])['statuses'] - if len(rel): - printNicely('Newest tweets:') - for i in reversed(xrange(SEARCH_MAX_RECORD)): - draw(t=rel[i], keyword=g['stuff'].strip()[1:]) - printNicely('') - else: - printNicely(magenta('I\'m afraid there is no result')) - else: - printNicely(red('A keyword should be a hashtag (like \'#AKB48\')')) - except: - printNicely(red('Sorry I can\'t understand.')) - - -def friend(): - """ - List of friend (following) - """ - t = Twitter(auth=authen()) - g['friends'] = t.friends.ids()['ids'] - for i in g['friends']: - name = t.users.lookup(user_id=i)[0]['name'] - screen_name = '@' + t.users.lookup(user_id=i)[0]['screen_name'] - user = cycle_color(name) + grey(' ' + screen_name + ' ') - print user - - -def follower(): - """ - List of follower - """ - t = Twitter(auth=authen()) - g['followers'] = t.followers.ids()['ids'] - for i in g['followers']: - name = t.users.lookup(user_id=i)[0]['name'] - screen_name = '@' + t.users.lookup(user_id=i)[0]['screen_name'] - user = cycle_color(name) + grey(' ' + screen_name + ' ') - print user - - -def help(): - """ - Help - """ - s = ' ' * 2 - h, w = os.popen('stty size', 'r').read().split() - - usage = '\n' - usage += s + 'Hi boss! I\'m ready to serve you right now!\n' - usage += s + '-' * (int(w) - 4) + '\n' - - usage += s + 'You are ' + yellow('already') + ' on your personal stream.\n' - usage += s * 2 + green('switch public #AKB') + \ - ' will switch to public stream and follow "' + \ - yellow('AKB') + '" keyword.\n' - usage += s * 2 + green('switch mine') + \ - ' will switch to your personal stream.\n' - usage += s * 2 + green('switch mine -f ') + \ - ' will prompt to enter the filter.\n' - usage += s * 3 + yellow('Only nicks') + \ - ' filter will decide nicks will be INCLUDE ONLY.\n' - usage += s * 3 + yellow('Ignore nicks') + \ - ' filter will decide nicks will be EXCLUDE.\n' - usage += s * 2 + green('switch mine -d') + \ - ' will use the config\'s ONLY_LIST and IGNORE_LIST.\n' - usage += s * 3 + '(see ' + grey('rainbowstream/config.py') + ').\n' - - usage += s + 'For more action: \n' - usage += s * 2 + green('home') + ' will show your timeline. ' + \ - green('home 7') + ' will show 7 tweet.\n' - usage += s * 2 + green('view @mdo') + \ - ' will show ' + yellow('@mdo') + '\'s home.\n' - usage += s * 2 + green('t oops ') + \ - 'will tweet "' + yellow('oops') + '" immediately.\n' - usage += s * 2 + \ - green('rt 12 ') + ' will retweet to tweet with ' + \ - yellow('[id=12]') + '.\n' - usage += s * 2 + \ - green('fav 12 ') + ' will favorite the tweet with ' + \ - yellow('[id=12]') + '.\n' - usage += s * 2 + green('rep 12 oops') + ' will reply "' + \ - yellow('oops') + '" to tweet with ' + yellow('[id=12]') + '.\n' - usage += s * 2 + \ - green('del 12 ') + ' will delete tweet with ' + \ - yellow('[id=12]') + '.\n' - usage += s * 2 + \ - green('ufav 12 ') + ' will unfavorite tweet with ' + \ - yellow('[id=12]') + '.\n' - usage += s * 2 + green('s #AKB48') + ' will search for "' + \ - yellow('AKB48') + '" and return 5 newest tweet.\n' - usage += s * 2 + green('fr') + ' will list out your following people.\n' - usage += s * 2 + green('fl') + ' will list out your follower.\n' - usage += s * 2 + green('h') + ' will show this help again.\n' - usage += s * 2 + green('c') + ' will clear the screen.\n' - usage += s * 2 + green('q') + ' will quit.\n' - - usage += s + '-' * (int(w) - 4) + '\n' - usage += s + 'Have fun and hang tight!\n' - printNicely(usage) - - -def clear(): - """ - Clear screen - """ - os.system('clear') - - -def quit(): - """ - Exit all - """ - os.system('rm -rf rainbow.db') - os.kill(g['stream_pid'], signal.SIGKILL) - sys.exit() - - -def reset(): - """ - Reset prefix of line - """ - if g['reset']: - printNicely(magenta('Need tips ? Type "h" and hit Enter key!')) - g['reset'] = False - - -def process(cmd): - """ - Process switch - """ - return dict(zip( - cmdset, - [ - switch, - home, - view, - tweet, - retweet, - favorite, - reply, - delete, - unfavorite, - search, - friend, - follower, - help, - clear, - quit - ] - )).get(cmd, reset) - - -def listen(): - """ - Listen to user's input - """ - d = dict(zip( - cmdset, - [ - ['public #', 'mine'], # switch - [], # home - ['@'], # view - [], # tweet - [], # retweet - [], # reply - [], # delete - ['#'], # search - [], # friend - [], # follower - [], # help - [], # clear - [], # quit - ] - )) - init_interactive_shell(d) - reset() - while True: - if g['prefix']: - line = raw_input(g['decorated_name']) - else: - line = raw_input() - try: - cmd = line.split()[0] - except: - cmd = '' - # Save cmd to global variable and call process - g['stuff'] = ' '.join(line.split()[1:]) - process(cmd)() - if cmd in ['switch', 't', 'rt', 'rep']: - g['prefix'] = False - else: - g['prefix'] = True - - -def stream(domain, args, name='Rainbow Stream'): - """ - Track the stream - """ - - # The Logo - art_dict = { - USER_DOMAIN: name, - PUBLIC_DOMAIN: args.track_keywords, - SITE_DOMAIN: 'Site Stream', - } - ascii_art(art_dict[domain]) - - # These arguments are optional: - stream_args = dict( - timeout=args.timeout, - block=not args.no_block, - heartbeat_timeout=args.heartbeat_timeout) - - # Track keyword - query_args = dict() - if args.track_keywords: - query_args['track'] = args.track_keywords - - # Get stream - stream = TwitterStream( - auth=authen(), - domain=domain, - **stream_args) - - if domain == USER_DOMAIN: - tweet_iter = stream.user(**query_args) - elif domain == SITE_DOMAIN: - tweet_iter = stream.site(**query_args) - else: - if args.track_keywords: - tweet_iter = stream.statuses.filter(**query_args) - else: - tweet_iter = stream.statuses.sample() - - # Iterate over the stream. - for tweet in tweet_iter: - if tweet is None: - printNicely("-- None --") - elif tweet is Timeout: - printNicely("-- Timeout --") - elif tweet is HeartbeatTimeout: - printNicely("-- Heartbeat Timeout --") - elif tweet is Hangup: - printNicely("-- Hangup --") - elif tweet.get('text'): - draw( - t=tweet, - keyword=args.track_keywords, - fil=args.filter, - ig=args.ignore) - - -def fly(): - """ - Main function - """ - # Spawn stream process - args = parse_arguments() - get_decorated_name() - p = Process(target=stream, args=(USER_DOMAIN, args, g['original_name'])) - p.start() - - # Start listen process - time.sleep(0.5) - g['reset'] = True - g['prefix'] = True - g['stream_pid'] = p.pid - listen() diff --git a/build/lib/rainbowstream/table_def.py b/build/lib/rainbowstream/table_def.py deleted file mode 100644 index c312bde..0000000 --- a/build/lib/rainbowstream/table_def.py +++ /dev/null @@ -1,20 +0,0 @@ -from sqlalchemy import * -from sqlalchemy.ext.declarative import declarative_base - -engine = create_engine('sqlite:///rainbow.db', echo=False) -Base = declarative_base() - - -class Map(Base): - - __tablename__ = "map" - - rainbow_id = Column(Integer, primary_key=True) - tweet_id = Column(Integer) - - def __init__(self, tweet_id): - self.tweet_id = tweet_id - - -def init_db(): - Base.metadata.create_all(engine) -- 2.25.1