增加aigc发布

.env.example

@@ -68,7 +68,7 @@ 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=false
+PIPELINE_EXTERNAL_EFFECTS_ENABLED=true
 # Aliyun OSS (agent 运行日志可视化上传；qwen 视频截断后片段也上传到此 bucket)
 ALIYUN_OSS_ACCESS_KEY_ID=
 ALIYUN_OSS_ACCESS_KEY_SECRET=
@@ -81,6 +81,5 @@ ALIYUN_OSS_PUBLIC_BASE_URL=http://rescdn.yishihui.com
 ALIYUN_OSS_CONNECT_TIMEOUT_SECONDS=30
 LOG_OSS_UPLOAD_ENABLED=true
 
-# AIGC platform (视频爬取/发布计划)
+# AIGC platform (视频爬取/发布计划；需配置 token，默认真实调用接口)
 AIGC_API_TOKEN=
-AIGC_DRY_RUN=true

api/services/pipeline.py

@@ -2,6 +2,7 @@ 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 (
@@ -56,10 +57,11 @@ 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": True,
+        "dry_run": not settings.pipeline_external_effects_enabled,
         "steps": len(PIPELINE_STEPS),
         "latest_run": latest,
         **pipeline_health_snapshot(),

deploy/README.md

@@ -9,7 +9,8 @@ Scheduler、多个 Pipeline Worker 和 Reconciler。
 2. 挂载 `/app/logs/pipeline` 到持久化目录；
 3. `SCHEDULER_ENABLED=true` 只影响独立 Scheduler 进程，API 不再内嵌调度器；
    默认调度时间为 `Asia/Shanghai` 每日 15:00；
-4. `AIGC_DRY_RUN=true` 且 `PIPELINE_EXTERNAL_EFFECTS_ENABLED=false`；
+4. `PIPELINE_EXTERNAL_EFFECTS_ENABLED=true` 并配置 `AIGC_API_TOKEN`，日批最后一步
+   会真实创建/绑定 AIGC 计划；紧急演练可临时设为 `false`；
 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 | None = None) -> None:
+    def __init__(self, token: str | None = None, *, dry_run: bool = False) -> None:
         settings = get_infra_settings()
         self.token = (
             token
@@ -34,19 +34,8 @@ class AigcClient:
             or settings.aigc_api_token
             or ""
         ).strip()
-        if dry_run is None:
-            env_dry_run = os.getenv("AIGC_DRY_RUN", "").strip().lower()
-            if env_dry_run:
-                dry_run = env_dry_run in {"1", "true", "yes", "on"}
-            else:
-                dry_run = settings.aigc_dry_run
-        # Phase one safety barrier: callers cannot enable real side effects unless
-        # the global control-plane switch is also explicitly enabled.
-        self.dry_run = (
-            bool(dry_run)
-            or settings.aigc_dry_run
-            or not settings.pipeline_external_effects_enabled
-        )
+        # 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,

supply_infra/config.py

@@ -155,7 +155,7 @@ class InfraSettings(BaseSettings):
     )
     pipeline_log_dir: str = Field(default="logs/pipeline", alias="PIPELINE_LOG_DIR")
     pipeline_external_effects_enabled: bool = Field(
-        default=False,
+        default=True,
         alias="PIPELINE_EXTERNAL_EFFECTS_ENABLED",
     )
 
@@ -183,7 +183,6 @@ class InfraSettings(BaseSettings):
 
     # AIGC platform
     aigc_api_token: str = Field(default="", alias="AIGC_API_TOKEN")
-    aigc_dry_run: bool = Field(default=True, alias="AIGC_DRY_RUN")
 
     @model_validator(mode="after")
     def validate_pipeline_runtime(self) -> "InfraSettings":
@@ -208,14 +207,6 @@ class InfraSettings(BaseSettings):
                 "Pipeline connection budget exceeded: "
                 f"required={required} budget={self.mysql_connection_budget}"
             )
-        if (
-            self.pipeline_external_effects_enabled
-            and not self.aigc_dry_run
-        ):
-            raise ValueError(
-                "Real AIGC effects are disabled in the current migration phase; "
-                "set AIGC_DRY_RUN=true"
-            )
         return self
 
     @property

supply_infra/db/models/pipeline_outbox.py

@@ -20,7 +20,7 @@ from supply_infra.db.base import Base
 
 
 class PipelineOutbox(Base):
-    """Dry-run ledger of intended AIGC writes; it is not dispatched in phase one."""
+    """Ledger of AIGC write effects (dry-run recorded or dispatched)."""
 
     __tablename__ = "pipeline_outbox"
     __table_args__ = (

supply_infra/db/repositories/pipeline_outbox_repo.py

@@ -26,6 +26,29 @@ class PipelineOutboxRepository(BaseRepository[PipelineOutbox]):
         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,
+        *,
+        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,
+        dry_run: bool = True,
     ) -> PipelineOutbox:
         existing = self.get_by_idempotency_key(idempotency_key)
         if existing is not None:
