316 lines
13 KiB
Python
316 lines
13 KiB
Python
"""خواندن محتوای سایت از 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 ""
|