1#!/usr/bin/env python
 2# -*- coding: utf-8 -*-
 3import base64
 4from io import BytesIO
 5from pathlib import Path
 6
 7from loguru import logger
 8from PIL import Image
 9
10from config import API, DOWNLOAD_DIR
11from database.pastbin import upload_pastbin
12from networking import hx_req
13from utils import rand_string
14
15
16async def generate_from_api(message_info: dict) -> str:
17    """Generate a quote image from API.
18
19    Docs: https://github.com/LyoSU/quote-api
20
21    message_info example:
22    {
23        "from": {
24            "uid": 123456,
25            "first_name": "First",
26            "last_name": "Last",
27            "username": "alice",
28            "photo": "local_image_path",
29        },
30        "text": "This is a sample message for testing only",
31        "avatar": True,
32        "media": "local_media_path",
33        "entities": [
34            {"type": "bold", "offset": 0, "length": 2},
35            {"type": "spoiler", "offset": 5, "length": 2},
36            {"type": "text_link", "offset": 7, "length": 2, "url": "https://github.com/"},
37            {"type": "strikethrough", "offset": 12, "length": 3},
38        ]
39    }
40    """
41    # Reconstruct the payload
42    if isinstance(message_info["from"]["photo"], str):  # local image path
43        avatar_url, _ = await upload_pastbin(message_info["from"]["photo"], ttl="1h")
44        message_info["from"]["photo"] = {"url": avatar_url}
45    if isinstance(message_info.get("media"), str):  # local media path
46        media_url, _ = await upload_pastbin(message_info["media"], ttl="1h")
47        message_info["media"] = {"url": media_url}
48
49    payload = {"type": "quote", "format": "webp", "messages": [message_info]}
50    logger.trace(f"payload: {payload}")
51    resp = await hx_req(
52        API.QUOTELY,
53        "POST",
54        json_data=payload,
55        headers={"content-type": "application/json"},
56        check_keys=["result.image"],
57        silent=True,
58    )
59    if not resp.get("hx_error"):
60        b64img = resp["result"]["image"]
61        image_data = base64.b64decode(b64img)
62        image = Image.open(BytesIO(image_data))
63        output_path = Path(DOWNLOAD_DIR) / f"{rand_string(16)}.webp"
64        image.save(output_path, "WEBP")
65        return output_path.as_posix()
66    return "/not-exist-path"