api, talent list structures

.gitignore

@@ -143,4 +143,4 @@ cython_debug/
 .vscode
 
 # project-specific
-secrets.ini
+/secrets.ini

README.md

@@ -2,6 +2,8 @@
 *Twitter bot that follows interactions between Nijisanji EN/ID and hololive EN/ID members.*  
 ...because some folks are that desperate. Like me!
 
+**This project is intended to run [this account](https://twitter.com/NijiHoloEN_Msgs).**
+
 ## Roadmap
 * Read past tweets of members from both companies
 * Track tweets in queue and history/log files

lists/holoen.txt

@@ -0,0 +1,28 @@
+# ----- hololive EN  -----
+
+# --- [Myth] ---
+gawrgura 1283657064410017793
+watsonameliaen 283656034305769472
+moricalliope 1283653858510598144
+ninomaeinanis 1283650008835743744
+takanashikiara 1283646922406760448
+
+# --- [HOPE] ---
+irys_en 1363705980261855232
+
+# --- [Council] ---
+hakosbaelz 1409783149211443200
+ourokronii 1409817096523968513
+ceresfauna 1409784760805650436
+tsukumosana 1409819816194576394
+nanashimumei_en 1409817941705515015
+
+# --- [TEMPUS] ---
+noirvesper_en 1536579341332516864
+axelsyrios 1536577295632441344
+magnidezmond 1536576325296996352
+regisaltare 1536575088996524032
+
+# --- [STAFF] ---
+omegaalpha_en 1397148959798226945
+hololivepro_EN 1540204458042621952

lists/nijien.txt

@@ -0,0 +1,42 @@
+# ----- [NIJISANJI EN] -----
+
+# --- [Lazulight] ---
+PomuRainpuff 1390637197167038464
+EliraPendora 1390620618001838086
+FinanaRyugu 1390209302120394754
+
+# --- [Obsydia] ---
+Petra_Gurin 1413339084076978179
+Selen_Tatsuki 1413318241804439552
+Rosemi_Lovelock 1413326894435602434
+
+# --- [Ethyria] ---
+MillieParfait 1437952405283426310
+EnnaAlouette 1437963160544284675
+NinaKosaka 1437959162651156484
+ReimuEndou 1437961007029227520
+
+# --- [Luxiem]---
+Vox_Akuma 1465851881180348425
+shu_amino 1465850835951357955
+ike_eveland 1465851188562345985
+Mysta_Rias 1465851243167895554
+luca_kaneshiro 1465858739970273281
+
+# --- [Noctyx] ---
+alban_knox 1490867613915828224
+uki_violeta 1491195742123397124
+Yugo_Asuma 1492604168145539072
+Fulgur_Ovid 1493392149664219138
+sonny_brisko 1493394108014292993
+
+# --- [ILUNA] ---
+MariaMari0nette 1545351225293426688
+AsterArcadia 1545352592884084736
+ScarleYonaguni 1545354510515654656
+KyoKanek0 1545552756773208066
+AiaAmare 1545562635650957312
+RenZott0 1546328834559340544
+
+# --- [STAFF] ---
+NIJISANJI_World 1214737620749578240

requirements.txt

@@ -1 +1,2 @@
 tweepy
+tweet-capture

src/api.py

@@ -1,27 +1,109 @@
+from lib2to3.pgen2 import token
 from math import inf
+from urllib import response
 import tweepy
 
 import secrets
 import util
 
-class API:
+class TwAPI:
     instance = None
+    TWEET_MEDIA_FIELDS = ['url']
+    TWEET_FIELDS = ['created_at', 'in_reply_to_user_id']
+    TWEET_EXPANSIONS = ['entities.mentions.username', 'referenced_tweets.id.author_id']
 
     def __init__(self):
-        API.instance = self
+        TwAPI.instance = self
         self.client = tweepy.Client(
             bearer_token=secrets.bearer_token(),
             consumer_key=secrets.api_key(), consumer_secret=secrets.api_secret(),
             access_token=secrets.access_token(), access_token_secret=secrets.access_secret()
         )
     
-    def get_user_tweets(self, id: int, count=inf):
+    # Returns a set of involved parties for a single tweet.
-        posts = list()
+    #
+    # Tweet must have been queried with these parameters:
+    # media_fields=['url'],
+    # tweet_fields=['created_at', 'in_reply_to_user_id'],
+    # expansions=['entities.mentions.username', 'referenced_tweets.id.author_id']
+    @staticmethod
+    def get_involved_parties(tweet, response):
+        involved_parties = set()
+        # mentions
+        try:
+            mention_list = tweet.entities['mentions']
+            for mention in mention_list:
+                involved_parties.add(int(mention['id']))
+        except: pass
+        # reply-to
+        if tweet.in_reply_to_user_id != None:
+            involved_parties.add(tweet.in_reply_to_user_id)
+        # qrt
+        if tweet.attachments:
+            for ref_tweet in tweet.attachments:
+                if ref_tweet.type == 'quoted':
+                    for incl_tweet in response.includes['tweets']:
+                        if incl_tweet.id == ref_tweet.id:
+                            involved_parties.add(incl_tweet.author_id)
+
+        return involved_parties
+    
+    # Returns a tweet and mention-set pair, given a tweet ID.
+    def get_tweet_mentions(self, id):
+        resp = self.client.get_tweet(id,
+            media_fields=TwAPI.TWEET_MEDIA_FIELDS,
+            tweet_fields=TwAPI.TWEET_FIELDS,
+            expansions=TwAPI.TWEET_EXPANSIONS)
+        
+        tweet = resp.data
+        mentions = TwAPI.get_involved_parties(tweet, resp)
+        return (tweet, mentions)
+    
+    # Returns a list (tweet, {mentions}) from a user.
+    # mentions- a set comprised of any other parties involved
+    # in this tweet (reply, mention, qrt)
+    def get_users_all_tweets_mentions(self, id: int, count=inf):
+        pairs = list()
 
         retrieve_size = util.clamp(count, 5, 100)
-        retrieved_tweets = 0
+        next_page_token = None
-        pagination_token = None
+        tokens_retrieved = 0
+        tweets_retrieved = 0
 
-        # while retrieved_tweets < count: # or we haven't reached the end of user's tweets
+        while tweets_retrieved < count:
-        resp = self.client.get_users_tweets(id, max_results=retrieve_size, media_fields=['url'], expansions=['entities.mentions.username', 'referenced_tweets.id.author_id'])
+            print(f'Retrieved {tokens_retrieved} tokens so far...')
-        return resp
+            resp = self.client.get_users_tweets(id, max_results=retrieve_size, pagination_token=next_page_token,
+                media_fields=TwAPI.TWEET_MEDIA_FIELDS,
+                tweet_fields=TwAPI.TWEET_FIELDS,
+                expansions=TwAPI.TWEET_EXPANSIONS)
+
+            for tweet in resp.data:
+                mentions = TwAPI.get_involved_parties(tweet, resp)
+                pairs.append((tweet, mentions))
+
+            # update counters and pagination token
+            tweets_retrieved += resp.meta['result_count']
+            if tweets_retrieved < count:
+                try:
+                    next_page_token = resp.meta['next_token']
+                    tokens_retrieved += 1
+                except KeyError:
+                    print("next_token wasn't provided; we've reached the end!")
+                    break  # reached end of user's tweets
+
+        print(f'Retrieved {tweets_retrieved} tweets using {tokens_retrieved} tokens.')
+        return pairs
+    
+    # returns a filtered list (tweet, [mentions]) from a user
+    def get_users_cross_tweets_mentions(self, id):
+        ret = list()
+        pairs = self.get_users_all_tweets_mentions(id)
+        for pair in pairs:
+            if util.is_cross_company(pair):
+                ret.append(pair)
+        
+        return ret
+
+    # Create a post that showcases given tweet and its mentions set.
+    def create_post(self, tweet, mentions):
+        pass

src/catchup.py

@@ -1,17 +1,21 @@
 ## The bot's catch-up mode
 # Scan all accounts for cross-company interactions.
 # Terminates when finished scanning and posting.
+#
+# We should post, at the fastest, one tweet per minute.
 
 import os
-import TwitterAPI as api
 
 from util import *
+from api import TwAPI
 
 ## Returns list of tweets present in queue.txt
 def get_local_queue():
-    f = open(os.path.join(get_project_dir(), 'queue.txt'))
+    # f = open(os.path.join(get_project_dir(), 'queue.txt'))
     pass
 
 def run():
     queue = get_local_queue()
-    pass
+    pairs = TwAPI.instance.get_users_all_tweets_mentions(1390620618001838086, count=5)
+    for (tweet, mentions) in pairs:
+        print_tweet(tweet, mentions)

src/listen.py

@@ -1,7 +1,5 @@
 ## The bot's listen mode
 # Continuously listen for cross-company interactions.
 
-import TwitterAPI as api
-
 def run():
     pass

src/main.py

@@ -2,19 +2,22 @@ import sys
 import argparse
 from argparse import RawTextHelpFormatter
 
+import talent_lists
 import secrets
 import catchup
 import listen
 
-from api import API
+from api import TwAPI
-import util
+from util import is_cross_company, print_tweet
+
+MODES_HELP_STR = '''mode to run the bot at:
+l,listen:       listen for new tweets from all accounts; will not terminate unless error occurs
+c,catchup:      scan all tweets from all accounts; will terminate when done'''
 
 def init_argparse():
     p = argparse.ArgumentParser(description='Twitter bot that follows interactions between Nijisanji EN/ID and hololive EN/ID members.', formatter_class=RawTextHelpFormatter)
     p.add_argument('mode', nargs='?', \
-        help='mode to run the bot at:\n\
+        help=MODES_HELP_STR)
-  l,listen:       listen for new tweets from all accounts; will not terminate unless error occurs\n\
-  c,catchup:      scan all tweets from all accounts; will terminate when done')
     p.add_argument('--show-tokens', action='store_true', help='[DO NOT USE IN PUBLIC SETTING] print stored tokens from secrets.ini')
     return p
 
@@ -31,11 +34,20 @@ def main():
 
     if args.mode is None: return
 
-    util.twAPI = API()
+    ## We expect to run in some mode now.
-    resp = util.twAPI.get_user_tweets(1390620618001838086, count=5)
-    print(resp.data)
 
-    # determine running mode
+    # Initialize shared API instance
+    twApi = TwAPI.instance = TwAPI()
+
+    # Initialize talent account lists
+    talent_lists.init()
+
+    ## TEST CODE ##
+    cross_pairs = twApi.get_users_cross_tweets_mentions(1390620618001838086)
+    for pair in cross_pairs:
+        print_tweet(pair)
+
+    ## Determine running mode
     match args.mode.lower():
         case 'l' | 'listen':
             print('RUNNING IN LISTEN MODE\n')

src/secrets.py

@@ -7,7 +7,7 @@ from util import *
 
 # returns dictionary of the Credentials section.
 # [NOT TO BE USED OUTSIDE OF THIS FILE.]
-def get_ini_credentials():
+def __get_ini_credentials():
     c = configparser.RawConfigParser()
     if len(c.read(os.path.join(get_project_dir(), 'secrets.ini'))) > 0 and c.has_section('Credentials'):
         return c['Credentials']
@@ -15,27 +15,27 @@ def get_ini_credentials():
 
 # returns the consumer api_key stored in secrets.ini
 def api_key():
-    c = get_ini_credentials()
+    c = __get_ini_credentials()
     return c.get(option='api_key', fallback='xxx') if c is not None else 'xxx'
 
 # returns the consumer api_secret stored in secrets.ini
 def api_secret():
-    c = get_ini_credentials()
+    c = __get_ini_credentials()
     return c.get(option='api_secret', fallback='yyy') if c is not None else 'yyy'
 
 # returns the bearer_token stored in secrets.ini
 def bearer_token():
-    c = get_ini_credentials()
+    c = __get_ini_credentials()
     return c.get(option='bearer_token', fallback='zzz') if c is not None else 'zzz'
 
 # returns the access_token stroed in secrets.ini
 def access_token():
-    c = get_ini_credentials()
+    c = __get_ini_credentials()
     return c.get(option='oauth1_access_token', fallback='zzz') if c is not None else 'aaa'
 
 # returns the access_secret stroed in secrets.ini
 def access_secret():
-    c = get_ini_credentials()
+    c = __get_ini_credentials()
     return c.get(option='oauth1_access_secret', fallback='zzz') if c is not None else 'bbb'
 
 def get_all_secrets():

src/talent_lists.py

@@ -0,0 +1,21 @@
+import util
+
+niji_en = dict()
+holo_en = dict()
+
+def __create_dict(file, _dict):
+    with open(file, 'r') as f:
+        for line in f:
+            words = line.split()
+            if len(words) == 2 and line[0] != '#':
+                name, id = line.split()
+                _dict[int(id)] = name
+
+def init():
+    global niji_en
+    global holo_en
+
+    # holoEN
+    __create_dict(f'{util.get_project_dir()}/lists/holoen.txt', holo_en)
+    # nijiEN
+    __create_dict(f'{util.get_project_dir()}/lists/nijien.txt', niji_en)

src/util.py

@@ -1,9 +1,7 @@
 ## Shared utility functions.
 
 import os
-
+import talent_lists
-# Twitter API instance to share throughout program
-twAPI = None
 
 # returns system path to this project, which is
 # up one level from this file's directory (src).
@@ -11,8 +9,32 @@ def get_project_dir():
     return os.path.join(os.path.dirname(__file__), os.pardir)
 
 # determine if tweet involves cross-company interaction
-def is_cross_company(tweet):
+def is_cross_company(pair: tuple):
-    pass
+    author_id, mentions = pair[0].author_id, pair[1]
+
+    for mention_id in mentions:
+        if author_id in talent_lists.niji_en:
+            if mention_id in talent_lists.holo_en:
+                return True
+        elif author_id in talent_lists.holo_en:
+            if mention_id in talent_lists.niji_en:
+                return True
+    return False
+
+def tweet_id_to_url(id):
+    return f'https://twitter.com/twitter/status/{id}'
+
+def print_tweet(pair: tuple):
+    tweet, mentions = pair
+    s = (
+        f'{tweet.id}: {tweet.created_at}: involves {mentions}\n'
+        f'{tweet.text}\n'
+        f'-----\n'
+        f'{tweet.entities}\n'
+        f'{tweet.referenced_tweets}\n'
+        f'================================================='
+    )
+    print(s)
 
 def clamp(n, smallest, largest):
     return max(smallest, min(n, largest))