aramesh-backend/chatbot/content.py

"""خواندن محتوای سایت از PostgreSQL و تبدیل آن به سندهای قابل ایندکس.

هر سند یک dict با کلیدهای text / type / title / url است. بدنه مقاله‌ها و
پروژه‌ها هم می‌تواند آرایه پاراگراف باشد و هم سند TipTap (JSON) — هر دو
پشتیبانی می‌شوند. مقاله‌های پیش‌نویس ایندکس نمی‌شوند.
"""

import logging
from typing import Any

from .persian import normalize

log = logging.getLogger(__name__)

# URL پایه صفحات — چت‌بات مسیر نسبی می‌دهد؛ فرانت‌اند روی همان دامنه است
_ARTICLE_URL = "/blog/{slug}"
_PROJECT_URL = "/projects/{slug}"


def _psycopg_dsn(database_url: str) -> str:
    """DATABASE_URL فرمت SQLAlchemy (postgresql+psycopg://) را به DSN psycopg3 تبدیل می‌کند."""
    return database_url.replace("+psycopg", "").replace("+asyncpg", "")


def _fetch_rows(dsn: str, sql: str) -> list[dict[str, Any]]:
    import psycopg
    from psycopg.rows import dict_row

    with psycopg.connect(dsn, row_factory=dict_row, connect_timeout=10) as conn:
        cur = conn.execute(sql)
        return cur.fetchall()


def tiptap_blocks(node: Any) -> list[str]:
    """متن بلوک‌به‌بلوک از سند TipTap (یا هر JSON تو در تو) بیرون می‌کشد."""
    if not isinstance(node, dict):
        return []
    ntype = node.get("type")
    if ntype in ("paragraph", "heading", "codeBlock"):
        text = _inline_text(node).strip()
        return [text] if text else []
    blocks: list[str] = []
    for child in node.get("content") or []:
        blocks.extend(tiptap_blocks(child))
    return blocks


def _inline_text(node: dict) -> str:
    parts: list[str] = []
    for child in node.get("content") or []:
        ctype = child.get("type")
        if ctype == "text":
            parts.append(child.get("text", ""))
        elif ctype == "hardBreak":
            parts.append(" ")
        elif ctype == "image":
            alt = (child.get("attrs") or {}).get("alt") or ""
            if alt:
                parts.append(f"[تصویر: {alt}]")
        else:
            parts.append(_inline_text(child))
    return "".join(parts)


def body_to_text(body: Any) -> str:
    """بدنه مقاله/پروژه (آرایه پاراگراف یا سند TipTap) → متن یکپارچه."""
    if not body:
        return ""
    if isinstance(body, list):
        # فرمت قدیمی: آرایه رشته‌ای پاراگراف‌ها
        return "\n".join(str(p).strip() for p in body if str(p).strip())
    blocks = tiptap_blocks(body)
    return "\n".join(b for b in blocks if b)


def _doc(text: str, dtype: str, title: str, url: str) -> dict | None:
    text = (text or "").strip()
    if not text:
        return None
    return {"text": text, "type": dtype, "title": title, "url": url}


def _clean(value: Any) -> str:
    return normalize(str(value or "")).strip()


