Text_formatting.py - master - github_better/GenAI

Text_formatting.py
607 строк21 KB
github_better
New
22 дек 2025, 08:52
22 дек 2025, 08:52b68d8c4
import re
import json
import uuid
from typing import Tuple
import ast
import logging
import os

from Lib.Debugger import debug_print

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def cut_text(text, number_of_words=20)->str:
    debug_print(logger)
    """
    Разбиение слов энтерами, чтобы было красивше

    :param text: исходный текст
    :param number_of_words: число слов после которых ставиться энтер
    :return: возрващает красивый текст
    """

    words = text.split()
    lines = [" ".join(words[i: i + number_of_words]) for i in range(0, len(words), number_of_words)]
    return "\n".join(lines)



def destroy_text_between_tag(text: str, tag: str) -> str:
    debug_print(logger)
    """
    Удаляет содержимое между тегами <tag>...</tag> вместе с самими тегами.
    
    :param text: текст 
    :param tag: теги из текста
    :return: текст без текста между тегами
    """

    pattern = fr'<{tag}>.*?</{tag}>'
    return re.sub(pattern, '', text, flags=re.DOTALL | re.IGNORECASE)

def take_tag(text, tag)->str:
    debug_print(logger)
    """
    выделяет текст между открывающимся и закрывающимеся тегами <TAG>...</TAG>

    :param text: исходный текст
    :param tag: нужный тег
    :return: возвращает текст между тегами
    """

    pattern = fr"<{tag}>(.*?)</{tag}>"
    match = re.search(pattern, text, re.DOTALL)
    if match:
        return match.group(1).strip()
    return ""


def kill_all_tags(text)->str:
    debug_print(logger)
    """
    :param text: исходный текст
    :return: удаляет все теги
    """

    return re.sub(r"</?[^>]+>", "", text)


def delete_string_after_double_dot(text: str, phrase: str) -> str:
    debug_print(logger)
    """
    Удаляет из текста строку, начинающуюся с "<phrase>:" вместе со всем после двоеточия до конца строки.
    :param text: исходный многострочный текст
    :param phrase: фраза до двоеточия, например "Launch event"
    :return: новый текст без этой строки
    """

    lines = text.splitlines()
    result_lines = []
    target = phrase + ":"
    for line in lines:
        # если строка начинается с "phrase:", пропускаем её
        if line.lstrip().startswith(target):
            continue
        result_lines.append(line)
    # Собираем обратно в один текст
    return "\n".join(result_lines)


def find_tag(text: str, tag: str) -> bool:
    debug_print(logger)
    """
    Проверяет, содержится ли в тексте открывающий или закрывающий тег <tag> или </tag>.

    :param text: Строка, в которой ведется поиск.
    :param tag: Имя тега без угловых скобок, например "ANSWER" или "LOCATION".
    :return: True, если тег найден, иначе False.
    """

    return f"<{tag}>" in text or f"</{tag}>" in text

def find_end_tag(text, tag) -> bool:
    debug_print(logger)
    """
    Ищет закрывающийся тег

    :param text: текст в котором будет поиск
    :param tag: тег который нужно найти
    :return: Найден/не найден
    """

    return f"</{tag}>" in text


def pretty_print_dialog(dialog):
    debug_print(logger)
    """
    :param dialog: красиво выводит диалог в формате:
    ответ_ИИ
    >>запрос_пользователя
    ответ_ИИ
    ...
    """

    if not dialog:
        debug_print(logger, "original dialog is empty")
        return

    i = 0
    while i < len(dialog):
        msg = dialog[i]
        role = msg.get('role', '')
        cont = kill_all_tags(msg.get('content', '')).strip()
        cont = cut_text(cont, 20).strip()

        if role == 'assistant':
            print(cont)
            i += 1
            if i < len(dialog) and dialog[i].get('role') == 'user':
                user_cont = kill_all_tags(dialog[i].get('content', '')).strip()
                user_cont = cut_text(user_cont, 20).strip()
                print(">> " + user_cont)
                i += 1
        elif role == 'user':
            print(">> " + cont)
            i += 1
            if i < len(dialog) and dialog[i].get('role') == 'assistant':
                ai_cont = kill_all_tags(dialog[i].get('content', '')).strip()
                ai_cont = cut_text(ai_cont, 20).strip()
                print(ai_cont)
                i += 1
        else:
            print(f"{role}: {cont}")
            i += 1