@@ -40,8 +63,8 @@ class PipelineOutboxRepository(BaseRepository[PipelineOutbox]):
                 payload_hash=payload_hash,
                 payload_json=payload,
                 payload_uri=payload_uri,
-                status="dry_run_recorded",
-                dry_run=True,
+                status="dry_run_recorded" if dry_run else "dispatched",
+                dry_run=dry_run,
             )
         )
 

supply_infra/pipeline/enums.py

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

supply_infra/pipeline/gates.py

@@ -32,17 +32,26 @@ def evaluate_step_gate(step_key: str, payload: dict[str, Any]) -> GateDecision:
             )
 
     if step_key == "aigc_write_record":
-        if payload.get("dry_run_recorded") is not True or not payload.get("payload_hash"):
+        if payload.get("effect_recorded") is not True or not payload.get("payload_hash"):
             return GateDecision(
                 False,
                 "aigc_record_incomplete",
-                "AIGC dry-run payload/hash was not durably recorded",
+                "AIGC publish payload/hash was not durably recorded",
             )
-        if payload.get("external_request_made") is True:
+        # 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 external request is forbidden in phase one",
+                "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(
+                False,
+                "aigc_publish_failed",
+                f"AIGC publish reported {failed_batches} failed batch(es)",
             )
 
     return GateDecision(True)

supply_infra/pipeline/registry.py

@@ -109,10 +109,14 @@ def _aigc_write_record(context: StepContext) -> dict[str, Any]:
         publish_videos_from_discovery,
     )
 
+    settings = get_infra_settings()
+    dry_run = not settings.pipeline_external_effects_enabled
     payload = publish_videos_from_discovery(
         biz_dt=context.biz_dt,
-        dry_run=True,
+        dry_run=dry_run,
     )
+    publish_success = bool(payload.get("success", True))
+    effective_dry_run = bool(payload.get("dry_run", dry_run))
     canonical = json.dumps(
         payload,
         ensure_ascii=False,
@@ -126,7 +130,7 @@ def _aigc_write_record(context: StepContext) -> dict[str, Any]:
     payload_uri: str | None = None
     stored_payload: dict[str, Any] | None = payload
     if len(encoded) > _MAX_INLINE_EFFECT_BYTES:
-        log_dir = Path(get_infra_settings().pipeline_log_dir)
+        log_dir = Path(settings.pipeline_log_dir)
         if not log_dir.is_absolute():
             log_dir = _REPO_ROOT / log_dir
         effect_dir = log_dir / context.run_id / "effects"
@@ -142,7 +146,7 @@ def _aigc_write_record(context: StepContext) -> dict[str, Any]:
             "payload_hash": payload_hash,
         }
     with get_session() as session:
-        record = PipelineOutboxRepository(session).record_dry_run(
+        record = PipelineOutboxRepository(session).record_effect(
             run_id=context.run_id,
             step_run_id=context.step_run_id,
             effect_type="aigc_write_record",
@@ -150,18 +154,30 @@ 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
-    return {
-        "success": True,
-        "dry_run_recorded": True,
-        "external_request_made": False,
+
+    result: dict[str, Any] = {
+        "success": publish_success,
+        "effect_recorded": True,
+        "dry_run_recorded": effective_dry_run,
+        "external_request_made": not effective_dry_run,
         "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(
+            payload.get("error")
+            or f"AIGC publish failed ({result['failed_batch_count']} batch(es))"
+        )
+        result["error_code"] = "aigc_publish_failed"
+    return result
 
 
 STEP_REGISTRY: dict[str, StepHandler] = {

supply_infra/pipeline/run_service.py

@@ -36,6 +36,7 @@ class RunSubmission:
     created: bool
     status: str
     biz_dt: str
+    dry_run: bool = False
 
     def to_dict(self) -> dict[str, Any]:
         return {
@@ -44,7 +45,7 @@ class RunSubmission:
             "created": self.created,
             "status": self.status,
             "biz_dt": self.biz_dt,
-            "dry_run": True,
+            "dry_run": self.dry_run,
         }
 
 
@@ -58,8 +59,7 @@ 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,
-        "aigc_dry_run": True,
-        "external_effects_enabled": False,
+        "external_effects_enabled": settings.pipeline_external_effects_enabled,
     }
 
 
@@ -93,6 +93,7 @@ def submit_pipeline_run(
                     created=False,
                     status=existing.status,
                     biz_dt=existing.biz_dt,
+                    dry_run=bool(existing.dry_run),
                 )
 
             # All entry points share the same no-overlap rule. Linking each new
@@ -104,6 +105,7 @@ 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,
@@ -115,7 +117,7 @@ def submit_pipeline_run(
                     "triggered_by": None,
                     "trigger_reason": trigger_reason,
                     "run_mode": "full",
-                    "dry_run": True,
+                    "dry_run": run_dry_run,
                     "status": initial_status,
                     "scheduled_for": scheduled_for,
                     "deadline_at": deadline_for_trigger(
@@ -170,6 +172,7 @@ def submit_pipeline_run(
                 created=True,
                 status=run.status,
                 biz_dt=run.biz_dt,
+                dry_run=run_dry_run,
             )
     except IntegrityError:
         with get_session() as session:
@@ -181,6 +184,7 @@ def submit_pipeline_run(
                 created=False,
                 status=existing.status,
                 biz_dt=existing.biz_dt,
+                dry_run=bool(existing.dry_run),
             )
 
 

supply_infra/scheduler/jobs/publish_videos_from_discovery.py

@@ -102,11 +102,7 @@ def publish_videos_from_discovery(
     仅处理 decision_bucket 为 primary / backup 且 aweme_id 非空的候选，不区分品类。
     """
     settings = get_infra_settings()
-    effective_dry_run = (
-        dry_run
-        or settings.aigc_dry_run
-        or not settings.pipeline_external_effects_enabled
-    )
+    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:

tests/supply_infra/pipeline/test_aigc_safety.py

@@ -7,6 +7,15 @@ 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,
+    )
+    monkeypatch.setattr(
+        "supply_infra.aigc.client.get_infra_settings",
+        lambda: settings,
+    )
+
     called = False
 
     def _unexpected_post(*_args, **_kwargs):
@@ -21,14 +30,37 @@ def test_aigc_client_forces_dry_run_when_external_effects_are_disabled(
     assert called is False
 
 
-def test_global_aigc_dry_run_cannot_be_disabled_by_caller(monkeypatch) -> None:
+def test_aigc_client_calls_api_by_default_when_effects_enabled(monkeypatch) -> None:
     settings = InfraSettings(
+        _env_file=None,
         PIPELINE_EXTERNAL_EFFECTS_ENABLED=True,
-        AIGC_DRY_RUN=True,
     )
     monkeypatch.setattr(
         "supply_infra.aigc.client.get_infra_settings",
         lambda: settings,
     )
-    client = AigcClient(token="secret", dry_run=False)
-    assert client.dry_run is True
+    client = AigcClient(token="secret")
+    assert client.dry_run is False
+
+
+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 _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=True)
+    result = client.create_video_crawler_plan(["123"])
+    assert result["dry_run"] is True
+    assert called is False

tests/supply_infra/pipeline/test_config_budget.py

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

tests/supply_infra/pipeline/test_dag_and_gates.py

@@ -13,19 +13,47 @@ def test_pipeline_has_twelve_strictly_ordered_steps() -> None:
         assert current.critical is True
 
 
-def test_aigc_gate_requires_durable_dry_run_record() -> None:
+def test_aigc_gate_requires_durable_effect_record() -> None:
     failed = evaluate_step_gate("aigc_write_record", {"success": True})
     assert failed.passed is False
-    passed = evaluate_step_gate(
+    live_ok = evaluate_step_gate(
         "aigc_write_record",
         {
             "success": True,
-            "dry_run_recorded": 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:
+    decision = evaluate_step_gate(
+        "aigc_write_record",
+        {
+            "success": True,
+            "effect_recorded": True,
+            "payload_hash": "abc",
+            "dry_run": True,
+            "external_request_made": True,
         },
     )
-    assert passed.passed is True
+    assert decision.passed is False
+    assert decision.error_code == "external_effect_detected"
 
 
 def test_explicit_step_failure_fails_closed() -> None:

tests/supply_infra/pipeline/test_dates.py

@@ -19,7 +19,6 @@ def _settings() -> InfraSettings:
         SCHEDULER_TIMEZONE="Asia/Shanghai",
         SCHEDULER_CRON_HOUR=15,
         SCHEDULER_CRON_MINUTE=0,
-        AIGC_DRY_RUN=True,
     )
 
 

tests/supply_infra/pipeline/test_health.py

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

tests/supply_infra/scheduler/test_run_supply_pipeline.py

@@ -20,7 +20,7 @@ def test_pipeline_wrapper_only_submits_durable_run(mock_submit) -> None:
 
     assert result["run_id"] == "run-1"
     assert result["status"] == "queued"
-    assert result["dry_run"] is True
+    assert result["dry_run"] is False
     mock_submit.assert_called_once()