Firepup650/fake-slack-app
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add statuses to logs

Browse source
This commit is contained in:
Firepup Sixfifty 2024-06-05 13:05:50 -05:00
parent 27fe210429
commit 45cbc73c1a
Signed by: Firepup650
GPG key ID: 7C92E2ABBBFAB9BA
1 changed files with 21 additions and 20 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

41
main.py
View file

@ -18,24 +18,25 @@ for requiredVar in ["SLACK_BOT_TOKEN", "SLACK_APP_TOKEN"]:
f'Missing required environment variable "{requiredVar}". Please create a .env file in the same directory as this script and define the missing variable.' f'Missing required environment variable "{requiredVar}". Please create a .env file in the same directory as this script and define the missing variable.'
) )
print("Establishing a connection to slack...") print("[INFO] Establishing a connection to slack...")
app = App(token=os.environ.get("SLACK_BOT_TOKEN")) app = App(token=os.environ.get("SLACK_BOT_TOKEN"))
client = app.client client = app.client
userMappings = {} userMappings = {}
try: try:
if "--no-cache" in sys.argv: if "--no-cache" in sys.argv:
print("Skipping cache on user request") print("[INFO] Skipping cache on user request")
raise ImportError("User requested to skip cache") raise ImportError("User requested to skip cache")
print("Trying to load user mappings from cache...") print("[INFO] Trying to load user mappings from cache...")
from cache import userMappings from cache import userMappings
print( print(
"Cache load OK. Reminder: If you need to regenerate the cache, call the script with `--no-cache`" """[INFO] Cache load OK.
[INFO] Reminder: If you need to regenerate the cache, call the script with `--no-cache`"""
) )
except ImportError: except ImportError:
users_list = [] users_list = []
print("Cache load failed, falling back to full load from slack...") print("[WARN] Cache load failed, falling back to full load from slack...")
cursor = "N/A" cursor = "N/A"
pages = 0 pages = 0
while ( while (
@ -49,9 +50,9 @@ except ImportError:
cursor = data["response_metadata"]["next_cursor"] cursor = data["response_metadata"]["next_cursor"]
users_list.extend(data["members"]) users_list.extend(data["members"])
pages += 1 pages += 1
print(f"Pages of users loaded: {pages}") print(f"[INFO] Pages of users loaded: {pages} (Estimated user count: {len(pages) * 1000}")
del pages del pages
print("Building user mappings now, this shouldn't take long...") print("[INFO] Building user mappings now, this shouldn't take long...")
for ( for (
user user
) in ( ) in (
@ -66,21 +67,21 @@ except ImportError:
else f"<@{user['id']}>" # User is missing a real name too... Fallback to ID else f"<@{user['id']}>" # User is missing a real name too... Fallback to ID
) )
) )
print("All mappings generated, writing cache file now...") print("[INFO] All mappings generated, writing cache file now...")
with open( with open(
"cache.py", "w" "cache.py", "w"
) as cacheFile: # It is many times faster to load from a local file instead of from slack ) as cacheFile: # It is many times faster to load from a local file instead of from slack
cacheFile.write(f"userMappings = {userMappings}") cacheFile.write(f"userMappings = {userMappings}")
print("Cache saved.") print("[INFO] Cache saved.")
print("User mappings loaded. User count:", len(userMappings)) print("[INFO] User mappings loaded. User count:", len(userMappings))
if __name__ == "__main__": if __name__ == "__main__":
print("^D at any time to terminate program") print("[INFO] ^D at any time to terminate program")
while 1: while 1:
chan = input("Channel ID") chan = input("Channel ID")
try: try:
print("^C to change channel") print("[INFO] ^C to change channel")
while 1: while 1:
thread = input("Reply to a thread? (y|N)").lower().startswith("y") thread = input("Reply to a thread? (y|N)").lower().startswith("y")
ts = None ts = None
@ -91,14 +92,14 @@ if __name__ == "__main__":
if not hasID: if not hasID:
try: try:
print( print(
"Getting the last 50 messages for threading options..." "[INFO] Getting the last 50 messages for threading options..."
) )
res = client.conversations_history( res = client.conversations_history(
channel=chan, inclusive=True, limit=50 channel=chan, inclusive=True, limit=50
) )
messages = res["messages"] messages = res["messages"]
texts = {} texts = {}
print("Building messages, this might take a little bit...") print("[INFO] Building messages, this might take a little bit...")
for i in range(len(messages)): for i in range(len(messages)):
label = f'{messages[i]["text"]} ({messages[i]["ts"]})' label = f'{messages[i]["text"]} ({messages[i]["ts"]})'
for user in userMappings: for user in userMappings:
@ -112,12 +113,12 @@ if __name__ == "__main__":
] ]
ts = found["ts"] ts = found["ts"]
except Exception as E: except Exception as E:
print(f"Exception: {E}") print(f"[WARN] Exception: {E}")
break break
else: else:
ts = input("TS ID") ts = input("TS ID")
print( print(
"^C to change/exit thread (^C twice if you want to change channel)" "[INFO] ^C to change/exit thread (^C twice if you want to change channel)"
) )
try: try:
while 1: while 1:
@ -128,9 +129,9 @@ if __name__ == "__main__":
client.chat_postMessage( client.chat_postMessage(
channel=chan, text=msg, thread_ts=ts channel=chan, text=msg, thread_ts=ts
) )
print("Message sent (to the thread)!") print("[INFO] Message sent (to the thread)!")
except Exception as E: except Exception as E:
print(f"Exception: {E}") print(f"[WARN] Exception: {E}")
except KeyboardInterrupt: except KeyboardInterrupt:
print() print()
if ts: if ts:
@ -140,8 +141,8 @@ if __name__ == "__main__":
) )
try: try:
client.chat_postMessage(channel=chan, text=msg) client.chat_postMessage(channel=chan, text=msg)
print("Message sent (to the channel)!") print("[INFO] Message sent (to the channel)!")
except Exception as E: except Exception as E:
print(f"Exception: {E}") print(f"[WARN] Exception: {E}")
except KeyboardInterrupt: except KeyboardInterrupt:
print() print()