aramesh-site deploy
d55c8cc1e5
- content.py: ستون desc در SQL بدون کوتیشن بود و PostgreSQL آن را کلمهٔ کلیدی می‌گیرد؛ نتیجه: «Initial index sync failed: syntax error at or near "desc"» در استارت و غیرقابل استفاده ماندن چت تا reindex. حالا "desc" کوتیشن شده. - Dockerfile: با این‌که سرور GPU ندارد، pip نسخهٔ CUDA تورچ را از PyPI می‌گرفت (ایمیج ۹.۷GB با کتابخانه‌های nvidia). اول torch از ایندکس CPU نصب می‌شود.
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 ""
|