def upload_doc(api, multipart_data, filename="image.png") -> Optional[Attachment]:
"""Upload file `multipart_data` and return Attachment for sending to user"""
sender = api.get_default_sender("docs.getWallUploadServer")
client = api.get_current_sender("docs.getWallUploadServer", sender=sender)
data = aiohttp.FormData()
data.add_field('file', multipart_data, filename=filename, content_type='multipart/form-data')
values = {}
if client.group_id:
values['group_id'] = client.group_id
response = await api(sender=sender).docs.getWallUploadServer(**values)
if not response or not response.get('upload_url'):
return None
upload_url = response['upload_url']
async with aiohttp.ClientSession() as sess:
async with sess.post(upload_url, data=data) as resp:
result = json.loads(await resp.text())
if not result:
return None
data = dict(file=result['file'])
result = (await api(sender=sender).docs.save(**data))[0]
return Attachment.from_upload_result(result, "doc")
评论列表
文章目录