1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
|
#!/usr/bin/env python3
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
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 send_image(client, room_id, image):
"""Send image to room.
Arguments:
---------
client : Client
room_id : str
image : str, file name of image
This is a working example for a JPG image.
"content": {
"body": "someimage.jpg",
"info": {
"size": 5420,
"mimetype": "image/jpeg",
"thumbnail_info": {
"w": 100,
"h": 100,
"mimetype": "image/jpeg",
"size": 2106
},
"w": 100,
"h": 100,
"thumbnail_url": "mxc://example.com/SomeStrangeThumbnailUriKey"
},
"msgtype": "m.image",
"url": "mxc://example.com/SomeStrangeUriKey"
}
"""
mime_type = magic.from_file(image, mime=True) # e.g. "image/jpeg"
if not mime_type.startswith("image/"):
print("Drop message because file does not have an image mime type.")
return
im = Image.open(image)
(width, height) = im.size # im.size returns (width,height) tuple
# first do an upload of image, then send URI of upload to room
file_stat = await aiofiles.os.stat(image)
async with aiofiles.open(image, "r+b") as f:
resp, _maybe_keys = await client.upload(
f,
content_type=mime_type, # image/jpeg
filename=os.path.basename(image),
filesize=file_stat.st_size,
)
if isinstance(resp, UploadResponse):
print("Image was uploaded successfully to server. ")
else:
print(f"Failed to upload image. Failure response: {resp}")
content = {
"body": os.path.basename(image), # descriptive title
"info": {
"size": file_stat.st_size,
"mimetype": mime_type,
"thumbnail_info": None, # TODO
"w": width, # width in pixel
"h": height, # height in pixel
"thumbnail_url": None, # TODO
},
"msgtype": "m.image",
"url": resp.content_uri,
}
try:
await client.room_send(room_id, message_type="m.room.message", content=content)
print("Image was sent successfully")
except Exception:
print(f"Image send of file {image} failed.")
async def main() -> None:
# 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}] ")
client = AsyncClient(homeserver, user_id)
pw = getpass.getpass()
resp = await client.login(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.",
"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"]
# Now we can send messages as the user
room_id = "!myfavouriteroomid:example.org"
room_id = input(f"Enter room id for image message: [{room_id}] ")
image = "exampledir/samplephoto.jpg"
image = input(f"Enter file name of image to send: [{image}] ")
await send_image(client, room_id, image)
print("Logged in using stored credentials. Sent a test message.")
# Close the client connection after we are done with it.
await client.close()
asyncio.run(main())
|
