2d17b98e
 tangwang
sugg
|
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
|
```json
"settings": {
"number_of_shards": 1,
"number_of_replicas": 0,
"refresh_interval": "1s",
},
"mappings": {
"properties": {
"tenant_id": {"type": "keyword"},
"active_alias": {"type": "keyword"},
"active_index": {"type": "keyword"},
"last_full_build_at": {"type": "date"},
"last_incremental_build_at": {"type": "date"},
"last_incremental_watermark": {"type": "date"},
"updated_at": {"type": "date"},
}
|
|
5b8f58c0
tangwang
sugg
|
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
|
- **settings**
```json
{
"number_of_shards": 1,
"number_of_replicas": 0,
"refresh_interval": "30s",
"analysis": {
"analyzer": {
"index_ik": {
"type": "custom",
"tokenizer": "ik_max_word",
"filter": ["lowercase", "asciifolding"]
},
"query_ik": {
"type": "custom",
"tokenizer": "ik_smart",
"filter": ["lowercase", "asciifolding"]
}
}
}
}
```
- **mappings.properties**
```json
{
"tenant_id": { "type": "keyword" },
"lang": { "type": "keyword" },
"text": { "type": "keyword" }, // 显示给用户的原始文案
"text_norm": { "type": "keyword" }, // 归一化后文本：小写+空白规整
"sources": { "type": "keyword" }, // 来源集合：["title", "qanchor", "query_log"]
"title_doc_count": { "type": "integer" },
"qanchor_doc_count": { "type": "integer" },
"query_count_7d": { "type": "integer" },
"query_count_30d": { "type": "integer" },
"rank_score": { "type": "float" }, // 排序打分
"lang_confidence": { "type": "float" },
"lang_source": { "type": "keyword" }, // 语言来源：log_field / request_params / script / default
"lang_conflict": { "type": "boolean" }, // 是否存在多来源语言冲突
"status": { "type": "byte" }, // 1 = 有效
"updated_at": { "type": "date" },
"completion": { // completion suggester 用
"properties": {
"<lang>": {
"type": "completion",
"analyzer": "...", // 见下文
"search_analyzer": "..." // zh 使用 query_ik
}
}
},
"sat": { // search_as_you_type 用于 bool_prefix fallback
"properties": {
"<lang>": {
"type": "search_as_you_type",
"analyzer": "..."
}
}
}
}
```
- **语言 → analyzer 映射（ANALYZER_BY_LANG）**（只列关键）：
- `"zh"`：`index_ik` / 搜索时 `query_ik`
- `"en"`：`english`
- 其他：按 ES 内建语言 analyzer 映射
> 构建索引时，必须根据每个租户的 `index_languages`（如 `["en", "zh"]`）动态构造 `completion.<lang>` 与 `sat.<lang>` 两棵子结构。
> index_languages永远包括 en
>
---
### 二、全量构建（Full Rebuild）
对应 `SuggestionIndexBuilder.rebuild_tenant_index(...)` 和 `_build_full_candidates(...)`。
#### 1. 输入参数
关键参数：
- `tenant_id: str`
- `days: int`：查询日志回溯天数
- `recreate: bool`：仅 legacy 索引模式下生效，是否先删除旧索引
- `batch_size: int`：从商品索引扫描商品的分页大小
- `min_query_len: int`：过滤短查询
- `publish_alias: bool`：是否构建完成后切 alias（只在 versioned 模式下起作用）
- `keep_versions: int`：保留多少个最新版本索引
|
|
5b8f58c0
tangwang
sugg
|
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
|
- 对每个商品文档：
1. 确定 `product_id`：
- `spu_id` 优先，否则 `id` / `_id`
2. `title` 字段结构：**多语言对象**，如：
```json
"title": { "en": "Running Shoes", "zh": "跑步鞋" }
```
3. 对于每个 `lang in index_languages`：
- **title 处理**：
- 从 `title[lang]` 取出字符串
- 经过 `_prepare_title_for_suggest` 做长度与截断控制：
- 若整体长度 ≤ 120，直接使用
- 若过长：尝试按常见分隔符（中文逗号、英文逗号、分号、竖线、斜杠、括号、方括号等）截取第一段；仍然过长则硬截断到 120 字符，并去掉末尾多余分隔符。
- 将结果记为 `text_raw`。若为空，跳过。
- 归一化得到 `text_norm = _normalize_text(text_raw)`：
- `NFKC` 归一化
- `strip()`
- `lower()`
- 连续空白压缩为单个空格
- 若 `_looks_noise(text_norm)` 为真（长度为 0 或 > 120，或全部为非字母数字符号），跳过。
- 使用键 `(lang, text_norm)` 在候选表中查找 / 新建 `SuggestionCandidate`：
- 初始字段：
- `text = text_raw`
- `text_norm = text_norm`
- `lang = lang`
- 调用 `add_product("title", spu_id=product_id)`：
- 记录来源 `"title"`
- 将 `product_id` 加入 `title_spu_ids`
- **qanchors 处理**：
- `qanchors` 字段同样为多语言对象：
```json
"qanchors": { "en": "...", "zh": "..." }
```
- 取 `q_raw = qanchors[lang]`
- 通过 `_split_qanchors(q_raw)` 拆分为若干字符串：
- 若原值为 list → 去空白后直接返回
- 若为字符串 → 按 `[,;|/\n\t]+` 拆分，去空白，保底返回原字符串
- 对每个 `q_text`：
- `text_norm = _normalize_text(q_text)`，再用 `_looks_noise` 过滤
- 同样按 `(lang, text_norm)` 合并为 `SuggestionCandidate`，调用 `add_product("qanchor", spu_id=product_id)`。
##### 4.2 从查询日志收集用户 query（Step 2）
对应 `_iter_query_log_rows` 与 `_build_full_candidates` 的后半段。
- 时间窗口：
- `now = 当前 UTC`
- `since = now - days`
- `since_7d = now - 7天`
- 数据源：**MySQL 表 `shoplazza_search_log`**：
- 字段：
- `tenant_id`
- `query`
- `language`
- `request_params`
- `create_time`
- `deleted`（0 表示有效）
- 条件：
- `tenant_id = :tenant_id`
- `deleted = 0`
- `query IS NOT NULL AND query <> ''`
- `create_time >= :since AND create_time < :now`
- 结果按 `create_time ASC` 排序，`fetch_size` 分页流式遍历。
- 对每一行：
1. `q = row.query.strip()`
- 若 `len(q) < min_query_len` 跳过。
2. 语言解析 `_resolve_query_language(...)`：
- 输入：
- `query=q`
- `log_language = row.language`
- `request_params = row.request_params`（可能是 JSON 字符串，内部再解析 `language` 字段）
- `index_languages` & `primary_language`
- 决策顺序：
1. **日志字段 request_params.language 有值**
2. 用检测方法进行检测：`_detect_script_language(query)`：
4. 若都失败 → 回退到 `primary_language`，`lang_confidence=0.3`，`lang_source="default"`
- 若日志里的 `language` 与 `request_params` 解析出的语言不一致，则 `lang_conflict=True`。
3. 文本归一化：
- `text_norm = _normalize_text(q)`
- 若 `_looks_noise(text_norm)` → 跳过。
4. 合并到候选表：
- key = `(lang, text_norm)`
- 若不存在，则新建 `SuggestionCandidate`：
- `text = q`
- `lang = lang`
- 其他字段默认。
- 更新：
- `lang_confidence = max(c.lang_confidence, conf)`
- `lang_source`：第一次从 `"default"` 变成具体来源，之后保持已有非 default 值
- `lang_conflict |= conflict`
- 根据 `create_time` 是否在 `since_7d` 之后判断 `is_7d`
- `add_query_log(is_7d)`：
- `query_count_30d += 1`
- 最近 7 天则 `query_count_7d += 1`
#### 5. rank_score 计算与文档成型
- **打分公式**（`_compute_rank_score` 与 `_compute_rank_score_from_candidate`）：
\[
rank\_score =
1.8 \cdot \log(1 + query\_count\_{30d}) +
1.2 \cdot \log(1 + query\_count\_{7d}) +
1.0 \cdot \log(1 + qanchor\_doc\_count) +
0.6 \cdot \log(1 + title\_doc\_count)
\]
- 构造最终文档（`_candidate_to_doc`）：
- `_id = f"{tenant_id}|{lang}|{text_norm}"`
- `completion` 字段：
```json
"completion": {
"<lang>": {
"input": [text], // 原始展示文案
"weight": int(max(rank_score, 1.0) * 100)
}
}
```
- `sat` 字段：
```json
"sat": {
"<lang>": text
}
```
- 其余字段直接来自 `SuggestionCandidate` 聚合结果（见 mapping 部分）。
- 将所有 doc 批量写入目标索引，然后执行 `refresh`。
#### 6. alias 发布与旧版本清理
- 若 `publish_alias = true` 且 `use_versioned_index = true`：
1. 通过 alias 名（`get_suggestion_alias_name`）查出当前挂载的索引列表 `current_indices`
2. 构造 `update_aliases` 操作：
- 为所有旧索引 `remove` alias
- 为新索引 `add` alias
3. 调用 `_cleanup_old_versions`：
- 通过通配符 `get_suggestion_versioned_index_pattern(tenant_id)` 列出所有版本索引
- 按名称排序，保留最新 `keep_versions` 个
- 删除其余版本索引，但保护当前新索引
4. 更新 meta 索引：
- `active_alias`
- `active_index`
#### 7. meta 索引更新
- 无论是否发布 alias，都会更新 meta：
```json
{
"tenant_id": "<tenant_id>",
"last_full_build_at": <now_utc_iso>,
"last_incremental_watermark": <now_utc_iso>,
"active_index": <index_name>, // 若使用 versioned + alias
"active_alias": "<alias_name>" // ditto
}
```
- 更新行为是「读旧值 → merge patch → index 文档，refresh wait_for」。
---
### 三、增量构建（Incremental Update）
对应 `incremental_update_tenant_index(...)` 与 `_build_incremental_deltas(...)`, `_build_incremental_actions(...)`。
#### 1. 输入参数
- `tenant_id: str`
- `min_query_len: int`：过滤短 query
- `fallback_days: int`：如果找不到上次水位线，从最近 N 天回补
- `overlap_minutes: int`：窗口向前重叠几分钟，防止日志延迟
- `bootstrap_if_missing: bool`：没有任何 suggestion 索引时是否先全量构建
- `bootstrap_days: int`：bootstrap 全量构建的日志回溯天数
- `batch_size: int`：仅在需要 bootstrap 全量构建时使用
#### 2. 目标索引解析
- `_resolve_incremental_target_index(tenant_id)`：
1. 首选：alias `get_suggestion_alias_name(tenant_id)` 所指向的索引（一般只会有一个）
2. 若 alias 不存在：回退到 legacy 索引名 `get_suggestion_legacy_index_name(tenant_id)`，且索引存在
3. 若都不存在 → 返回 `None`
- 若 `target_index is None`：
- 若 `bootstrap_if_missing = false` → 抛错，提示先跑全量或打开 bootstrap。
- 否则执行一次全量构建 `rebuild_tenant_index(...)`：
- 参数：`days=bootstrap_days`, `batch_size=batch_size`, `min_query_len`, `publish_alias=true`, `use_versioned_index=true`
- 返回 `{"mode": "incremental", "bootstrapped": true, "bootstrap_result": ...}`
#### 3. 增量时间窗口计算
- 从 meta 索引 `_get_meta(tenant_id)` 中读取：
- `last_incremental_watermark` 或 `last_full_build_at`
- 若是 ISO 字符串（可能带 `Z` 后缀）→ 解析为 UTC `since`。
- 若解析失败或不存在 → `since = now - fallback_days`
- 然后：
- `since = since - overlap_minutes`
- 但不得早于 `now - fallback_days`（防止 overlap 过大）
- 最终时间窗口：
```text
[since, now)
```
> 时间计算与 UTC 处理逻辑保持一致（字符串格式可兼容 ISO 8601）。
#### 4. 构建 QueryDelta（按 query + 语言累积增量）
- `_build_incremental_deltas(...)` 同样从表 `shoplazza_search_log` 读取，只是时间窗口换成增量窗口 `since` ～ `now`。
- 对每一条日志：
1. `q = row.query.strip()`，若 `len(q) < min_query_len` → 跳过。
2. 仍然通过 `_resolve_query_language` 计算 `(lang, conf, source, conflict)`。
3. `text_norm = _normalize_text(q)`，`_looks_noise` 过滤。
4. 键 `(lang, text_norm)`，合并为 `QueryDelta`：
- 初始：
- `tenant_id`
- `lang`
- `text = q`
- `text_norm`
- `lang_confidence = conf`
- `lang_source = source`
- `lang_conflict = conflict`
- 更新逻辑：
- `delta_30d += 1`
- 若 `create_time >= now - 7d` → `delta_7d += 1`
- 若新的 `conf > lang_confidence` → 更新 `lang_confidence` & `lang_source`
- `lang_conflict |= conflict`
#### 5. 构建增量更新动作（Bulk Update with Scripted Upsert）
- 将每个 `QueryDelta` 转换为 upsert 文档（`_delta_to_upsert_doc`）：
- 使用与全量同一套打分公式，但只基于增量的 `delta_7d / delta_30d`。
- 设置：
- `sources = ["query_log"]`
- `title_doc_count = 0`
- `qanchor_doc_count = 0`
- `completion.<lang>.input = [text]`
- `completion.<lang>.weight = int(max(rank_score, 1.0) * 100)`
- `sat.<lang> = text`
- `status = 1`
- `updated_at = now_iso`（本次执行时间）
- 为每个 delta 构造一个 bulk **update** 动作（`_build_incremental_actions`）：
- `_op_type = "update"`
- `_index = target_index`
- `_id = f"{tenant_id}|{lang}|{text_norm}"`
- `scripted_upsert = true`
- `script.lang = "painless"`
- `script.source` 为一段 Painless 脚本（见 `SuggestionIndexBuilder._build_incremental_update_script`），其要点：
- 若文档不存在 → 直接 `ctx._source = params.upsert`
- 若存在：
- 初始化空字段
- `query_count_30d += params.delta_30d`
- `query_count_7d += params.delta_7d`
- 将 `"query_log"` 加入 `sources`
- `lang_conflict` 与 `params.lang_conflict` 取或
- 若 `params.lang_confidence > ctx._source.lang_confidence` 则更新 `lang_confidence` 和 `lang_source`
- 基于更新后的 `query_count_7d/30d` + `qanchor_doc_count` + `title_doc_count` 重新计算 `rank_score`
- `status = 1`
- `updated_at = params.now_iso`
- 同步更新 `text / lang / text_norm`
- 更新 `completion[lang]` 的 `input` 与 `weight`
- 更新 `sat[lang] = text`
- `script.params` 包含：
- `delta_30d`, `delta_7d`, `lang_confidence`, `lang_source`, `lang_conflict`
- `now_iso`, `lang`, `text`, `text_norm`
- `completion_input`, `completion_weight`
- `upsert`（完整 upsert 文档）
- 整个增量批次调 `bulk` / `bulk_actions`，执行完成后 `refresh(target_index)`。
#### 6. 更新 meta 索引
- 更新字段：
```json
{
"last_incremental_build_at": now_iso,
"last_incremental_watermark": now_iso,
"active_index": target_index,
"active_alias": "<alias_name>"
}
```
> 注意：即便外部不依赖 `active_alias`，最好仍然保持与现有 Python 实现一致，方便混合环境切换。
---
### 四、语言解析与噪音过滤
#### 1. 文本归一化 `_normalize_text(value: str)`
- `unicodedata.normalize("NFKC", value or "")`
- `strip()`
- 转小写
- 用正则 `\s+` 将多空白压缩成单个空格
#### 2. 噪音检测 `_looks_noise(text_value: str)`
- 任何一条为真即视为噪音，候选会被忽略：
- 空字符串
- 长度 > 120
- `re.fullmatch(r"[\W_]+", text_value)`：全是符号/下划线，没有文字
#### 3. 语言标准化 `_normalize_lang(lang)`
- 输入 lang 为空 → None
- 转小写，`-` 替换成 `_`
- 若是 `"zh_tw"` / `"pt_br"` → 保留全量
- 其他 → 取 `_` 前缀（例如 `"en_US"` → `"en"`）
|