~nova/fletcher

fletcher/netcode.py -rw-r--r-- 1.9 KiB
ece10afa — Novalinium f-string issue 9 days ago
                                                                                
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
import io
import aiohttp
from sys import exc_info
import traceback
import logging

logger = logging.getLogger("fletcher")


async def simple_get_image(url):
    try:
        async with aiohttp.ClientSession(
            headers={"User-Agent": "Fletcher/0.1 (operator@noblejury.com)"}
        ) as session:
            logger.debug(url)
            async with session.get(str(url)) as resp:
                buffer = io.BytesIO(await resp.read())
                if resp.status != 200:
                    raise Exception(
                        f"HttpProcessingError: {resp.status} Retrieving image failed!"
                    )
                return buffer
    except Exception as e:
        if str(e).startswith("HttpProcessingError"):
            raise e
        if type(e) is aiohttp.InvalidURL:
            raise e
        logger.debug(traceback.format_exc())
        exc_type, exc_obj, exc_tb = exc_info()
        logger.error("SGI[{}]: {} {}".format(exc_tb.tb_lineno, type(e).__name__, e))


async def simple_post_image(post_url, image, filename, image_format, field_name="file"):
    try:
        async with aiohttp.ClientSession(
            headers={"User-Agent": "Fletcher/0.1 (operator@noblejury.com)",}
        ) as session:
            logger.debug(f"SPI: {post_url}")
            fd = aiohttp.FormData()
            fd.add_field(
                field_name, image, filename=filename, content_type=image_format
            )
            async with session.post(str(post_url), data=fd) as resp:
                buffer = await resp.text()
                if resp.status != 200:
                    raise Exception(
                        f"HttpProcessingError: {resp.status} Retrieving response failed!\n"
                    )
                return buffer
    except Exception as e:
        exc_type, exc_obj, exc_tb = exc_info()
        logger.error("SPI[{}]: {} {}".format(exc_tb.tb_lineno, type(e).__name__, e))