给微信加一套多模态搜索：Elasticsearch 9.5 `semantic` 字段上手

~/Library/Containers/com.tencent.xinWeChat/Data/Documents/
  xwechat_files/wxid_xxxxx/
    ├── temp/           # 截图、应用缓存图片
    ├── msg/video/      # 视频，按月份分目录
    ├── msg/file/       # 音频、文档
    └── msg/attach/     # .dat 加密容器

微信本地文件 ──压缩/裁剪──▶ PUT _doc ──▶ semantic 字段自动向量化
                                              │
图片 ─┐                                       ▼
视频 ─┼──▶ 同一个模型 ──▶ 同一个 1024 维空间 ◀── 查询也编码进这里
音频 ─┘                                       │
                                              ▼
                                        最近邻 = 搜索结果

GET _inference/embedding/.jina-embeddings-v5-omni-small

PUT wechat-multimodal
{
  "mappings": {
    "properties": {
      "embedding": {
        "type": "semantic",
        "inference_id": ".jina-embeddings-v5-omni-small"
      },
      "file_path":  { "type": "keyword" },
      "file_name":  { "type": "keyword" },
      "media_type": { "type": "keyword" },
      "indexed_at": { "type": "date" }
    }
  }
}

"model_settings": {
  "task_type": "embedding",
  "dimensions": 1024,
  "similarity": "cosine",
  "element_type": "float"
}

# 图片
{"embedding": {"type": "image", "value": f"data:image/jpeg;base64,{b64}"}}
# 视频
{"embedding": {"type": "video", "value": f"data:video/mp4;base64,{b64}"}}
# 音频
{"embedding": {"type": "audio", "value": f"data:audio/mpeg;base64,{b64}"}}

GET wechat-multimodal/_search
{
  "query": { "match": { "embedding": "夜景 灯光" } },
  "_source": { "excludes": ["embedding"] }
}

GET wechat-multimodal/_search
{
  "query": {
    "knn": {
      "field": "embedding",
      "query_vector_builder": {
        "embedding": {
          "input": { "type": "image",
                     "value": "data:image/jpeg;base64,..." }
        }
      },
      "num_candidates": 100,
      "k": 20
    }
  },
  "_source": { "excludes": ["embedding"] }
}

#!/usr/bin/env python3
"""微信本地媒体 → ES 9.5 semantic 字段。依赖：ffmpeg、Pillow"""
import base64, json, subprocess, tempfile, urllib.request, urllib.error
from datetime import datetime, timezone
from pathlib import Path

ES     = "https://your-cluster.es.cloud.es.io"
APIKEY = "your-api-key"
INDEX  = "wechat-multimodal"
IID    = ".jina-embeddings-v5-omni-small"

MAX_BYTES = 900 * 1024   # 上限 1MB，留余量
MAX_SECS  = 110          # 上限 120s，留余量 —— 见下文"坑 2"

WECHAT = Path.home() / ("Library/Containers/com.tencent.xinWeChat/Data/Documents"
                        "/xwechat_files/wxid_xxxxx")


def es(method, path, body=None, timeout=120):
    data = json.dumps(body).encode() if body is not None else None
    req = urllib.request.Request(
        f"{ES}{path}", data=data, method=method,
        headers={"Authorization": f"ApiKey {APIKEY}",
                 "Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=timeout) as r:
        return json.load(r)


def duration(p):
    out = subprocess.run(["ffprobe", "-v", "error", "-show_entries",
                          "format=duration", "-of", "csv=p=0", str(p)],
                         capture_output=True, text=True).stdout.strip()
    return float(out or 0)


def shrink_image(p):
    from PIL import Image
    import io
    img = Image.open(p).convert("RGB")
    q, scale = 85, 1.0
    while True:
        buf = io.BytesIO()
        im = img.resize((int(img.width*scale), int(img.height*scale)),
                        Image.Resampling.LANCZOS) if scale < 1 else img
        im.save(buf, "JPEG", quality=q)
        if buf.tell() <= MAX_BYTES:
            return buf.getvalue()
        if q > 40: q -= 15
        else: scale *= 0.75


def shrink_av(p, kind):
    """视频/音频：先裁时长，再压体积。时长必须回读校验。"""
    suffix = ".mp4" if kind == "video" else ".mp3"
    ladder = ([(320, 40), (240, 45), (192, 48)] if kind == "video"
              else [(0, 32), (0, 24)])
    for a, b in ladder:
        out = tempfile.mktemp(suffix=suffix)
        cmd = ["ffmpeg", "-y", "-i", str(p), "-t", str(MAX_SECS)]
        cmd += (["-vf", f"scale='min({a},iw)':-2", "-crf", str(b),
                 "-preset", "fast", "-an"] if kind == "video"
                else ["-ar", "16000", "-ac", "1", "-b:a", f"{b}k"])
        subprocess.run(cmd + [out], capture_output=True, check=True)
        data, secs = Path(out).read_bytes(), duration(out)
        Path(out).unlink(missing_ok=True)
        if len(data) <= MAX_BYTES and secs < 120:
            return data
    return None


MIME = {"image": "image/jpeg", "video": "video/mp4", "audio": "audio/mpeg"}


def index_file(p: Path, kind: str):
    raw = p.read_bytes()
    if kind == "image":
        data = raw if len(raw) <= MAX_BYTES else shrink_image(p)
    else:
        data = (raw if len(raw) <= MAX_BYTES and duration(p) < 120
                else shrink_av(p, kind))
    if data is None:
        return f"跳过 {p.name}（压不到限制以内）"

    b64 = base64.b64encode(data).decode()
    doc = {
        "embedding":  {"type": kind,
                       "value": f"data:{MIME[kind]};base64,{b64}"},
        "file_path":  str(p),
        "file_name":  p.name,
        "media_type": kind,
        "indexed_at": datetime.now(timezone.utc).isoformat(),
    }
    try:
        es("PUT", f"/{INDEX}/_doc/{p.stem[:64]}", doc)
        return f"OK  {p.name}"
    except urllib.error.HTTPError as e:
        # 关键：必须读 body，否则只能看到 "HTTP Error 400"
        try:
            reason = json.loads(e.read())["error"]["root_cause"][0]["reason"]
        except Exception:
            reason = str(e)
        return f"ERR {p.name}: {reason}"


def main():
    jobs = []
    for f in sorted((WECHAT / "temp").rglob("*")):
        if f.suffix.lower() in (".jpg", ".jpeg", ".png"):
            jobs.append((f, "image"))
    for f in sorted((WECHAT / "msg/video").rglob("*.mp4")):
        jobs.append((f, "video"))
    for f in sorted((WECHAT / "msg/file").rglob("*")):
        if f.suffix.lower() in (".mp3", ".m4a", ".wav", ".aac"):
            jobs.append((f, "audio"))

    print(f"共 {len(jobs)} 个文件")
    for i, (f, kind) in enumerate(jobs, 1):
        print(f"[{i}/{len(jobs)}] {index_file(f, kind)}")


if __name__ == "__main__":
    main()

def search_text(q, size=20):
    return es("POST", f"/{INDEX}/_search", {
        "query": {"match": {"embedding": q}},
        "_source": {"excludes": ["embedding"]},
        "size": size})


def search_media(kind, data_uri, size=20):
    return es("POST", f"/{INDEX}/_search", {
        "query": {"knn": {
            "field": "embedding",
            "query_vector_builder": {
                "embedding": {"input": {"type": kind, "value": data_uri}}},
            "num_candidates": size * 5, "k": size}},
        "_source": {"excludes": ["embedding"]},
        "size": size})

Video duration 300.0s exceeds 120s cap. Trim the clip and retry.

POST _inference/embedding/.jina-embeddings-v5-omni-small
{ "input": [{"type": "video", "value": "data:video/mp4;base64,..."}] }

{ "error": { "reason": "unknown field [type]" }, "status": 400 }

GET wechat-multimodal/_search
{ "query": {"match_all": {}}, "_source": {"exclude_vectors": false} }