def pretty_print_dialog_whith_tags(dialog):
    debug_print(logger)
    """
    :param dialog: красиво выводит диалог в формате:
    ответ_ИИ
    >>запрос_пользователя
    ответ_ИИ
    ...
    """

    if not dialog:
        print("[DB] original_dialog is empty.")
        return

    i = 0
    while i < len(dialog):
        msg = dialog[i]
        role = msg.get('role', '')
        cont = msg.get('content', '')

        if role == 'assistant':
            print(cont)
            i += 1
            if i < len(dialog) and dialog[i].get('role') == 'user':
                user_cont = dialog[i].get('content', '')
                print(">> " + user_cont)
                i += 1
        elif role == 'user':
            print(">> " + cont)
            i += 1
            if i < len(dialog) and dialog[i].get('role') == 'assistant':
                ai_cont = dialog[i].get('content', '')
                print(ai_cont)
                i += 1
        else:
            print(f"{role}: {cont}")
            i += 1

def parce_location_data(original_dialog) -> str:
    debug_print(logger)
    """
    парсим последнюю локацию из диалога

    :param original_dialog: диалог, с которого будем парсить
    :return: последний запарсенный LOCATION
    """

    # --- пустой вход ---
    if not original_dialog:
        return ""

    # --- если пришла строка — пробуем сразу ---
    if isinstance(original_dialog, str):
        loc = take_tag(original_dialog, "LOCATION")
        return kill_all_tags(loc).strip() if loc else ""

    # --- если список — пробегаем с конца (чтобы найти самую последнюю локацию) ---
    if isinstance(original_dialog, list):
        for item in reversed(original_dialog):
            content = ""
            if isinstance(item, dict):
                content = item.get("content", "") or ""
            elif isinstance(item, str):
                content = item
            else:
                # неизвестный элемент — приводим к строке
                content = str(item)

            if not content:
                continue

            loc = take_tag(content, "LOCATION")
            if loc:
                return kill_all_tags(loc).strip()

    return ""

def parse_max_weight_and_slots(stats: str) -> Tuple[float, int]:
    debug_print(logger)
    """
    Из строки со статами достаёт MAX_WEIGHT (float) и MAX_SLOTS (int).
    Поддерживает python-репрезентацию dict, JSON или произвольный текст.
    Возвращает (max_weight, max_slots). По умолчанию (0.0, 0).

    :param stats: статы персонажа
    :return: (max_weight, max_slots)
    """

    s = str(stats).strip()
    if s.startswith("```"):
        s = re.sub(r"^```[a-zA-Z]*\n", "", s)  # убираем ```json
        s = s.rstrip("`")  # убираем хвост ```

    def __to_float_safe(x):
        """Пытаемся привести к float, аккуратно убирая лишние символы."""
        if x is None:
            return 0.0
        if isinstance(x, (int, float)):
            return float(x)
        s = str(x).strip()
        # заменить запятую на точку и удалить всё, кроме цифр, точки и знака минус
        s = s.replace(',', '.')
        m = re.search(r'[-+]?\d*\.?\d+', s)
        return float(m.group(0)) if m else 0.0

    def __to_int_safe(x):
        """Пытаемся привести к int (через float если нужно)."""
        if x is None:
            return 0
        if isinstance(x, int):
            return x
        try:
            return int(x)
        except Exception:
            f = __to_float_safe(x)
            return int(f)

    if not stats:
        return 0.0, 0

    # если передано что-то вроде "STATS: {...}" — обрежем всё до первого '{'
    if 'STATS' in s.upper() and '{' in s:
        idx = s.find('{')
        s = s[idx:]

    # Попробуем безопасно распарсить как python-literal (случай одинарных кавычек)
    try:
        parsed = ast.literal_eval(s)
        if isinstance(parsed, dict):
            max_weight = None
            max_slots = None
            for k, v in parsed.items():
                if isinstance(k, str):
                    if re.search(r'weight', k, re.I) and max_weight is None:
                        max_weight = __to_float_safe(v)
                    if re.search(r'slot', k, re.I) and max_slots is None:
                        max_slots = __to_int_safe(v)
            return (max_weight or 0.0, max_slots or 0)
    except Exception:
        pass

    # Попробуем распарсить как JSON (заменив одинарные кавычки — на случай простой миграции)
    try:
        j = json.loads(s.replace("'", '"'))
        if isinstance(j, dict):
            max_weight = None
            max_slots = None
            for k, v in j.items():
                if isinstance(k, str):
                    if re.search(r'weight', k, re.I) and max_weight is None:
                        max_weight = __to_float_safe(v)
                    if re.search(r'slot', k, re.I) and max_slots is None:
                        max_slots = __to_int_safe(v)
            return (max_weight or 0.0, max_slots or 0)
    except Exception:
        pass

    # Регэксп-фоллбэк по тексту: ищем шаблоны для MAX_WEIGHT и MAX_SLOTS
    weight_match = re.search(
        r'(?:MAX[_\s\-]*WEIGHT|MAXWEIGHT|max[_\s\-]*weight|maxWeight)\s*[:=]\s*([-\d\.,]+)',
        s,
        re.I
    )
    slots_match = re.search(
        r'(?:MAX[_\s\-]*SLOTS|MAXSLOTS|max[_\s\-]*slots|maxSlots)\s*[:=]\s*([-\d\.,]+)',
        s,
        re.I
    )

    if weight_match or slots_match:
        w = __to_float_safe(weight_match.group(1)) if weight_match else 0.0
        sl = __to_int_safe(slots_match.group(1)) if slots_match else 0
        return w, sl

    # Ещё один общий вариант: найти любые ключевые пары weight: X и slots: Y (иногда пишут без MAX_)
    weight_match2 = re.search(r'(?:MAX[_\s\-]*)?weight\s*[:=]\s*([-\d\.,]+)', s, re.I)
    slots_match2 = re.search(r'(?:MAX[_\s\-]*)?slots?\s*[:=]\s*([-\d\.,]+)', s, re.I)
    if weight_match2 or slots_match2:
        w = __to_float_safe(weight_match2.group(1)) if weight_match2 else 0.0
        sl = __to_int_safe(slots_match2.group(1)) if slots_match2 else 0
        return w, sl

    return 0.0, 0


