dom/getbot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

dump

Browse Source
This commit is contained in:
DomNomNom 2025-02-15 22:42:38 +13:00
parent bea22dfa70
commit 81c98e18c0
4 changed files with 160 additions and 133 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

96
getbot/access_token_login.py
View File

@ -1,71 +1,38 @@
#!/usr/bin/env python3
import asyncio
import getpass
import json
import os
import sys
import aiofiles
from nio import AsyncClient, LoginResponse
CONFIG_FILE = "credentials.json"
# Check out main() below to see how it's done.
def write_details_to_disk(resp: LoginResponse, homeserver) -> None:
"""Writes the required login details to disk so we can log in later without
using a password.
Arguments:
resp {LoginResponse} -- the successful client login response.
homeserver -- URL of homeserver, e.g. "https://matrix.example.org"
"""
# open the config file in write-mode
with open(CONFIG_FILE, "w") as f:
# write the login details to disk
json.dump(
{
"homeserver": homeserver, # e.g. "https://matrix.example.org"
"user_id": resp.user_id, # e.g. "@user:example.org"
"device_id": resp.device_id, # device ID, 10 uppercase letters
"access_token": resp.access_token, # cryptogr. access token
},
f,
)
async def main() -> None:
# If there are no previously-saved credentials, we'll use the password
@ -75,114 +42,53 @@ async def main() -> None:
print(
"First time use. Did not find credential file. Asking for "
"homeserver, user, and password to create credential file."
)
# homeserver = "https://matrix.example.org"
# homeserver = input(f"Enter your homeserver URL: [{homeserver}] ")
# if not (homeserver.startswith("https://") or homeserver.startswith("http://")):
# homeserver = "https://" + homeserver
# user_id = "@user:example.org"
# user_id = input(f"Enter your full user ID: [{user_id}] ")
# device_name = "matrix-nio"
# device_name = input(f"Choose a name for this device: [{device_name}] ")
# client = AsyncClient(homeserver, user_id)
# pw = getpass.getpass()
homeserver = "https://matrix.domnomnom.com"
user_id = "@getbot:matrix.domnomnom.com"
room_id = "!sZpfYzLsRbnIOKJlPH:matrix.domnomnom.com"
client = AsyncClient(homeserver, user_id)
# client.add_event_callback(message_callback, RoomMessageText)
resp = await client.login(password)
# check that we logged in successfully
if isinstance(resp, LoginResponse):
write_details_to_disk(resp, homeserver)
else:
print(f'homeserver = "{homeserver}"; user = "{user_id}"')
print(f"Failed to log in: {resp}")
sys.exit(1)
print(
"Logged in using a password. Credentials were stored.",
"Try running the script again to login with credentials.",
)
# Otherwise the config file exists, so we'll use the stored credentials
else:
# open the file in read-only mode
async with aiofiles.open(CONFIG_FILE) as f:
contents = await f.read()
config = json.loads(contents)
client = AsyncClient(config["homeserver"])
client.access_token = config["access_token"]
client.user_id = config["user_id"]
client.device_id = config["device_id"]
room_id = config["room_id"]
# Now we can send messages as the user
await client.room_send(
room_id,
message_type="m.room.message",
content={"msgtype": "m.text", "body": "Hello world!"},
)
print("Logged in using stored credentials. Sent a test message.")
# Either way we're logged in here, too
await client.close()
asyncio.run(main())

110
getbot/login.py Executable file
View File

@ -0,0 +1,110 @@
import asyncio
import getpass
import json
import os
import sys
import traceback
import aiofiles
from nio import (
AsyncClient,
AsyncClientConfig,
KeyVerificationCancel,
KeyVerificationEvent,
KeyVerificationKey,
KeyVerificationMac,
KeyVerificationStart,
LocalProtocolError,
LoginResponse,
ToDeviceError,
ToDeviceMessage,
UnknownToDeviceEvent,
)
CONFIG_FILE = "credentials.json"
STORE_PATH = "./store/" # local directory
async def login() -> AsyncClient:
"""Handle login with or without stored credentials."""
# Configuration options for the AsyncClient
client_config = AsyncClientConfig(
max_limit_exceeded=0,
max_timeouts=0,
store_sync_tokens=True,
encryption_enabled=True,
)
# If there are no previously-saved credentials, we'll use the password
if not os.path.exists(CONFIG_FILE):
# print(
# "First time use. Did not find credential file. Asking for "
# "homeserver, user, and password to create credential file."
# )
# homeserver = "https://matrix.example.org"
# homeserver = input(f"Enter your homeserver URL: [{homeserver}] ")
# if not (homeserver.startswith("https://") or homeserver.startswith("http://")):
# homeserver = "https://" + homeserver
# user_id = "@user:example.org"
# user_id = input(f"Enter your full user ID: [{user_id}] ")
# device_name = "matrix-nio"
# device_name = input(f"Choose a name for this device: [{device_name}] ")
homeserver = "https://matrix.domnomnom.com"
user_id = "@getbot:matrix.domnomnom.com"
room_id = "!sZpfYzLsRbnIOKJlPH:matrix.domnomnom.com"
device_name = "whitebox-nio"
if not os.path.exists(STORE_PATH):
os.makedirs(STORE_PATH)
# Initialize the matrix client
client = AsyncClient(
homeserver,
user_id,
store_path=STORE_PATH,
config=client_config,
)
pw = getpass.getpass()
resp = await client.login(password=pw, device_name=device_name)
# check that we logged in successfully
if isinstance(resp, LoginResponse):
write_details_to_disk(resp, homeserver)
else:
print(f'homeserver = "{homeserver}"; user = "{user_id}"')
print(f"Failed to log in: {resp}")
sys.exit(1)
print(
"Logged in using a password. Credentials were stored. "
"On next execution the stored login credentials will be used."
)
# Otherwise the config file exists, so we'll use the stored credentials
else:
# open the file in read-only mode
async with aiofiles.open(CONFIG_FILE, "r") as f:
contents = await f.read()
config = json.loads(contents)
# Initialize the matrix client based on credentials from file
client = AsyncClient(
config["homeserver"],
config["user_id"],
device_id=config["device_id"],
store_path=STORE_PATH,
config=client_config,
)
client.restore_login(
user_id=config["user_id"],
device_id=config["device_id"],
access_token=config["access_token"],
)
print(f"Logged in using stored credentials as {client.user_id}. device_id={client.device_id}")
return client

