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

Compare commits

base: Firepup650:c8cb828b7009104570433d6f200374bf5f64c932
Branches Tags
Firepup650:master
..
compare: Firepup650:994d47a6907efec15e175e3515ad6b4d16f4ad2d
Branches Tags
Firepup650:master

8 commits
c8cb828b70 ... 994d47a690

Author SHA1 Message Date
Firepup Sixfifty
994d47a690
Formatting 2024-06-05 13:05:57 -05:00
Firepup Sixfifty
45cbc73c1a
Add statuses to logs 2024-06-05 13:05:50 -05:00
Firepup Sixfifty
27fe210429
Formatting 2024-06-05 13:01:23 -05:00
Firepup Sixfifty
deb52bf348
Add some comments 2024-06-05 13:01:10 -05:00
Firepup Sixfifty
0c34cfc843
Tweak wording 2024-06-05 12:52:56 -05:00
Firepup Sixfifty
b2a4e34148
Switch to normalized names 2024-06-05 12:50:37 -05:00
Firepup Sixfifty
edfe7cd5d5
Formatting 2024-06-05 12:48:35 -05:00
Firepup Sixfifty
889fd9be49
Fallback to real_names if a user has no display_name 2024-06-05 12:48:18 -05:00
1 changed files with 49 additions and 28 deletions
Show stats Expand all files Collapse all files

77
main.py
View file

@ -15,25 +15,33 @@ load_dotenv()
for requiredVar in ["SLACK_BOT_TOKEN", "SLACK_APP_TOKEN"]:
if not os.environ.get(requiredVar):
raise ValueError(
f'Missing required environment variable "{requiredVar}". Please create a .env file in the same directory as this script and define it.'
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"))
client = app.client
userMappings = {}
try:
if "--no-cache" in sys.argv:
print("[INFO] Skipping cache on user request")
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
print(
"""[INFO] Cache load OK.
[INFO] Reminder: If you need to regenerate the cache, call the script with `--no-cache`"""
)
except ImportError:
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"
pages = 0
while cursor:
while (
cursor
): # If slack gives us a cursor, then we ain't done loading user data yet
data = ""
if cursor != "N/A":
data = client.users_list(cursor=cursor, limit=1000)
@ -42,29 +50,40 @@ except ImportError:
cursor = data["response_metadata"]["next_cursor"]
users_list.extend(data["members"])
pages += 1
print(f"Pages of users loaded: {pages}")
del pages
print("Building user mappings now, this shouldn't take long...")
userMappings = {}
for user in users_list:
userMappings[f"<@{user['id']}>"] = (
f"<@{user['profile']['display_name']}>"
if user["profile"]["display_name"]
else f"<@{user['id']}>"
print(
f"[INFO] Pages of users loaded: {pages} (Estimated user count: {len(pages) * 1000}"
)
print("All mappings generated, writing cache file now...")
with open("cache.py", "w") as cacheFile:
del pages
print("[INFO] Building user mappings now, this shouldn't take long...")
for (
user
) in (
users_list
): # Map user ID mentions to user name mentions, it's nicer when printing messages for thread selection.
userMappings[f"<@{user['id']}>"] = (
f"<@{user['profile']['display_name_normalized']}>"
if user["profile"]["display_name_normalized"]
else ( # User is missing a display name for some reason, fallback to real names
f"<@{user['profile']['real_name_normalized']}>"
if user["profile"]["real_name_normalized"]
else f"<@{user['id']}>" # User is missing a real name too... Fallback to ID
)
)
print("[INFO] All mappings generated, writing cache file now...")
with open(
"cache.py", "w"
) as cacheFile: # It is many times faster to load from a local file instead of from slack
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__":
print("^D at any time to terminate program")
print("[INFO] ^D at any time to terminate program")
while 1:
chan = input("Channel ID")
try:
print("^C to change channel")
print("[INFO] ^C to change channel")
while 1:
thread = input("Reply to a thread? (y|N)").lower().startswith("y")
ts = None
@ -75,14 +94,16 @@ if __name__ == "__main__":
if not hasID:
try:
print(
"Getting the last 50 messages for threading options..."
"[INFO] Getting the last 50 messages for threading options..."
)
res = client.conversations_history(
channel=chan, inclusive=True, limit=50
)
messages = res["messages"]
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)):
label = f'{messages[i]["text"]} ({messages[i]["ts"]})'
for user in userMappings:
@ -96,12 +117,12 @@ if __name__ == "__main__":
]
ts = found["ts"]
except Exception as E:
print(f"Exception: {E}")
print(f"[WARN] Exception: {E}")
break
else:
ts = input("TS ID")
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:
while 1:
@ -112,9 +133,9 @@ if __name__ == "__main__":
client.chat_postMessage(
channel=chan, text=msg, thread_ts=ts
)
print("Message sent (to the thread)!")
print("[INFO] Message sent (to the thread)!")
except Exception as E:
print(f"Exception: {E}")
print(f"[WARN] Exception: {E}")
except KeyboardInterrupt:
print()
if ts:
@ -124,8 +145,8 @@ if __name__ == "__main__":
)
try:
client.chat_postMessage(channel=chan, text=msg)
print("Message sent (to the channel)!")
print("[INFO] Message sent (to the channel)!")
except Exception as E:
print(f"Exception: {E}")
print(f"[WARN] Exception: {E}")
except KeyboardInterrupt:
print()