Improve online mining record conversion

skills/online-mining-v2/README.md

@@ -16,23 +16,48 @@
 | `scripts/elk_profile_index.py` | 采样索引 schema 和关键字段，帮助 Agent 判断字段可用性。 |
 | `scripts/elk_search_cases.py` | 按 query/domain/prompt/model output/candidate domain 等条件搜索候选 case。 |
 | `scripts/elk_fetch_by_request_ids.py` | 批量按 request id 拉取 main/pre_processing 原始摘要。 |
-| `scripts/elk_join_request_logs.py` | 把 `arch-flat-nlp-log-f-*` 和 `pre-processing-info*` 按 request id 合并。 |
+| `scripts/elk_join_request_logs.py` | 把 `arch-flat-nlp-log-f-*` 和 `pre-processing*` 按 request id 合并。 |
 | `scripts/build_dataset_draft.py` | 把 review 后的线上 case 转成 product-data dataset draft text。 |
+| `scripts/build_online_records.py` | 把 rid/候选 case 直接转成 canonical records，并可导出 records.csv、training.jsonl、eval_planning.csv。 |
 
 ## 数据源
 
 - `main`：`arch-flat-nlp-log-f-*`
-- `pre_processing`：`pre-processing-info*`
+- `pre_processing`：`pre-processing*`
 
 字段说明见 `knowledge/数据源字段说明.md`。
 
 ## 后处理
 
-后处理复用 `skills/product-data/scripts/`：
+优先使用 `scripts/build_online_records.py` 一步完成线上候选到标准数据：
+
+```json
+{
+  "request_ids": ["<rid>"],
+  "date": "YYYYMMDD",
+  "dataset_label": "线上挖掘样本",
+  "target": "Agent(tag=\"xxx\")",
+  "complex": false,
+  "output_dir": "output",
+  "export_records": true,
+  "export_training": false,
+  "export_eval": false
+}
+```
+
+它会补齐：
+
+- `request_id`：主表 `request_id` / 前处理表 `requestId`
+- `timestamp`：优先主表 `timestamp`
+- `query`：优先前处理 query，缺失时用主表 query
+- `prev_session`：主表同 `session_id` 且早于当前 timestamp 的最多 10 轮
+- `target`：优先用户输入，否则用前处理 `code/planningOriginalResult`，再用主表仲裁信息兜底
+- `complex`：优先用户输入，默认 false
+
+如果需要拆开调试，后处理仍可复用 `skills/product-data/scripts/`：
 
 1. `normalize_dataset_draft.py`
 2. `validate_dataset_records.py`
 3. `export_dataset_records.py`
 4. 可选 `export_training_jsonl.py`
 5. 可选 `export_planning_eval_csv.py`
-

skills/online-mining-v2/SKILL.md

@@ -28,11 +28,12 @@ skills/online-mining-v2/
     elk_fetch_by_request_ids.py
     elk_join_request_logs.py
     build_dataset_draft.py
+    build_online_records.py
 ```
 
 默认数据源：
 
-- `pre-processing-info*`：前处理日志，适合拿模型 prompt、模型输出、候选 domain、`excellent_domains_result`。
+- `pre-processing*`：前处理日志，适合拿模型 prompt、模型输出、候选 domain、`excellent_domains_result`。
 - `arch-flat-nlp-log-f-*`：主 NLP 日志，适合拿最终 `query`、`domain`、`func`、`session_id`、`device_id`、`device`、`tts/text/to_speak`。
 
 后处理全部复用 `product-data`：
@@ -128,12 +129,51 @@ skills/online-mining-v2/
 }
 ```
 