def fetch_site_documents(database_url: str) -> list[dict]:
    """همه محتوای عمومی سایت را به سند تبدیل می‌کند (بدون وابستگی به ORM بک‌اند)."""
    dsn = _psycopg_dsn(database_url)
    docs: list[dict] = []

    # --- برند و اطلاعات تماس (رکورد یگانه) ---
    brand_rows = _fetch_rows(dsn, "SELECT name, phone, email, address FROM brand LIMIT 1")
    if brand_rows:
        b = brand_rows[0]
        contact = "\n".join(
            filter(None, [
                f"نام مجموعه: {_clean(b['name'])}",
                f"تلفن تماس: {_clean(b['phone'])}",
                f"ایمیل: {_clean(b['email'])}",
                f"آدرس: {_clean(b['address'])}",
            ])
        )
        d = _doc(contact, "contact", "اطلاعات تماس", "/contact")
        if d:
            docs.append(d)

    # --- دفاتر ---
    for row in _fetch_rows(dsn, "SELECT city, address, hours FROM offices ORDER BY position"):
        d = _doc(
            f"دفتر {_clean(row['city'])} — آدرس: {_clean(row['address'])} — ساعت کاری: {_clean(row['hours'])}",
            "office", f"دفتر {_clean(row['city'])}", "/contact",
        )
        if d:
            docs.append(d)

    # --- خدمات ---
    for row in _fetch_rows(dsn, "SELECT title, desc FROM services ORDER BY position"):
        d = _doc(
            f"خدمت «{_clean(row['title'])}»: {_clean(row['desc'])}",
            "service", _clean(row["title"]), "/services",
        )
        if d:
            docs.append(d)

    # --- پرسش‌های متداول ---
    for row in _fetch_rows(dsn, "SELECT question, answer FROM faqs ORDER BY position"):
        d = _doc(
            f"پرسش: {_clean(row['question'])}\nپاسخ: {_clean(row['answer'])}",
            "faq", _clean(row["question"]), "/faq",
        )
        if d:
            docs.append(d)

    # --- پروژه‌ها ---
    for row in _fetch_rows(
        dsn,
        "SELECT slug, title, location, year, status, area, floors, units, client, summary, body FROM projects ORDER BY position",
    ):
        meta = "\n".join(filter(None, [
            f"مکان: {_clean(row['location'])}" if row["location"] else "",
            f"سال: {_clean(row['year'])}" if row["year"] else "",
            f"وضعیت: {'در حال ساخت' if row['status'] == 'progress' else 'تکمیل‌شده'}",
            f"متراژ: {_clean(row['area'])}" if row["area"] else "",
            f"طبقات: {_clean(row['floors'])}" if row["floors"] else "",
            f"واحدها: {_clean(row['units'])}" if row["units"] else "",
            f"کارفرما: {_clean(row['client'])}" if row["client"] else "",
        ]))
        text = "\n".join(filter(None, [meta, _clean(row["summary"]), body_to_text(row["body"])]))
        d = _doc(text, "project", _clean(row["title"]), _PROJECT_URL.format(slug=row["slug"]))
        if d:
            docs.append(d)

    # --- مقاله‌ها (فقط منتشرشده) ---
    for row in _fetch_rows(
        dsn,
        "SELECT slug, title, date, category, excerpt, body FROM articles WHERE status = 'published' ORDER BY position",
    ):
        text = "\n".join(filter(None, [
            f"دسته‌بندی: {_clean(row['category'])}" if row["category"] else "",
            _clean(row["excerpt"]),
            body_to_text(row["body"]),
        ]))
        d = _doc(text, "article", _clean(row["title"]), _ARTICLE_URL.format(slug=row["slug"]))
        if d:
            docs.append(d)

    # --- مدل‌های سرمایه‌گذاری ---
    for row in _fetch_rows(dsn, "SELECT title, desc, points FROM investment_models ORDER BY position"):
        points = "؛ ".join(_clean(p) for p in (row["points"] or []))
        d = _doc(
            "\n".join(filter(None, [
                f"«{_clean(row['title'])}»: {_clean(row['desc'])}",
                f"نکات: {points}" if points else "",
            ])),
            "investment", _clean(row["title"]), "/investment",
        )
        if d:
            docs.append(d)

    # --- مراحل همکاری ---
    for row in _fetch_rows(dsn, "SELECT step, title, desc FROM process_steps ORDER BY position"):
        d = _doc(
            f"مرحله {_clean(row['step'])} — {_clean(row['title'])}: {_clean(row['desc'])}",
            "process", _clean(row["title"]), "/investment",
        )
        if d:
            docs.append(d)

    # --- چرا ما / تیم / نظرسنجی / تایم‌لاین / داستان ما ---
    for row in _fetch_rows(dsn, "SELECT title, desc FROM why_us ORDER BY position"):
        d = _doc(f"{_clean(row['title'])}: {_clean(row['desc'])}", "about", _clean(row["title"]), "/about")
        if d:
            docs.append(d)

    for row in _fetch_rows(dsn, "SELECT name, role, bio FROM team ORDER BY position"):
        d = _doc(
            f"{_clean(row['name'])} — {_clean(row['role'])}. {_clean(row['bio'])}",
            "team", _clean(row["name"]), "/about",
        )
        if d:
            docs.append(d)

    for row in _fetch_rows(dsn, "SELECT name, role, quote FROM testimonials ORDER BY position"):
        d = _doc(
            f"نظر {_clean(row['name'])} ({_clean(row['role'])}): {_clean(row['quote'])}",
            "testimonial", _clean(row["name"]), "/",
        )
        if d:
            docs.append(d)

    for row in _fetch_rows(dsn, "SELECT year, title, desc FROM timeline ORDER BY position"):
        d = _doc(
            f"سال {_clean(row['year'])} — {_clean(row['title'])}: {_clean(row['desc'])}",
            "timeline", _clean(row["title"]), "/about",
        )
        if d:
            docs.append(d)

    story_rows = _fetch_rows(dsn, "SELECT title, paragraphs, quote FROM story LIMIT 1")
    if story_rows and story_rows[0]:
        s = story_rows[0]
        paragraphs = "\n".join(_clean(p) for p in (s["paragraphs"] or []) if _clean(p))
        text = "\n".join(filter(None, [_clean(s["title"]), paragraphs, _clean(s["quote"])]))
        d = _doc(text, "about", _clean(s["title"]) or "درباره ما", "/about")
        if d:
            docs.append(d)

    log.info("Fetched %d documents from site database", len(docs))
    return docs


def chunk_documents(docs: list[dict], chunk_size: int = 700, overlap: int = 120) -> list[dict]:
    """متن‌های بلند (مقاله/پروژه) را به قطعات هم‌پوشان تقسیم می‌کند.

    هر قطعه یک doc_id پایدار می‌گیرد («type:ترتیب در همان url:شماره قطعه») تا
    همگام‌سازی افزایشی pgvector بتواند فقط سطرهای تغییر یافته را امبد کند.
    """
    from langchain_text_splitters import RecursiveCharacterTextSplitter

    from .vectorstore import chunk_content_hash

    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=overlap,
        separators=["\n", "۔", ".", "،", " ", ""],
    )

    # شماره ردیف هر سند در میان سندهای هم‌type و هم-url (برای FAQs که url یکسان دارند)
    occurrence: dict[str, int] = {}

    out: list[dict] = []
    for doc in docs:
        key = f"{doc['type']}|{doc['url']}"
        occ = occurrence.get(key, 0)
        occurrence[key] = occ + 1

        pieces = splitter.split_text(doc["text"]) if len(doc["text"]) > chunk_size * 1.3 else [doc["text"]]
        for i, piece in enumerate(pieces, 1):
            chunk = {
                "doc_id": f"{doc['type']}:{occ}:{i}",
                "text": piece if len(pieces) == 1 else f"[{doc['title']} — بخش {i} از {len(pieces)}]\n{piece}",
                "type": doc["type"],
                "title": doc["title"],
                "url": doc["url"],
            }
            chunk["content_hash"] = chunk_content_hash(chunk)
            out.append(chunk)

    log.info("Chunked %d documents into %d pieces", len(docs), len(out))
    return out


def content_fingerprint(docs: list[dict]) -> str:
    """هش کل محتوا — فقط برای لاگ؛ تشخیص تغییر ردیفی در vectorstore انجام می‌شود."""
    import hashlib
    import json

    payload = json.dumps(docs, ensure_ascii=False, sort_keys=True)
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:16]


# --- زمینه همیشه‌تازه سایت (بلوک کوتاه تماس/خدمات برای پرامپت) ---
_site_context_cache: tuple[float, str] | None = None


def fetch_site_context(database_url: str, ttl: int = 120) -> str:
    """نام برند، تماس و فهرست خدمات — با کش کوتاه‌مدت تا هر پیام یک کوئری سبک بزند."""
    global _site_context_cache
    import time as _time

    now = _time.time()
    if _site_context_cache and now - _site_context_cache[0] < ttl:
        return _site_context_cache[1]
    try:
        dsn = _psycopg_dsn(database_url)
        brand = _fetch_rows(dsn, "SELECT name, phone, email, address FROM brand LIMIT 1")
        services = _fetch_rows(dsn, "SELECT title FROM services ORDER BY position")
        lines: list[str] = []
        if brand:
            b = brand[0]
            lines += [
                f"نام مجموعه: {_clean(b['name'])}",
                f"تلفن تماس: {_clean(b['phone'])}",
                f"ایمیل: {_clean(b['email'])}",
                f"آدرس: {_clean(b['address'])}",
            ]
        if services:
            lines.append("خدمات سایت: " + "، ".join(_clean(s["title"]) for s in services))
        ctx = "\n".join(lines)
        _site_context_cache = (now, ctx)
        return ctx
    except Exception as e:
        log.warning("Site context fetch failed: %s", e)
        return _site_context_cache[1] if _site_context_cache else ""