| """
|
| qdrant_alias.py — ตั้ง/ดู alias ของ collection Modutech (28 ต.ค.)
|
|
|
| บอทเรียก collection ข้อความของ Modutech ผ่าน alias modutech_live (brands/modutech/brand.yaml)
|
| ตั้ง alias ให้ชี้ collection จริง แล้วสลับเวอร์ชัน / ย้อนกลับได้ด้วยคำสั่งเดียว ไม่ต้องแก้โค้ด
|
|
|
| & $py qdrant_alias.py show
|
| & $py qdrant_alias.py set --to modutech_belt_chunks # ครั้งแรก: ชี้ collection เดิม (ไม่คัดลอก ไม่สร้างใหม่)
|
| & $py qdrant_alias.py set --to modutech_belt_chunks_v31 # วันหน้า: สลับไปเวอร์ชันใหม่ที่ build แยกไว้แล้ว
|
| & $py qdrant_alias.py set --to modutech_belt_chunks # ย้อนกลับ
|
|
|
| กติกา
|
| · ปิดบอทก่อนรัน (Qdrant แบบไฟล์ล็อกโฟลเดอร์)
|
| · แตะเฉพาะ alias — ไม่สร้าง ไม่ลบ ไม่เขียนจุดใน collection ไหนเลย
|
| · ห้ามใช้กับ ./qdrant_db ของ Movex และห้ามชื่อที่ขึ้นต้นด้วย movex (สคริปต์ปฏิเสธเอง)
|
| · สำรอง meta.json เป็น meta.json.bak-<เวลา> ก่อนแก้ทุกครั้ง แล้วเปิดใหม่ตรวจผลหลังแก้
|
| · collection ปลายทางต้องมีจุด และชื่อ/ขนาดเวกเตอร์ต้องตรงกับตัวที่ alias ชี้อยู่ (ไม่ตรง = ไม่สลับ)
|
|
|
| หลังตั้งเสร็จ meta.json ในโฟลเดอร์เปลี่ยน (ไฟล์อยู่ใน Git LFS) → commit แล้วอัปโหลดโฟลเดอร์ขึ้น Space
|
| """
|
| from __future__ import annotations
|
|
|
| import argparse
|
| import os
|
| import shutil
|
| import sys
|
| import time
|
|
|
| BASE = os.path.dirname(os.path.abspath(__file__))
|
| DEFAULT_PATH = os.path.join(BASE, "qdrant_modutech_db")
|
| DEFAULT_ALIAS = "modutech_live"
|
|
|
|
|
| def _refuse_movex(path: str, *names: str) -> str | None:
|
| real = os.path.realpath(path)
|
| if os.path.basename(real.rstrip("\\/")) == "qdrant_db" or real == os.path.realpath(os.path.join(BASE, "qdrant_db")):
|
| return "นี่คือฐานข้อมูลของ Movex — ห้ามแตะ (กฎโปรเจกต์)"
|
| for n in names:
|
| if n and n.lower().startswith("movex"):
|
| return f"ชื่อ {n} เป็นของ Movex — ห้ามแตะ"
|
| return None
|
|
|
|
|
| def _vectors(c, name: str) -> dict:
|
| """ชื่อเวกเตอร์ → (ขนาด, distance) เทียบโครงสร้างก่อนสลับ"""
|
| vc = c.get_collection(name).config.params.vectors
|
| if isinstance(vc, dict):
|
| return {k: (v.size, str(v.distance)) for k, v in vc.items()}
|
| return {"": (vc.size, str(vc.distance))}
|
|
|
|
|
| def _open(path: str):
|
| from qdrant_client import QdrantClient
|
| if not os.path.isfile(os.path.join(path, "meta.json")):
|
| raise SystemExit(f"❌ ไม่พบ {os.path.join(path, 'meta.json')}")
|
| try:
|
| return QdrantClient(path=path)
|
| except Exception as e:
|
| raise SystemExit(f"❌ เปิด {path} ไม่ได้ ({type(e).__name__}: {e}) — ปิดบอทก่อนแล้วรันใหม่")
|
|
|
|
|
| def show(path: str) -> dict:
|
| c = _open(path)
|
| try:
|
| aliases = {a.alias_name: a.collection_name for a in c.get_aliases().aliases}
|
| print(f"โฟลเดอร์: {path}")
|
| for col in sorted(x.name for x in c.get_collections().collections):
|
| n = c.get_collection(col).points_count
|
| via = [a for a, t in aliases.items() if t == col]
|
| print(f" collection {col}: {n} จุด · เวกเตอร์ {_vectors(c, col)}"
|
| + (f" ← alias {', '.join(via)}" if via else ""))
|
| if not aliases:
|
| print(" (ยังไม่มี alias — บอทใช้ชื่อสำรองใน brand.yaml)")
|
| return aliases
|
| finally:
|
| c.close()
|
|
|
|
|
| def set_alias(path: str, alias: str, to: str, dry: bool = False) -> int:
|
| from qdrant_client import models as M
|
| why = _refuse_movex(path, alias, to)
|
| if why:
|
| print(f"❌ {why}")
|
| return 2
|
| c = _open(path)
|
| try:
|
| names = {x.name for x in c.get_collections().collections}
|
| aliases = {a.alias_name: a.collection_name for a in c.get_aliases().aliases}
|
| if to not in names:
|
| print(f"❌ ไม่มี collection {to} (มี: {', '.join(sorted(names))})")
|
| return 2
|
| if alias in names:
|
| print(f"❌ {alias} เป็นชื่อ collection อยู่แล้ว ตั้งเป็น alias ซ้ำไม่ได้")
|
| return 2
|
| n = c.get_collection(to).points_count or 0
|
| if n <= 0:
|
| print(f"❌ {to} ไม่มีจุด — ไม่สลับ")
|
| return 2
|
| old = aliases.get(alias)
|
| if old == to:
|
| print(f"✅ {alias} ชี้ {to} อยู่แล้ว ไม่ต้องทำอะไร")
|
| return 0
|
| if old:
|
| a, b = _vectors(c, old), _vectors(c, to)
|
| if a != b:
|
| print(f"❌ โครงสร้างเวกเตอร์ไม่ตรงกัน {old} {a} ≠ {to} {b} — retriever จะค้นไม่ได้ ไม่สลับ")
|
| return 2
|
| print(f"{alias}: {old or '(ยังไม่มี)'} → {to} ({n} จุด)")
|
| if dry:
|
| print("(--dry-run ไม่ได้เขียน)")
|
| return 0
|
| finally:
|
| c.close()
|
| del c
|
|
|
| meta = os.path.join(path, "meta.json")
|
| bak = f"{meta}.bak-{time.strftime('%Y%m%d-%H%M%S')}"
|
| shutil.copy2(meta, bak)
|
| print(f"สำรอง → {os.path.basename(bak)}")
|
|
|
| c = _open(path)
|
| try:
|
| ops = []
|
| if old:
|
| ops.append(M.DeleteAliasOperation(delete_alias=M.DeleteAlias(alias_name=alias)))
|
| ops.append(M.CreateAliasOperation(create_alias=M.CreateAlias(collection_name=to, alias_name=alias)))
|
| c.update_collection_aliases(change_aliases_operations=ops)
|
| finally:
|
| c.close()
|
| del c
|
|
|
| c = _open(path)
|
| try:
|
| got = {a.alias_name: a.collection_name for a in c.get_aliases().aliases}.get(alias)
|
| cnt = c.get_collection(alias).points_count if got else 0
|
| finally:
|
| c.close()
|
| del c
|
| if got != to or cnt != n:
|
| print(f"❌ ตรวจหลังเขียนไม่ผ่าน (alias → {got}, {cnt} จุด) — คืนไฟล์จาก {os.path.basename(bak)}")
|
| shutil.copy2(bak, meta)
|
| return 1
|
| print(f"✅ {alias} → {to} · ค้นผ่าน alias ได้ {cnt} จุด · commit qdrant_modutech_db/meta.json แล้วอัปโหลดขึ้น Space")
|
| return 0
|
|
|
|
|
| def main() -> int:
|
| ap = argparse.ArgumentParser()
|
| ap.add_argument("cmd", choices=["show", "set"])
|
| ap.add_argument("--path", default=DEFAULT_PATH)
|
| ap.add_argument("--alias", default=DEFAULT_ALIAS)
|
| ap.add_argument("--to", help="collection ปลายทาง (คำสั่ง set)")
|
| ap.add_argument("--dry-run", action="store_true")
|
| a = ap.parse_args()
|
| why = _refuse_movex(a.path)
|
| if why:
|
| print(f"❌ {why}")
|
| return 2
|
| if a.cmd == "show":
|
| show(a.path)
|
| return 0
|
| if not a.to:
|
| print("❌ ต้องระบุ --to <collection>")
|
| return 2
|
| return set_alias(a.path, a.alias, a.to, a.dry_run)
|
|
|
|
|
| if __name__ == "__main__":
|
| sys.exit(main())
|
|
|