-### 5. 给用户 review
+### 5. 单条 rid 或已确认候选直接转元数据
+
+如果用户给了明确 rid，或已经确认一批线上候选要直接作为样本，优先使用
+`build_online_records.py`。这个脚本会完成：
+
+- 主表 `arch-flat-nlp-log-f-*` 和前处理表 `pre-processing*` 双表拉取。
+- 用主表 `session_id + timestamp` 补齐当前请求前最多 10 轮 session。
+- 优先使用用户指定 `target`；没有指定时，按 `pre_processing.code` -> `planningOriginalResult` -> 主表 `llm_agent_info.agentType` -> 主表 `domain` 推断标签。
+- 生成 product-data canonical records。
+- 默认导出 `output/records.jsonl` 和 `output/records.csv`，可选导出 `training.jsonl` 和 `eval_planning.csv`。
+
+示例：
+
+```json
+{
+  "script_path": "skills/online-mining-v2/scripts/build_online_records.py",
+  "stdin": {
+    "request_ids": ["6656271eedfe4b478ac716448a3ad310"],
+    "date": "20260514",
+    "dataset_label": "线上挖掘样本",
+    "target": "Agent(tag=\"地图导航\")",
+    "complex": false,
+    "output_dir": "output",
+    "export_records": true,
+    "export_training": false,
+    "export_eval": false
+  },
+  "timeout_seconds": 300,
+  "max_output_chars": 30000
+}
+```
+
+注意：
+
+- 老 rid 超出近 48 小时时必须让用户补日期，传 `date=YYYYMMDD`。
+- 如果用户没有指定 `target`，脚本可以从线上模型输出推断，但最终仍建议展示 `target_source` 给用户确认。
+- 如果脚本返回 `case has no query`，说明该 rid 在当前日期窗口没有命中主表/前处理表，不要继续伪造元数据。
+
+### 6. 给用户 review
 
 展示候选样本时只展示必要字段：
 
 - request_id
 - query
+- target / target_source（如果已经转换）
 - main.domain / main.func
 - pre.planning_result
 - pre.hit_rules
@@ -143,9 +183,10 @@ skills/online-mining-v2/
 
 不要一次贴大量 prompt。需要看 prompt 时只展示截断摘要，或保存到会话 output 目录。
 
-### 6. 用户确认后转成 product-data draft
+### 7. 用户确认后转成 product-data draft
 
-线上候选直接作为样本时，先生成 dataset draft text：
+一般情况下不要再手动接 normalize/export。只有当用户明确要看 draft text，
+或需要和 `product-data` 的人工生成流程混合时，才先生成 dataset draft text：
 
 ```json
 {
@@ -162,7 +203,7 @@ skills/online-mining-v2/
 }
 ```
 
-然后必须调用 product-data：
+然后再调用 product-data：
 
 ```json
 {
@@ -179,7 +220,9 @@ skills/online-mining-v2/
 }
 ```
 
-再执行校验和导出：
+更推荐直接使用第 5 步的 `build_online_records.py`，减少模型手动接线出错。
+
+如果使用 draft 分支，再执行校验和导出：
 
 - `validate_dataset_records.py`
 - `export_dataset_records.py`，输出固定 `output/records.jsonl`
@@ -201,4 +244,3 @@ skills/online-mining-v2/
 - canonical records：通过 product-data 导出到 `output/records.jsonl`
 
 不要写项目根目录、`tasks/`、`src/` 或 `skills/`。
-

skills/online-mining-v2/knowledge/工作流.md

@@ -16,11 +16,36 @@
 3. 用 `elk_search_cases.py` 拉候选。
 4. 用 `elk_join_request_logs.py` 补全 main/pre_processing 字段。
 5. 展示小批候选给用户 review。
-6. 用户确认后，用 `build_dataset_draft.py` 转 dataset draft。
+6. 用户确认后，优先用 `build_online_records.py` 直接生成 canonical records 和流通表格。
-7. 用 `product-data` 脚本 normalize、validate、export。
+7. 只有需要人工查看 draft 或混合生成数据时，才用 `build_dataset_draft.py` 转 dataset draft，再接 product-data。
 
 这个分支不生成新 query。
 
+### rid 快速转样本流程
+
+适用表达：
+
+- “把这个 rid 的数据拉下来，作为测试数据”
+- “这几个 request_id 转成我们的元数据”
+- “线上命中的这批样本直接导出”
+
+流程：
+
+1. 整理 `request_ids`、`dataset_label`、`target`、`complex`、日期。
+2. 如果没有日期，先按近 48 小时查；查不到要问用户补日期。
+3. 调用 `build_online_records.py`。
+4. 检查返回的 `summaries`：
+   - `query`
+   - `target`
+   - `target_source`
+   - `prev_session_count`
+   - `main_domain`
+   - `planning_result`
+5. 告诉用户输出路径：
+   - `output/records.jsonl`
+   - `output/records.csv`
+   - 用户要求训练/评测时，再导出 `training.jsonl` / `eval_planning.csv`
+
 ## 分支 B：基于线上问题补充生成数据
 
 适用表达：
@@ -51,4 +76,3 @@ tts:
 ```
 
 不要默认展示完整 prompt。prompt 很长时只展示命中片段。
-

skills/online-mining-v2/knowledge/数据源字段说明.md

@@ -28,7 +28,7 @@
 - `session_id`、`device_id`、设备类型。
 - `intention.intent_arbitrator_info.llm_agent_info.agentType`。
 
-## pre_processing：`pre-processing-info*`
+## pre_processing：`pre-processing*`
 
 适合拿前处理和规划模型判断过程。
 
@@ -65,6 +65,13 @@
 main.request_id == pre_processing.requestId
 ```
 
+精确查询时优先使用 keyword 字段：
+
+```text
+main.request_id.keyword
+pre_processing.requestId.keyword
+```
+
 join 后优先使用：
 
 - query：优先 `pre_processing` 的 query，缺失时用 `main.query`。
@@ -72,3 +79,33 @@ join 后优先使用：
 - tts：优先 `main.to_speak`，再 `main.text` 或 `display_text`。
 - label 参考：模型输出用 `pre_processing.planning_result`，线上最终结果用 `main.domain` / `main.func`。
 
+## 转 canonical record 的字段映射
+
+`build_online_records.py` 固定执行以下映射，模型不要手写转换逻辑：
+
+| canonical 字段 | 来源 |
+| --- | --- |
+| `source.type` | 固定 `online` |
+| `source.request_id` | 主表 `request_id` 或前处理表 `requestId` |
+| `source.timestamp` | 优先主表 `timestamp`，缺失时用前处理表 `timestamp` |
+| `turn.query` | 优先前处理 `responseBoby.nodes.0.core.query.query`，缺失时用主表 `query` |
+| `turn.timestamp` | 同 `source.timestamp` |
+| `prev_session` | 主表同 `session_id` 且 timestamp 早于当前请求的最多 10 轮，按时间升序排列 |
+| `prev_session[].query` | 主表历史轮 `query` |
+| `prev_session[].tts` | 主表历史轮 `to_speak/text/display_text`，没有时保留空字符串 |
+| `prev_session[].timestamp` | 主表历史轮 `timestamp` |
+| `context` | 先置 `{}`，后续由专用工具补充 |
+| `label.dataset_label` | 用户输入的批次标签 |
+| `label.target` | 用户输入 `target`；否则依次用前处理 `code`、`planningOriginalResult`、主表 `llm_agent_info.agentType`、主表 `domain` 推断 |
+| `dimensions.complex` | 用户输入，默认 false |
+| `meta` | 保存 `session_id/device_id/device/main_domain/main_func/planning_result/planning_code/hit_rules/candidate_domains/target_source` |
+
+## 按 rid 转数据的固定流程
+
+当用户说“把这个 rid 的数据拉下来作为测试数据”时：
+
+1. 如果用户没给日期，先用近 48 小时查；查不到就问用户日期，不要继续生成空数据。
+2. 调用 `build_online_records.py`，传 `request_ids`、`dataset_label`、必要时传 `target/complex/date`。
+3. 检查返回的 `summaries[].target_source` 和 `prev_session_count`。
+4. 展示 query、target、target_source、prev_session_count 给用户确认。
+5. 产物默认在当前会话 `output/records.jsonl` 和 `output/records.csv`。

skills/online-mining-v2/scripts/build_dataset_draft.py

@@ -35,6 +35,69 @@ def case_value(case: dict, key: str):
     return case.get(key) or pre.get(key) or main.get(key)
 
 
+def build_dataset_draft_text(
+    *,
+    dataset_label: str,
+    target: str,
+    complex_value: bool,
+    cases: list[dict],
+) -> str:
+    """把 review 后的线上 case 转成 product-data dataset draft text。
+
+    这个 draft 是中间格式，不是最终数据。后续必须再经过
+    product-data 的 normalize/validate/export。
+    """
+
+    lines = [f"# dataset_label: {dataset_label}", ""]
+    for index, case in enumerate(cases, start=1):
+        query = str(case_value(case, "query") or "").strip()
+        if not query:
+            continue
+        request_id = str(case_value(case, "request_id") or "").strip()
+        timestamp = case_value(case, "timestamp")
+        tts = str(case_value(case, "tts") or "").strip()
+        planning_result = str(case_value(case, "planning_result") or "").strip()
+        case_target = str(case_value(case, "target") or target).strip()
+        case_complex = case_value(case, "complex")
+        if isinstance(case_complex, str):
+            case_complex_text = case_complex.strip().lower()
+            item_complex = case_complex_text in {"true", "1", "yes", "y", "是", "复杂", "complex"}
+        elif isinstance(case_complex, bool):
+            item_complex = case_complex
+        else:
+            item_complex = complex_value
+        notes = []
+        if planning_result:
+            notes.append(f"线上模型输出: {planning_result}")
+        main_domain = case_value(case, "domain") or case.get("main_domain")
+        if main_domain:
+            notes.append(f"线上 domain: {main_domain}")
+        lines.append(f"### case: online_{index:04d}")
+        prev_session = case_value(case, "prev_session")
+        if isinstance(prev_session, list):
+            for item in prev_session[-10:]:
+                if not isinstance(item, dict):
+                    continue
+                prev_query = str(item.get("query") or "").strip()
+                prev_tts = str(item.get("tts") or "").strip()
+                if prev_query and prev_tts:
+                    lines.append(f"用户: {prev_query}")
+                    lines.append(f"小爱: {prev_tts}")
+        lines.append(f"用户: {query}")
+        lines.append(f"complex: {'true' if item_complex else 'false'}")
+        lines.append(f"target: {case_target}")
+        if request_id:
+            lines.append(f"request_id: {request_id}")
+        if timestamp:
+            lines.append(f"timestamp: {timestamp}")
+        if tts:
+            lines.append(f"notes: tts={tts}" + (f"；{'；'.join(notes)}" if notes else ""))
+        elif notes:
+            lines.append(f"notes: {'；'.join(notes)}")
+        lines.append("")
+    return "\n".join(lines).strip() + "\n"
+
+
 def main() -> int:
     try:
         payload = load_json_payload()
@@ -49,36 +112,12 @@ def main() -> int:
         if not cases:
             raise ValueError("cases must not be empty")
 
-        lines = [f"# dataset_label: {dataset_label}", ""]
+        draft_text = build_dataset_draft_text(
-        for index, case in enumerate(cases, start=1):
+            dataset_label=dataset_label,
-            query = str(case_value(case, "query") or "").strip()
+            target=target,
-            if not query:
+            complex_value=complex_value,
-                continue
+            cases=cases,
-            request_id = str(case_value(case, "request_id") or "").strip()
+        )
-            timestamp = case_value(case, "timestamp")
-            tts = str(case_value(case, "tts") or "").strip()
-            planning_result = str(case_value(case, "planning_result") or "").strip()
-            notes = []
-            if planning_result:
-                notes.append(f"线上模型输出: {planning_result}")
-            main_domain = case_value(case, "domain") or case.get("main_domain")
-            if main_domain:
-                notes.append(f"线上 domain: {main_domain}")
-            lines.append(f"### case: online_{index:04d}")
-            lines.append(f"用户: {query}")
-            lines.append(f"complex: {'true' if complex_value else 'false'}")
-            lines.append(f"target: {target}")
-            if request_id:
-                lines.append(f"request_id: {request_id}")
-            if timestamp:
-                lines.append(f"timestamp: {timestamp}")
-            if tts:
-                lines.append(f"notes: tts={tts}" + (f"；{'；'.join(notes)}" if notes else ""))
-            elif notes:
-                lines.append(f"notes: {'；'.join(notes)}")
-            lines.append("")
-
-        draft_text = "\n".join(lines).strip() + "\n"
         output_path = None
         if payload.get("output_path"):
             output = resolve_portable_path(str(payload["output_path"]))
@@ -94,4 +133,3 @@ def main() -> int:
 
 if __name__ == "__main__":
     raise SystemExit(main())
-

skills/online-mining-v2/scripts/build_online_records.py

