技能修改

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: