From e05fb2824629fb7767165bd01849d478a232144d Mon Sep 17 00:00:00 2001 From: Kirill Kirilenko Date: Sun, 1 Mar 2026 22:59:35 +0300 Subject: [PATCH] =?UTF-8?q?=D0=A1=D0=B2=D0=BE=D0=B1=D0=BE=D0=B4=D0=BD?= =?UTF-8?q?=D1=8B=D0=B5=20=D1=84=D1=83=D0=BD=D0=BA=D1=86=D0=B8=D0=B8=20?= =?UTF-8?q?=D0=BF=D0=B5=D1=80=D0=B5=D0=BC=D0=B5=D1=89=D0=B5=D0=BD=D1=8B=20?= =?UTF-8?q?=D0=B2=20=D0=BA=D0=BE=D0=BD=D0=B5=D1=86=20=D1=84=D0=B0=D0=B9?= =?UTF-8?q?=D0=BB=D0=B0.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- ai_agent.py | 118 ++++++++++++++++++++++++++-------------------------- 1 file changed, 59 insertions(+), 59 deletions(-) diff --git a/ai_agent.py b/ai_agent.py index 8259faa..053df0d 100644 --- a/ai_agent.py +++ b/ai_agent.py @@ -40,65 +40,6 @@ class Message: message_id: int = None -def _add_message_prefix(text: Optional[str], username: Optional[str] = None) -> str: - current_time = datetime.datetime.now().strftime("%d.%m.%Y %H:%M") - prefix = f"[{current_time}, {username}]" if username is not None else f"[{current_time}]" - return f"{prefix}: {text}" if text is not None else f"{prefix}:" - - -def _encode_image(image: bytes) -> str: - encoded_image = base64.b64encode(image).decode('utf-8') - return f"data:image/jpeg;base64,{encoded_image}" - - -def _serialize_message(role: str, text: Optional[str], image: Optional[bytes]) -> dict: - serialized = {"role": role, "content": []} - if text is not None: - serialized["content"].append({"type": "text", "text": text}) - if image is not None: - serialized["content"].append({"type": "image_url", "image_url": {"url": _encode_image(image)}}) - return serialized - - -def _remove_none_recursive(data: Union[dict, list, any]) -> Union[dict, list, any]: - if isinstance(data, dict): - return { - k: _remove_none_recursive(v) - for k, v in data.items() - if v is not None - } - elif isinstance(data, list): - return [ - _remove_none_recursive(item) - for item in data - if item is not None - ] - else: - return data - - -def _serialize_assistant_message(message: AssistantMessage) -> AssistantMessageTypedDict: - return _remove_none_recursive(message.model_dump(by_alias=True)) - - -def _get_resolution_for_aspect_ratio(aspect_ratio: str) -> Tuple[int, int]: - aspect_ratio_resolution_map = { - "1:1": (1280, 1280), - "4:3": (1280, 1024), - "3:4": (1024, 1280), - "16:9": (1280, 720), - "9:16": (720, 1280) - } - return aspect_ratio_resolution_map.get(aspect_ratio, (1280, 1024)) - - -def _convert_image_to_jpeg(image: bytes) -> bytes: - img = Image.open(BytesIO(image)).convert("RGB") - output = BytesIO() - img.save(output, format='JPEG', quality=95, optimize=True) - return output.getvalue() - - class AiAgent: def __init__(self, openrouter_token: str, openrouter_model: str, @@ -423,3 +364,62 @@ def create_ai_agent(openrouter_token: str, openrouter_model: str, db: BasicDatabase, platform: str): global agent agent = AiAgent(openrouter_token, openrouter_model, fal_token, replicate_token, db, platform) + + +def _add_message_prefix(text: Optional[str], username: Optional[str] = None) -> str: + current_time = datetime.datetime.now().strftime("%d.%m.%Y %H:%M") + prefix = f"[{current_time}, {username}]" if username is not None else f"[{current_time}]" + return f"{prefix}: {text}" if text is not None else f"{prefix}:" + + +def _encode_image(image: bytes) -> str: + encoded_image = base64.b64encode(image).decode('utf-8') + return f"data:image/jpeg;base64,{encoded_image}" + + +def _serialize_message(role: str, text: Optional[str], image: Optional[bytes]) -> dict: + serialized = {"role": role, "content": []} + if text is not None: + serialized["content"].append({"type": "text", "text": text}) + if image is not None: + serialized["content"].append({"type": "image_url", "image_url": {"url": _encode_image(image)}}) + return serialized + + +def _remove_none_recursive(data: Union[dict, list, any]) -> Union[dict, list, any]: + if isinstance(data, dict): + return { + k: _remove_none_recursive(v) + for k, v in data.items() + if v is not None + } + elif isinstance(data, list): + return [ + _remove_none_recursive(item) + for item in data + if item is not None + ] + else: + return data + + +def _serialize_assistant_message(message: AssistantMessage) -> AssistantMessageTypedDict: + return _remove_none_recursive(message.model_dump(by_alias=True)) + + +def _get_resolution_for_aspect_ratio(aspect_ratio: str) -> Tuple[int, int]: + aspect_ratio_resolution_map = { + "1:1": (1280, 1280), + "4:3": (1280, 1024), + "3:4": (1024, 1280), + "16:9": (1280, 720), + "9:16": (720, 1280) + } + return aspect_ratio_resolution_map.get(aspect_ratio, (1280, 1024)) + + +def _convert_image_to_jpeg(image: bytes) -> bytes: + img = Image.open(BytesIO(image)).convert("RGB") + output = BytesIO() + img.save(output, format='JPEG', quality=95, optimize=True) + return output.getvalue()