@@ -0,0 +1,353 @@
+from __future__ import annotations
+
+"""把线上 ELK case 直接转换为 product-data canonical records。
+
+这个脚本承接 online-mining-v2 和 product-data：
+1. 按 request_id 拉主 NLP 表和前处理表。
+2. 用主表 session_id 补齐当前请求前最多 10 轮上下文。
+3. 从用户指定 target 或线上 planning/code 结果推断标签。
+4. 直接产出 canonical records，并可同步导出流通表格、训练 jsonl、评测 csv。
+"""
+
+import json
+import re
+import sys
+from pathlib import Path
+from typing import Any
+
+from online_mining_common import (
+    OnlineMiningError,
+    compact_text,
+    emit_error,
+    emit_success,
+    extract_case,
+    fetch_one_by_request_id,
+    fetch_session_turns,
+    load_json_payload,
+    resolve_portable_path,
+    string_values,
+)
+
+PRODUCT_DATA_SCRIPT_DIR = Path(__file__).resolve().parents[1].parent / "product-data" / "scripts"
+sys.path.insert(0, str(PRODUCT_DATA_SCRIPT_DIR))
+
+from product_data_portable import (  # noqa: E402
+    canonical_target,
+    export_dataset_records,
+    export_planning_eval_csv,
+    export_training_jsonl,
+    record_id,
+    target_type,
+    validate_dataset_records,
+)
+
+
+def load_cases(payload: dict[str, Any]) -> list[dict[str, Any]]:
+    if isinstance(payload.get("cases"), list):
+        return [item for item in payload["cases"] if isinstance(item, dict)]
+    cases_path = payload.get("cases_path")
+    if cases_path:
+        rows = []
+        text = resolve_portable_path(str(cases_path)).read_text(encoding="utf-8")
+        for line in text.splitlines():
+            if not line.strip():
+                continue
+            item = json.loads(line)
+            if isinstance(item, dict):
+                rows.append(item)
+        return rows
+    return []
+
+
+def merge_case(request_id: str, main_case: dict[str, Any] | None, pre_case: dict[str, Any] | None) -> dict[str, Any]:
+    return {
+        "request_id": request_id,
+        "query": (pre_case or {}).get("query") or (main_case or {}).get("query"),
+        "timestamp": (main_case or {}).get("timestamp") or (pre_case or {}).get("timestamp"),
+        "session_id": (main_case or {}).get("session_id"),
+        "device_id": (main_case or {}).get("device_id"),
+        "device": (main_case or {}).get("device"),
+        "tts": (main_case or {}).get("text"),
+        "main": main_case,
+        "pre_processing": pre_case,
+    }
+
+
+def fetch_cases_by_request_ids(request_ids: list[str], *, date: str | None) -> list[dict[str, Any]]:
+    rows = []
+    for request_id in request_ids:
+        main_doc = fetch_one_by_request_id("main", request_id, date)
+        pre_doc = fetch_one_by_request_id("pre_processing", request_id, date)
+        main_case = extract_case("main", main_doc) if main_doc else None
+        pre_case = extract_case("pre_processing", pre_doc) if pre_doc else None
+        rows.append(merge_case(request_id, main_case, pre_case))
+    return rows
+
+
+def case_value(case: dict[str, Any], key: str) -> Any:
+    if key in case:
+        return case.get(key)
+    main = case.get("main") if isinstance(case.get("main"), dict) else {}
+    pre = case.get("pre_processing") if isinstance(case.get("pre_processing"), dict) else {}
+    return case.get(key) or pre.get(key) or main.get(key)
+
+
+def enrich_prev_session(case: dict[str, Any], *, date: str | None, limit: int) -> list[dict[str, Any]]:
+    explicit = case.get("prev_session")
+    if isinstance(explicit, list):
+        return normalize_prev_session(explicit)[-limit:]
+    session_id = str(case_value(case, "session_id") or "").strip()
+    timestamp = optional_int(case_value(case, "timestamp"))
+    if not session_id or timestamp is None:
+        return []
+    turns = fetch_session_turns(session_id=session_id, before_timestamp=timestamp, date=date, limit=limit)
+    return [
+        {
+            "query": str(turn.get("query") or ""),
+            "tts": str(turn.get("text") or ""),
+            "timestamp": optional_int(turn.get("timestamp")) or 0,
+        }
+        for turn in turns
+        if str(turn.get("query") or "").strip()
+    ][-limit:]
+
+
+def normalize_prev_session(items: list[Any]) -> list[dict[str, Any]]:
+    rows = []
+    for item in items:
+        if not isinstance(item, dict):
+            continue
+        query = str(item.get("query") or "").strip()
+        if not query:
+            continue
+        rows.append(
+            {
+                "query": query,
+                "tts": str(item.get("tts") or ""),
+                "timestamp": optional_int(item.get("timestamp")) or 0,
+            }
+        )
+    rows.sort(key=lambda item: int(item.get("timestamp") or 0))
+    return rows
+
+
+def infer_target(case: dict[str, Any], explicit_target: str) -> tuple[str, str]:
+    if explicit_target.strip():
+        return canonical_target(explicit_target.strip()), "input.target"
+    case_target = str(case_value(case, "target") or "").strip()
+    if case_target:
+        return canonical_target(case_target), "case.target"
+
+    pre = case.get("pre_processing") if isinstance(case.get("pre_processing"), dict) else {}
+    main = case.get("main") if isinstance(case.get("main"), dict) else {}
+    for key in ("code", "planning_result"):
+        value = str(pre.get(key) or "").strip()
+        if value:
+            return canonical_target(value), f"pre_processing.{key}"
+
+    llm_agent_info = main.get("llm_agent_info")
+    if isinstance(llm_agent_info, dict):
+        agent_type = str(llm_agent_info.get("agentType") or "").strip()
+        if agent_type:
+            return f'Agent(tag="{agent_type}")', "main.intention.intent_arbitrator_info.llm_agent_info.agentType"
+
+    domain = str(main.get("domain") or case_value(case, "domain") or "").strip()
+    if domain:
+        if re.fullmatch(r"[A-Z][A-Za-z0-9_]*", domain):
+            return f"{domain}()", "main.domain"
+        return f'Agent(tag="{domain}")', "main.domain"
+    raise OnlineMiningError("target is required because online logs do not contain code/planning_result/domain")
+
+
+def case_to_record(
+    case: dict[str, Any],
+    *,
+    dataset_label: str,
+    target: str,
+    complex_value: bool,
+    index: int,
+    date: str | None,
+    session_limit: int,
+) -> tuple[dict[str, Any], dict[str, Any]]:
+    query = str(case_value(case, "query") or "").strip()
+    if not query:
+        raise OnlineMiningError(f"case {index} has no query")
+    request_id = str(case_value(case, "request_id") or "").strip()
+    if not request_id:
+        raise OnlineMiningError(f"case {index} has no request_id")
+    timestamp = optional_int(case_value(case, "timestamp"))
+    if timestamp is None:
+        raise OnlineMiningError(f"case {index} has no timestamp")
+
+    final_target, target_source = infer_target(case, target)
+    prev_session = enrich_prev_session(case, date=date, limit=session_limit)
+    main = case.get("main") if isinstance(case.get("main"), dict) else {}
+    pre = case.get("pre_processing") if isinstance(case.get("pre_processing"), dict) else {}
+    record = {
+        "record_id": record_id(source_type="online", batch_id="online", request_id=request_id, index=index),
+        "source": {
+            "type": "online",
+            "request_id": request_id,
+            "timestamp": timestamp,
+        },
+        "turn": {
+            "query": query,
+            "timestamp": timestamp,
+        },
+        "prev_session": prev_session,
+        "context": {},
+        "label": {
+            "dataset_label": dataset_label,
+            "target": final_target,
+            "target_type": target_type(final_target),
+        },
+        "dimensions": {
+            "complex": parse_complex(case_value(case, "complex"), default=complex_value),
+        },
+        "meta": {
+            "session_id": str(case_value(case, "session_id") or ""),
+            "device_id": str(case_value(case, "device_id") or ""),
+            "device": compact_text(case_value(case, "device"), 500),
+            "online_tts": str(case_value(case, "tts") or ""),
+            "main_domain": str(main.get("domain") or ""),
+            "main_func": str(main.get("func") or ""),
+            "planning_result": str(pre.get("planning_result") or ""),
+            "planning_code": str(pre.get("code") or ""),
+            "hit_rules": pre.get("hit_rules") or main.get("hit_rules") or [],
+            "candidate_domains": pre.get("candidate_domains") or [],
+            "target_source": target_source,
+        },
+    }
+    summary = {
+        "request_id": request_id,
+        "query": query,
+        "target": final_target,
+        "target_source": target_source,
+        "prev_session_count": len(prev_session),
+        "main_domain": record["meta"]["main_domain"],
+        "planning_result": record["meta"]["planning_result"],
+    }
+    return record, summary
+
+
+def optional_int(value: Any) -> int | None:
+    if value is None or value == "":
+        return None
+    try:
+        return int(value)
+    except (TypeError, ValueError):
+        return None
+
+
+def parse_complex(value: Any, *, default: bool) -> bool:
+    if value is None or value == "":
+        return default
+    if isinstance(value, bool):
+        return value
+    text = str(value).strip().lower()
+    if text in {"true", "1", "yes", "y", "是", "复杂", "complex"}:
+        return True
+    if text in {"false", "0", "no", "n", "否", "不复杂", "简单", "simple"}:
+        return False
+    raise OnlineMiningError("complex must be true/false")
+
+
+def write_records_jsonl(records: list[dict[str, Any]], path: str) -> str:
+    output = resolve_portable_path(path)
+    output.parent.mkdir(parents=True, exist_ok=True)
+    with output.open("w", encoding="utf-8") as fh:
+        for record in records:
+            fh.write(json.dumps(record, ensure_ascii=False, separators=(",", ":")) + "\n")
+    return str(output)
+
+
+def main() -> int:
+    try:
+        payload = load_json_payload()
+        dataset_label = str(payload.get("dataset_label") or "").strip()
+        if not dataset_label:
+            raise OnlineMiningError("dataset_label is required")
+        target = str(payload.get("target") or "").strip()
+        complex_value = parse_complex(payload.get("complex"), default=False)
+        date = str(payload.get("date") or "").strip() or None
+        session_limit = int(payload.get("session_limit") or 10)
+        if session_limit < 0 or session_limit > 10:
+            raise OnlineMiningError("session_limit must be between 0 and 10")
+
+        request_ids = string_values(payload.get("request_ids"))
+        cases = load_cases(payload)
+        if request_ids:
+            cases = fetch_cases_by_request_ids(request_ids, date=date)
+        if not cases:
+            raise OnlineMiningError("request_ids, cases or cases_path is required")
+
+        records = []
+        summaries = []
+        skipped = []
+        for index, case in enumerate(cases, start=1):
+            try:
+                record, summary = case_to_record(
+                    case,
+                    dataset_label=dataset_label,
+                    target=target,
+                    complex_value=complex_value,
+                    index=index,
+                    date=date,
+                    session_limit=session_limit,
+                )
+            except Exception as exc:  # noqa: BLE001
+                skipped.append({"index": index, "request_id": case.get("request_id"), "error": str(exc)})
+                continue
+            records.append(record)
+            summaries.append(summary)
+
+        if not records:
+            raise OnlineMiningError(f"no records built; skipped={skipped}")
+
+        records_path = write_records_jsonl(records, str(payload.get("records_output_path") or "scratchpad/online_records.jsonl"))
+        validation = validate_dataset_records(records)
+        output_dir = str(payload.get("output_dir") or "output")
+        exports: dict[str, Any] = {}
+        if bool(payload.get("export_records", True)):
+            exports["records"] = export_dataset_records(
+                records,
+                output_path=str(resolve_portable_path(output_dir) / "records.jsonl"),
+                output_format="jsonl",
+                require_validation_ok=bool(payload.get("require_validation_ok", True)),
+                overwrite=bool(payload.get("overwrite", True)),
+                export_table=bool(payload.get("export_table", True)),
+            )
+        if bool(payload.get("export_training", False)):
+            exports["training"] = export_training_jsonl(
+                records,
+                output_path=str(resolve_portable_path(output_dir) / "training.jsonl"),
+                require_validation_ok=bool(payload.get("require_validation_ok", True)),
+                overwrite=bool(payload.get("overwrite", True)),
+            )
+        if bool(payload.get("export_eval", False)):
+            exports["eval"] = export_planning_eval_csv(
+                records,
+                output_path=str(resolve_portable_path(output_dir) / "eval_planning.csv"),
+                require_validation_ok=bool(payload.get("require_validation_ok", True)),
+                overwrite=bool(payload.get("overwrite", True)),
+            )
+
+        emit_success(
+            {
+                "date": date or "past-48h",
+                "record_count": len(records),
+                "records_path": records_path,
+                "validation": validation,
+                "exports": exports,
+                "summaries": summaries,
+                "skipped": skipped,
+            }
+        )
+        return 0
+    except Exception as exc:  # noqa: BLE001
+        emit_error(exc)
+        return 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