def parse_weight_and_slots_new_objs(new_obj_in_inv: str) -> tuple[float, int]:
    debug_print(logger)
    """
    Парсит вес и слоты из множества новых объектов.
    Может быть JSON-список, JSON-объект или текст с weight:/slots:.
    Возвращает суммарные (weight, slots).

    :param new_obj_in_inv: новые объекты
    :return: вес и слоты в новых объектах
    """

    if not new_obj_in_inv:
        return 0.0, 0

    total_weight = 0.0
    total_slots = 0

    # --- Попробуем как JSON ---
    try:
        obj = json.loads(new_obj_in_inv.replace("'", '"'))

        if isinstance(obj, dict):
            total_weight += float(obj.get("weight", 0))
            total_slots += int(obj.get("slots", 0))

        elif isinstance(obj, list):
            for o in obj:
                if isinstance(o, dict):
                    total_weight += float(o.get("weight", 0))
                    total_slots += int(o.get("slots", 0))
    except Exception:
        # --- Если не JSON — парсим текст регулярками ---
        weights = re.findall(r'weight\s*:\s*([\d.]+)', new_obj_in_inv, re.IGNORECASE)
        slots = re.findall(r'slots?\s*:\s*(\d+)', new_obj_in_inv, re.IGNORECASE)

        total_weight = sum(float(w) for w in weights)
        total_slots = sum(int(s) for s in slots)

    return total_weight, total_slots


def add_new_obj_to_inv(current_inv: str, new_obj: str) -> str:
    debug_print(logger)
    """
    Добавляет новый объект в инвентарь (JSON-список).
    current_inv — строка с JSON или список.
    new_obj — строка с JSON-объектом или словарь.
    Возвращает новый JSON-список (строкой).

    :param current_inv: текущий инвентарь
    :param new_obj: новые объекты которые надо добавить
    :return: инвентарь с новыми предметами
    """

    # --- Загружаем текущий инвентарь ---
    try:
        if isinstance(current_inv, str):
            inv = json.loads(current_inv.replace("'", '"'))
        else:
            inv = current_inv
    except Exception:
        inv = []

    # --- Загружаем новый объект ---
    try:
        if isinstance(new_obj, str):
            obj = json.loads(new_obj.replace("'", '"'))
        else:
            obj = new_obj
    except Exception:
        return json.dumps(inv, ensure_ascii=False)

    if isinstance(obj, dict):
        obj["id"] = obj.get("id") or str(uuid.uuid4())
        inv.append(obj)
    elif isinstance(obj, list):
        for o in obj:
            o["id"] = o.get("id") or str(uuid.uuid4())
            inv.append(o)

    return json.dumps(inv, ensure_ascii=False)

def sanitize_inventory_for_history(inv_data) -> list:
    debug_print(logger)
    """
    Универсально чистит инвентарь перед записью в history:
    - принимает str / list / dict / None;
    - удаляет поле 'id' (если есть);
    - возвращает список словарей.

    :param inv_data: инвентарь
    :return: чистый инвентарь
    """

    # --- Нормализуем вход ---
    if inv_data is None:
        return []

    if isinstance(inv_data, str):
        try:
            # попытка распарсить JSON-строку
            inv = json.loads(inv_data.replace("'", '"'))
        except Exception:
            # fallback: вручную искать объекты {"name": ..., "count": ...}
            matches = re.findall(r'\{[^{}]*\}', inv_data)
            inv = []
            for m in matches:
                try:
                    item = json.loads(m.replace("'", '"'))
                    inv.append(item)
                except Exception:
                    continue
    elif isinstance(inv_data, dict):
        inv = [inv_data]
    elif isinstance(inv_data, list):
        inv = inv_data
    else:
        return []

    # --- Убираем поле 'id' и чистим от None ---
    cleaned = []
    for item in inv:
        if not isinstance(item, dict):
            continue
        new_item = {k: v for k, v in item.items() if k != "id" and v is not None}
        cleaned.append(new_item)
    return cleaned

def format_dialog_preview(dialog_list, max_per_message=3500, cut_len=200):
    debug_print(logger)
    """
    Форматирует превью диалога, объединяя пары USER/ASSISTANT сообщений.

    :param dialog_list: список сообщений диалога
    :param max_per_message: максимальная длина одного чанка
    :param cut_len: максимальная длина отдельного сообщения
    :return: список чанков текста для отправки
    """

    pairs = []
    i = 0
    L = len(dialog_list)
    while i < L:
        user_item = dialog_list[i] if i < L else None
        assistant_item = dialog_list[i + 1] if (i + 1) < L else None
        user_text = ''
        assistant_text = ''
        if user_item:
            user_text = kill_all_tags(user_item.get('content', '') or '')
            user_text = cut_text(user_text, cut_len)
        if assistant_item:
            assistant_text = kill_all_tags(assistant_item.get('content', '') or '')
            assistant_text = cut_text(assistant_text, cut_len)
        pair_text = ""
        if user_item:
            pair_text += f"<b>ASK:</b>\n{user_text}"
        if assistant_item:
            pair_text += f"\n\n<b>ANSWER:</b>\n{assistant_text}"
        pairs.append(pair_text)
        i += 2
    full = "\n\n---\n\n".join(pairs) if pairs else ""
    chunks = [full[i:i + max_per_message] for i in range(0, len(full), max_per_message)] if full else []
    return chunks if chunks else ["No dialog to preview."]

def take_game_guide(lang: str) -> str:
    debug_print(logger)
    """
    Парсит из txt игровые правила

    :param lang: язык юзера
    :return: строка с правилами
    """

    file_path = os.path.join("Text", f"Game_guide_{lang}.txt")
    try:
        with open(file_path, "r", encoding="utf-8") as f:
            text = f.read()
        # Экранируем < и > чтобы Telegram не путал с HTML
        safe_text = text.replace("<", "&lt;").replace(">", "&gt;")
        return safe_text
    except FileNotFoundError:
        return f"⚠️ Guide file not found for language '{lang}' ({file_path})"
    except Exception as e:
        return f"⚠️ Error reading guide file: {e}"

def take_polit() -> str:
    debug_print(logger)
    """
    :return: Берет политику использования (она тока на английском)
    """

    file_path = os.path.join("Text", f"Rules_of_usage.txt")
    try:
        with open(file_path, "r", encoding="utf-8") as f:
            text = f.read()
        return text
    except Exception as e:
        return f"⚠️ Error reading guide file: {e}"


def get_last_assistant_message(original_dialog) -> str:
    debug_print(logger)
    """
    Извлекает последнее сообщение ассистента из диалога.

    param original_dialog: список словарей с диалогом или JSON-строка
    return: текст последнего сообщения ассистента без тегов, или пустая строка
    """

    if not original_dialog:
        return ""

    # Если пришла строка - парсим JSON
    if isinstance(original_dialog, str):
        try:
            dialog = json.loads(original_dialog)
        except Exception:
            return ""
    else:
        dialog = original_dialog

    if not isinstance(dialog, list):
        return ""

    # Ищем последнее сообщение ассистента с конца
    for item in reversed(dialog):
        if not isinstance(item, dict):
            continue

        role = item.get("role", "")
        if role == "assistant":
            content = item.get("content", "")
            # Убираем все теги
            clean_content = kill_all_tags(content)
            return clean_content.strip()

    return ""

def split_text_into_chunks(text: str, max_len: int = 4096) -> list:
    debug_print(logger)
    """
    Бьет текст на чанки, чтобы потом их отправить нормально в тг
    
    :param text: текст
    :param max_len: максимальная длина чанка (в символах)
    :return: список чанков
    """

    lines = (text or "").split("\n")
    chunks = []
    buf = ""
    for line in lines:
        add = (("\n" if buf else "") + line)
        if len(buf) + len(add) <= max_len:
            buf += add
        else:
            if buf:
                chunks.append(buf)
            buf = line
            # Если одна строка превышает лимит — жёстко режем
            while len(buf) > max_len:
                chunks.append(buf[:max_len])
                buf = buf[max_len:]
    if buf:
        chunks.append(buf)
    return chunks