增加aigc发布

.env.example

@@ -68,7 +68,6 @@ PIPELINE_WARN_AFTER_HOURS=12
 PIPELINE_CRITICAL_BEFORE_NEXT_MINUTES=120
 PIPELINE_FINAL_WARN_BEFORE_NEXT_MINUTES=30
 PIPELINE_LOG_DIR=logs/pipeline
-PIPELINE_EXTERNAL_EFFECTS_ENABLED=true
 # Aliyun OSS (agent 运行日志可视化上传；qwen 视频截断后片段也上传到此 bucket)
 ALIYUN_OSS_ACCESS_KEY_ID=
 ALIYUN_OSS_ACCESS_KEY_SECRET=

api/services/pipeline.py

@@ -2,7 +2,6 @@ from __future__ import annotations
 
 from typing import Any
 
-from supply_infra.config import get_infra_settings
 from supply_infra.pipeline.dag import PIPELINE_STEPS
 from supply_infra.pipeline.health import pipeline_health_snapshot
 from supply_infra.pipeline.run_service import (
@@ -57,11 +56,9 @@ def retry_step(run_id: str, step_key: str) -> bool:
 def pipeline_health() -> dict[str, Any]:
     recent = list_pipeline_runs(limit=1)
     latest = recent[0] if recent else None
-    settings = get_infra_settings()
     return {
         "scheduler_embedded_in_api": False,
         "pipeline_key": "supply_pipeline",
-        "dry_run": not settings.pipeline_external_effects_enabled,
         "steps": len(PIPELINE_STEPS),
         "latest_run": latest,
         **pipeline_health_snapshot(),

deploy/README.md

@@ -9,8 +9,7 @@ Scheduler、多个 Pipeline Worker 和 Reconciler。
 2. 挂载 `/app/logs/pipeline` 到持久化目录；
 3. `SCHEDULER_ENABLED=true` 只影响独立 Scheduler 进程，API 不再内嵌调度器；
    默认调度时间为 `Asia/Shanghai` 每日 15:00；
-4. `PIPELINE_EXTERNAL_EFFECTS_ENABLED=true` 并配置 `AIGC_API_TOKEN`，日批最后一步
-   会真实创建/绑定 AIGC 计划；紧急演练可临时设为 `false`；
+4. 配置 `AIGC_API_TOKEN`，日批最后一步会真实创建/绑定 AIGC 计划；
 5. Docker 停止窗口至少 300 秒。
 6. 容器和 Alembic 连接 MySQL 后会将会话时区固定为 `+08:00`（中国标准时间）；定时日批有次日
    调度 deadline，手工/API/CLI 补数不设置 deadline。

supply_infra/aigc/client.py

@@ -26,7 +26,7 @@ _INPUT_SOURCE_CHECK_KEYS = ("inputSourceModal", "inputSourceChannel", "contentTy
 class AigcClient:
     """AIGC 平台 HTTP 客户端（视频爬取计划 + 生成计划绑定）。"""
 
-    def __init__(self, token: str | None = None, *, dry_run: bool = False) -> None:
+    def __init__(self, token: str | None = None) -> None:
         settings = get_infra_settings()
         self.token = (
             token
@@ -34,8 +34,6 @@ class AigcClient:
             or settings.aigc_api_token
             or ""
         ).strip()
-        # Explicit dry_run (e.g. CLI) or global external-effects kill switch.
-        self.dry_run = bool(dry_run) or not settings.pipeline_external_effects_enabled
 
     def create_video_crawler_plan(
         self,
@@ -69,15 +67,6 @@ class AigcClient:
             "voiceExtractFlag": 1,
         }
 
-        if self.dry_run:
-            return {
-                "success": True,
-                "dry_run": True,
-                "crawler_plan_id": "dry-run-crawler-plan-id",
-                "crawler_plan_name": crawler_plan_name,
-                "aweme_ids": aweme_ids,
-            }
-
         response_json = self._post(CRAWLER_PLAN_CREATE_URL, params)
         if response_json.get("code") != 0:
             message = response_json.get("msg", "创建爬取计划失败")
@@ -105,14 +94,6 @@ class AigcClient:
         if not crawler_plan_id or not produce_plan_id:
             raise ValueError("crawler_plan_id 与 produce_plan_id 均不能为空")
 
-        if self.dry_run:
-            return {
-                "success": True,
-                "dry_run": True,
-                "produce_plan_id": produce_plan_id,
-                "msg": "成功",
-            }
-
         input_source_info = {
             "contentType": 1,
             "inputSourceType": 2,

supply_infra/config.py

@@ -154,10 +154,6 @@ class InfraSettings(BaseSettings):
         alias="PIPELINE_FINAL_WARN_BEFORE_NEXT_MINUTES",
     )
     pipeline_log_dir: str = Field(default="logs/pipeline", alias="PIPELINE_LOG_DIR")
-    pipeline_external_effects_enabled: bool = Field(
-        default=True,
-        alias="PIPELINE_EXTERNAL_EFFECTS_ENABLED",
-    )
 
     # Aliyun OSS (agent run log publishing)
     aliyun_oss_access_key_id: str = Field(default="", alias="ALIYUN_OSS_ACCESS_KEY_ID")

supply_infra/db/models/pipeline_outbox.py

@@ -20,7 +20,7 @@ from supply_infra.db.base import Base
 
 
 class PipelineOutbox(Base):
-    """Ledger of AIGC write effects (dry-run recorded or dispatched)."""
+    """Ledger of dispatched AIGC write effects."""
 
     __tablename__ = "pipeline_outbox"
     __table_args__ = (
@@ -48,9 +48,9 @@ class PipelineOutbox(Base):
     status: Mapped[str] = mapped_column(
         String(32),
         nullable=False,
-        default="dry_run_recorded",
+        default="dispatched",
     )
-    dry_run: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
+    dry_run: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
     record_error: Mapped[str | None] = mapped_column(Text, nullable=True)
     created_at: Mapped[datetime] = mapped_column(
         DateTime,

supply_infra/db/models/pipeline_run.py

@@ -34,7 +34,7 @@ class PipelineRun(Base):
     trigger_reason: Mapped[str | None] = mapped_column(Text, nullable=True)
     parent_run_id: Mapped[str | None] = mapped_column(String(36), nullable=True)
     run_mode: Mapped[str] = mapped_column(String(24), nullable=False, default="full")
-    dry_run: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
+    dry_run: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
     status: Mapped[str] = mapped_column(String(32), nullable=False, default="queued")
     current_step: Mapped[str | None] = mapped_column(String(64), nullable=True)
     scheduled_for: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)

supply_infra/db/repositories/pipeline_outbox_repo.py

@@ -16,28 +16,6 @@ class PipelineOutboxRepository(BaseRepository[PipelineOutbox]):
         stmt = select(PipelineOutbox).where(PipelineOutbox.idempotency_key == key)
         return self.session.scalar(stmt)
 
-    def record_dry_run(
-        self,
-        *,
-        run_id: str,
-        step_run_id: str,
-        effect_type: str,
-        idempotency_key: str,
-        payload_hash: str,
-        payload: dict[str, Any] | None,
-        payload_uri: str | None = None,
-    ) -> PipelineOutbox:
-        return self.record_effect(
-            run_id=run_id,
-            step_run_id=step_run_id,
-            effect_type=effect_type,
-            idempotency_key=idempotency_key,
-            payload_hash=payload_hash,
-            payload=payload,
-            payload_uri=payload_uri,
-            dry_run=True,
-        )
-
     def record_effect(
         self,
         *,
@@ -48,7 +26,6 @@ class PipelineOutboxRepository(BaseRepository[PipelineOutbox]):
         payload_hash: str,
         payload: dict[str, Any] | None,
         payload_uri: str | None = None,
-        dry_run: bool = True,
     ) -> PipelineOutbox:
         existing = self.get_by_idempotency_key(idempotency_key)
         if existing is not None:
@@ -63,8 +40,8 @@ class PipelineOutboxRepository(BaseRepository[PipelineOutbox]):
                 payload_hash=payload_hash,
                 payload_json=payload,
                 payload_uri=payload_uri,
-                status="dry_run_recorded" if dry_run else "dispatched",
-                dry_run=dry_run,
+                status="dispatched",
+                dry_run=False,
             )
         )
 

supply_infra/pipeline/enums.py

@@ -28,6 +28,5 @@ class StepStatus(StrEnum):
 
 
 class OutboxStatus(StrEnum):
-    DRY_RUN_RECORDED = "dry_run_recorded"
     DISPATCHED = "dispatched"
     RECORD_FAILED = "record_failed"

supply_infra/pipeline/gates.py

@@ -38,14 +38,6 @@ def evaluate_step_gate(step_key: str, payload: dict[str, Any]) -> GateDecision:
                 "aigc_record_incomplete",
                 "AIGC publish payload/hash was not durably recorded",
             )
-        # Dry-run mode still records to outbox; live mode must have attempted API calls
-        # when there were candidates (external_request_made may be False if empty).
-        if payload.get("dry_run") is True and payload.get("external_request_made") is True:
-            return GateDecision(
-                False,
-                "external_effect_detected",
-                "AIGC dry-run must not make external requests",
-            )
         failed_batches = int(payload.get("failed_batch_count", 0) or 0)
         if failed_batches > 0:
             return GateDecision(

supply_infra/pipeline/registry.py

@@ -110,13 +110,8 @@ def _aigc_write_record(context: StepContext) -> dict[str, Any]:
     )
 
     settings = get_infra_settings()
-    dry_run = not settings.pipeline_external_effects_enabled
-    payload = publish_videos_from_discovery(
-        biz_dt=context.biz_dt,
-        dry_run=dry_run,
-    )
+    payload = publish_videos_from_discovery(biz_dt=context.biz_dt)
     publish_success = bool(payload.get("success", True))
-    effective_dry_run = bool(payload.get("dry_run", dry_run))
     canonical = json.dumps(
         payload,
         ensure_ascii=False,
@@ -154,22 +149,19 @@ def _aigc_write_record(context: StepContext) -> dict[str, Any]:
             payload_hash=payload_hash,
             payload=stored_payload,
             payload_uri=payload_uri,
-            dry_run=effective_dry_run,
         )
         outbox_id = record.outbox_id
 
     result: dict[str, Any] = {
         "success": publish_success,
         "effect_recorded": True,
-        "dry_run_recorded": effective_dry_run,
-        "external_request_made": not effective_dry_run,
+        "external_request_made": int(payload.get("candidate_count", 0) or 0) > 0,
         "outbox_id": outbox_id,
         "payload_hash": payload_hash,
         "payload_uri": payload_uri,
         "candidate_count": payload.get("candidate_count", 0),
         "batch_count": payload.get("batch_count", 0),
         "failed_batch_count": payload.get("failed_batch_count", 0),
-        "dry_run": effective_dry_run,
     }
     if not publish_success:
         result["error"] = str(

supply_infra/pipeline/run_service.py

@@ -59,7 +59,6 @@ def _config_snapshot(settings: InfraSettings) -> dict[str, Any]:
         "worker_processes": settings.pipeline_worker_processes,
         "max_active_steps": settings.pipeline_max_active_steps,
         "mysql_connection_budget": settings.mysql_connection_budget,
-        "external_effects_enabled": settings.pipeline_external_effects_enabled,
     }
 
 
@@ -93,7 +92,7 @@ def submit_pipeline_run(
                     created=False,
                     status=existing.status,
                     biz_dt=existing.biz_dt,
-                    dry_run=bool(existing.dry_run),
+                    dry_run=False,
                 )
 
             # All entry points share the same no-overlap rule. Linking each new
@@ -105,7 +104,6 @@ def submit_pipeline_run(
                 if previous is not None
                 else RunStatus.QUEUED.value
             )
-            run_dry_run = not active.pipeline_external_effects_enabled
             run = run_repo.create(
                 {
                     "run_id": run_id,
@@ -117,7 +115,7 @@ def submit_pipeline_run(
                     "triggered_by": None,
                     "trigger_reason": trigger_reason,
                     "run_mode": "full",
-                    "dry_run": run_dry_run,
+                    "dry_run": False,
                     "status": initial_status,
                     "scheduled_for": scheduled_for,
                     "deadline_at": deadline_for_trigger(
@@ -172,7 +170,7 @@ def submit_pipeline_run(
                 created=True,
                 status=run.status,
                 biz_dt=run.biz_dt,
-                dry_run=run_dry_run,
+                dry_run=False,
             )
     except IntegrityError:
         with get_session() as session:
@@ -184,7 +182,7 @@ def submit_pipeline_run(
                 created=False,
                 status=existing.status,
                 biz_dt=existing.biz_dt,
-                dry_run=bool(existing.dry_run),
+                dry_run=False,
             )
 
 

supply_infra/scheduler/jobs/publish_videos_from_discovery.py

@@ -45,7 +45,6 @@ class PublishBatchResult:
     bind_success: bool = False
     bind_error: str | None = None
     error: str | None = None
-    dry_run: bool = False
 
     def to_dict(self) -> dict[str, Any]:
         return {
@@ -60,7 +59,6 @@ class PublishBatchResult:
             "bind_success": self.bind_success,
             "bind_error": self.bind_error,
             "error": self.error,
-            "dry_run": self.dry_run,
         }
 
 
@@ -94,7 +92,6 @@ def publish_videos_from_discovery(
     run_id: str | None = None,
     skip_published: bool = True,
     limit: int | None = None,
-    dry_run: bool = False,
 ) -> dict[str, Any]:
     """
     从 video_discovery_candidate 读取视频，按轮询均匀分配到各 AIGC 计划对。
@@ -102,7 +99,6 @@ def publish_videos_from_discovery(
     仅处理 decision_bucket 为 primary / backup 且 aweme_id 非空的候选，不区分品类。
     """
     settings = get_infra_settings()
-    effective_dry_run = dry_run or not settings.pipeline_external_effects_enabled
     resolved_biz_dt = _resolve_biz_dt(biz_dt)
     plan_pairs = list_unique_plan_pairs()
     if not plan_pairs:
@@ -130,7 +126,6 @@ def publish_videos_from_discovery(
             "plan_count": len(plan_pairs),
             "candidate_count": 0,
             "decision_buckets": list(_PUBLISHABLE_BUCKETS),
-            "dry_run": effective_dry_run,
             "batches": [],
         }
 
@@ -140,7 +135,7 @@ def publish_videos_from_discovery(
         [bucket for _, bucket in grouped],
     )
 
-    client = AigcClient(dry_run=effective_dry_run)
+    client = AigcClient()
     timezone = ZoneInfo(settings.scheduler_timezone)
     timestamp = datetime.now(timezone).strftime("%Y%m%d%H%M%S")
     batch_results: list[PublishBatchResult] = []
@@ -164,7 +159,6 @@ def publish_videos_from_discovery(
                 publish_plan_id=plan_pair.publish_plan_id,
                 aweme_ids=aweme_batch,
                 candidate_ids=id_batch,
-                dry_run=effective_dry_run,
             )
 
             if not create_result.get("success"):
@@ -186,7 +180,7 @@ def publish_videos_from_discovery(
             if not batch.bind_success:
                 batch.bind_error = str(bind_result.get("error") or "绑定生成计划失败")
 
-            if not effective_dry_run and crawler_plan_id and batch.bind_success:
+            if crawler_plan_id and batch.bind_success:
                 with get_session() as session:
                     updated = VideoDiscoveryRepository(session).mark_candidates_aigc_plans(
                         id_batch,
@@ -216,7 +210,6 @@ def publish_videos_from_discovery(
         "distribution": distribution_preview,
         "batch_count": len(batch_results),
         "failed_batch_count": len(failed_batches),
-        "dry_run": effective_dry_run,
         "batches": [item.to_dict() for item in batch_results],
     }
 
@@ -233,11 +226,6 @@ if __name__ == "__main__":
         parser.add_argument("biz_dt", nargs="?", help="业务日 YYYYMMDD，默认取最新")
         parser.add_argument("--run-id", dest="run_id", help="仅处理指定 run_id")
         parser.add_argument("--limit", type=int, help="最多处理候选视频数")
-        parser.add_argument(
-            "--dry-run",
-            action="store_true",
-            help="只演练分配与请求，不写库",
-        )
         parser.add_argument(
             "--force",
             action="store_true",
@@ -252,7 +240,6 @@ if __name__ == "__main__":
             run_id=_args.run_id,
             skip_published=not _args.force,
             limit=_args.limit,
-            dry_run=_args.dry_run,
         ),
         label="publish_videos_from_discovery",
     )

tests/supply_infra/pipeline/test_aigc_safety.py

@@ -4,63 +4,44 @@ from supply_infra.aigc.client import AigcClient
 from supply_infra.config import InfraSettings
 
 
-def test_aigc_client_forces_dry_run_when_external_effects_are_disabled(
-    monkeypatch,
-) -> None:
-    settings = InfraSettings(
-        _env_file=None,
-        PIPELINE_EXTERNAL_EFFECTS_ENABLED=False,
-    )
+def test_aigc_client_requires_token_for_live_calls(monkeypatch) -> None:
+    settings = InfraSettings(_env_file=None)
     monkeypatch.setattr(
         "supply_infra.aigc.client.get_infra_settings",
         lambda: settings,
     )
+    client = AigcClient(token="")
+    try:
+        client.create_video_crawler_plan(["123"])
+    except RuntimeError as exc:
+        assert "AIGC_API_TOKEN" in str(exc)
+    else:
+        raise AssertionError("missing token must raise")
 
-    called = False
-
-    def _unexpected_post(*_args, **_kwargs):
-        nonlocal called
-        called = True
-        raise AssertionError("network must not be called")
-
-    monkeypatch.setattr("supply_infra.aigc.client.requests.post", _unexpected_post)
-    client = AigcClient(token="secret", dry_run=False)
-    result = client.create_video_crawler_plan(["123"])
-    assert result["dry_run"] is True
-    assert called is False
 
-
-def test_aigc_client_calls_api_by_default_when_effects_enabled(monkeypatch) -> None:
-    settings = InfraSettings(
-        _env_file=None,
-        PIPELINE_EXTERNAL_EFFECTS_ENABLED=True,
-    )
+def test_aigc_client_posts_to_platform(monkeypatch) -> None:
+    settings = InfraSettings(_env_file=None)
     monkeypatch.setattr(
         "supply_infra.aigc.client.get_infra_settings",
         lambda: settings,
     )
-    client = AigcClient(token="secret")
-    assert client.dry_run is False
 
+    class _Response:
+        def raise_for_status(self) -> None:
+            return None
 
-def test_explicit_dry_run_still_skips_network(monkeypatch) -> None:
-    settings = InfraSettings(
-        _env_file=None,
-        PIPELINE_EXTERNAL_EFFECTS_ENABLED=True,
-    )
-    monkeypatch.setattr(
-        "supply_infra.aigc.client.get_infra_settings",
-        lambda: settings,
-    )
-    called = False
+        def json(self) -> dict:
+            return {"code": 0, "data": {"id": "plan-1"}}
+
+    calls: list[dict] = []
 
-    def _unexpected_post(*_args, **_kwargs):
-        nonlocal called
-        called = True
-        raise AssertionError("network must not be called")
+    def _post(*, url, json, headers, timeout):
+        calls.append({"url": url, "json": json})
+        return _Response()
 
-    monkeypatch.setattr("supply_infra.aigc.client.requests.post", _unexpected_post)
-    client = AigcClient(token="secret", dry_run=True)
+    monkeypatch.setattr("supply_infra.aigc.client.requests.post", _post)
+    client = AigcClient(token="secret")
     result = client.create_video_crawler_plan(["123"])
-    assert result["dry_run"] is True
-    assert called is False
+    assert result["success"] is True
+    assert result["crawler_plan_id"] == "plan-1"
+    assert len(calls) == 1

tests/supply_infra/pipeline/test_config_budget.py

@@ -16,7 +16,6 @@ def _settings(**overrides: object) -> InfraSettings:
         "MYSQL_MAX_OVERFLOW": 0,
         "PIPELINE_WORKER_PROCESSES": 4,
         "PIPELINE_MAX_ACTIVE_STEPS": 4,
-        "PIPELINE_EXTERNAL_EFFECTS_ENABLED": True,
     }
     values.update(overrides)
     return InfraSettings(_env_file=None, **values)

tests/supply_infra/pipeline/test_dag_and_gates.py

@@ -22,38 +22,24 @@ def test_aigc_gate_requires_durable_effect_record() -> None:
             "success": True,
             "effect_recorded": True,
             "payload_hash": "abc",
-            "dry_run": False,
             "external_request_made": True,
         },
     )
     assert live_ok.passed is True
-    empty_live_ok = evaluate_step_gate(
-        "aigc_write_record",
-        {
-            "success": True,
-            "effect_recorded": True,
-            "payload_hash": "abc",
-            "dry_run": False,
-            "external_request_made": False,
-            "candidate_count": 0,
-        },
-    )
-    assert empty_live_ok.passed is True
 
 
-def test_aigc_gate_rejects_dry_run_with_external_request() -> None:
+def test_aigc_gate_rejects_failed_batches() -> None:
     decision = evaluate_step_gate(
         "aigc_write_record",
         {
             "success": True,
             "effect_recorded": True,
             "payload_hash": "abc",
-            "dry_run": True,
-            "external_request_made": True,
+            "failed_batch_count": 2,
         },
     )
     assert decision.passed is False
-    assert decision.error_code == "external_effect_detected"
+    assert decision.error_code == "aigc_publish_failed"
 
 
 def test_explicit_step_failure_fails_closed() -> None:

tests/supply_infra/pipeline/test_health.py

@@ -8,10 +8,7 @@ from supply_infra.pipeline.health import run_alert_level
 
 
 def _settings() -> InfraSettings:
-    return InfraSettings(
-        _env_file=None,
-        PIPELINE_EXTERNAL_EFFECTS_ENABLED=True,
-    )
+    return InfraSettings(_env_file=None)
 
 
 def test_alert_level_escalates_against_next_schedule_deadline() -> None: