feat: 为DAG触发添加运行预览详情，优化配置加载与默认数据集 (7a28f4fe) · Commits · csst-cicd / csst-dag

csst_dag/v2/config_loader.py

+15 −6

Original line number	Diff line number	Diff line
		@@ -62,16 +62,17 @@ def _extract_field_default(field_cfg: dict) -> object:
		return _MISSING


		def _normalize_field_group(field_group: dict \| None) -> tuple[dict, dict, dict, dict, dict, dict]:
		"""将字段中心的 submission 配置展开为 defaults/fixed/options/allow_empty/option_sources/default_strategies 六张表。"""
		def _normalize_field_group(field_group: dict \| None) -> tuple[dict, dict, dict, dict, dict, dict, dict]:
		"""将字段中心的 submission 配置展开为 defaults/fixed/options/allow_empty/option_sources/default_strategies/placeholders 七张表。"""
		defaults: dict = {}
		fixed: dict = {}
		options: dict = {}
		allow_empty: dict = {}
		option_sources: dict = {}
		default_strategies: dict = {}
		placeholders: dict = {}
		if not isinstance(field_group, dict):
		return defaults, fixed, options, allow_empty, option_sources, default_strategies
		return defaults, fixed, options, allow_empty, option_sources, default_strategies, placeholders

		for field_name, field_cfg in field_group.items():
		if isinstance(field_cfg, dict):
		@@ -91,10 +92,12 @@ def _normalize_field_group(field_group: dict \| None) -> tuple[dict, dict, dict,
		default_strategy = str(field_cfg.get("default_strategy") or "").strip()
		if default_strategy:
		default_strategies[field_name] = default_strategy
		if "placeholder" in field_cfg:
		placeholders[field_name] = "" if field_cfg.get("placeholder") is None else str(field_cfg.get("placeholder"))
		continue
		if field_cfg is not None:
		defaults[field_name] = field_cfg
		return defaults, fixed, options, allow_empty, option_sources, default_strategies
		return defaults, fixed, options, allow_empty, option_sources, default_strategies, placeholders


		def _normalize_match_options(match_cfg: dict \| None) -> dict:
		@@ -177,21 +180,23 @@ def _normalize_submission_section(submission_cfg: dict \| None) -> dict:
		"data_allow_empty": {},
		"data_option_sources": {},
		"data_default_strategies": {},
		"data_placeholders": {},
		"proc_defaults": {},
		"proc_fixed": {},
		"proc_options": {},
		"proc_allow_empty": {},
		"proc_option_sources": {},
		"proc_default_strategies": {},
		"proc_placeholders": {},
		}
		if not isinstance(submission_cfg, dict):
		return normalized

		# 新结构：submission.data.<field>.default/fixed 与 submission.proc.<field>.default/fixed
		data_defaults, data_fixed, data_options, data_allow_empty, data_option_sources, data_default_strategies = _normalize_field_group(
		data_defaults, data_fixed, data_options, data_allow_empty, data_option_sources, data_default_strategies, data_placeholders = _normalize_field_group(
		submission_cfg.get("data")
		)
		proc_defaults, proc_fixed, proc_options, proc_allow_empty, proc_option_sources, proc_default_strategies = _normalize_field_group(
		proc_defaults, proc_fixed, proc_options, proc_allow_empty, proc_option_sources, proc_default_strategies, proc_placeholders = _normalize_field_group(
		submission_cfg.get("proc")
		)
		if (
		@@ -201,12 +206,14 @@ def _normalize_submission_section(submission_cfg: dict \| None) -> dict:
		or data_allow_empty
		or data_option_sources
		or data_default_strategies
		or data_placeholders
		or proc_defaults
		or proc_fixed
		or proc_options
		or proc_allow_empty
		or proc_option_sources
		or proc_default_strategies
		or proc_placeholders
		):
		normalized["data_defaults"] = data_defaults
		normalized["data_fixed"] = data_fixed
		@@ -214,12 +221,14 @@ def _normalize_submission_section(submission_cfg: dict \| None) -> dict:
		normalized["data_allow_empty"] = data_allow_empty
		normalized["data_option_sources"] = data_option_sources
		normalized["data_default_strategies"] = data_default_strategies
		normalized["data_placeholders"] = data_placeholders
		normalized["proc_defaults"] = proc_defaults
		normalized["proc_fixed"] = proc_fixed
		normalized["proc_options"] = proc_options
		normalized["proc_allow_empty"] = proc_allow_empty
		normalized["proc_option_sources"] = proc_option_sources
		normalized["proc_default_strategies"] = proc_default_strategies
		normalized["proc_placeholders"] = proc_placeholders
		return normalized

		# 兼容旧结构：data_defaults/data_fixed/proc_defaults/proc_fixed

csst_dag/v2/dag/base.py

+126 −3

Original line number	Diff line number	Diff line
		@@ -26,6 +26,45 @@ class DependencyBundle:
		upstream: Optional[list[dict[str, Any]]]


		def _preview_value(value: Any) -> Any:
		if value is None or isinstance(value, (str, int, float, bool)):
		return value
		if hasattr(value, "item"):
		try:
		item = value.item()
		if item is None or isinstance(item, (str, int, float, bool)):
		return item
		except Exception:
		pass
		if isinstance(value, (list, tuple)):
		return [_preview_value(item) for item in value[:5]]
		if isinstance(value, dict):
		return {str(key): _preview_value(item) for key, item in list(value.items())[:8]}
		return str(value)


		def _preview_table_sample(table_obj: Optional[Table], *, limit: int = 3) -> list[dict[str, Any]]:
		if table_obj is None or len(table_obj) == 0:
		return []
		columns = list(table_obj.colnames[:8])
		sample: list[dict[str, Any]] = []
		for row in table_obj[:limit]:
		sample.append({name: _preview_value(row[name]) for name in columns})
		return sample


		def _preview_query(query: dict[str, Any]) -> dict[str, Any]:
		return {
		key: _preview_value(value)
		for key, value in query.items()
		if key not in {"page", "limit"} and value not in (None, "", [], {})
		}


		def _table_size(table_obj: Optional[Table]) -> int:
		return len(table_obj) if table_obj is not None else 0


		class BaseDAG(ABC):
		"""
		通用 DAG 基类（v2）。
		@@ -150,6 +189,76 @@ class BaseDAG(ABC):
		"""
		return deps

		def build_preview(
		self,
		*,
		plan_query: dict[str, Any],
		data_query: dict[str, Any],
		upstream_query: dict[str, Any],
		raw_deps: DependencyBundle,
		filtered_deps: DependencyBundle,
		dag_group: DagRunGroup,
		) -> dict[str, Any]:
		preview = {
		"plan": {
		"enabled": self.require_plan,
		"query": _preview_query(plan_query),
		"count": _table_size(raw_deps.plan),
		"sample": _preview_table_sample(raw_deps.plan),
		},
		"data": {
		"enabled": self.require_data,
		"query": _preview_query(data_query),
		"count": _table_size(raw_deps.data),
		"sample": _preview_table_sample(raw_deps.data),
		},
		"upstream": {
		"enabled": self.require_upstream,
		"query": _preview_query(upstream_query),
		"count": len(raw_deps.upstream) if raw_deps.upstream is not None else 0,
		},
		"match": {
		"plan_before": _table_size(raw_deps.plan),
		"plan_after": _table_size(filtered_deps.plan),
		"data_before": _table_size(raw_deps.data),
		"data_after": _table_size(filtered_deps.data),
		"upstream_before": len(raw_deps.upstream) if raw_deps.upstream is not None else 0,
		"upstream_after": len(filtered_deps.upstream) if filtered_deps.upstream is not None else 0,
		},
		"expand": {
		"dag_run_count": (
		len(dag_group.get("dag_runs") or [])
		if isinstance(dag_group, dict) and "dag_runs" in dag_group
		else None
		),
		},
		}
		expand_preview = dag_group.pop("_preview_expand", None)
		if isinstance(expand_preview, dict):
		preview["expand"].update(expand_preview)

		zero_stage = None
		zero_reason = None
		if isinstance(preview["expand"]["dag_run_count"], int) and preview["expand"]["dag_run_count"] == 0:
		if self.require_plan and preview["plan"]["count"] == 0:
		zero_stage = "plan"
		zero_reason = "plan.find 未命中任何记录"
		elif self.require_data and preview["data"]["count"] == 0:
		zero_stage = "data"
		zero_reason = "data.find 未命中任何记录"
		elif preview["match"]["plan_before"] > 0 and preview["match"]["plan_after"] == 0:
		zero_stage = "match"
		zero_reason = "match/filter 后 plan 结果归零"
		elif preview["match"]["data_before"] > 0 and preview["match"]["data_after"] == 0:
		zero_stage = "match"
		zero_reason = "match/filter 后 data 结果归零"
		else:
		zero_stage = "expand"
		zero_reason = str(preview["expand"].get("zero_reason") or "依赖查询有结果，但未展开出任何 dag_run")
		preview["zero_stage"] = zero_stage
		preview["zero_reason"] = zero_reason
		return preview

		def trigger(
		self,
		*,
		@@ -173,7 +282,7 @@ class BaseDAG(ABC):
		data=data, proc=proc
		)

		deps = DependencyBundle(
		raw_deps = DependencyBundle(
		plan=self.resolve_plan_dependency(plan_query) if self.require_plan else None,
		data=self.resolve_data_dependency(data_query) if self.require_data else None,
		upstream=(
		@@ -182,8 +291,22 @@ class BaseDAG(ABC):
		else None
		),
		)
		deps = self.filter_dependencies(deps)
		return self.build_trigger_payload(data=data, proc=proc, docker_images=docker_images, deps=deps)
		filtered_deps = self.filter_dependencies(raw_deps)
		dag_group = self.build_trigger_payload(
		data=data,
		proc=proc,
		docker_images=docker_images,
		deps=filtered_deps,
		)
		dag_group["details"] = self.build_preview(
		plan_query=plan_query,
		data_query=data_query,
		upstream_query=upstream_query,
		raw_deps=raw_deps,
		filtered_deps=filtered_deps,
		dag_group=dag_group,
		)
		return dag_group

		@abstractmethod
		def build_trigger_payload(

csst_dag/v2/dag/brick.py

+24 −0

Original line number	Diff line number	Diff line
		@@ -161,10 +161,21 @@ class BaseBrickDAG(BaseDAG):
		docker_images=docker_images,
		)
		if deps.data is None or len(deps.data) == 0:
		dag_group["_preview_expand"] = {
		"unique_healpix_count": 0,
		"multiplier_count": 0,
		"dag_run_sample": [],
		}
		return dag_group

		run_multiplier_combinations = self._resolve_run_multiplier_combinations(data)
		if not run_multiplier_combinations:
		dag_group["_preview_expand"] = {
		"unique_healpix_count": len(deps.data),
		"multiplier_count": 0,
		"dag_run_sample": [],
		"zero_reason": "brick multiplier 解析后为空",
		}
		return dag_group
		unique_healpix = [
		self._normalize_healpix(value) for value in deps.data["healpix"]
		@@ -206,6 +217,19 @@ class BaseBrickDAG(BaseDAG):
		run = DagRun(**run_kwargs)
		dag_group.append_dag_run(run)

		dag_group["_preview_expand"] = {
		"unique_healpix_count": sum(1 for value in unique_healpix if value is not None),
		"multiplier_count": len(run_multiplier_combinations),
		"dag_run_sample": [
		{
		"healpix": run.get("data", {}).get("healpix"),
		"dataset": run.get("data", {}).get("dataset"),
		"data_model": run.get("data", {}).get("data_model"),
		}
		for run in [item.to_payload() if hasattr(item, "to_payload") else item for item in (dag_group.get("dag_runs") or [])[:3]]
		if isinstance(run, dict)
		],
		}
		return dag_group

csst_dag/v2/dag/file.py

+57 −8

Original line number	Diff line number	Diff line
		@@ -169,14 +169,63 @@ class FileDAG(BaseDAG):
		docker_images=docker_images,
		)
		if deps.plan is None or deps.data is None or len(deps.plan) == 0 or len(deps.data) == 0:
		dag_group["_preview_expand"] = {
		"task_candidate_count": 0,
		"successful_task_count": 0,
		"dag_run_sample": [],
		}
		return dag_group

		dag_group = self.schedule(
		deps=deps,
		data=data,
		proc=proc,
		dag_group=dag_group,
		force_success=bool(data.get("force_success", False)),
		return_data_list=bool(data.get("return_data_list", True)),
		task_list = self.dispatcher(deps.plan, deps.data)
		force_success = bool(data.get("force_success", False))
		return_data_list = bool(data.get("return_data_list", True))
		for this_task in task_list:
		if not (force_success or this_task.get("success")):
		continue
		task = dict(this_task.get("task") or {})
		run = DagRun(
		dag=dag_group["dag"],
		dag_run_group=dag_group["dag_run_group"],
		batch_id=dag_group.get("batch_id"),
		priority=dag_group.get("priority", "low"),
		data_model=task.get("data_model"),
		dataset=task.get("dataset"),
		source_batch_id=(
		str(data.get("source_batch_id"))
		if data.get("source_batch_id") not in (None, "")
		else None
		),
		instrument=task.get("instrument"),
		obs_type=task.get("obs_type"),
		obs_group=task.get("obs_group"),
		obs_id=task.get("obs_id"),
		detector=task.get("detector"),
		filter=task.get("filter"),
		custom_id=task.get("custom_id"),
		prc_status=task.get("prc_status"),
		qc_status=task.get("qc_status"),
		n_file_expected=task.get("n_file_expected", -1),
		n_file_found=task.get("n_file_found", -1),
		data_list=(this_task.get("relevant_data_id_list") if return_data_list else []),
		pmapname=proc.get("pmapname"),
		ref_cat=proc.get("ref_cat"),
		extra_kwargs=proc.get("extra_kwargs"),
		)
		dag_group.append_dag_run(run)
		successful_task_count = sum(1 for item in task_list if bool(item.get("success")))
		dag_group["_preview_expand"] = {
		"task_candidate_count": len(task_list),
		"successful_task_count": successful_task_count,
		"dag_run_sample": [
		{
		"obs_id": run.get("data", {}).get("obs_id"),
		"detector": run.get("data", {}).get("detector"),
		"dataset": run.get("data", {}).get("dataset"),
		}
		for run in [item.to_payload() if hasattr(item, "to_payload") else item for item in (dag_group.get("dag_runs") or [])[:3]]
		if isinstance(run, dict)
		],
		}
		if successful_task_count == 0:
		dag_group["_preview_expand"]["zero_reason"] = "dispatcher 未产出任何 success task"
		return dag_group

csst_dag/v2/dag/standalone.py

+11 −0

Original line number	Diff line number	Diff line
		@@ -60,4 +60,15 @@ class StandaloneDAG(BaseDAG):
		extra_kwargs=proc.get("extra_kwargs"),
		)
		)
		dag_group["_preview_expand"] = {
		"task_candidate_count": 1,
		"successful_task_count": 1,
		"dag_run_sample": [
		{
		"dataset": data.get("dataset"),
		"data_model": data.get("data_model"),
		"obs_id": data.get("obs_id"),
		}
		],
		}
		return dag_group