技能提交

skills-data/xdfx-single-agent-full-flow/SKILL.md

@@ -26,20 +26,20 @@ scripts/xdfx_flow.py
 }
 ```
 
-默认会在缺少 `Authorization` 或接口返回 401 时，自动用 `admin / ch@user` 调用：
+鉴权不能通过命令行传递 token。脚本会在启动阶段先 `import scripts/xdfx_auth.py`，用 `admin / ch@user` 获取访问令牌，并在第一步业务接口调用前写入请求头。
+
+默认调用：
 
 ```text
 POST /api/v1/auth/login/username?password=ch%40user
 body: {"username":"admin"}
 ```
 
-拿到 `access_token` 后会写入 `Authorization: Bearer ...` 并重试原接口一次。登录接口和 401 重试都会输出 `OP_FEEDBACK` / `ERROR_FEEDBACK`，但反馈中的 `password`、`token` 会脱敏。默认关闭 HTTPS 证书校验，适配 UAT/内网自签证书；如果环境证书链正常，可以追加 `--verify-ssl`。
-
-如果已经有可用登录态，也可以在命令里追加 `--auth-token TOKEN` 或 `--header "Authorization: Bearer TOKEN"`；此时脚本优先使用传入的 token。
+拿到 `access_token` 后会写入内部 `Authorization: Bearer ...`。不要通过任何命令行参数传 token；脚本会忽略外部 Authorization，统一由 `xdfx_auth.py` 获取和刷新。登录接口和 401 重试都会输出 `操作反馈` / `错误反馈`，但反馈中的 `password`、`token` 会脱敏。默认关闭 HTTPS 证书校验，适配 UAT/内网自签证书；如果环境证书链正常，可以追加 `--verify-ssl`。
 
 ## 执行命令
 
-从本 Skill 目录逐步执行。虽然这是“一个技能做完整流程”的版本，但为了给用户真实的逐步反馈，智能体必须按 step1-step6 分六个命令运行，每个命令结束后先把 `STEP_FEEDBACK` / 新增错误反馈给用户，再继续下一步：
+从本 Skill 目录逐步执行。虽然这是“一个技能做完整流程”的版本，但为了给用户真实的逐步反馈，智能体必须按 step1-step6 分六个命令运行，每个命令结束后先把 `步骤反馈` / 新增错误反馈给用户，再继续下一步：
 
 ```bash
 python scripts/xdfx_flow.py --mode step1 --input xdfx-input.json --state xdfx-flow-state.json --summary xdfx-flow-summary.md
@@ -52,13 +52,29 @@ python scripts/xdfx_flow.py --mode step6 --input xdfx-input.json --state xdfx-fl
 
 `--mode full` 仅作为无人值守或本地排障兜底，不作为默认执行方式；因为外层工具通常会等长命令结束才返回输出，无法保证用户看到实时反馈。逐步命令共享同一个 `xdfx-flow-state.json`，id 传递和断点续跑不受影响。
 
-每一步完成后都会输出一行 `STEP_FEEDBACK {...}`。智能体必须在每个 step 命令结束后给用户一次简短反馈，说明该步骤状态、已入库数量和新增错误数，然后再执行下一步。
+如果确实想用一条命令编排完整流程，优先使用轻量包装脚本，而不是 `--mode full`。它会在内部逐个启动 step 命令并转发每一行反馈：
 
-脚本每次调用接口前还会输出 `OP_FEEDBACK {...}`，包含当前操作、HTTP 方法和 URL。智能体必须用简短中文告诉用户“正在做什么、调用哪个接口”。
+```bash
+python scripts/xdfx_run_full_steps.py --input xdfx-input.json --state xdfx-flow-state.json --summary xdfx-flow-summary.md
+```
 
-脚本出现错误时会立即输出 `ERROR_FEEDBACK {...}`，包含阶段、接口 URL、状态码、请求摘要、响应摘要和异常类型。智能体必须立刻告诉用户错误细节，不能等到最后才说。
+每一步完成后都会输出一行 `步骤反馈 {...}`。智能体必须在每个 step 命令结束后给用户一次简短反馈，说明该步骤状态、已入库数量和新增错误数，然后再执行下一步。
 
-如果模型返回内容经过多候选 JSON 提取和修复后仍无法解析，脚本会输出 `JSON_PARSE_REVIEW {...}`。智能体必须查看其中的 `responseExcerpt`，判断是否能识别出可用 JSON；无法确认时，重跑当前步骤或重新调用对应接口。
+脚本每次调用接口前还会输出 `操作反馈 {...}`，包含当前操作、HTTP 方法和 URL。智能体必须用简短中文告诉用户“正在做什么、调用哪个接口”。
+
+每次保存接口成功拿到返回 ID 后会输出 `入库反馈 {...}`，包含 `入库对象类型`、`入库对象名称`、`返回ID` 和 `接口路径`。智能体必须把它转述成“已入库 xxx 受众 / xxx 脆弱点 / xxx 策略”等明确日志。
+
+脚本出现错误时会立即输出 `错误反馈 {...}`，包含阶段、接口 URL、状态码、请求摘要、响应摘要和异常类型。智能体必须立刻告诉用户错误细节，不能等到最后才说。
+
+如果模型返回内容经过多候选 JSON 提取和修复后仍无法解析，脚本会输出 `JSON解析复核 {...}`。智能体必须查看其中的 `响应片段`，判断是否能识别出可用 JSON；无法确认时，重跑当前步骤或重新调用对应接口。
+
+## 失败重试规则
+
+- 任意步骤出现 `错误反馈`、`JSON解析复核`，或 `步骤反馈` 中 `状态` 为 `部分完成` / `已跳过` / `本步骤新增错误数` 大于 0 时，必须先重跑当前失败的同一条 `--mode stepN` 命令。
+- 重试前必须告诉用户失败阶段、接口 URL、状态码、响应摘要和“正在重试当前步骤”，不能静默重试。
+- 当前步骤未重试成功前，不允许继续执行下一步；不要因为某一步失败就跳过进入后续分析或入库。
+- 同一步重试后仍失败时，停止自动推进，读取 `xdfx-flow-state.json.errors` 汇报失败细节，并等待用户处理或确认。
+- 只有用户明确要求忽略该失败，才可以继续后续步骤；回复中必须说明这是用户确认后的带风险继续。
 
 只做参数和脚本检查、不调用接口：
 
@@ -68,36 +84,44 @@ python scripts/xdfx_flow.py --mode full --input xdfx-input.json --dry-run
 
 ## 脚本保证
 
-- 受众只取最优 3 个继续。
-- 关键传播节点全局去重后只取最优 3 个继续。
+- 受众目标默认入库最优 3 个，可用 `--max-audiences` 调整。
+- 关键传播节点全局去重后默认入库最优 3 个，可用 `--max-pathways` 调整。
+- 下游脆弱点/策略默认只从已入库结果中选择前 2 个受众、前 2 个传播节点继续，可用 `--downstream-audiences` / `--downstream-pathways` 调成 1 或 2。
+- 每个被选中的受众/传播节点默认生成最多 3 个脆弱点，可用 `--max-vulns` 调整。
+- 每个被选中的受众/传播节点默认生成最多 3 个策略，可用 `--max-strategies` 调整。
 - 每个保存接口必须拿到返回 `id` 才进入下游。
 - 保存后会同步前端 `task/task-action.step` 步骤状态；如只想入库不更新步骤条，追加 `--no-step-status`。
 - 策略保存会同步写入 `task-action-channel`、`base-narrate`、`task-action-message`、`task-action-design-strategy-relation`，兼容策略表格/编排页面读取。
+- 策略 `信息形式/模态` 入库前会读取 `sys/sys-dict-data` 的 `modelType` 字典，把模型返回的“图片,视频/文本,图片”等文字转成前端需要的 value 逗号串；字典读取失败时使用内置兜底映射。
+- Step1 的受众目标抽取结果里，`用户编号列表` / `编号id` / `id` 必须对应上一轮 `zhanghao` 账号数组的索引值；脚本会按这个索引依次找到账号并保存受众账号关系。
 - 模型流式 `{"answer":"..."}` 会自动拼接。
 - 流式模型接口第一次返回空响应、`y模型没有响应`、`模型没有响应` 或超时类字符串时，会自动再调用一次；仍失败才写入 `errors`。
 - 模型输出会自动去掉 `<think>`，从前后混杂文本、多个 JSON 代码块、无换行流式文本、连续 JSON 对象、非工整 JSON 中提取多个候选，并按当前步骤所需字段选择最像的 JSON。
 - 模型 JSON 会尝试修复中文引号、全角冒号/逗号、尾随逗号、未加引号的键、Python 风格 `True/False/None`。
 - 模型 JSON 解析失败会自动重试 1 次。
-- 单个受众、账号、节点或策略失败不会拖垮全流程。
-- 每完成一步都会输出 `STEP_FEEDBACK`，包含 `step`、`label`、`status`、累计入库计数和新增错误数。
-- 每次接口调用前都会输出 `OP_FEEDBACK`，说明当前操作和接口 URL。
-- 每个错误都会立即输出 `ERROR_FEEDBACK`，说明失败阶段、接口、状态码、请求摘要和响应摘要。
+- 单个受众、账号、节点或策略失败会被脚本记录；智能体仍必须按“失败重试规则”先重跑当前步骤，不能直接跳到下一步。
+- 每完成一步都会输出 `步骤反馈`，包含 `步骤`、`步骤名称`、`状态`、`累计入库数量` 和 `本步骤新增错误数`。
+- 每次接口调用前都会输出 `操作反馈`，说明当前操作和接口 URL。
+- 每次保存成功都会输出 `入库反馈`，说明已入库对象类型、名称和返回 ID。
+- 每个错误都会立即输出 `错误反馈`，说明失败阶段、接口、状态码、请求摘要和响应摘要。
 - 所有接口错误都会写入 `xdfx-flow-state.json.errors`，包含阶段、URL、状态码、请求摘要、响应正文和异常类型。
 
 ## 回复用户
 
-执行过程中，每出现一行 `STEP_FEEDBACK`，都要立刻向用户反馈一次：
+执行过程中，每出现一行 `步骤反馈`，都要立刻向用户反馈一次：
 
 - 当前完成的步骤名称。
-- 状态：`ok`、`partial` 或 `skipped`。
+- 状态：`成功`、`部分完成` 或 `已跳过`。
 - 当前累计入库数量。
 - 本步骤新增错误数。
 
-每出现一行 `OP_FEEDBACK`，都要简短告诉用户当前正在执行的操作和接口，例如“正在保存受众目标，调用 POST /api/task/task-action-design”。
+每出现一行 `操作反馈`，都要简短告诉用户当前正在执行的操作和接口，例如“正在保存受众目标，调用 POST /api/task/task-action-design”。
 
-每出现一行 `ERROR_FEEDBACK`，都要立刻告诉用户失败阶段、接口 URL、状态码和响应摘要。
+每出现一行 `入库反馈`，都要简短告诉用户当前已入库的对象，例如“已入库受众目标：xxx，designId=123”或“已入库受众脆弱点：xxx，id=456”。
 
-每出现一行 `JSON_PARSE_REVIEW`，都要查看 `responseExcerpt`。如果能明显识别出目标 JSON，说明需要人工复核；如果不能识别，重跑当前 step 或重新调用对应模型接口，并告诉用户“模型 JSON 不工整，正在复核并重试”。
+每出现一行 `错误反馈`，都要立刻告诉用户失败阶段、接口 URL、状态码和响应摘要。
+
+每出现一行 `JSON解析复核`，都要查看 `响应片段`。如果能明显识别出目标 JSON，说明需要人工复核；如果不能识别，重跑当前 step 或重新调用对应模型接口，并告诉用户“模型 JSON 不工整，正在复核并重试”。
 
 全部执行结束后再读取 `xdfx-flow-summary.md` 和命令输出，向用户说明：
 
@@ -109,3 +133,4 @@ python scripts/xdfx_flow.py --mode full --input xdfx-input.json --dry-run
 - 失败明细
 
 只要 `errors` 不为空，就说“部分完成”，不要说“全部完成”。
+

skills-data/xdfx-single-agent-full-flow/scripts/xdfx_auth.py

@@ -0,0 +1,149 @@
+#!/usr/bin/env python3
+"""Authentication helper for xdfx skill scripts.
+
+The caller should not pass tokens through command-line arguments.  This module
+owns the fixed admin login flow and returns a Bearer token before business
+interfaces are called.
+"""
+
+from __future__ import annotations
+
+import json
+import re
+import ssl
+import urllib.error
+import urllib.parse
+import urllib.request
+from dataclasses import dataclass
+from typing import Any
+
+
+@dataclass
+class AuthResult:
+    token: str
+    url: str
+    status: int | None
+    raw: str
+
+
+class AuthError(RuntimeError):
+    def __init__(
+        self,
+        message: str,
+        *,
+        url: str,
+        status: int | None = None,
+        response: str = "",
+        exception: BaseException | None = None,
+    ) -> None:
+        super().__init__(message)
+        self.url = url
+        self.status = status
+        self.response = response
+        self.exception = exception
+
+
+def auth_header_value(token: str) -> str:
+    token = token.strip()
+    return token if token.lower().startswith("bearer ") else f"Bearer {token}"
+
+
+def redact_url(url: str) -> str:
+    return re.sub(r"([?&](?:password|token|authToken|access_token)=)[^&]+", r"\1***", url, flags=re.IGNORECASE)
+
+
+def ssl_context(*, verify_ssl: bool) -> ssl.SSLContext | None:
+    if verify_ssl:
+        return None
+    return ssl._create_unverified_context()
+
+
+def build_login_url(api_base: str, path: str, password: str) -> str:
+    base = path if re.match(r"^https?://", path, flags=re.IGNORECASE) else api_base.rstrip("/") + "/" + path.lstrip("/")
+    if password:
+        separator = "&" if "?" in base else "?"
+        base = f"{base}{separator}password={urllib.parse.quote(password, safe='')}"
+    return base
+
+
+def extract_access_token(data: Any) -> str | None:
+    if isinstance(data, dict):
+        for key in ("access_token", "accessToken", "token", "jwt", "id_token", "authorization", "Authorization"):
+            value = data.get(key)
+            if isinstance(value, str) and value.strip():
+                return value.strip()
+        for value in data.values():
+            nested = extract_access_token(value)
+            if nested:
+                return nested
+    elif isinstance(data, list):
+        for value in data:
+            nested = extract_access_token(value)
+            if nested:
+                return nested
+    return None
+
+
+def parse_json_loose(text: str) -> Any:
+    try:
+        return json.loads(text)
+    except Exception:
+        pass
+    match = re.search(r"[\[{]", text or "")
+    if not match:
+        return {}
+    start = match.start()
+    for end in range(len(text), start, -1):
+        try:
+            return json.loads(text[start:end])
+        except Exception:
+            continue
+    return {}
+
+
+def fetch_access_token(
+    *,
+    api_base: str,
+    username: str = "admin",
+    password: str = "ch@user",
+    login_path: str = "/api/v1/auth/login/username",
+    timeout: float = 90,
+    verify_ssl: bool = False,
+) -> AuthResult:
+    """Login once and return the extracted access token."""
+    url = build_login_url(api_base, login_path, password)
+    payload = {"username": username}
+    data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
+    headers = {"Content-Type": "application/json", "Accept": "application/json"}
+    req = urllib.request.Request(url, data=data, headers=headers, method="POST")
+    try:
+        with urllib.request.urlopen(req, timeout=timeout, context=ssl_context(verify_ssl=verify_ssl)) as resp:
+            raw = resp.read().decode("utf-8", errors="replace")
+            parsed = parse_json_loose(raw)
+            token = extract_access_token(parsed)
+            if not token:
+                raise AuthError(
+                    "登录接口已响应但未找到 access_token",
+                    url=redact_url(url),
+                    status=getattr(resp, "status", None),
+                    response=raw,
+                )
+            return AuthResult(token=token, url=redact_url(url), status=getattr(resp, "status", None), raw=raw)
+    except urllib.error.HTTPError as exc:
+        body = exc.read().decode("utf-8", errors="replace") if exc.fp else ""
+        raise AuthError(
+            "登录接口返回错误",
+            url=redact_url(url),
+            status=exc.code,
+            response=body,
+            exception=exc,
+        ) from exc
+    except AuthError:
+        raise
+    except Exception as exc:  # noqa: BLE001 - auth errors must preserve detail for user feedback.
+        raise AuthError(
+            "登录接口调用异常",
+            url=redact_url(url),
+            response=str(exc),
+            exception=exc,
+        ) from exc

skills-data/xdfx-single-agent-full-flow/scripts/xdfx_run_full_steps.py

@@ -0,0 +1,95 @@
+#!/usr/bin/env python3
+"""Run the xdfx flow one step at a time.
+
+The heavy business logic intentionally stays in xdfx_flow.py.  This wrapper only
+keeps "full flow" orchestration readable and preserves per-step feedback: every
+subprocess line is flushed immediately, so OP/SAVE/ERROR/STEP feedback remains
+visible to the calling agent.
+"""
+
+from __future__ import annotations
+
+import argparse
+import os
+import subprocess
+import sys
+from pathlib import Path
+
+
+STEPS = ("step1", "step2", "step3", "step4", "step5", "step6")
+
+STEP_LABELS = {
+    "step1": "受众目标分析与入库",
+    "step2": "关键传播节点分析与入库",
+    "step3": "受众脆弱点分析与入库",
+    "step4": "传播节点脆弱点分析与入库",
+    "step5": "受众策略分析与入库",
+    "step6": "传播节点策略分析与入库",
+}
+
+
+def build_parser() -> argparse.ArgumentParser:
+    """Parse only wrapper-owned flags; pass all other flags to xdfx_flow.py."""
+    parser = argparse.ArgumentParser(
+        description="按六个步骤运行 xdfx_flow.py，并实时转发中文反馈日志。",
+        add_help=True,
+    )
+    parser.add_argument("--from-step", choices=STEPS, default="step1")
+    parser.add_argument("--to-step", choices=STEPS, default="step6")
+    parser.add_argument(
+        "--continue-on-error",
+        action="store_true",
+        help="某一步返回非零退出码时仍继续执行后续步骤。",
+    )
+    return parser
+
+
+def selected_steps(first: str, last: str) -> list[str]:
+    """Return an inclusive step slice for resume/partial reruns."""
+    start = STEPS.index(first)
+    end = STEPS.index(last)
+    if start > end:
+        raise SystemExit("起始步骤必须小于或等于结束步骤")
+    return list(STEPS[start : end + 1])
+
+
+def stream_command(step: str, cmd: list[str]) -> int:
+    """Run one step and forward stdout/stderr lines as they are produced."""
+    print(f"运行步骤：{STEP_LABELS.get(step, step)}；执行命令：" + " ".join(cmd), flush=True)
+    env = os.environ.copy()
+    env["PYTHONIOENCODING"] = "utf-8"
+    process = subprocess.Popen(
+        cmd,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+        text=True,
+        encoding="utf-8",
+        errors="replace",
+        env=env,
+    )
+    assert process.stdout is not None
+    for line in process.stdout:
+        sys.stdout.buffer.write(line.encode("utf-8", errors="replace"))
+        sys.stdout.buffer.flush()
+    return process.wait()
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = build_parser()
+    args, passthrough = parser.parse_known_args(argv)
+    flow_script = Path(__file__).with_name("xdfx_flow.py")
+    exit_code = 0
+    for step in selected_steps(args.from_step, args.to_step):
+        # Keep all business flags unchanged so the wrapper behaves like a thin
+        # streaming version of repeated xdfx_flow.py --mode stepN calls.
+        cmd = [sys.executable, str(flow_script), "--mode", step, *passthrough]
+        code = stream_command(step, cmd)
+        if code:
+            exit_code = code
+            if not args.continue_on_error:
+                return code
+    return exit_code
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())