58
getbot/main.py
View File

@ -1,50 +1,56 @@
import asyncio
import aiofiles.os
from nio import AsyncClient, MatrixRoom, RoomMessageText
import asyncio
import getpass
import json
import os
import sys
import aiofiles.os
import magic
from PIL import Image
from nio import AsyncClient, LoginResponse, UploadResponse
from login import login
async def message_callback(room: MatrixRoom, event: RoomMessageText) -> None:
print(
f"Message received in room {room.display_name}\n"
f"{room.user_name(event.sender)} | {event.body}"
f"{room.user_name(event.sender) or event.sender} | {event.body}"
)
async def main() -> None:
client = AsyncClient("https://matrix.domnomnom.com", "@getbot:matrix.domnomnom.com")
# async with aiofiles.open(CONFIG_FILE) as f:
# contents = await f.read()
# config = json.loads(contents)
# client = AsyncClient(config["homeserver"])
# client.access_token = config["access_token"]
# client.user_id = config["user_id"]
# client.device_id = config["device_id"]
# room_id = "!sZpfYzLsRbnIOKJlPH:matrix.domnomnom.com"
# image = "/home/get/getbot_profile_pic_256.png"
# # await send_image(client, room_id, image)
# if client.should_upload_keys:
# await client.keys_upload()
client = await login()
client.add_event_callback(message_callback, RoomMessageText)
print(await client.login(password))
# "Logged in as @alice:example.org device id: RANDOMDID"
# If you made a new room and haven't joined as that user, you can use
# await client.join("your-room-id")
await client.room_send(
room_id="!sZpfYzLsRbnIOKJlPH:matrix.domnomnom.com",
message_type="m.room.message",
content={"msgtype": "m.text", "body": "Hello world!"},
)
await client.sync_forever(timeout=30000) # milliseconds
asyncio.run(main())
if __name__ == '__main__':
asyncio.run(main())

29
getbot/verify_with_emoji.py
View File

@ -359,21 +359,26 @@ async def login() -> AsyncClient:
# If there are no previously-saved credentials, we'll use the password
if not os.path.exists(CONFIG_FILE):
print(
"First time use. Did not find credential file. Asking for "
"homeserver, user, and password to create credential file."
)
homeserver = "https://matrix.example.org"
homeserver = input(f"Enter your homeserver URL: [{homeserver}] ")
# print(
# "First time use. Did not find credential file. Asking for "
# "homeserver, user, and password to create credential file."
# )
# homeserver = "https://matrix.example.org"
# homeserver = input(f"Enter your homeserver URL: [{homeserver}] ")
if not (homeserver.startswith("https://") or homeserver.startswith("http://")):
homeserver = "https://" + homeserver
# if not (homeserver.startswith("https://") or homeserver.startswith("http://")):
# homeserver = "https://" + homeserver
user_id = "@user:example.org"
user_id = input(f"Enter your full user ID: [{user_id}] ")
# user_id = "@user:example.org"
# user_id = input(f"Enter your full user ID: [{user_id}] ")
# device_name = "matrix-nio"
# device_name = input(f"Choose a name for this device: [{device_name}] ")
homeserver = "https://matrix.domnomnom.com"
user_id = "@getbot:matrix.domnomnom.com"
room_id = "!sZpfYzLsRbnIOKJlPH:matrix.domnomnom.com"
device_name = "whitebox-nio"
device_name = "matrix-nio"
device_name = input(f"Choose a name for this device: [{device_name}] ")
if not os.path.exists(STORE_PATH):
os.makedirs(STORE_PATH)