aramesh-backend/chatbot/content.py

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

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

به‌جز جدول‌های محتوایی، متن‌های مدیریت محتوا (آمار، هیرو، هیرو صفحه‌ها،
تیتر بخش‌ها، فوتر، قبل/بعد) هم ایندکس می‌شوند — ادعاهای مرکزی سایت
(سال تجربه، تعداد پروژه و…) فقط آنجا ثبت شده‌اند.
"""

import logging
from typing import Any

from .persian import normalize

log = logging.getLogger(__name__)

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

# بخش‌های صفحه اصلی و صفحات داخلی — برای نسبت دادن URL به تیتر بخش‌ها
_SECTION_URL = {
    "services": "/services",
    "featured_projects": "/projects",
    "articles": "/blog",
    "faq": "/faq",
    "investment_models": "/investment",
    "process": "/investment",
}

# سندهای کوتاه‌تر از این جای‌گاه‌خالی‌اند («خدمت جدید» و امثال آن) — ایندکس نمی‌شوند
_MIN_DOC_CHARS = 30


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 len(text) < _MIN_DOC_CHARS:
        if text:
            log.info("Skipped tiny doc (%d chars, type=%s, title=%s)", len(text), dtype, title)
        return None
    return {"text": text, "type": dtype, "title": title, "url": url}


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


def _format_stat_rows(rows: list[dict[str, Any]]) -> list[str]:
    """سطرهای جدول stats → «۲۳+ سال تجربه ساخت» (پسوند تک‌حرفی می‌چسبد، بقیه با فاصله)."""
    parts: list[str] = []
    for r in rows:
        suffix = _clean(r.get("suffix"))
        value = str(r.get("value") or "").strip()
        if not value:
            continue
        num = f"{value}{suffix}" if len(suffix) == 1 else f"{value} {suffix}".strip()
        parts.append(f"{num} {_clean(r.get('label'))}".strip())
    return parts


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)

    # --- آمار و افتخارات — اعداد مرکزی سایت فقط اینجا ثبت‌اند ---
    stats = _format_stat_rows(_fetch_rows(dsn, "SELECT value, suffix, label FROM stats ORDER BY position"))
    if stats:
        d = _doc("آمار و افتخارات مجموعه: " + "؛ ".join(stats), "stats", "آمار و افتخارات", "/")
        if d:
            docs.append(d)

    # --- هیرو صفحه اصلی (رکورد یگانه) ---
    hero_rows = _fetch_rows(
        dsn,
        "SELECT kicker, title, title_highlight, subtitle, badge_value, badge_label FROM hero LIMIT 1",
    )
    if hero_rows:
        h = hero_rows[0]
        title = _clean(h["title"])
        if _clean(h["title_highlight"]):
            title = f"{title} (کلمهٔ برجسته: {_clean(h['title_highlight'])})"
        lines = list(filter(None, [
            f"شعار بالای تیتر: {_clean(h['kicker'])}" if _clean(h["kicker"]) else "",
            f"تیتر اصلی: {title}" if title else "",
            f"زیرتیتر: {_clean(h['subtitle'])}" if _clean(h["subtitle"]) else "",
            (
                f"نشان: {_clean(h['badge_value'])} {_clean(h['badge_label'])}"
                if _clean(h["badge_value"]) or _clean(h["badge_label"]) else ""
            ),
        ]))
        d = _doc("\n".join(lines), "hero", "معرفی صفحه اصلی", "/")
        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, land_area, pilot_floors,
               units_residential, units_commercial, units_office, client,
               project_type, usage, summary, body
        FROM projects ORDER BY position
        """,
    ):
        units = "، ".join(
            f"{v} واحد {label}"
            for label, col in (("مسکونی", row["units_residential"]),
                               ("تجاری", row["units_commercial"]),
                               ("اداری", row["units_office"]))
            if (v := _clean(col)) and v != "0"
        )
        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['project_type'])}" if row["project_type"] else "",
            f"کاربری: {_clean(row['usage'])}" if row["usage"] else "",
            f"متراژ زیربنا: {_clean(row['area'])}" if row["area"] else "",
            f"عرصه: {_clean(row['land_area'])}" if row["land_area"] else "",
            f"طبقات: {_clean(row['floors'])}" if row["floors"] else "",
            f"طبقات پیلوت: {_clean(row['pilot_floors'])}" if row["pilot_floors"] else "",
            f"واحدها: {units}" if units else "",
            f"کارفرما: {_clean(row['client'])}" if row["client"] else "",
        ]))
        text = "\n".join(filter(None, [
            f"پروژه «{_clean(row['title'])}»",
            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, faqs 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)

        article_url = _ARTICLE_URL.format(slug=row["slug"])
        for f in row["faqs"] or []:
            if not isinstance(f, dict):
                continue
            q, a = _clean(f.get("question")), _clean(f.get("answer"))
            d = _doc(f"پرسش: {q}\nپاسخ: {a}", "faq", q, article_url)
            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 key, title, lead FROM section_headings ORDER BY key"):
        text = "\n".join(filter(None, [_clean(row["title"]), _clean(row["lead"])]))
        d = _doc(text, "section", _clean(row["title"]), _SECTION_URL.get(row["key"], "/"))
        if d:
            docs.append(d)

    # --- هیرو صفحه‌های داخلی (about/services/projects/… ) ---
    for row in _fetch_rows(dsn, "SELECT key, title, lead FROM page_heroes ORDER BY key"):
        text = "\n".join(filter(None, [_clean(row["title"]), _clean(row["lead"])]))
        d = _doc(text, "page_hero", _clean(row["title"]), f"/{row['key']}")
        if d:
            docs.append(d)

    # --- قبل/بعد (بخش صفحه اصلی) ---
    ba_rows = _fetch_rows(dsn, "SELECT title, lead FROM before_after LIMIT 1")
    if ba_rows:
        text = "\n".join(filter(None, [_clean(ba_rows[0]["title"]), _clean(ba_rows[0]["lead"])]))
        d = _doc(text, "before_after", _clean(ba_rows[0]["title"]) or "قبل و بعد", "/")
        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)

    testimonial_rows = _fetch_rows(dsn, "SELECT name, role, quote FROM testimonials ORDER BY position")
    for row in testimonial_rows:
        d = _doc(
            f"نظر {_clean(row['name'])} ({_clean(row['role'])}): {_clean(row['quote'])}",
            "testimonial", _clean(row["name"]), "/",
        )
        if d:
            docs.append(d)
    # سند تجمیعی — پرسش‌های طبیعی مثل «نظرات مشتریان چیست؟» به تک‌نفرها نمی‌رسند
    # (بی‌ریشه‌سازی BM25: «نظرات» با «نظر» و «مشتریان» با «کارفرما» هیچ اشتراکی ندارند)
    if len(testimonial_rows) >= 2:
        agg = "؛ ".join(
            f"{_clean(t['name'])} ({_clean(t['role'])}): {_clean(t['quote'])}"
            for t in testimonial_rows
        )
        d = _doc(f"نظرات مشتریان و کارفرمایان پروژه‌ها:\n{agg}", "testimonial", "نظرات مشتریان", "/")
        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)

    footer_rows = _fetch_rows(dsn, "SELECT about_text FROM footer LIMIT 1")
    if footer_rows and _clean(footer_rows[0]["about_text"]):
        d = _doc(
            _clean(footer_rows[0]["about_text"]),
            "about", "درباره مجموعه (فوتر)", "/",
        )
        if d:
            docs.append(d)

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


def _url_key(url: str) -> str:
    """URL → کلید یکتا برای doc_id («/projects/x» → «projects-x»)."""
    return url.strip("/").replace("/", "-") or "home"


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

    هر قطعه یک doc_id پایدار می‌گیرد («type:url-key:ترتیب در همان url:شماره قطعه») تا
    همگام‌سازی افزایشی pgvector بتواند فقط سطرهای تغییر یافته را امبد کند.
    URL بخشی از doc_id است — بدون آن همه پروژه‌ها/مقاله‌ها که url یکتا دارند
    روی همان id می‌افتند و در جدول بردار یکدیگر را بازنویسی می‌کنند.
    """
    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):
            if len(pieces) == 1:
                # تیتر باید در متن امبدشده باشد تا BM25/معنایی پیدایش کند —
                # در سندهای تک‌قطعه‌ای که تیترشان در متن نیست، به ابتدای قطعه می‌چسبد
                text = piece if not doc["title"] or doc["title"] in piece[:150] else f"{doc['title']}\n{piece}"
            else:
                text = f"[{doc['title']} — بخش {i} از {len(pieces)}]\n{piece}"
            chunk = {
                "doc_id": f"{doc['type']}:{_url_key(doc['url'])}:{occ}:{i}",
                "text": text,
                "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")
        stats = _format_stat_rows(
            _fetch_rows(dsn, "SELECT value, suffix, label FROM stats ORDER BY position")
        )
        offices = _fetch_rows(dsn, "SELECT city, address FROM offices 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))
        if stats:
            lines.append("آمار: " + "؛ ".join(stats))
        if offices:
            lines.append("دفاتر: " + "؛ ".join(
                f"{_clean(o['city'])} ({_clean(o['address'])})" for o in offices
            ))
        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 ""