diff --git a/skills-data/xdfx-full-flow-skill-guide.md b/skills-data/xdfx-full-flow-skill-guide.md index 014a6300..ec7ed043 100644 --- a/skills-data/xdfx-full-flow-skill-guide.md +++ b/skills-data/xdfx-full-flow-skill-guide.md @@ -194,7 +194,8 @@ Step1 受众目标分析完成:当前已入库受众 X 个,本步骤新增 - 单个受众、账号、传播节点、脆弱点或策略失败,不中断整个流程。 - 保存接口必须返回 `id`,否则不会进入下游步骤。 - 受众目标默认入库 3 个,关键传播节点默认入库 3 个。 -- 下游脆弱点/策略默认只选择前 2 个受众和前 2 个传播节点继续,可用 `--downstream-audiences` / `--downstream-pathways` 调成 1 或 2。 +- 下游受众脆弱点/策略只能从“已在 Step2 分析出并关联关键传播节点”的受众中选择,默认取前 2 个;没有产生传播节点的受众不得继续分析脆弱点和策略。 +- 下游传播节点脆弱点/策略默认只选择前 2 个已入库传播节点继续,可用 `--downstream-audiences` / `--downstream-pathways` 调成 1 或 2。 - 每个被选中的对象默认最多生成 3 个脆弱点、3 个策略。 - 模型流式 `answer` 会自动拼接。 - 流式模型接口第一次返回空响应、`y模型没有响应`、`模型没有响应` 或超时类字符串时,会自动再调用一次;仍失败才写入 `errors`。 diff --git a/skills-data/xdfx-single-agent-full-flow/SKILL.md b/skills-data/xdfx-single-agent-full-flow/SKILL.md index 2cdce48e..b0a45773 100644 --- a/skills-data/xdfx-single-agent-full-flow/SKILL.md +++ b/skills-data/xdfx-single-agent-full-flow/SKILL.md @@ -86,7 +86,8 @@ python scripts/xdfx_flow.py --mode full --input xdfx-input.json --dry-run - 受众目标默认入库最优 3 个,可用 `--max-audiences` 调整。 - 关键传播节点全局去重后默认入库最优 3 个,可用 `--max-pathways` 调整。 -- 下游脆弱点/策略默认只从已入库结果中选择前 2 个受众、前 2 个传播节点继续,可用 `--downstream-audiences` / `--downstream-pathways` 调成 1 或 2。 +- 下游受众脆弱点/策略只能从“已在 Step2 分析出并关联关键传播节点”的受众中选择,默认取前 2 个;没有产生传播节点的受众不得继续分析脆弱点和策略。 +- 下游传播节点脆弱点/策略默认从已入库传播节点中选择前 2 个继续;可用 `--downstream-audiences` / `--downstream-pathways` 调成 1 或 2。 - 每个被选中的受众/传播节点默认生成最多 3 个脆弱点,可用 `--max-vulns` 调整。 - 每个被选中的受众/传播节点默认生成最多 3 个策略,可用 `--max-strategies` 调整。 - 每个保存接口必须拿到返回 `id` 才进入下游。 diff --git a/skills-data/xdfx-single-agent-full-flow/scripts/xdfx_flow.py b/skills-data/xdfx-single-agent-full-flow/scripts/xdfx_flow.py index d9580e2f..8e62e51e 100644 --- a/skills-data/xdfx-single-agent-full-flow/scripts/xdfx_flow.py +++ b/skills-data/xdfx-single-agent-full-flow/scripts/xdfx_flow.py @@ -368,6 +368,7 @@ def default_state() -> dict[str, Any]: "topAudiences": [], "pathwayCandidates": [], "topPathways": [], + "audiencesWithPathways": [], "audienceVulns": {}, "pathwayVulns": {}, "audienceStrategies": [], @@ -2055,6 +2056,88 @@ def selected_accounts_for_item(item: dict[str, Any], *, audience: bool) -> list[ return [item.get("account") or item] +def id_ref_key(value: Any) -> str: + number = as_int(value) + if number is not None: + return str(number) + return str(value or "").strip() + + +def audience_name_key(value: Any) -> str: + return re.sub(r"\s+", "", str(value or "")).lower() + + +def compact_pathway_ref(pathway: dict[str, Any]) -> dict[str, Any]: + return { + "pathwayAccountId": pathway.get("pathwayAccountId") or pathway.get("pathwayId"), + "accountId": pathway.get("accountId"), + "accountName": pathway.get("accountName"), + } + + +def add_unique_pathway(target: list[dict[str, Any]], pathway: dict[str, Any]) -> None: + key = id_ref_key(pathway.get("pathwayAccountId") or pathway.get("pathwayId") or pathway.get("accountId") or pathway.get("accountName")) + if not key: + return + seen = { + id_ref_key(item.get("pathwayAccountId") or item.get("pathwayId") or item.get("accountId") or item.get("accountName")) + for item in target + } + if key not in seen: + target.append(pathway) + + +def audience_pathway_pairs(state: dict[str, Any]) -> list[tuple[dict[str, Any], list[dict[str, Any]]]]: + """Return only audiences that produced at least one saved key pathway.""" + id_links: dict[str, list[dict[str, Any]]] = {} + name_links: dict[str, list[dict[str, Any]]] = {} + for pathway in state.get("topPathways") or []: + if not isinstance(pathway, dict): + continue + for design_id in pathway.get("designIds") or []: + key = id_ref_key(design_id) + if key: + id_links.setdefault(key, []).append(pathway) + for name in pathway.get("relatedAudiences") or []: + key = audience_name_key(name) + if key: + name_links.setdefault(key, []).append(pathway) + + pairs: list[tuple[dict[str, Any], list[dict[str, Any]]]] = [] + for audience_item in state.get("topAudiences") or []: + if not isinstance(audience_item, dict): + continue + linked: list[dict[str, Any]] = [] + design_key = id_ref_key(audience_item.get("designId")) + for pathway in id_links.get(design_key, []): + add_unique_pathway(linked, pathway) + name_key = audience_name_key(audience_item.get("targetAudience")) + for pathway in name_links.get(name_key, []): + add_unique_pathway(linked, pathway) + if linked: + item = {**audience_item, "downstreamPathways": [compact_pathway_ref(pathway) for pathway in linked]} + pairs.append((item, linked)) + return pairs + + +def audience_pathway_summary(state: dict[str, Any]) -> list[dict[str, Any]]: + summary: list[dict[str, Any]] = [] + for audience_item, pathways in audience_pathway_pairs(state): + summary.append( + { + "designId": audience_item.get("designId"), + "targetAudience": audience_item.get("targetAudience"), + "pathwayCount": len(pathways), + "pathways": [compact_pathway_ref(pathway) for pathway in pathways], + } + ) + return summary + + +def has_audiences_with_pathways(state: dict[str, Any]) -> bool: + return bool(audience_pathway_pairs(state)) + + def downstream_limit(ctx: FlowContext, *, audience: bool) -> int: """Limit expensive downstream analysis while keeping 3 upstream rows stored.""" raw = ctx.downstream_audiences if audience else ctx.downstream_pathways @@ -2066,22 +2149,35 @@ def downstream_items(ctx: FlowContext, state: dict[str, Any], stage: str, *, aud source_key = "topAudiences" if audience else "topPathways" id_key = "designId" if audience else "pathwayAccountId" selected_key = "selectedDownstreamAudiences" if audience else "selectedDownstreamPathways" - items = [item for item in (state.get(source_key) or []) if isinstance(item, dict)] + all_items = [item for item in (state.get(source_key) or []) if isinstance(item, dict)] + if audience: + state["audiencesWithPathways"] = audience_pathway_summary(state) + items = [item for item, _ in audience_pathway_pairs(state)] + if not items: + state[selected_key] = [] + record_error(state, stage, "没有已分析出关键传播节点的受众,不能继续分析受众脆弱点或策略;请先重试关键传播节点分析") + return [] + else: + items = all_items limit = downstream_limit(ctx, audience=audience) selected = items[:limit] - state[selected_key] = [ - { + selected_rows: list[dict[str, Any]] = [] + for item in selected: + row = { "id": item.get(id_key), "name": item.get("targetAudience") if audience else item.get("accountName"), } - for item in selected - ] - if items: + if audience: + row["pathwayCount"] = len(item.get("downstreamPathways") or []) + row["pathways"] = item.get("downstreamPathways") or [] + selected_rows.append(row) + state[selected_key] = selected_rows + if selected: record_event( state, stage, - f"下游分析选择 {len(selected)}/{len(items)} 个{'受众' if audience else '传播节点'}", - {"limit": limit, "selected": state[selected_key]}, + f"下游分析选择 {len(selected)}/{len(items)} 个{'已产生传播节点的受众' if audience else '传播节点'}", + {"limit": limit, "selected": state[selected_key], "candidateCount": len(items), "totalStoredCount": len(all_items)}, ) return selected @@ -2183,6 +2279,7 @@ def run_step2(ctx: FlowContext, state: dict[str, Any]) -> None: record_error(state, stage, "没有已入库受众,无法分析传播节点") return if state.get("topPathways") and not ctx.dry_run: + state["audiencesWithPathways"] = audience_pathway_summary(state) record_event(state, stage, "topPathways 已存在,跳过。如需重跑请清空 state 中该字段。") return all_candidates: list[dict[str, Any]] = [] @@ -2278,9 +2375,10 @@ def run_step2(ctx: FlowContext, state: dict[str, Any]) -> None: http_post_json(ctx, state, "save_relation", relation_payload, stage + "/save-relation") saved.append(item) state["topPathways"] = saved + state["audiencesWithPathways"] = audience_pathway_summary(state) if saved: mark_step_status(ctx, state, ["step-analysis-keypoint", "step-analysis-weakpoint"], stage + "/step-status") - record_event(state, stage, f"完成,保存传播节点 {len(saved)} 个") + record_event(state, stage, f"完成,保存传播节点 {len(saved)} 个,可继续下游分析的受众 {len(state.get('audiencesWithPathways') or [])} 个") def run_step3(ctx: FlowContext, state: dict[str, Any]) -> None: diff --git a/skills-data/xdfx-split-controller/SKILL.md b/skills-data/xdfx-split-controller/SKILL.md index 1ee180ab..71ad1d51 100644 --- a/skills-data/xdfx-split-controller/SKILL.md +++ b/skills-data/xdfx-split-controller/SKILL.md @@ -77,8 +77,9 @@ python scripts/xdfx_run_full_steps.py --input xdfx-input.json --state xdfx-flow- - Step1 的受众目标抽取结果里,`用户编号列表` / `编号id` / `id` 必须对应上一轮 `zhanghao` 账号数组的索引值;脚本会按这个索引依次找到账号并保存受众账号关系。 - Step2 成功后会写入 `topPathways`。 - Step1 默认入库最优 3 个受众;Step2 默认入库最优 3 个传播节点。 -- Step3/Step4 默认只选择前 2 个已入库受众/传播节点继续,每个生成最多 3 个脆弱点;可用 `--downstream-audiences` / `--downstream-pathways` 调成 1 或 2。 -- Step5/Step6 默认对上述选中的每个受众/传播节点生成最多 3 个策略。 +- Step3/Step5 的受众下游分析只能从“已在 Step2 分析出并关联关键传播节点”的受众中选择,默认取前 2 个;没有产生传播节点的受众不得继续分析脆弱点和策略。 +- Step4/Step6 的传播节点下游分析默认从已入库传播节点中选择前 2 个继续;可用 `--downstream-audiences` / `--downstream-pathways` 调成 1 或 2。 +- Step3-Step6 对上述选中的每个受众/传播节点默认生成最多 3 个脆弱点或策略。 - 各步骤会同步前端 `task/task-action.step`;需要禁用时每条命令追加 `--no-step-status`。 - 策略步骤会同步写入 channel、narrate、message、策略与受众/传播节点关系表,保证原前端策略表格/编排页可继续读取。 - 策略步骤保存 `信息形式/模态` 时,必须使用脚本读取 `sys/sys-dict-data` 的 `modelType` 字典并把文字转成 value 逗号串,不能把“图片,视频”等文字直接写入 `task-action-message.modelType`。 diff --git a/skills-data/xdfx-split-controller/scripts/xdfx_flow.py b/skills-data/xdfx-split-controller/scripts/xdfx_flow.py index d9580e2f..8e62e51e 100644 --- a/skills-data/xdfx-split-controller/scripts/xdfx_flow.py +++ b/skills-data/xdfx-split-controller/scripts/xdfx_flow.py @@ -368,6 +368,7 @@ def default_state() -> dict[str, Any]: "topAudiences": [], "pathwayCandidates": [], "topPathways": [], + "audiencesWithPathways": [], "audienceVulns": {}, "pathwayVulns": {}, "audienceStrategies": [], @@ -2055,6 +2056,88 @@ def selected_accounts_for_item(item: dict[str, Any], *, audience: bool) -> list[ return [item.get("account") or item] +def id_ref_key(value: Any) -> str: + number = as_int(value) + if number is not None: + return str(number) + return str(value or "").strip() + + +def audience_name_key(value: Any) -> str: + return re.sub(r"\s+", "", str(value or "")).lower() + + +def compact_pathway_ref(pathway: dict[str, Any]) -> dict[str, Any]: + return { + "pathwayAccountId": pathway.get("pathwayAccountId") or pathway.get("pathwayId"), + "accountId": pathway.get("accountId"), + "accountName": pathway.get("accountName"), + } + + +def add_unique_pathway(target: list[dict[str, Any]], pathway: dict[str, Any]) -> None: + key = id_ref_key(pathway.get("pathwayAccountId") or pathway.get("pathwayId") or pathway.get("accountId") or pathway.get("accountName")) + if not key: + return + seen = { + id_ref_key(item.get("pathwayAccountId") or item.get("pathwayId") or item.get("accountId") or item.get("accountName")) + for item in target + } + if key not in seen: + target.append(pathway) + + +def audience_pathway_pairs(state: dict[str, Any]) -> list[tuple[dict[str, Any], list[dict[str, Any]]]]: + """Return only audiences that produced at least one saved key pathway.""" + id_links: dict[str, list[dict[str, Any]]] = {} + name_links: dict[str, list[dict[str, Any]]] = {} + for pathway in state.get("topPathways") or []: + if not isinstance(pathway, dict): + continue + for design_id in pathway.get("designIds") or []: + key = id_ref_key(design_id) + if key: + id_links.setdefault(key, []).append(pathway) + for name in pathway.get("relatedAudiences") or []: + key = audience_name_key(name) + if key: + name_links.setdefault(key, []).append(pathway) + + pairs: list[tuple[dict[str, Any], list[dict[str, Any]]]] = [] + for audience_item in state.get("topAudiences") or []: + if not isinstance(audience_item, dict): + continue + linked: list[dict[str, Any]] = [] + design_key = id_ref_key(audience_item.get("designId")) + for pathway in id_links.get(design_key, []): + add_unique_pathway(linked, pathway) + name_key = audience_name_key(audience_item.get("targetAudience")) + for pathway in name_links.get(name_key, []): + add_unique_pathway(linked, pathway) + if linked: + item = {**audience_item, "downstreamPathways": [compact_pathway_ref(pathway) for pathway in linked]} + pairs.append((item, linked)) + return pairs + + +def audience_pathway_summary(state: dict[str, Any]) -> list[dict[str, Any]]: + summary: list[dict[str, Any]] = [] + for audience_item, pathways in audience_pathway_pairs(state): + summary.append( + { + "designId": audience_item.get("designId"), + "targetAudience": audience_item.get("targetAudience"), + "pathwayCount": len(pathways), + "pathways": [compact_pathway_ref(pathway) for pathway in pathways], + } + ) + return summary + + +def has_audiences_with_pathways(state: dict[str, Any]) -> bool: + return bool(audience_pathway_pairs(state)) + + def downstream_limit(ctx: FlowContext, *, audience: bool) -> int: """Limit expensive downstream analysis while keeping 3 upstream rows stored.""" raw = ctx.downstream_audiences if audience else ctx.downstream_pathways @@ -2066,22 +2149,35 @@ def downstream_items(ctx: FlowContext, state: dict[str, Any], stage: str, *, aud source_key = "topAudiences" if audience else "topPathways" id_key = "designId" if audience else "pathwayAccountId" selected_key = "selectedDownstreamAudiences" if audience else "selectedDownstreamPathways" - items = [item for item in (state.get(source_key) or []) if isinstance(item, dict)] + all_items = [item for item in (state.get(source_key) or []) if isinstance(item, dict)] + if audience: + state["audiencesWithPathways"] = audience_pathway_summary(state) + items = [item for item, _ in audience_pathway_pairs(state)] + if not items: + state[selected_key] = [] + record_error(state, stage, "没有已分析出关键传播节点的受众,不能继续分析受众脆弱点或策略;请先重试关键传播节点分析") + return [] + else: + items = all_items limit = downstream_limit(ctx, audience=audience) selected = items[:limit] - state[selected_key] = [ - { + selected_rows: list[dict[str, Any]] = [] + for item in selected: + row = { "id": item.get(id_key), "name": item.get("targetAudience") if audience else item.get("accountName"), } - for item in selected - ] - if items: + if audience: + row["pathwayCount"] = len(item.get("downstreamPathways") or []) + row["pathways"] = item.get("downstreamPathways") or [] + selected_rows.append(row) + state[selected_key] = selected_rows + if selected: record_event( state, stage, - f"下游分析选择 {len(selected)}/{len(items)} 个{'受众' if audience else '传播节点'}", - {"limit": limit, "selected": state[selected_key]}, + f"下游分析选择 {len(selected)}/{len(items)} 个{'已产生传播节点的受众' if audience else '传播节点'}", + {"limit": limit, "selected": state[selected_key], "candidateCount": len(items), "totalStoredCount": len(all_items)}, ) return selected @@ -2183,6 +2279,7 @@ def run_step2(ctx: FlowContext, state: dict[str, Any]) -> None: record_error(state, stage, "没有已入库受众,无法分析传播节点") return if state.get("topPathways") and not ctx.dry_run: + state["audiencesWithPathways"] = audience_pathway_summary(state) record_event(state, stage, "topPathways 已存在,跳过。如需重跑请清空 state 中该字段。") return all_candidates: list[dict[str, Any]] = [] @@ -2278,9 +2375,10 @@ def run_step2(ctx: FlowContext, state: dict[str, Any]) -> None: http_post_json(ctx, state, "save_relation", relation_payload, stage + "/save-relation") saved.append(item) state["topPathways"] = saved + state["audiencesWithPathways"] = audience_pathway_summary(state) if saved: mark_step_status(ctx, state, ["step-analysis-keypoint", "step-analysis-weakpoint"], stage + "/step-status") - record_event(state, stage, f"完成,保存传播节点 {len(saved)} 个") + record_event(state, stage, f"完成,保存传播节点 {len(saved)} 个,可继续下游分析的受众 {len(state.get('audiencesWithPathways') or [])} 个") def run_step3(ctx: FlowContext, state: dict[str, Any]) -> None: diff --git a/skills-data/xdfx-split-flow-skill-guide.md b/skills-data/xdfx-split-flow-skill-guide.md index 62d52a6e..f09c6203 100644 --- a/skills-data/xdfx-split-flow-skill-guide.md +++ b/skills-data/xdfx-split-flow-skill-guide.md @@ -134,8 +134,9 @@ python scripts/xdfx_run_full_steps.py --input xdfx-input.json --state xdfx-flow- - Step1 入库最优 3 个受众。 - Step2 入库最优 3 个传播节点。 -- Step3/Step4 只从已入库结果中默认选择前 2 个受众/传播节点继续,每个生成最多 3 个脆弱点;可用 `--downstream-audiences` / `--downstream-pathways` 调成 1 或 2。 -- Step5/Step6 对上述选中的每个受众/传播节点默认生成最多 3 个策略。 +- Step3/Step5 的受众下游分析只能从“已在 Step2 分析出并关联关键传播节点”的受众中选择,默认取前 2 个;没有产生传播节点的受众不得继续分析脆弱点和策略。 +- Step4/Step6 的传播节点下游分析默认从已入库传播节点中选择前 2 个继续;可用 `--downstream-audiences` / `--downstream-pathways` 调成 1 或 2。 +- Step3-Step6 对上述选中的每个受众/传播节点默认生成最多 3 个脆弱点或策略。 - Step5/Step6 的策略模态必须按前端 `modelType` 字典保存为 value 逗号串;脚本会自动把“图片,视频/文本,图片”等文字转成 `task-action-message.modelType` 需要的值。 总控反馈用户时建议使用这种格式: diff --git a/skills-data/xdfx-step1-audience/scripts/xdfx_flow.py b/skills-data/xdfx-step1-audience/scripts/xdfx_flow.py index d9580e2f..8e62e51e 100644 --- a/skills-data/xdfx-step1-audience/scripts/xdfx_flow.py +++ b/skills-data/xdfx-step1-audience/scripts/xdfx_flow.py @@ -368,6 +368,7 @@ def default_state() -> dict[str, Any]: "topAudiences": [], "pathwayCandidates": [], "topPathways": [], + "audiencesWithPathways": [], "audienceVulns": {}, "pathwayVulns": {}, "audienceStrategies": [], @@ -2055,6 +2056,88 @@ def selected_accounts_for_item(item: dict[str, Any], *, audience: bool) -> list[ return [item.get("account") or item] +def id_ref_key(value: Any) -> str: + number = as_int(value) + if number is not None: + return str(number) + return str(value or "").strip() + + +def audience_name_key(value: Any) -> str: + return re.sub(r"\s+", "", str(value or "")).lower() + + +def compact_pathway_ref(pathway: dict[str, Any]) -> dict[str, Any]: + return { + "pathwayAccountId": pathway.get("pathwayAccountId") or pathway.get("pathwayId"), + "accountId": pathway.get("accountId"), + "accountName": pathway.get("accountName"), + } + + +def add_unique_pathway(target: list[dict[str, Any]], pathway: dict[str, Any]) -> None: + key = id_ref_key(pathway.get("pathwayAccountId") or pathway.get("pathwayId") or pathway.get("accountId") or pathway.get("accountName")) + if not key: + return + seen = { + id_ref_key(item.get("pathwayAccountId") or item.get("pathwayId") or item.get("accountId") or item.get("accountName")) + for item in target + } + if key not in seen: + target.append(pathway) + + +def audience_pathway_pairs(state: dict[str, Any]) -> list[tuple[dict[str, Any], list[dict[str, Any]]]]: + """Return only audiences that produced at least one saved key pathway.""" + id_links: dict[str, list[dict[str, Any]]] = {} + name_links: dict[str, list[dict[str, Any]]] = {} + for pathway in state.get("topPathways") or []: + if not isinstance(pathway, dict): + continue + for design_id in pathway.get("designIds") or []: + key = id_ref_key(design_id) + if key: + id_links.setdefault(key, []).append(pathway) + for name in pathway.get("relatedAudiences") or []: + key = audience_name_key(name) + if key: + name_links.setdefault(key, []).append(pathway) + + pairs: list[tuple[dict[str, Any], list[dict[str, Any]]]] = [] + for audience_item in state.get("topAudiences") or []: + if not isinstance(audience_item, dict): + continue + linked: list[dict[str, Any]] = [] + design_key = id_ref_key(audience_item.get("designId")) + for pathway in id_links.get(design_key, []): + add_unique_pathway(linked, pathway) + name_key = audience_name_key(audience_item.get("targetAudience")) + for pathway in name_links.get(name_key, []): + add_unique_pathway(linked, pathway) + if linked: + item = {**audience_item, "downstreamPathways": [compact_pathway_ref(pathway) for pathway in linked]} + pairs.append((item, linked)) + return pairs + + +def audience_pathway_summary(state: dict[str, Any]) -> list[dict[str, Any]]: + summary: list[dict[str, Any]] = [] + for audience_item, pathways in audience_pathway_pairs(state): + summary.append( + { + "designId": audience_item.get("designId"), + "targetAudience": audience_item.get("targetAudience"), + "pathwayCount": len(pathways), + "pathways": [compact_pathway_ref(pathway) for pathway in pathways], + } + ) + return summary + + +def has_audiences_with_pathways(state: dict[str, Any]) -> bool: + return bool(audience_pathway_pairs(state)) + + def downstream_limit(ctx: FlowContext, *, audience: bool) -> int: """Limit expensive downstream analysis while keeping 3 upstream rows stored.""" raw = ctx.downstream_audiences if audience else ctx.downstream_pathways @@ -2066,22 +2149,35 @@ def downstream_items(ctx: FlowContext, state: dict[str, Any], stage: str, *, aud source_key = "topAudiences" if audience else "topPathways" id_key = "designId" if audience else "pathwayAccountId" selected_key = "selectedDownstreamAudiences" if audience else "selectedDownstreamPathways" - items = [item for item in (state.get(source_key) or []) if isinstance(item, dict)] + all_items = [item for item in (state.get(source_key) or []) if isinstance(item, dict)] + if audience: + state["audiencesWithPathways"] = audience_pathway_summary(state) + items = [item for item, _ in audience_pathway_pairs(state)] + if not items: + state[selected_key] = [] + record_error(state, stage, "没有已分析出关键传播节点的受众,不能继续分析受众脆弱点或策略;请先重试关键传播节点分析") + return [] + else: + items = all_items limit = downstream_limit(ctx, audience=audience) selected = items[:limit] - state[selected_key] = [ - { + selected_rows: list[dict[str, Any]] = [] + for item in selected: + row = { "id": item.get(id_key), "name": item.get("targetAudience") if audience else item.get("accountName"), } - for item in selected - ] - if items: + if audience: + row["pathwayCount"] = len(item.get("downstreamPathways") or []) + row["pathways"] = item.get("downstreamPathways") or [] + selected_rows.append(row) + state[selected_key] = selected_rows + if selected: record_event( state, stage, - f"下游分析选择 {len(selected)}/{len(items)} 个{'受众' if audience else '传播节点'}", - {"limit": limit, "selected": state[selected_key]}, + f"下游分析选择 {len(selected)}/{len(items)} 个{'已产生传播节点的受众' if audience else '传播节点'}", + {"limit": limit, "selected": state[selected_key], "candidateCount": len(items), "totalStoredCount": len(all_items)}, ) return selected @@ -2183,6 +2279,7 @@ def run_step2(ctx: FlowContext, state: dict[str, Any]) -> None: record_error(state, stage, "没有已入库受众,无法分析传播节点") return if state.get("topPathways") and not ctx.dry_run: + state["audiencesWithPathways"] = audience_pathway_summary(state) record_event(state, stage, "topPathways 已存在,跳过。如需重跑请清空 state 中该字段。") return all_candidates: list[dict[str, Any]] = [] @@ -2278,9 +2375,10 @@ def run_step2(ctx: FlowContext, state: dict[str, Any]) -> None: http_post_json(ctx, state, "save_relation", relation_payload, stage + "/save-relation") saved.append(item) state["topPathways"] = saved + state["audiencesWithPathways"] = audience_pathway_summary(state) if saved: mark_step_status(ctx, state, ["step-analysis-keypoint", "step-analysis-weakpoint"], stage + "/step-status") - record_event(state, stage, f"完成,保存传播节点 {len(saved)} 个") + record_event(state, stage, f"完成,保存传播节点 {len(saved)} 个,可继续下游分析的受众 {len(state.get('audiencesWithPathways') or [])} 个") def run_step3(ctx: FlowContext, state: dict[str, Any]) -> None: diff --git a/skills-data/xdfx-step2-pathway/scripts/xdfx_flow.py b/skills-data/xdfx-step2-pathway/scripts/xdfx_flow.py index d9580e2f..8e62e51e 100644 --- a/skills-data/xdfx-step2-pathway/scripts/xdfx_flow.py +++ b/skills-data/xdfx-step2-pathway/scripts/xdfx_flow.py @@ -368,6 +368,7 @@ def default_state() -> dict[str, Any]: "topAudiences": [], "pathwayCandidates": [], "topPathways": [], + "audiencesWithPathways": [], "audienceVulns": {}, "pathwayVulns": {}, "audienceStrategies": [], @@ -2055,6 +2056,88 @@ def selected_accounts_for_item(item: dict[str, Any], *, audience: bool) -> list[ return [item.get("account") or item] +def id_ref_key(value: Any) -> str: + number = as_int(value) + if number is not None: + return str(number) + return str(value or "").strip() + + +def audience_name_key(value: Any) -> str: + return re.sub(r"\s+", "", str(value or "")).lower() + + +def compact_pathway_ref(pathway: dict[str, Any]) -> dict[str, Any]: + return { + "pathwayAccountId": pathway.get("pathwayAccountId") or pathway.get("pathwayId"), + "accountId": pathway.get("accountId"), + "accountName": pathway.get("accountName"), + } + + +def add_unique_pathway(target: list[dict[str, Any]], pathway: dict[str, Any]) -> None: + key = id_ref_key(pathway.get("pathwayAccountId") or pathway.get("pathwayId") or pathway.get("accountId") or pathway.get("accountName")) + if not key: + return + seen = { + id_ref_key(item.get("pathwayAccountId") or item.get("pathwayId") or item.get("accountId") or item.get("accountName")) + for item in target + } + if key not in seen: + target.append(pathway) + + +def audience_pathway_pairs(state: dict[str, Any]) -> list[tuple[dict[str, Any], list[dict[str, Any]]]]: + """Return only audiences that produced at least one saved key pathway.""" + id_links: dict[str, list[dict[str, Any]]] = {} + name_links: dict[str, list[dict[str, Any]]] = {} + for pathway in state.get("topPathways") or []: + if not isinstance(pathway, dict): + continue + for design_id in pathway.get("designIds") or []: + key = id_ref_key(design_id) + if key: + id_links.setdefault(key, []).append(pathway) + for name in pathway.get("relatedAudiences") or []: + key = audience_name_key(name) + if key: + name_links.setdefault(key, []).append(pathway) + + pairs: list[tuple[dict[str, Any], list[dict[str, Any]]]] = [] + for audience_item in state.get("topAudiences") or []: + if not isinstance(audience_item, dict): + continue + linked: list[dict[str, Any]] = [] + design_key = id_ref_key(audience_item.get("designId")) + for pathway in id_links.get(design_key, []): + add_unique_pathway(linked, pathway) + name_key = audience_name_key(audience_item.get("targetAudience")) + for pathway in name_links.get(name_key, []): + add_unique_pathway(linked, pathway) + if linked: + item = {**audience_item, "downstreamPathways": [compact_pathway_ref(pathway) for pathway in linked]} + pairs.append((item, linked)) + return pairs + + +def audience_pathway_summary(state: dict[str, Any]) -> list[dict[str, Any]]: + summary: list[dict[str, Any]] = [] + for audience_item, pathways in audience_pathway_pairs(state): + summary.append( + { + "designId": audience_item.get("designId"), + "targetAudience": audience_item.get("targetAudience"), + "pathwayCount": len(pathways), + "pathways": [compact_pathway_ref(pathway) for pathway in pathways], + } + ) + return summary + + +def has_audiences_with_pathways(state: dict[str, Any]) -> bool: + return bool(audience_pathway_pairs(state)) + + def downstream_limit(ctx: FlowContext, *, audience: bool) -> int: """Limit expensive downstream analysis while keeping 3 upstream rows stored.""" raw = ctx.downstream_audiences if audience else ctx.downstream_pathways @@ -2066,22 +2149,35 @@ def downstream_items(ctx: FlowContext, state: dict[str, Any], stage: str, *, aud source_key = "topAudiences" if audience else "topPathways" id_key = "designId" if audience else "pathwayAccountId" selected_key = "selectedDownstreamAudiences" if audience else "selectedDownstreamPathways" - items = [item for item in (state.get(source_key) or []) if isinstance(item, dict)] + all_items = [item for item in (state.get(source_key) or []) if isinstance(item, dict)] + if audience: + state["audiencesWithPathways"] = audience_pathway_summary(state) + items = [item for item, _ in audience_pathway_pairs(state)] + if not items: + state[selected_key] = [] + record_error(state, stage, "没有已分析出关键传播节点的受众,不能继续分析受众脆弱点或策略;请先重试关键传播节点分析") + return [] + else: + items = all_items limit = downstream_limit(ctx, audience=audience) selected = items[:limit] - state[selected_key] = [ - { + selected_rows: list[dict[str, Any]] = [] + for item in selected: + row = { "id": item.get(id_key), "name": item.get("targetAudience") if audience else item.get("accountName"), } - for item in selected - ] - if items: + if audience: + row["pathwayCount"] = len(item.get("downstreamPathways") or []) + row["pathways"] = item.get("downstreamPathways") or [] + selected_rows.append(row) + state[selected_key] = selected_rows + if selected: record_event( state, stage, - f"下游分析选择 {len(selected)}/{len(items)} 个{'受众' if audience else '传播节点'}", - {"limit": limit, "selected": state[selected_key]}, + f"下游分析选择 {len(selected)}/{len(items)} 个{'已产生传播节点的受众' if audience else '传播节点'}", + {"limit": limit, "selected": state[selected_key], "candidateCount": len(items), "totalStoredCount": len(all_items)}, ) return selected @@ -2183,6 +2279,7 @@ def run_step2(ctx: FlowContext, state: dict[str, Any]) -> None: record_error(state, stage, "没有已入库受众,无法分析传播节点") return if state.get("topPathways") and not ctx.dry_run: + state["audiencesWithPathways"] = audience_pathway_summary(state) record_event(state, stage, "topPathways 已存在,跳过。如需重跑请清空 state 中该字段。") return all_candidates: list[dict[str, Any]] = [] @@ -2278,9 +2375,10 @@ def run_step2(ctx: FlowContext, state: dict[str, Any]) -> None: http_post_json(ctx, state, "save_relation", relation_payload, stage + "/save-relation") saved.append(item) state["topPathways"] = saved + state["audiencesWithPathways"] = audience_pathway_summary(state) if saved: mark_step_status(ctx, state, ["step-analysis-keypoint", "step-analysis-weakpoint"], stage + "/step-status") - record_event(state, stage, f"完成,保存传播节点 {len(saved)} 个") + record_event(state, stage, f"完成,保存传播节点 {len(saved)} 个,可继续下游分析的受众 {len(state.get('audiencesWithPathways') or [])} 个") def run_step3(ctx: FlowContext, state: dict[str, Any]) -> None: diff --git a/skills-data/xdfx-step3-audience-vulns/scripts/xdfx_flow.py b/skills-data/xdfx-step3-audience-vulns/scripts/xdfx_flow.py index d9580e2f..8e62e51e 100644 --- a/skills-data/xdfx-step3-audience-vulns/scripts/xdfx_flow.py +++ b/skills-data/xdfx-step3-audience-vulns/scripts/xdfx_flow.py @@ -368,6 +368,7 @@ def default_state() -> dict[str, Any]: "topAudiences": [], "pathwayCandidates": [], "topPathways": [], + "audiencesWithPathways": [], "audienceVulns": {}, "pathwayVulns": {}, "audienceStrategies": [], @@ -2055,6 +2056,88 @@ def selected_accounts_for_item(item: dict[str, Any], *, audience: bool) -> list[ return [item.get("account") or item] +def id_ref_key(value: Any) -> str: + number = as_int(value) + if number is not None: + return str(number) + return str(value or "").strip() + + +def audience_name_key(value: Any) -> str: + return re.sub(r"\s+", "", str(value or "")).lower() + + +def compact_pathway_ref(pathway: dict[str, Any]) -> dict[str, Any]: + return { + "pathwayAccountId": pathway.get("pathwayAccountId") or pathway.get("pathwayId"), + "accountId": pathway.get("accountId"), + "accountName": pathway.get("accountName"), + } + + +def add_unique_pathway(target: list[dict[str, Any]], pathway: dict[str, Any]) -> None: + key = id_ref_key(pathway.get("pathwayAccountId") or pathway.get("pathwayId") or pathway.get("accountId") or pathway.get("accountName")) + if not key: + return + seen = { + id_ref_key(item.get("pathwayAccountId") or item.get("pathwayId") or item.get("accountId") or item.get("accountName")) + for item in target + } + if key not in seen: + target.append(pathway) + + +def audience_pathway_pairs(state: dict[str, Any]) -> list[tuple[dict[str, Any], list[dict[str, Any]]]]: + """Return only audiences that produced at least one saved key pathway.""" + id_links: dict[str, list[dict[str, Any]]] = {} + name_links: dict[str, list[dict[str, Any]]] = {} + for pathway in state.get("topPathways") or []: + if not isinstance(pathway, dict): + continue + for design_id in pathway.get("designIds") or []: + key = id_ref_key(design_id) + if key: + id_links.setdefault(key, []).append(pathway) + for name in pathway.get("relatedAudiences") or []: + key = audience_name_key(name) + if key: + name_links.setdefault(key, []).append(pathway) + + pairs: list[tuple[dict[str, Any], list[dict[str, Any]]]] = [] + for audience_item in state.get("topAudiences") or []: + if not isinstance(audience_item, dict): + continue + linked: list[dict[str, Any]] = [] + design_key = id_ref_key(audience_item.get("designId")) + for pathway in id_links.get(design_key, []): + add_unique_pathway(linked, pathway) + name_key = audience_name_key(audience_item.get("targetAudience")) + for pathway in name_links.get(name_key, []): + add_unique_pathway(linked, pathway) + if linked: + item = {**audience_item, "downstreamPathways": [compact_pathway_ref(pathway) for pathway in linked]} + pairs.append((item, linked)) + return pairs + + +def audience_pathway_summary(state: dict[str, Any]) -> list[dict[str, Any]]: + summary: list[dict[str, Any]] = [] + for audience_item, pathways in audience_pathway_pairs(state): + summary.append( + { + "designId": audience_item.get("designId"), + "targetAudience": audience_item.get("targetAudience"), + "pathwayCount": len(pathways), + "pathways": [compact_pathway_ref(pathway) for pathway in pathways], + } + ) + return summary + + +def has_audiences_with_pathways(state: dict[str, Any]) -> bool: + return bool(audience_pathway_pairs(state)) + + def downstream_limit(ctx: FlowContext, *, audience: bool) -> int: """Limit expensive downstream analysis while keeping 3 upstream rows stored.""" raw = ctx.downstream_audiences if audience else ctx.downstream_pathways @@ -2066,22 +2149,35 @@ def downstream_items(ctx: FlowContext, state: dict[str, Any], stage: str, *, aud source_key = "topAudiences" if audience else "topPathways" id_key = "designId" if audience else "pathwayAccountId" selected_key = "selectedDownstreamAudiences" if audience else "selectedDownstreamPathways" - items = [item for item in (state.get(source_key) or []) if isinstance(item, dict)] + all_items = [item for item in (state.get(source_key) or []) if isinstance(item, dict)] + if audience: + state["audiencesWithPathways"] = audience_pathway_summary(state) + items = [item for item, _ in audience_pathway_pairs(state)] + if not items: + state[selected_key] = [] + record_error(state, stage, "没有已分析出关键传播节点的受众,不能继续分析受众脆弱点或策略;请先重试关键传播节点分析") + return [] + else: + items = all_items limit = downstream_limit(ctx, audience=audience) selected = items[:limit] - state[selected_key] = [ - { + selected_rows: list[dict[str, Any]] = [] + for item in selected: + row = { "id": item.get(id_key), "name": item.get("targetAudience") if audience else item.get("accountName"), } - for item in selected - ] - if items: + if audience: + row["pathwayCount"] = len(item.get("downstreamPathways") or []) + row["pathways"] = item.get("downstreamPathways") or [] + selected_rows.append(row) + state[selected_key] = selected_rows + if selected: record_event( state, stage, - f"下游分析选择 {len(selected)}/{len(items)} 个{'受众' if audience else '传播节点'}", - {"limit": limit, "selected": state[selected_key]}, + f"下游分析选择 {len(selected)}/{len(items)} 个{'已产生传播节点的受众' if audience else '传播节点'}", + {"limit": limit, "selected": state[selected_key], "candidateCount": len(items), "totalStoredCount": len(all_items)}, ) return selected @@ -2183,6 +2279,7 @@ def run_step2(ctx: FlowContext, state: dict[str, Any]) -> None: record_error(state, stage, "没有已入库受众,无法分析传播节点") return if state.get("topPathways") and not ctx.dry_run: + state["audiencesWithPathways"] = audience_pathway_summary(state) record_event(state, stage, "topPathways 已存在,跳过。如需重跑请清空 state 中该字段。") return all_candidates: list[dict[str, Any]] = [] @@ -2278,9 +2375,10 @@ def run_step2(ctx: FlowContext, state: dict[str, Any]) -> None: http_post_json(ctx, state, "save_relation", relation_payload, stage + "/save-relation") saved.append(item) state["topPathways"] = saved + state["audiencesWithPathways"] = audience_pathway_summary(state) if saved: mark_step_status(ctx, state, ["step-analysis-keypoint", "step-analysis-weakpoint"], stage + "/step-status") - record_event(state, stage, f"完成,保存传播节点 {len(saved)} 个") + record_event(state, stage, f"完成,保存传播节点 {len(saved)} 个,可继续下游分析的受众 {len(state.get('audiencesWithPathways') or [])} 个") def run_step3(ctx: FlowContext, state: dict[str, Any]) -> None: diff --git a/skills-data/xdfx-step4-pathway-vulns/scripts/xdfx_flow.py b/skills-data/xdfx-step4-pathway-vulns/scripts/xdfx_flow.py index d9580e2f..8e62e51e 100644 --- a/skills-data/xdfx-step4-pathway-vulns/scripts/xdfx_flow.py +++ b/skills-data/xdfx-step4-pathway-vulns/scripts/xdfx_flow.py @@ -368,6 +368,7 @@ def default_state() -> dict[str, Any]: "topAudiences": [], "pathwayCandidates": [], "topPathways": [], + "audiencesWithPathways": [], "audienceVulns": {}, "pathwayVulns": {}, "audienceStrategies": [], @@ -2055,6 +2056,88 @@ def selected_accounts_for_item(item: dict[str, Any], *, audience: bool) -> list[ return [item.get("account") or item] +def id_ref_key(value: Any) -> str: + number = as_int(value) + if number is not None: + return str(number) + return str(value or "").strip() + + +def audience_name_key(value: Any) -> str: + return re.sub(r"\s+", "", str(value or "")).lower() + + +def compact_pathway_ref(pathway: dict[str, Any]) -> dict[str, Any]: + return { + "pathwayAccountId": pathway.get("pathwayAccountId") or pathway.get("pathwayId"), + "accountId": pathway.get("accountId"), + "accountName": pathway.get("accountName"), + } + + +def add_unique_pathway(target: list[dict[str, Any]], pathway: dict[str, Any]) -> None: + key = id_ref_key(pathway.get("pathwayAccountId") or pathway.get("pathwayId") or pathway.get("accountId") or pathway.get("accountName")) + if not key: + return + seen = { + id_ref_key(item.get("pathwayAccountId") or item.get("pathwayId") or item.get("accountId") or item.get("accountName")) + for item in target + } + if key not in seen: + target.append(pathway) + + +def audience_pathway_pairs(state: dict[str, Any]) -> list[tuple[dict[str, Any], list[dict[str, Any]]]]: + """Return only audiences that produced at least one saved key pathway.""" + id_links: dict[str, list[dict[str, Any]]] = {} + name_links: dict[str, list[dict[str, Any]]] = {} + for pathway in state.get("topPathways") or []: + if not isinstance(pathway, dict): + continue + for design_id in pathway.get("designIds") or []: + key = id_ref_key(design_id) + if key: + id_links.setdefault(key, []).append(pathway) + for name in pathway.get("relatedAudiences") or []: + key = audience_name_key(name) + if key: + name_links.setdefault(key, []).append(pathway) + + pairs: list[tuple[dict[str, Any], list[dict[str, Any]]]] = [] + for audience_item in state.get("topAudiences") or []: + if not isinstance(audience_item, dict): + continue + linked: list[dict[str, Any]] = [] + design_key = id_ref_key(audience_item.get("designId")) + for pathway in id_links.get(design_key, []): + add_unique_pathway(linked, pathway) + name_key = audience_name_key(audience_item.get("targetAudience")) + for pathway in name_links.get(name_key, []): + add_unique_pathway(linked, pathway) + if linked: + item = {**audience_item, "downstreamPathways": [compact_pathway_ref(pathway) for pathway in linked]} + pairs.append((item, linked)) + return pairs + + +def audience_pathway_summary(state: dict[str, Any]) -> list[dict[str, Any]]: + summary: list[dict[str, Any]] = [] + for audience_item, pathways in audience_pathway_pairs(state): + summary.append( + { + "designId": audience_item.get("designId"), + "targetAudience": audience_item.get("targetAudience"), + "pathwayCount": len(pathways), + "pathways": [compact_pathway_ref(pathway) for pathway in pathways], + } + ) + return summary + + +def has_audiences_with_pathways(state: dict[str, Any]) -> bool: + return bool(audience_pathway_pairs(state)) + + def downstream_limit(ctx: FlowContext, *, audience: bool) -> int: """Limit expensive downstream analysis while keeping 3 upstream rows stored.""" raw = ctx.downstream_audiences if audience else ctx.downstream_pathways @@ -2066,22 +2149,35 @@ def downstream_items(ctx: FlowContext, state: dict[str, Any], stage: str, *, aud source_key = "topAudiences" if audience else "topPathways" id_key = "designId" if audience else "pathwayAccountId" selected_key = "selectedDownstreamAudiences" if audience else "selectedDownstreamPathways" - items = [item for item in (state.get(source_key) or []) if isinstance(item, dict)] + all_items = [item for item in (state.get(source_key) or []) if isinstance(item, dict)] + if audience: + state["audiencesWithPathways"] = audience_pathway_summary(state) + items = [item for item, _ in audience_pathway_pairs(state)] + if not items: + state[selected_key] = [] + record_error(state, stage, "没有已分析出关键传播节点的受众,不能继续分析受众脆弱点或策略;请先重试关键传播节点分析") + return [] + else: + items = all_items limit = downstream_limit(ctx, audience=audience) selected = items[:limit] - state[selected_key] = [ - { + selected_rows: list[dict[str, Any]] = [] + for item in selected: + row = { "id": item.get(id_key), "name": item.get("targetAudience") if audience else item.get("accountName"), } - for item in selected - ] - if items: + if audience: + row["pathwayCount"] = len(item.get("downstreamPathways") or []) + row["pathways"] = item.get("downstreamPathways") or [] + selected_rows.append(row) + state[selected_key] = selected_rows + if selected: record_event( state, stage, - f"下游分析选择 {len(selected)}/{len(items)} 个{'受众' if audience else '传播节点'}", - {"limit": limit, "selected": state[selected_key]}, + f"下游分析选择 {len(selected)}/{len(items)} 个{'已产生传播节点的受众' if audience else '传播节点'}", + {"limit": limit, "selected": state[selected_key], "candidateCount": len(items), "totalStoredCount": len(all_items)}, ) return selected @@ -2183,6 +2279,7 @@ def run_step2(ctx: FlowContext, state: dict[str, Any]) -> None: record_error(state, stage, "没有已入库受众,无法分析传播节点") return if state.get("topPathways") and not ctx.dry_run: + state["audiencesWithPathways"] = audience_pathway_summary(state) record_event(state, stage, "topPathways 已存在,跳过。如需重跑请清空 state 中该字段。") return all_candidates: list[dict[str, Any]] = [] @@ -2278,9 +2375,10 @@ def run_step2(ctx: FlowContext, state: dict[str, Any]) -> None: http_post_json(ctx, state, "save_relation", relation_payload, stage + "/save-relation") saved.append(item) state["topPathways"] = saved + state["audiencesWithPathways"] = audience_pathway_summary(state) if saved: mark_step_status(ctx, state, ["step-analysis-keypoint", "step-analysis-weakpoint"], stage + "/step-status") - record_event(state, stage, f"完成,保存传播节点 {len(saved)} 个") + record_event(state, stage, f"完成,保存传播节点 {len(saved)} 个,可继续下游分析的受众 {len(state.get('audiencesWithPathways') or [])} 个") def run_step3(ctx: FlowContext, state: dict[str, Any]) -> None: diff --git a/skills-data/xdfx-step5-audience-strategy/scripts/xdfx_flow.py b/skills-data/xdfx-step5-audience-strategy/scripts/xdfx_flow.py index d9580e2f..8e62e51e 100644 --- a/skills-data/xdfx-step5-audience-strategy/scripts/xdfx_flow.py +++ b/skills-data/xdfx-step5-audience-strategy/scripts/xdfx_flow.py @@ -368,6 +368,7 @@ def default_state() -> dict[str, Any]: "topAudiences": [], "pathwayCandidates": [], "topPathways": [], + "audiencesWithPathways": [], "audienceVulns": {}, "pathwayVulns": {}, "audienceStrategies": [], @@ -2055,6 +2056,88 @@ def selected_accounts_for_item(item: dict[str, Any], *, audience: bool) -> list[ return [item.get("account") or item] +def id_ref_key(value: Any) -> str: + number = as_int(value) + if number is not None: + return str(number) + return str(value or "").strip() + + +def audience_name_key(value: Any) -> str: + return re.sub(r"\s+", "", str(value or "")).lower() + + +def compact_pathway_ref(pathway: dict[str, Any]) -> dict[str, Any]: + return { + "pathwayAccountId": pathway.get("pathwayAccountId") or pathway.get("pathwayId"), + "accountId": pathway.get("accountId"), + "accountName": pathway.get("accountName"), + } + + +def add_unique_pathway(target: list[dict[str, Any]], pathway: dict[str, Any]) -> None: + key = id_ref_key(pathway.get("pathwayAccountId") or pathway.get("pathwayId") or pathway.get("accountId") or pathway.get("accountName")) + if not key: + return + seen = { + id_ref_key(item.get("pathwayAccountId") or item.get("pathwayId") or item.get("accountId") or item.get("accountName")) + for item in target + } + if key not in seen: + target.append(pathway) + + +def audience_pathway_pairs(state: dict[str, Any]) -> list[tuple[dict[str, Any], list[dict[str, Any]]]]: + """Return only audiences that produced at least one saved key pathway.""" + id_links: dict[str, list[dict[str, Any]]] = {} + name_links: dict[str, list[dict[str, Any]]] = {} + for pathway in state.get("topPathways") or []: + if not isinstance(pathway, dict): + continue + for design_id in pathway.get("designIds") or []: + key = id_ref_key(design_id) + if key: + id_links.setdefault(key, []).append(pathway) + for name in pathway.get("relatedAudiences") or []: + key = audience_name_key(name) + if key: + name_links.setdefault(key, []).append(pathway) + + pairs: list[tuple[dict[str, Any], list[dict[str, Any]]]] = [] + for audience_item in state.get("topAudiences") or []: + if not isinstance(audience_item, dict): + continue + linked: list[dict[str, Any]] = [] + design_key = id_ref_key(audience_item.get("designId")) + for pathway in id_links.get(design_key, []): + add_unique_pathway(linked, pathway) + name_key = audience_name_key(audience_item.get("targetAudience")) + for pathway in name_links.get(name_key, []): + add_unique_pathway(linked, pathway) + if linked: + item = {**audience_item, "downstreamPathways": [compact_pathway_ref(pathway) for pathway in linked]} + pairs.append((item, linked)) + return pairs + + +def audience_pathway_summary(state: dict[str, Any]) -> list[dict[str, Any]]: + summary: list[dict[str, Any]] = [] + for audience_item, pathways in audience_pathway_pairs(state): + summary.append( + { + "designId": audience_item.get("designId"), + "targetAudience": audience_item.get("targetAudience"), + "pathwayCount": len(pathways), + "pathways": [compact_pathway_ref(pathway) for pathway in pathways], + } + ) + return summary + + +def has_audiences_with_pathways(state: dict[str, Any]) -> bool: + return bool(audience_pathway_pairs(state)) + + def downstream_limit(ctx: FlowContext, *, audience: bool) -> int: """Limit expensive downstream analysis while keeping 3 upstream rows stored.""" raw = ctx.downstream_audiences if audience else ctx.downstream_pathways @@ -2066,22 +2149,35 @@ def downstream_items(ctx: FlowContext, state: dict[str, Any], stage: str, *, aud source_key = "topAudiences" if audience else "topPathways" id_key = "designId" if audience else "pathwayAccountId" selected_key = "selectedDownstreamAudiences" if audience else "selectedDownstreamPathways" - items = [item for item in (state.get(source_key) or []) if isinstance(item, dict)] + all_items = [item for item in (state.get(source_key) or []) if isinstance(item, dict)] + if audience: + state["audiencesWithPathways"] = audience_pathway_summary(state) + items = [item for item, _ in audience_pathway_pairs(state)] + if not items: + state[selected_key] = [] + record_error(state, stage, "没有已分析出关键传播节点的受众,不能继续分析受众脆弱点或策略;请先重试关键传播节点分析") + return [] + else: + items = all_items limit = downstream_limit(ctx, audience=audience) selected = items[:limit] - state[selected_key] = [ - { + selected_rows: list[dict[str, Any]] = [] + for item in selected: + row = { "id": item.get(id_key), "name": item.get("targetAudience") if audience else item.get("accountName"), } - for item in selected - ] - if items: + if audience: + row["pathwayCount"] = len(item.get("downstreamPathways") or []) + row["pathways"] = item.get("downstreamPathways") or [] + selected_rows.append(row) + state[selected_key] = selected_rows + if selected: record_event( state, stage, - f"下游分析选择 {len(selected)}/{len(items)} 个{'受众' if audience else '传播节点'}", - {"limit": limit, "selected": state[selected_key]}, + f"下游分析选择 {len(selected)}/{len(items)} 个{'已产生传播节点的受众' if audience else '传播节点'}", + {"limit": limit, "selected": state[selected_key], "candidateCount": len(items), "totalStoredCount": len(all_items)}, ) return selected @@ -2183,6 +2279,7 @@ def run_step2(ctx: FlowContext, state: dict[str, Any]) -> None: record_error(state, stage, "没有已入库受众,无法分析传播节点") return if state.get("topPathways") and not ctx.dry_run: + state["audiencesWithPathways"] = audience_pathway_summary(state) record_event(state, stage, "topPathways 已存在,跳过。如需重跑请清空 state 中该字段。") return all_candidates: list[dict[str, Any]] = [] @@ -2278,9 +2375,10 @@ def run_step2(ctx: FlowContext, state: dict[str, Any]) -> None: http_post_json(ctx, state, "save_relation", relation_payload, stage + "/save-relation") saved.append(item) state["topPathways"] = saved + state["audiencesWithPathways"] = audience_pathway_summary(state) if saved: mark_step_status(ctx, state, ["step-analysis-keypoint", "step-analysis-weakpoint"], stage + "/step-status") - record_event(state, stage, f"完成,保存传播节点 {len(saved)} 个") + record_event(state, stage, f"完成,保存传播节点 {len(saved)} 个,可继续下游分析的受众 {len(state.get('audiencesWithPathways') or [])} 个") def run_step3(ctx: FlowContext, state: dict[str, Any]) -> None: diff --git a/skills-data/xdfx-step6-pathway-strategy/scripts/xdfx_flow.py b/skills-data/xdfx-step6-pathway-strategy/scripts/xdfx_flow.py index d9580e2f..8e62e51e 100644 --- a/skills-data/xdfx-step6-pathway-strategy/scripts/xdfx_flow.py +++ b/skills-data/xdfx-step6-pathway-strategy/scripts/xdfx_flow.py @@ -368,6 +368,7 @@ def default_state() -> dict[str, Any]: "topAudiences": [], "pathwayCandidates": [], "topPathways": [], + "audiencesWithPathways": [], "audienceVulns": {}, "pathwayVulns": {}, "audienceStrategies": [], @@ -2055,6 +2056,88 @@ def selected_accounts_for_item(item: dict[str, Any], *, audience: bool) -> list[ return [item.get("account") or item] +def id_ref_key(value: Any) -> str: + number = as_int(value) + if number is not None: + return str(number) + return str(value or "").strip() + + +def audience_name_key(value: Any) -> str: + return re.sub(r"\s+", "", str(value or "")).lower() + + +def compact_pathway_ref(pathway: dict[str, Any]) -> dict[str, Any]: + return { + "pathwayAccountId": pathway.get("pathwayAccountId") or pathway.get("pathwayId"), + "accountId": pathway.get("accountId"), + "accountName": pathway.get("accountName"), + } + + +def add_unique_pathway(target: list[dict[str, Any]], pathway: dict[str, Any]) -> None: + key = id_ref_key(pathway.get("pathwayAccountId") or pathway.get("pathwayId") or pathway.get("accountId") or pathway.get("accountName")) + if not key: + return + seen = { + id_ref_key(item.get("pathwayAccountId") or item.get("pathwayId") or item.get("accountId") or item.get("accountName")) + for item in target + } + if key not in seen: + target.append(pathway) + + +def audience_pathway_pairs(state: dict[str, Any]) -> list[tuple[dict[str, Any], list[dict[str, Any]]]]: + """Return only audiences that produced at least one saved key pathway.""" + id_links: dict[str, list[dict[str, Any]]] = {} + name_links: dict[str, list[dict[str, Any]]] = {} + for pathway in state.get("topPathways") or []: + if not isinstance(pathway, dict): + continue + for design_id in pathway.get("designIds") or []: + key = id_ref_key(design_id) + if key: + id_links.setdefault(key, []).append(pathway) + for name in pathway.get("relatedAudiences") or []: + key = audience_name_key(name) + if key: + name_links.setdefault(key, []).append(pathway) + + pairs: list[tuple[dict[str, Any], list[dict[str, Any]]]] = [] + for audience_item in state.get("topAudiences") or []: + if not isinstance(audience_item, dict): + continue + linked: list[dict[str, Any]] = [] + design_key = id_ref_key(audience_item.get("designId")) + for pathway in id_links.get(design_key, []): + add_unique_pathway(linked, pathway) + name_key = audience_name_key(audience_item.get("targetAudience")) + for pathway in name_links.get(name_key, []): + add_unique_pathway(linked, pathway) + if linked: + item = {**audience_item, "downstreamPathways": [compact_pathway_ref(pathway) for pathway in linked]} + pairs.append((item, linked)) + return pairs + + +def audience_pathway_summary(state: dict[str, Any]) -> list[dict[str, Any]]: + summary: list[dict[str, Any]] = [] + for audience_item, pathways in audience_pathway_pairs(state): + summary.append( + { + "designId": audience_item.get("designId"), + "targetAudience": audience_item.get("targetAudience"), + "pathwayCount": len(pathways), + "pathways": [compact_pathway_ref(pathway) for pathway in pathways], + } + ) + return summary + + +def has_audiences_with_pathways(state: dict[str, Any]) -> bool: + return bool(audience_pathway_pairs(state)) + + def downstream_limit(ctx: FlowContext, *, audience: bool) -> int: """Limit expensive downstream analysis while keeping 3 upstream rows stored.""" raw = ctx.downstream_audiences if audience else ctx.downstream_pathways @@ -2066,22 +2149,35 @@ def downstream_items(ctx: FlowContext, state: dict[str, Any], stage: str, *, aud source_key = "topAudiences" if audience else "topPathways" id_key = "designId" if audience else "pathwayAccountId" selected_key = "selectedDownstreamAudiences" if audience else "selectedDownstreamPathways" - items = [item for item in (state.get(source_key) or []) if isinstance(item, dict)] + all_items = [item for item in (state.get(source_key) or []) if isinstance(item, dict)] + if audience: + state["audiencesWithPathways"] = audience_pathway_summary(state) + items = [item for item, _ in audience_pathway_pairs(state)] + if not items: + state[selected_key] = [] + record_error(state, stage, "没有已分析出关键传播节点的受众,不能继续分析受众脆弱点或策略;请先重试关键传播节点分析") + return [] + else: + items = all_items limit = downstream_limit(ctx, audience=audience) selected = items[:limit] - state[selected_key] = [ - { + selected_rows: list[dict[str, Any]] = [] + for item in selected: + row = { "id": item.get(id_key), "name": item.get("targetAudience") if audience else item.get("accountName"), } - for item in selected - ] - if items: + if audience: + row["pathwayCount"] = len(item.get("downstreamPathways") or []) + row["pathways"] = item.get("downstreamPathways") or [] + selected_rows.append(row) + state[selected_key] = selected_rows + if selected: record_event( state, stage, - f"下游分析选择 {len(selected)}/{len(items)} 个{'受众' if audience else '传播节点'}", - {"limit": limit, "selected": state[selected_key]}, + f"下游分析选择 {len(selected)}/{len(items)} 个{'已产生传播节点的受众' if audience else '传播节点'}", + {"limit": limit, "selected": state[selected_key], "candidateCount": len(items), "totalStoredCount": len(all_items)}, ) return selected @@ -2183,6 +2279,7 @@ def run_step2(ctx: FlowContext, state: dict[str, Any]) -> None: record_error(state, stage, "没有已入库受众,无法分析传播节点") return if state.get("topPathways") and not ctx.dry_run: + state["audiencesWithPathways"] = audience_pathway_summary(state) record_event(state, stage, "topPathways 已存在,跳过。如需重跑请清空 state 中该字段。") return all_candidates: list[dict[str, Any]] = [] @@ -2278,9 +2375,10 @@ def run_step2(ctx: FlowContext, state: dict[str, Any]) -> None: http_post_json(ctx, state, "save_relation", relation_payload, stage + "/save-relation") saved.append(item) state["topPathways"] = saved + state["audiencesWithPathways"] = audience_pathway_summary(state) if saved: mark_step_status(ctx, state, ["step-analysis-keypoint", "step-analysis-weakpoint"], stage + "/step-status") - record_event(state, stage, f"完成,保存传播节点 {len(saved)} 个") + record_event(state, stage, f"完成,保存传播节点 {len(saved)} 个,可继续下游分析的受众 {len(state.get('audiencesWithPathways') or [])} 个") def run_step3(ctx: FlowContext, state: dict[str, Any]) -> None: