Spaces:
Running
Running
| """工作流自动派发与历史回填。 | |
| 这里集中处理上传派发、项目绑定后的历史回填和幂等判断，避免上传路由、 | |
| 手工重处理接口各自维护一套略有差异的匹配规则。工作流定义仍是共享的； | |
| 项目绑定只覆盖输入设备名，真正入队时才生成项目级 graph snapshot。 | |
| """ | |
| from __future__ import annotations | |
| import hashlib | |
| import json | |
| from datetime import datetime, timezone | |
| from pathlib import Path | |
| from uuid import uuid4 | |
| from app.localstore import ( | |
| add_exception, | |
| delete_exception, | |
| get_workflow, | |
| list_runs, | |
| list_exceptions, | |
| read_episode_state, | |
| scan_sessions, | |
| save_run_if_absent, | |
| set_episode_status, | |
| ) | |
| from app.workflow_bindings import ( | |
| apply_bindings, | |
| ) | |
| from app.device_naming import camera_profile, is_depth_only_key | |
| from app.glove_sources import group_glove_source_keys | |
| from app.lerobot_v21 import canonical_source_key | |
| _CAMERA_INPUTS = { | |
| "mono_camera", "rgbd_camera", "fisheye_camera", "rgb_camera", | |
| "stereo_camera", "stereo_rgbd_camera", | |
| } | |
| _COMPOSITE_INPUTS = {"gripper_device"} | |
| def project_workflow_ids(project: dict | None) -> list[str]: | |
| """Return the one workflow bound to a project. | |
| ``workflow_ids`` remains readable for old project files, but the product | |
| model is one project → one workflow. Prefer the canonical singular field | |
| when present and ignore legacy extra entries. | |
| """ | |
| if not project: | |
| return [] | |
| value = project.get("workflow_id") | |
| if not value: | |
| raw = project.get("workflow_ids") or [] | |
| if not isinstance(raw, list): | |
| raw = [raw] if raw else [] | |
| value = raw[0] if raw else None | |
| value = str(value or "") | |
| return [value] if value else [] | |
| def workflow_revision(workflow: dict, bindings: dict | None = None) -> str: | |
| """生成稳定的工作流版本指纹。 | |
| 绑定后的 graph 也进入指纹，所以同一工作流在两个项目使用不同设备 | |
| 名称时不会互相去重。字段排序保证 JSON 文件重排不会导致重复运行。 | |
| """ | |
| payload = { | |
| "workflow_id": workflow.get("id"), | |
| "graph": apply_bindings(workflow.get("graph") or {}, bindings or {}), | |
| "node_configs": workflow.get("node_configs") or {}, | |
| } | |
| raw = json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")) | |
| return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:24] | |
| def _config_source_keys(config: dict) -> list[str]: | |
| values = config.get("source_keys") or config.get("source_key") or config.get("position") | |
| if isinstance(values, str): | |
| values = [v.strip() for v in values.split(",") if v.strip()] | |
| if not isinstance(values, (list, tuple, set)): | |
| return [] | |
| return [str(value).strip() for value in values if str(value).strip()] | |
| def _node_source_groups(workflow: dict, bindings: dict | None) -> tuple[list[str], list[str]]: | |
| """Return source keys from connected input cards only. | |
| Input cards may be placed on a canvas before the upload is known. A card | |
| with no outgoing edge is design-time content and must not make dispatch | |
| reject an otherwise compatible upload. | |
| """ | |
| graph = apply_bindings(workflow.get("graph") or {}, bindings or {}) | |
| connected = { | |
| str(edge.get("source")) for edge in graph.get("edges", []) | |
| if edge.get("source") | |
| } | |
| camera_keys: list[str] = [] | |
| sensor_keys: list[str] = [] | |
| for node in graph.get("nodes", []): | |
| if str(node.get("id")) not in connected: | |
| continue | |
| data = node.get("data") or {} | |
| node_type = str(data.get("nodeType") or "") | |
| if node_type not in _CAMERA_INPUTS | _COMPOSITE_INPUTS and node_type != "glove_sensor": | |
| continue | |
| values = _config_source_keys(data.get("config") or {}) | |
| target = sensor_keys if node_type in {"glove_sensor", "gripper_device"} else camera_keys | |
| target.extend(values) | |
| return camera_keys, sensor_keys | |
| def _episode_sources(episode: dict) -> tuple[set[str], set[str]]: | |
| cameras = { | |
| str(value).lower() | |
| for value in (episode.get("camera_names") or []) | |
| if not str(value).lower().endswith("_aux") | |
| } | |
| # A workflow may be configured with the physical device name from | |
| # meta/info.json (for example D435_depth), while the uploaded video is | |
| # stored under its slot/source key (D435_depth_rgb). Include aliases for | |
| # matching only; actual processing continues to use camera_names. | |
| for device_name in (episode.get("device_names") or {}).values(): | |
| value = str(device_name).strip().lower() | |
| if value: | |
| cameras.add(value) | |
| sensors = {str(value).lower() for value in (episode.get("sensors") or [])} | |
| # 采集端有时没有把 sensors 写进 info.json，但会按 data/<sensor>/ 保存 | |
| # parquet。只从目录名补充手套/触觉类，IMU/深度仍只是附属能力。 | |
| batch_path = Path(str(episode.get("path") or "")) | |
| data_root = batch_path / "data" | |
| if data_root.is_dir(): | |
| for child in data_root.iterdir(): | |
| name = child.name.lower() | |
| if child.is_dir() and any(k in name for k in ("glove", "tactile", "sensor")): | |
| sensors.add(name) | |
| return cameras, sensors | |
| def _episode_camera_values(episode: dict) -> list[str]: | |
| return [str(value) for value in (episode.get("camera_names") or []) | |
| if not str(value).lower().endswith("_aux")] | |
| def _episode_input_groups(episode: dict) -> list[dict]: | |
| """Build physical input groups from the uploaded episode metadata. | |
| A group is the matching unit: one RGB stream, one Stereo pair, one RGB-D | |
| device, one UMI gripper, or one Stereo RGB-D device with an RGB pair plus | |
| depth metadata. | |
| The real source | |
| keys are kept for the worker snapshot, while ``input_type`` is the stable | |
| workflow contract. | |
| """ | |
| metadata: dict = {} | |
| batch_path = Path(str(episode.get("path") or "")) | |
| for relative in ("meta/info.json", "metadata.json"): | |
| candidate = batch_path / relative | |
| if candidate.is_file(): | |
| try: | |
| metadata = json.loads(candidate.read_text(encoding="utf-8")) | |
| break | |
| except (OSError, ValueError): | |
| continue | |
| camera_values = _episode_camera_values(episode) | |
| if not camera_values: | |
| camera_values = [str(value) for value in (metadata.get("cameras") or {}) | |
| if not str(value).lower().endswith("_aux")] | |
| camera_lowers = {value.lower() for value in camera_values} | |
| device_names = episode.get("device_names") or metadata.get("device_names") or {} | |
| if not isinstance(device_names, dict): | |
| device_names = {} | |
| devices = episode.get("devices") or metadata.get("devices") or [] | |
| groups: list[dict] = [] | |
| used: set[str] = set() | |
| def add_group(name: str, kind: str, slots: list[str]) -> None: | |
| clean: list[str] = [] | |
| seen: set[str] = set() | |
| for value in slots: | |
| item = str(value).strip() | |
| low = item.lower() | |
| # State camera_names usually lists only RGB video streams; the | |
| # physical device metadata may additionally list a pure depth | |
| # slot. Keep that slot for RGB-D classification and pairing. | |
| if (not item or low in seen | |
| or (low not in camera_lowers and not is_depth_only_key(item))): | |
| continue | |
| seen.add(low) | |
| clean.append(item) | |
| if not clean: | |
| return | |
| if all(key.lower() in used for key in clean): | |
| return | |
| rgb_keys = [key for key in clean if not is_depth_only_key(key)] | |
| if not rgb_keys: | |
| return | |
| profile, _lens = camera_profile(kind, name, clean) | |
| groups.append({ | |
| "input_type": profile, | |
| "source_keys": rgb_keys, | |
| "depth_keys": [key for key in clean if is_depth_only_key(key)], | |
| "aliases": [str(name).strip(), *clean], | |
| }) | |
| used.update(key.lower() for key in clean) | |
| # New collector metadata provides the authoritative physical grouping. | |
| if isinstance(devices, list): | |
| for device in devices: | |
| if not isinstance(device, dict): | |
| continue | |
| name = str(device.get("name") or device.get("key") or "").strip() | |
| slots = [str(value) for value in (device.get("slots") or [])] | |
| if name: | |
| slots.extend( | |
| str(stream) for stream, mapped in device_names.items() | |
| if str(mapped).strip().lower() == name.lower() | |
| ) | |
| add_group(name, str(device.get("kind") or ""), slots) | |
| # Compatibility path for older episode state that has only stream → name. | |
| by_device: dict[str, list[str]] = {} | |
| for stream, name in device_names.items(): | |
| key = str(stream).strip() | |
| if (not key or (key.lower() not in camera_lowers | |
| and not is_depth_only_key(key))): | |
| continue | |
| group_name = str(name).strip() or key | |
| by_device.setdefault(group_name.lower(), [group_name]).append(key) | |
| for values in by_device.values(): | |
| add_group(values[0], "", values[1:]) | |
| # Final fallback for old state without physical device metadata: pair | |
| # conventional left/right streams and classify the remaining stream. | |
| remaining = [key for key in camera_values if key.lower() not in used] | |
| remaining_lowers = {key.lower() for key in remaining} | |
| for key in sorted(remaining, key=str.lower): | |
| if key.lower() not in remaining_lowers: | |
| continue | |
| pair = None | |
| low = key.lower() | |
| replacements = ( | |
| ("_left_", "_right_"), ("_right_", "_left_"), | |
| ("_left", "_right"), ("_right", "_left"), | |
| ) | |
| for old, new in replacements: | |
| if old not in low: | |
| continue | |
| candidate = low.replace(old, new) | |
| pair = next((value for value in remaining | |
| if value.lower() == candidate), None) | |
| if pair: | |
| break | |
| add_group("", "stereo" if pair else "", [key, pair] if pair else [key]) | |
| remaining_lowers.discard(key.lower()) | |
| if pair: | |
| remaining_lowers.discard(pair.lower()) | |
| sensors = set(str(value) for value in (episode.get("sensors") or [])) | |
| sensors.update(str(value) for value in (metadata.get("sensors") or [])) | |
| # Keep the existing directory fallback for batches whose state file is | |
| # older than the uploaded metadata. | |
| sensors.update(_episode_sources(episode)[1]) | |
| for keys in group_glove_source_keys(sorted(sensors)): | |
| groups.append({ | |
| "input_type": "glove_sensor", | |
| "source_keys": [str(key) for key in keys], | |
| "depth_keys": [], | |
| "aliases": [str(key) for key in keys], | |
| }) | |
| return groups | |
| def _workflow_input_specs(workflow: dict, bindings: dict | None) -> list[dict]: | |
| """Return connected input cards with their fixed semantic categories.""" | |
| graph = apply_bindings(workflow.get("graph") or {}, bindings or {}) | |
| connected = { | |
| str(edge.get("source")) for edge in graph.get("edges", []) | |
| if edge.get("source") | |
| } | |
| specs: list[dict] = [] | |
| for node in graph.get("nodes", []): | |
| node_id = str(node.get("id") or "") | |
| if not node_id or node_id not in connected: | |
| continue | |
| data = node.get("data") or {} | |
| node_type = str(data.get("nodeType") or "") | |
| if node_type not in _CAMERA_INPUTS | _COMPOSITE_INPUTS and node_type != "glove_sensor": | |
| continue | |
| semantic = ( | |
| "glove_sensor" if node_type == "glove_sensor" | |
| else "gripper_device" if node_type == "gripper_device" | |
| else "stereo_rgbd_camera" if node_type == "stereo_rgbd_camera" | |
| else "rgbd_camera" if node_type == "rgbd_camera" | |
| else "stereo_rgb" if node_type == "stereo_camera" | |
| else "mono_rgb" | |
| ) | |
| specs.append({ | |
| "node_id": node_id, | |
| "node_type": node_type, | |
| "semantic_type": semantic, | |
| "keys": _config_source_keys(data.get("config") or {}), | |
| }) | |
| return specs | |
| def _key_matches_group(key: str, group: dict) -> bool: | |
| wanted = str(key).strip().lower() | |
| if not wanted: | |
| return False | |
| canonical_wanted = canonical_source_key(key).lower() | |
| aliases = [str(alias).strip() for alias in group.get("aliases") or [] | |
| if str(alias).strip()] | |
| if any(canonical_source_key(alias).lower() == canonical_wanted | |
| for alias in aliases): | |
| return True | |
| return any( | |
| wanted in alias.lower() or alias.lower() in wanted | |
| for alias in aliases | |
| ) | |
| def _auto_bindings_for_episode(workflow: dict, episode: dict, | |
| bindings: dict | None) -> tuple[dict, set[str], list[dict], list[dict]]: | |
| """Match connected fixed cards to real input groups. | |
| A persisted source name is preferred only when it matches the uploaded | |
| group. A stale name from an older workflow is treated as a hint and is | |
| replaced by a same-category source in the run snapshot. | |
| """ | |
| specs = _workflow_input_specs(workflow, bindings) | |
| groups = _episode_input_groups(episode) | |
| remaining = set(range(len(groups))) | |
| auto: dict = {} | |
| matched: set[str] = set() | |
| for spec in specs: | |
| candidates = [i for i in remaining | |
| if groups[i].get("input_type") == spec["semantic_type"]] | |
| if not candidates: | |
| continue | |
| named = [i for i in candidates | |
| if any(_key_matches_group(key, groups[i]) for key in spec["keys"])] | |
| selected = sorted( | |
| named or candidates, | |
| key=lambda i: str(groups[i].get("aliases") or ""), | |
| )[0] | |
| group = groups[selected] | |
| remaining.remove(selected) | |
| keys = [str(value) for value in group.get("source_keys") or []] | |
| if not keys: | |
| continue | |
| auto[spec["node_id"]] = ( | |
| {"source_keys": ",".join(keys)} | |
| if len(keys) > 1 or spec["semantic_type"] in {"glove_sensor", "gripper_device"} | |
| else {"source_key": keys[0]} | |
| ) | |
| matched.add(spec["node_id"]) | |
| missing = [spec for spec in specs if spec["node_id"] not in matched] | |
| return auto, matched, missing, groups | |
| def _clear_unmatched_input_configs(graph: dict, matched: set[str]) -> None: | |
| """Remove stale source names from skipped input branches in a run copy.""" | |
| for node in graph.get("nodes", []): | |
| node_id = str(node.get("id") or "") | |
| data = node.get("data") or {} | |
| node_type = str(data.get("nodeType") or "") | |
| if node_id in matched or (node_type not in _CAMERA_INPUTS | _COMPOSITE_INPUTS and node_type != "glove_sensor"): | |
| continue | |
| config = data.get("config") | |
| if not isinstance(config, dict): | |
| continue | |
| for field in ("source_key", "source_keys", "position"): | |
| if field in config: | |
| config[field] = "" | |
| def _record_mismatch(project: dict | None, episode: dict, workflow: dict, | |
| wanted: list[str], available: set[str], message: str) -> None: | |
| try: | |
| add_exception({ | |
| "id": str(uuid4()), | |
| "kind": "upload_mismatch", | |
| "project_id": project.get("id") if project else None, | |
| "project_name": project.get("name") if project else None, | |
| "episode_id": episode.get("id"), | |
| "workflow_id": workflow.get("id"), | |
| "workflow_name": workflow.get("name"), | |
| "wanted": list(wanted), | |
| "available": sorted(available), | |
| "message": message, | |
| "created_at": datetime.now(timezone.utc).isoformat(), | |
| }) | |
| except Exception as exc: | |
| print(f"[WorkflowDispatch] Failed to record mismatch: {exc}") | |
| def _clear_episode_mismatches(episode_id: str) -> None: | |
| """Remove stale upload-mismatch records after a compatible run exists.""" | |
| try: | |
| for exc in list_exceptions(): | |
| if (exc.get("episode_id") == episode_id | |
| and exc.get("kind") == "upload_mismatch"): | |
| delete_exception(exc.get("id")) | |
| except Exception as exc: | |
| print(f"[WorkflowDispatch] Failed to clear stale mismatches: {exc}") | |
| def _record_mismatch_summary(project: dict | None, episode: dict, | |
| mismatches: list[dict]) -> None: | |
| """Store one mismatch for the batch, not one for every candidate workflow.""" | |
| if not mismatches: | |
| return | |
| available = mismatches[0].get("available") or [] | |
| wanted: list[str] = [] | |
| for item in mismatches: | |
| for value in item.get("wanted") or []: | |
| if value not in wanted: | |
| wanted.append(value) | |
| _clear_episode_mismatches(str(episode.get("id"))) | |
| try: | |
| add_exception({ | |
| "id": str(uuid4()), | |
| "kind": "upload_mismatch", | |
| "project_id": project.get("id") if project else None, | |
| "project_name": project.get("name") if project else None, | |
| "episode_id": episode.get("id"), | |
| "workflow_id": None, | |
| "workflow_name": "No compatible workflow", | |
| "wanted": wanted, | |
| "available": sorted(available), | |
| "message": ( | |
| f"No bound workflow matches episode inputs " | |
| f"{sorted(available)} (checked {len(mismatches)} workflow(s))" | |
| ), | |
| "created_at": datetime.now(timezone.utc).isoformat(), | |
| }) | |
| except Exception as exc: | |
| print(f"[WorkflowDispatch] Failed to record mismatch summary: {exc}") | |
| def _record_partial_input_warnings(project: dict | None, episode: dict, | |
| partials: list[dict]) -> None: | |
| """Persist missing inputs while still allowing matched inputs to run.""" | |
| for item in partials: | |
| missing = [str(value) for value in item.get("missing") or [] if str(value).strip()] | |
| if not missing: | |
| continue | |
| available = [str(value) for value in item.get("available") or []] | |
| matched = [str(value) for value in item.get("matched") or []] | |
| try: | |
| add_exception({ | |
| "id": str(uuid4()), | |
| "kind": "input_missing", | |
| "project_id": project.get("id") if project else None, | |
| "project_name": project.get("name") if project else None, | |
| "episode_id": episode.get("id"), | |
| "workflow_id": item.get("workflow_id"), | |
| "workflow_name": item.get("workflow_name"), | |
| "wanted": list(item.get("wanted") or []), | |
| "matched": matched, | |
| "missing": missing, | |
| "available": available, | |
| "message": ( | |
| f"Workflow partially matched; missing inputs: {missing}. " | |
| f"Only available inputs will be processed: {matched or available}." | |
| ), | |
| "created_at": datetime.now(timezone.utc).isoformat(), | |
| }) | |
| except Exception as exc: | |
| print(f"[WorkflowDispatch] Failed to record partial input warning: {exc}") | |
| def _clear_episode_input_warnings(episode_id: str) -> None: | |
| """Clear partial-input warnings after a complete input match exists.""" | |
| try: | |
| for exc in list_exceptions(): | |
| if (exc.get("episode_id") == episode_id | |
| and exc.get("kind") in {"upload_mismatch", "input_missing"}): | |
| delete_exception(exc.get("id")) | |
| except Exception as exc: | |
| print(f"[WorkflowDispatch] Failed to clear input warnings: {exc}") | |
| def prepare_workflow_run(workflow: dict, episode: dict, project: dict | None = None, | |
| bindings: dict | None = None, | |
| record_mismatch: bool = True, | |
| mismatch_records: list[dict] | None = None, | |
| partial_records: list[dict] | None = None, | |
| ) -> tuple[dict | None, str | None]: | |
| """Match real episode inputs and return an immutable run snapshot. | |
| Matching is type-first (RGB / Stereo RGB / RGB-D / Stereo RGB-D / Glove Sensor), | |
| then source-name-aware. This lets a workflow be assembled before upload | |
| and keeps stale source names from an older workflow from blocking a run. | |
| """ | |
| bindings = bindings if isinstance(bindings, dict) else {} | |
| cameras, sensors = _episode_sources(episode) | |
| specs = _workflow_input_specs(workflow, bindings) | |
| if not specs: | |
| return None, "no_supported_inputs" | |
| auto, matched, missing, groups = _auto_bindings_for_episode( | |
| workflow, episode, bindings) | |
| if not matched: | |
| wanted = [spec["semantic_type"] for spec in specs] | |
| message = f"Workflow expects input types {wanted}, episode has {sorted(cameras | sensors)}" | |
| if mismatch_records is not None: | |
| mismatch_records.append({ | |
| "wanted": wanted, | |
| "available": sorted(cameras | sensors), | |
| }) | |
| elif record_mismatch: | |
| _record_mismatch(project, episode, workflow, wanted, cameras | sensors, message) | |
| return None, "input_type_mismatch" | |
| bindings = {**bindings, **auto} | |
| if partial_records is not None and missing: | |
| partial_records.append({ | |
| "workflow_id": workflow.get("id"), | |
| "workflow_name": workflow.get("name"), | |
| "wanted": [spec["semantic_type"] for spec in specs], | |
| "matched": [spec["semantic_type"] for spec in specs | |
| if spec["node_id"] in matched], | |
| "missing": [spec["semantic_type"] for spec in missing], | |
| "available": sorted(cameras | sensors), | |
| }) | |
| bound = dict(workflow) | |
| bound["graph"] = apply_bindings(workflow.get("graph") or {}, bindings) | |
| _clear_unmatched_input_configs(bound["graph"], matched) | |
| return bound, None | |
| def enqueue_workflow_once(workflow: dict, episode: dict, project: dict | None = None, | |
| bindings: dict | None = None, trigger: str = "", | |
| record_mismatch: bool = True, | |
| mismatch_records: list[dict] | None = None, | |
| partial_records: list[dict] | None = None, | |
| force_rerun: bool = False) -> dict: | |
| """匹配并以工作流版本幂等入队。返回统一结果。""" | |
| bound, reason = prepare_workflow_run( | |
| workflow, episode, project, bindings, | |
| record_mismatch=record_mismatch, | |
| mismatch_records=mismatch_records, | |
| partial_records=partial_records, | |
| ) | |
| if bound is None: | |
| return {"status": "skipped", "reason": reason, "workflow_id": workflow.get("id")} | |
| # A forced reprocess starts a new annotation generation. Do not carry | |
| # old manual or AI segments into the new run. | |
| episode_id = str(episode["id"]) | |
| if force_rerun: | |
| try: | |
| from app.ai_annotation import invalidate_ai_annotation_tasks | |
| invalidate_ai_annotation_tasks(episode_id) | |
| except Exception as exc: | |
| # Annotation cleanup remains valid even if the optional AI module | |
| # is unavailable during a worker-only dispatch. | |
| print(f"[WorkflowDispatch] AI task invalidation skipped: {exc}") | |
| from app.localstore import save_annotations | |
| save_annotations(episode_id, []) | |
| # 延迟导入避免 workflow API 与派发器互相导入。 | |
| from app.api.workflows import _build_run_record | |
| # ``bound`` 已包含名称匹配产生的自动绑定；指纹必须基于最终 snapshot， | |
| # 否则同一工作流切换到另一台同类型相机时会被错误去重。 | |
| revision = workflow_revision(bound, {}) | |
| run = _build_run_record(bound, str(episode["id"]), | |
| project_id=project.get("id") if project else None, | |
| workflow_revision=revision, trigger=trigger) | |
| previous_status = read_episode_state(episode_id).get("status") | |
| # Mark processing before publishing the run record. This closes the | |
| # race where a fast worker completed the run and the dispatcher then | |
| # wrote processing after the worker had already written to_review. | |
| if previous_status != "processing": | |
| set_episode_status(episode_id, "processing") | |
| try: | |
| saved, created = save_run_if_absent( | |
| run, | |
| allow_completed_rerun=force_rerun, | |
| supersede_active=force_rerun, | |
| ) | |
| except Exception: | |
| # Do not leave a batch stuck in processing if enqueue itself fails. | |
| if read_episode_state(episode_id).get("status") == "processing": | |
| set_episode_status(episode_id, previous_status or "to_review") | |
| raise | |
| if not created and saved.get("status") in ("completed", "failed"): | |
| # A normal upload/backfill may find an already completed run. Restore | |
| # the old state; only a newly created run should move it to processing. | |
| if read_episode_state(episode_id).get("status") == "processing": | |
| set_episode_status(episode_id, previous_status or "to_review") | |
| if created: | |
| return {"status": "queued", "run": saved, "workflow_id": workflow.get("id")} | |
| return {"status": "already_scheduled", "run": saved, "workflow_id": workflow.get("id")} | |
| def dispatch_project_episode(project: dict | None, episode: dict, | |
| trigger: str = "upload", | |
| workflow_ids: list[str] | None = None, | |
| force_rerun: bool = False) -> dict: | |
| """派发一个项目批次的全部 active 工作流。""" | |
| if not project or project.get("status", "active") != "active": | |
| return {"queued": 0, "matched": 0, "skipped": 0} | |
| bindings_all = project.get("workflow_bindings") or {} | |
| if not isinstance(bindings_all, dict): | |
| bindings_all = {} | |
| stats = {"queued": 0, "matched": 0, "skipped": 0, "already_scheduled": 0} | |
| candidate_ids = project_workflow_ids(project) | |
| if workflow_ids is not None: | |
| allowed = {str(value) for value in workflow_ids} | |
| candidate_ids = [value for value in candidate_ids if value in allowed] | |
| mismatches: list[dict] = [] | |
| partials: list[dict] = [] | |
| for workflow_id in candidate_ids: | |
| workflow = get_workflow(workflow_id) | |
| if not workflow or workflow.get("status", "draft") != "active": | |
| stats["skipped"] += 1 | |
| continue | |
| result = enqueue_workflow_once( | |
| workflow, episode, project, | |
| bindings_all.get(workflow_id) if isinstance(bindings_all.get(workflow_id), dict) else {}, | |
| trigger=trigger, | |
| record_mismatch=False, | |
| mismatch_records=mismatches, | |
| partial_records=partials, | |
| force_rerun=force_rerun, | |
| ) | |
| if result["status"] == "queued": | |
| stats["queued"] += 1 | |
| stats["matched"] += 1 | |
| elif result["status"] == "already_scheduled": | |
| stats["already_scheduled"] += 1 | |
| stats["matched"] += 1 | |
| else: | |
| stats["skipped"] += 1 | |
| if stats["matched"]: | |
| # One valid workflow is enough for an episode. Other candidate | |
| # workflows are alternatives, not errors for this same batch. A | |
| # partial match is different: keep a visible warning for review. | |
| if partials: | |
| _clear_episode_mismatches(str(episode.get("id"))) | |
| _record_partial_input_warnings(project, episode, partials) | |
| else: | |
| _clear_episode_input_warnings(str(episode.get("id"))) | |
| elif mismatches: | |
| _record_mismatch_summary(project, episode, mismatches) | |
| return stats | |
| def backfill_project(project: dict, workflow_ids: list[str] | None = None) -> dict: | |
| """工作流首次绑定/设备绑定改变后，回填项目内未完成的历史批次。""" | |
| wanted = workflow_ids or project_workflow_ids(project) | |
| wanted = {str(value) for value in wanted if value} | |
| stats = {"episodes": 0, "queued": 0, "already_scheduled": 0, "skipped": 0} | |
| # Backfill is for batches that have never entered this workflow. A user | |
| # who wants to apply a changed workflow to an already processed batch uses | |
| # the explicit Reprocess action, which can force a new run safely. | |
| existing = { | |
| (str(run.get("workflow_id") or ""), str(run.get("episode_id") or "")) | |
| for run in list_runs() | |
| if run.get("workflow_id") and run.get("episode_id") | |
| } | |
| for episode in scan_sessions(): | |
| if episode.get("project") != project.get("name"): | |
| continue | |
| stats["episodes"] += 1 | |
| if any((workflow_id, str(episode.get("id") or "")) in existing | |
| for workflow_id in wanted): | |
| stats["skipped"] += 1 | |
| continue | |
| result = dispatch_project_episode( | |
| project, episode, trigger="project_backfill", workflow_ids=list(wanted) | |
| ) | |
| for key in ("queued", "already_scheduled", "skipped"): | |
| stats[key] += result.get(key, 0) | |
| return stats | |