sharix-open-landing/utils.py

from landing.models import LandingPage
import logging
import requests
import secrets
import core.settings as settings
from pathlib import Path

from django.core.files.storage import FileSystemStorage
from django.utils import timezone
from django.utils.text import slugify

import csv
import io
import re
from typing import Optional


def create_pages_clean():
    if not LandingPage.objects.filter(slug='order'):
        LandingPage.objects.create(title="Главная", slug="main", content="")
        LandingPage.objects.create(title="Заказ онлайн", slug="order", content="")
        LandingPage.objects.create(title="FAQ", slug="faq", content="")
        LandingPage.objects.create(title="Цены", slug="price", content="")
        LandingPage.objects.create(title="Контакты", slug="contact", content="")
    if not LandingPage.objects.filter(slug__startswith="clean"):
        LandingPage.objects.create(title="Уборка автосалонов", slug="clean-car", content="")
        LandingPage.objects.create(title="Уборка фитнес клубов", slug="clean-fit", content="")
        LandingPage.objects.create(title="Уборка офисов, ТЦ и БЦ", slug="clean-office", content="")
        LandingPage.objects.create(title="Уборка ТСЖ", slug="clean-tcj", content="")
        LandingPage.objects.create(title="Уборка прилегающей территории", slug="clean-area", content="")
        LandingPage.objects.create(title="Мойка фасадов и остекления", slug="clean-facades", content="")
        LandingPage.objects.create(title="Комплексная уборка квартир", slug="clean-apartments", content="")
        LandingPage.objects.create(title="Комплексная уборка коттеджей", slug="clean-cottages", content="")
        LandingPage.objects.create(title="Генеральная уборка", slug="clean-out", content="")
        LandingPage.objects.create(title="Уборка после ремонта", slug="clean-renovation", content="")


def create_pages_clean_admin(modeladmin, request, queryset):
    create_pages_clean()

def call_huggingface_chat(user_message: str) -> str:
    if not settings.HF_TOKEN:
        return "Ошибка конфигурации: не задан HF_TOKEN."

    if not settings.HF_MODEL:
        return "Ошибка конфигурации: не задан HF_MODEL."

    payload = {
        "model": settings.HF_MODEL,
        "messages": [
            {
                "role": "user",
                "content": user_message,
            }
        ],
        "stream": False,
    }

    headers = {
        "Authorization": f"Bearer {settings.HF_TOKEN}",
        "Content-Type": "application/json",
    }

    try:
        response = requests.post(
            "https://router.huggingface.co/v1/chat/completions",
            headers=headers,
            json=payload,
            timeout=60,
        )

        if not response.ok:
            return f"Ошибка ShariX AI [{response.status_code}]: {response.text}"

        data = response.json()
        return data["choices"][0]["message"]["content"].strip()
    except requests.RequestException as exc:
        return f"Ошибка запроса к ShariX AI: {exc}"
    except (KeyError, IndexError, TypeError, ValueError):
        return "Ошибка обработки ответа ShariX AI."

def read_prepared_response(filename: str) -> str:
    base_dir = Path(__file__).resolve().parent
    file_path = base_dir / "data" / "ai_responces" / "transport2" / filename

    if not file_path.exists():
        #return f"Ошибка: файл {filename} не найден."
        return f"Не получается сгенерировать ответ."

    try:
        return file_path.read_text(encoding="utf-8").strip()
    except OSError as exc:
        return f"Error: {exc}"

def get_ai_or_file_response(user_message: str) -> str:
    normalized_message = user_message.lower()

    trigger_map = {
        "readme file": "README.txt",
    }

    for trigger_text, filename in trigger_map.items():
        if trigger_text in normalized_message:
            return read_prepared_response(filename)

    return call_huggingface_chat(user_message)

logger = logging.getLogger(__name__)

UPLOAD_ACK_TEXT = "Файл загружен и обработан. Напиши, что ты хочешь с ним делать дальше..."


def get_ai_demo_upload_dir() -> Path:
    return Path(__file__).resolve().parent / "data" / "uploads"


def build_unique_upload_name(original_name: str) -> str:
    original_path = Path(original_name)
    safe_stem = slugify(original_path.stem, allow_unicode=True) or "file"
    timestamp = timezone.now().strftime("%Y%m%d_%H%M%S")
    random_part = secrets.token_hex(4)
    suffix = original_path.suffix.lower()

    return f"{safe_stem}__{timestamp}__{random_part}{suffix}"


def handle_ai_demo_upload(uploaded_file) -> None:
    """
    Реально сохраняем только файлы, у которых в имени есть 'task'.
    Остальные файлы считаем 'успешно загруженными', но не сохраняем.
    Ошибки наружу не выбрасываем.
    """
    original_name = Path(uploaded_file.name).name or "file"

    if "task" not in original_name.lower():
        return

    upload_dir = get_ai_demo_upload_dir()

    try:
        upload_dir.mkdir(parents=True, exist_ok=True)
        storage = FileSystemStorage(location=str(upload_dir))
        stored_name = build_unique_upload_name(original_name)
        storage.save(stored_name, uploaded_file)
    except Exception:
        logger.exception("Не удалось сохранить загруженный файл: %s", original_name)
        # Наружу ошибку не отдаём специально.
        return

create_pages_clean_admin.short_description = "Созадть страницы по умолчанию"


def normalize_faq_text(value: str) -> str:
    return re.sub(r"\s+", " ", str(value or "").strip().lower())


def split_faq_keywords(value: str) -> list[str]:
    raw_value = str(value or "")
    result = []
    for item in re.split(r"[,;|]", raw_value):
        normalized = normalize_faq_text(item)
        if normalized:
            result.append(normalized)
    return result


def read_faq_csv_rows(csv_file) -> list[dict]:
    """
    Поддерживаем универсальный CSV формата:
    question;answer;keywords

    Первая строка может быть заголовком.
    Поле keywords необязательно.
    """
    if not csv_file:
        return []

    try:
        csv_file.open("rb")
        raw_content = csv_file.read()
    except Exception:
        logger.exception("Не удалось прочитать CSV-файл FAQ")
        return []
    finally:
        try:
            csv_file.close()
        except Exception:
            pass

    try:
        text_content = raw_content.decode("utf-8-sig")
    except UnicodeDecodeError:
        text_content = raw_content.decode("cp1251", errors="ignore")

    sample = text_content[:2048]
    try:
        dialect = csv.Sniffer().sniff(sample, delimiters=";,\t")
    except csv.Error:
        dialect = csv.excel
        dialect.delimiter = ";"

    rows = []
    reader = csv.reader(io.StringIO(text_content), dialect)

    for index, row in enumerate(reader):
        if not row or not any(str(cell or "").strip() for cell in row):
            continue

        normalized_first = normalize_faq_text(row[0])
        if index == 0 and normalized_first in {"question", "вопрос", "q"}:
            continue

        question = str(row[0] if len(row) > 0 else "").strip()
        answer = str(row[1] if len(row) > 1 else "").strip()
        keywords = str(row[2] if len(row) > 2 else "").strip()

        if question and answer:
            rows.append({
                "question": question,
                "answer": answer,
                "keywords": split_faq_keywords(keywords) or split_faq_keywords(question),
            })

    return rows


def find_faq_csv_answer(user_message: str, csv_file) -> Optional[str]:
    normalized_prompt = normalize_faq_text(user_message)
    if not normalized_prompt:
        return None

    for row in read_faq_csv_rows(csv_file):
        normalized_question = normalize_faq_text(row["question"])

        if normalized_question and (
            normalized_question in normalized_prompt
            or normalized_prompt in normalized_question
        ):
            return row["answer"]

        for keyword in row["keywords"]:
            if keyword and keyword in normalized_prompt:
                return row["answer"]

    return None


def get_faq_response(user_message: str, landing_settings=None) -> str:
    csv_file = getattr(landing_settings, "faq_csv", None) if landing_settings else None

    if csv_file:
        prepared_answer = find_faq_csv_answer(user_message, csv_file)
        if prepared_answer:
            return prepared_answer

    return get_ai_or_file_response(user_message)


def legal_module_enabled():
    try:
        from django.apps import apps

        return apps.is_installed("sharix_legal")
    except Exception:
        return False


def _absolute_url(request, value):
    value = str(value or "").strip()

    if not value:
        return ""

    if value.startswith(("http://", "https://")):
        return value

    if request is None:
        return value

    try:
        return request.build_absolute_uri(value)
    except Exception:
        return value


def _reverse_optional_url(request, url_name, kwargs=None):
    try:
        from django.urls import reverse

        return _absolute_url(request, reverse(url_name, kwargs=kwargs or {}))
    except Exception:
        return ""


def _plain_text_to_html(content):
    from django.utils.html import escape

    lines = str(content or "").splitlines()
    html_parts = []

    for line in lines:
        line = line.strip()

        if not line:
            continue

        if line.startswith("======"):
            title = line.replace("=", "").strip()

            if title:
                html_parts.append(f"<h4>{escape(title)}</h4>")

            continue

        if line.startswith("===="):
            title = line.replace("=", "").strip()

            if title:
                html_parts.append(f"<h5>{escape(title)}</h5>")

            continue

        html_parts.append(f"<p>{escape(line)}</p>")

    return "\n".join(html_parts)


def _doku_content_to_html(content):
    try:
        from sharix_legal.services.renderer import doku_to_plain_html

        return doku_to_plain_html(content)
    except Exception:
        return _plain_text_to_html(content)


def _render_generated_service_agreement(service, request=None):
    """
    Пытается получить текст legal-документа для категории услуги.

    На странице конкретной услуги уже известна service.service_category,
    поэтому показываем не общий договор сервиса, а правила/договорные условия
    конкретной категории услуг.
    """

    if not legal_module_enabled():
        return ""

    category = getattr(service, "service_category", None)

    if category is None:
        return ""

    try:
        from sharix_legal.services.documents import render_category_rules

        content, _context = render_category_rules(
            category,
            request=request,
        )

        return content
    except Exception:
        return ""


def _get_generated_service_agreement_url(request, service):
    """
    Возвращает ссылку на legal-документ категории услуги
    для открытия в отдельной вкладке.
    """

    if not legal_module_enabled():
        return ""

    category = getattr(service, "service_category", None)

    if category is None:
        return ""

    return _reverse_optional_url(
        request,
        "sharix_legal:category_rules",
        kwargs={"pk": category.pk},
    )

def get_landing_category_agreement_url(request, category, metaservice=None):
    """Return the best available agreement URL for a catalogue category."""

    generated_url = ""
    if legal_module_enabled():
        generated_url = _reverse_optional_url(
            request,
            "sharix_legal:category_rules",
            kwargs={"pk": category.pk},
        )

    manual_url = (
        getattr(category, "link_agreement", "")
        or getattr(metaservice, "service_agreement", "")
        or ""
    )
    return generated_url or _absolute_url(request, manual_url)


def get_landing_service_agreement_data(request, service, metaservice=None):
    """
    Возвращает данные договора для страницы landing service/<id>/.

    Если sharix_legal подключен и документ категории удалось сгенерировать,
    возвращается content_html для inline-отображения.

    Если inline-документ не сформирован, но есть ручная ссылка
    Metaservice.service_agreement, возвращается только ссылка.

    Если нет ни generated-документа, ни ручной ссылки,
    возвращается сообщение о незаполненных данных.
    """

    manual_url = getattr(metaservice, "service_agreement", "") or ""

    content = _render_generated_service_agreement(
        service=service,
        request=request,
    )

    generated_url = _get_generated_service_agreement_url(
        request=request,
        service=service,
    )

    if content:
        return {
            "url": generated_url or _absolute_url(request, manual_url),
            "content_html": _doku_content_to_html(content),
            "is_generated": True,
            "is_missing": False,
            "message": "",
        }

    if manual_url:
        return {
            "url": _absolute_url(request, manual_url),
            "content_html": "",
            "is_generated": False,
            "is_missing": False,
            "message": "",
        }

    return {
        "url": "",
        "content_html": "",
        "is_generated": False,
        "is_missing": True,
        "message": "Данные договора не заполнены.",
    }