skills/online-mining-v2/scripts/online_mining_common.py

@@ -8,6 +8,7 @@ from __future__ import annotations
 import argparse
 import importlib.util
 import json
+import os
 import re
 import sys
 from copy import deepcopy
@@ -35,7 +36,7 @@ SOURCE_CONFIGS: dict[str, dict[str, Any]] = {
     },
     "pre_processing": {
         "profile": "default",
-        "index": "pre-processing-info*",
+        "index": "pre-processing*",
         "id_field": "requestId",
         "time_field": "timestamp",
     },
@@ -282,21 +283,70 @@ def search_docs(
 
 def fetch_one_by_request_id(source: str, request_id: str, date: str | None = None) -> dict[str, Any] | None:
     config = source_config(source)
-    docs = search_docs(
+    id_field = str(config["id_field"])
-        source=source,
+    # 精确查优先用 .keyword；部分索引字段本身就是 keyword，所以再兜底原字段。
-        date=date,
+    attempts = [
-        size=1,
+        {"term": {f"{id_field}.keyword": request_id}},
-        query_filters=[{"term": {str(config["id_field"]): request_id}}],
+        {"term": {id_field: request_id}},
-    )
+        {"wildcard": {f"{id_field}.keyword": {"value": f"{request_id}*"}}},
-    if not docs:
+        {"wildcard": {id_field: {"value": f"{request_id}*"}}},
-        # 部分 request id 会带后缀，兜底用 wildcard。
+    ]
-        docs = search_docs(
+    for query_filter in attempts:
-            source=source,
+        docs = search_docs(source=source, date=date, size=1, query_filters=[query_filter])
-            date=date,
+        if docs:
-            size=1,
+            return docs[0]
-            query_filters=[{"wildcard": {str(config["id_field"]): {"value": f"{request_id}*"}}}],
+    return None
-        )
+
-    return docs[0] if docs else None
+
+def fetch_session_turns(
+    *,
+    session_id: str,
+    before_timestamp: int,
+    date: str | None = None,
+    limit: int = 10,
+) -> list[dict[str, Any]]:
+    """按主表 session_id 拉当前请求之前的多轮上下文。"""
+
+    if not session_id or not before_timestamp:
+        return []
+    filters = [
+        {"term": {"session_id.keyword": session_id}},
+        {"range": {"timestamp": {"lt": before_timestamp}}},
+    ]
+    docs = search_docs(source="main", date=date, size=max(limit * 3, limit), query_filters=filters)
+    cases = [extract_main_case(doc) for doc in docs]
+    cases = [
+        case
+        for case in cases
+        if str(case.get("session_id") or "") == session_id
+        and isinstance(case.get("timestamp"), (int, str))
+        and int(case.get("timestamp") or 0) < before_timestamp
+        and str(case.get("query") or "").strip()
+    ]
+    cases.sort(key=lambda item: int(item.get("timestamp") or 0))
+    return cases[-limit:]
+
+
+def fetch_session_turns_for_case(case: dict[str, Any], *, date: str | None = None, limit: int = 10) -> list[dict[str, Any]]:
+    session_id = str(case.get("session_id") or "").strip()
+    try:
+        timestamp = int(case.get("timestamp") or 0)
+    except (TypeError, ValueError):
+        timestamp = 0
+    return fetch_session_turns(session_id=session_id, before_timestamp=timestamp, date=date, limit=limit)
+
+
+def fetch_by_terms(source: str, *, field: str, value: str, date: str | None = None, size: int = 10) -> list[dict[str, Any]]:
+    """小范围精确字段查询，供脚本探测 session/request 字段时使用。"""
+
+    filters = [{"term": {field: value}}]
+    return search_docs(source=source, date=date, size=size, query_filters=filters)
+
+
+def fetch_one_by_request_id_legacy(source: str, request_id: str, date: str | None = None) -> dict[str, Any] | None:
+    """保留旧函数名兼容外部脚本；新代码请用 fetch_one_by_request_id。"""
+
+    return fetch_one_by_request_id(source, request_id, date)
 
 
 def write_optional_jsonl(path: str | None, rows: list[dict[str, Any]]) -> str | None:
@@ -311,8 +361,27 @@ def write_optional_jsonl(path: str | None, rows: list[dict[str, Any]]) -> str |
 
 
 def resolve_portable_path(path: str) -> Path:
+    """解析 online-mining-v2 脚本路径，并优先映射到当前会话目录。
+
+    在 ZK Data Agent 中，python_exec 会注入 PYTHON_EXEC_SCRATCHPAD。
+    用户和 skill 文档里写的 scratchpad/、output/、input/ 都应该落在
+    当前会话下，不能误写到项目根目录。
+    """
+
     raw = Path(path).expanduser()
     if raw.is_absolute():
         return raw
+    scratchpad = os.environ.get("PYTHON_EXEC_SCRATCHPAD")
+    if scratchpad:
+        session_root = Path(scratchpad).expanduser().parent
+        parts = raw.parts
+        if parts:
+            head, *tail = parts
+            tail_path = Path(*tail) if tail else Path()
+            if head in {"scratchpad", "scratch"}:
+                return (Path(scratchpad).expanduser() / tail_path).resolve()
+            if head in {"output", "outputs"}:
+                return (session_root / "output" / tail_path).resolve()
+            if head in {"input", "inputs"}:
+                return (session_root / "input" / tail_path).resolve()
     return (REPO_ROOT / raw).resolve()
-