From 51f236776a10192166f4aacf0482541536eb6a20 Mon Sep 17 00:00:00 2001 From: lingying Date: Mon, 14 Sep 2026 01:13:52 +0800 Subject: [PATCH 1/2] =?UTF-8?q?fix(worker):=20=E4=BF=AE=E5=A4=8D=E6=89=B9?= =?UTF-8?q?=E9=87=8F=E5=8F=98=E4=BD=933=E4=B8=AAP0=20bug=E2=80=94=E2=80=94?= =?UTF-8?q?=E9=85=8D=E9=9F=B3=E6=97=B6=E9=95=BF=E5=8E=BB=E9=87=8D/?= =?UTF-8?q?=E7=B4=A0=E6=9D=90=E5=8C=BA=E9=97=B4=E6=9C=AA=E4=BC=A0=E9=80=92?= =?UTF-8?q?/=E8=8A=82=E5=A5=8F=E6=A8=A1=E6=9D=BF=E6=97=B6=E6=9C=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- apps/api/app/api/routes/generation_tasks.py | 57 ++++- apps/api/app/services/edit_plan_service.py | 229 ++++++++++++++------ 2 files changed, 209 insertions(+), 77 deletions(-) diff --git a/apps/api/app/api/routes/generation_tasks.py b/apps/api/app/api/routes/generation_tasks.py index c322a4f80..152974ac7 100755 --- a/apps/api/app/api/routes/generation_tasks.py +++ b/apps/api/app/api/routes/generation_tasks.py @@ -61,24 +61,38 @@ def _query_voice_durations(db: Session, voice_ids: list[str]) -> list[float]: """批量查询配音素材时长（秒），#1749 配音时长分配用。 逐项 try/float 硬化：MagicMock/异常/缺失 → 0.0（无配音不分配，不阻断）。 + + #1855 P0修复：不再对 voice_ids 去重，保持与调用方传入顺序/长度一致， + 允许同配音id多次出现时返回相同时长（支持"同配音N变体"的时长对齐）。 """ - ids = [v for v in dict.fromkeys(voice_ids or []) if v] - if not ids: + # 先去重查询（IN 查询性能优化），但最终按原始 voice_ids 顺序返回 + raw_ids = list(voice_ids or []) + if not raw_ids: return [] + # 去重且保序，用于 SQL IN 查询；空字符串/None 视为无效id → 0.0 + unique_ids: list[str] = [] + _seen: set[str] = set() + for v in raw_ids: + if v and v not in _seen: + _seen.add(v) + unique_ids.append(v) + if not unique_ids: + return [0.0 for _ in raw_ids] try: from packages.adapters.sqlalchemy_impl.models import AssetModel - rows = db.query(AssetModel.id, AssetModel.duration).filter(AssetModel.id.in_(ids)).all() + rows = db.query(AssetModel.id, AssetModel.duration).filter(AssetModel.id.in_(unique_ids)).all() dur_map: dict[str, float] = {} for row in rows: try: dur_map[row[0]] = float(row[1] or 0.0) except (TypeError, ValueError): dur_map[row[0]] = 0.0 - return [dur_map.get(v, 0.0) for v in ids] + # 按原始 voice_ids 顺序返回，保持长度一致；空/None/未查到 → 0.0 + return [dur_map.get(v, 0.0) if v else 0.0 for v in raw_ids] except Exception: logger.warning("[生成任务] 配音时长查询失败（按无配音处理，不阻断）", exc_info=True) - return [0.0 for _ in ids] + return [0.0 for _ in raw_ids] def _to_generation_task_response(task) -> GenerationTaskResponse: @@ -552,7 +566,26 @@ def create_generation_task( ) from clone_err variant_plan_ids.append(_plan0.id) - # 变体 1..N-1 独立选片 + # #1855 P0：批次区间避让表，从变体0实际clips构建初始值 + def _collect_segments(pid): + segs = {} + _sk, _pg = 0, 500 + while True: + _b = _plan_svc._clip_repo.list_by_plan(pid, skip=_sk, limit=_pg) + if not _b: + break + for _c in _b: + if _c.asset_id and float(_c.duration or 0) > 0: + _st = float(_c.start_time or 0.0) + segs.setdefault(_c.asset_id, []).append((_st, _st + float(_c.duration))) + if len(_b) < _pg: + break + _sk += _pg + return segs + + _batch_segments = _collect_segments(_plan0.id) + + # 变体 1..N-1 独立选片（传入累积batch_segments做素材区间避让） for task_index in range(1, count): variant = None last_err: Exception | None = None @@ -564,6 +597,7 @@ def create_generation_task( created_by_user_id=user_id, name_suffix=f"批量{task_index + 1}", voice_duration=voice_durations[task_index] if task_index < len(voice_durations) else 0.0, + batch_segments=_batch_segments, ) break except ValueError as ve: @@ -595,7 +629,16 @@ def create_generation_task( ) from last_err variant_plan_ids.append(variant.id) - # ③ 配音时长分配（回传 plan / clone 变体0 均需幂等分配；reselect 已在选片时分配） + # #1855 P0：把新变体的clips区间追加到batch_segments，供下一变体避让 + try: + _new_segs = _collect_segments(variant.id) + for _aid, _ivs in _new_segs.items(): + _batch_segments.setdefault(_aid, []).extend(_ivs) + except Exception: + logger.exception("[生成任务] 变体%d 区间收集失败（不阻断）", task_index) + + # ③ 配音时长分配（回传 plan / clone 变体0 均需幂等分配；reselect 已在选片时分配， + # #1855：apply_voice_duration_to_plan 已内置幂等判断，重复调用安全） for _vi, _pid in enumerate(variant_plan_ids): _vd = voice_durations[_vi] if _vi < len(voice_durations) else 0.0 if _vd > 0: diff --git a/apps/api/app/services/edit_plan_service.py b/apps/api/app/services/edit_plan_service.py index 1d66f366d..81a962b3b 100755 --- a/apps/api/app/services/edit_plan_service.py +++ b/apps/api/app/services/edit_plan_service.py @@ -473,6 +473,7 @@ class EditPlanService: name_suffix: str = "变体", voice_duration: float = 0.0, rng=None, + batch_segments: dict[str, list[tuple[float, float]]] | None = None, ) -> EditPlan: """为批量变体生成独立 plan：完整重跑单视频选片流程（#1743）。 @@ -489,6 +490,8 @@ class EditPlanService: created_by_user_id: 新 plan 归属用户。 name_suffix: plan 名后缀。 rng: 可选随机数（测试注入种子）。 + batch_segments: 可选，外部传入的批次内已使用素材区间（前序变体避让用）。 + 传入时作为初始避让对象；未传则保持原逻辑从源 plan clips 自建（向后兼容）。 Raises: ValueError: 源 plan 不存在/无片段、素材池为空或时长全未知。 @@ -537,16 +540,26 @@ class EditPlanService: voice = float(voice_duration or 0.0) except (TypeError, ValueError): voice = 0.0 + rhythm_template_for_reselect = None + if source.config: + rhythm_template_for_reselect = source.config.get("rhythm_template") + if voice > 0 and source_clips_data: from packages.domain.voice_duration_planner import plan_clip_durations _effects: list[str | None] = [c.get("transition_effect") for c in source_clips_data] _tdurs: list[float] = [float(c.get("transition_duration") or 0.0) for c in source_clips_data] + # #1855 P0：先占位durations为空dict，真正查durations在后面pool_ids确定后执行； + # plan_clip_durations 的 asset_durations 参数在该函数中仅作最大段长钳制， + # 这里先不依赖它（durations 还没查），传 None 让planner用默认策略； + # 真正的asset_durations会在后面 clips_data 生成时传入 reselect_clips_for_variant target_durations = plan_clip_durations( len(source_clips_data), voice, transition_effects=_effects, transition_durations=_tdurs, + rhythm_template=rhythm_template_for_reselect, + asset_durations=None, ) if target_durations: for _c, _d in zip(source_clips_data, target_durations, strict=False): @@ -582,12 +595,18 @@ class EditPlanService: created_by_user_id=created_by_user_id or (source.created_by_user_id or ""), ) - # 批次内区间：以源 plan（变体 0）片段为初始避让对象 - batch_segments: dict[str, list[tuple[float, float]]] = {} - for c in clips: - if c.asset_id and float(c.duration or 0) > 0: - st = float(c.start_time or 0.0) - batch_segments.setdefault(c.asset_id, []).append((st, st + float(c.duration))) + # 批次内区间：外部传入时使用外部传入（含前序变体已用区间）； + # 否则保持原逻辑从源 plan clips 自建（向后兼容） + if batch_segments is not None: + batch_segments_resolved: dict[str, list[tuple[float, float]]] = { + k: list(v) for k, v in batch_segments.items() + } + else: + batch_segments_resolved = {} + for c in clips: + if c.asset_id and float(c.duration or 0) > 0: + st = float(c.start_time or 0.0) + batch_segments_resolved.setdefault(c.asset_id, []).append((st, st + float(c.duration))) clips_data = reselect_clips_for_variant( source_clips_data, @@ -595,7 +614,7 @@ class EditPlanService: asset_durations=durations, asset_scene_points=scene_points, historical_used_segments=historical, - batch_segments=batch_segments, + batch_segments=batch_segments_resolved, target_durations=target_durations, rng=rng, ) @@ -767,6 +786,17 @@ class EditPlanService: if plan is None: return None + # #1855 P0：幂等判断——如果已成功分配过且当前 total_duration 已接近 voice_duration，直接返回 + try: + existing_mark = None + if plan.config: + existing_mark = plan.config.get("voice_duration_applied") + cur_total = float(plan.total_duration or 0.0) + if existing_mark is not None and abs(existing_mark - voice) < 1e-6 and abs(cur_total - voice) < 0.5: + return plan + except Exception: + pass + clips: List[EditPlanClip] = [] skip, page = 0, 500 while True: @@ -838,6 +868,10 @@ class EditPlanService: ) try: plan.total_duration = net + # #1855 P0：写入幂等标记，避免二次调用时只重分配 duration 不重算 start_time + new_cfg = dict(plan.config or {}) + new_cfg["voice_duration_applied"] = voice + plan.config = new_cfg db = self._clip_repo.session db.commit() except Exception: @@ -878,6 +912,71 @@ class EditPlanService: rng = rng or _random.Random() plan_ids: list[str] = [] + # #1855 P0：先确定片段数 clip_count（用于节奏模板生成长度匹配） + from packages.domain.voice_duration_planner import RHYTHM_TEMPLATES, adapt_template_length + from packages.domain.bgm_pool import allocate_bgm_pool_for_variants + from packages.domain.variant_plan_selector import ( + generate_visual_perturbation, + generate_pixel_perturbation, + ) + + clip_count = 0 + # 从源 plan 获取片段数（分页读，避免关系加载问题） + _sclips: list = [] + _sk, _pg = 0, 500 + while True: + _b = self._clip_repo.list_by_plan(source_plan_id, skip=_sk, limit=_pg) + if not _b: + break + _sclips.extend(_b) + if len(_b) < _pg: + break + _sk += _pg + clip_count = len(_sclips) + + # 预先生成所有 N 个变体的节奏模板/BGM/扰动参数（时机提前到选片前写入config） + rhythm_templates_for_variants: list = [] + for _idx in range(count): + if clip_count > 0: + variant_seed = rng.randint(0, 999999) + _tpl = adapt_template_length( + RHYTHM_TEMPLATES[variant_seed % len(RHYTHM_TEMPLATES)], clip_count + ) + rhythm_templates_for_variants.append(_tpl) + else: + rhythm_templates_for_variants.append(None) + + source_bgm_config: dict = {} + source_plan = self.get_plan(source_plan_id) + if source_plan and source_plan.config: + source_bgm_config = source_plan.config.get("bgm", {}) or {} + variant_seeds_for_bgm = [rng.randint(0, 999999) for _ in range(count)] + bgm_pool_assignments = allocate_bgm_pool_for_variants(source_bgm_config, variant_seeds_for_bgm) + + def _build_variant_config_update(idx: int) -> dict: + """构建单个变体的 config 更新（节奏模板/BGM/视觉/像素扰动）。""" + upd: dict = {} + try: + perturbation = generate_visual_perturbation(rng) + if idx == 0: + perturbation["hflip"] = False + upd["visual_perturbation"] = perturbation + except Exception: + logger.exception("变体 %d 视觉扰动生成失败（不阻断）", idx) + try: + pixel_pert = generate_pixel_perturbation(rng) + upd["pixel_perturbation"] = pixel_pert + except Exception: + logger.exception("变体 %d 像素扰动生成失败（不阻断）", idx) + rt = rhythm_templates_for_variants[idx] if idx < len(rhythm_templates_for_variants) else None + if rt is not None: + upd["rhythm_template"] = rt + if idx < len(bgm_pool_assignments): + existing_bgm = dict((source_plan.config or {}).get("bgm", {}) or {}) + existing_bgm.update(bgm_pool_assignments[idx]) + upd["bgm"] = existing_bgm + return upd + # 变体 0：clone（片段结构同源 plan，起点重算），不污染源 plan plan0 = self.clone_plan_for_variant( source_plan_id, @@ -890,6 +989,15 @@ class EditPlanService: v0_voice = float(voice_durations[0] or 0.0) except (TypeError, ValueError): v0_voice = 0.0 + + # #1855 P0：在配音分配前先写入变体0的节奏模板/扰动/BGM，确保 apply_voice_duration_to_plan 能读到 rhythm_template + try: + _cfg0 = _build_variant_config_update(0) + if _cfg0: + self.update_plan_config(plan0.id, _cfg0) + except Exception: + logger.exception("变体0 配置写入失败（不阻断）: plan=%s", plan0.id) + if v0_voice > 0: try: self.apply_voice_duration_to_plan(plan0.id, v0_voice) @@ -897,7 +1005,27 @@ class EditPlanService: logger.exception("变体0 配音分配失败（不阻断）: plan=%s", plan0.id) plan_ids.append(plan0.id) - # 变体 1..N-1：独立选片 + # #1855 P0：批次内素材区间避让表——从变体0实际落库的clips构建初始值 + def _collect_plan_segments(pid: str) -> dict[str, list[tuple[float, float]]]: + """分页读取 plan 所有 clips，构建 {asset_id: [(start, end), ...]} 区间表。""" + segs: dict[str, list[tuple[float, float]]] = {} + _sk2, _pg2 = 0, 500 + while True: + _b2 = self._clip_repo.list_by_plan(pid, skip=_sk2, limit=_pg2) + if not _b2: + break + for _c in _b2: + if _c.asset_id and float(_c.duration or 0) > 0: + _st = float(_c.start_time or 0.0) + segs.setdefault(_c.asset_id, []).append((_st, _st + float(_c.duration))) + if len(_b2) < _pg2: + break + _sk2 += _pg2 + return segs + + batch_segments_acc: dict[str, list[tuple[float, float]]] = _collect_plan_segments(plan0.id) + + # 变体 1..N-1：独立选片（传入累积的 batch_segments 做区间避让） for i in range(1, count): voice = 0.0 if voice_durations and i < len(voice_durations): @@ -905,6 +1033,14 @@ class EditPlanService: voice = float(voice_durations[i] or 0.0) except (TypeError, ValueError): voice = 0.0 + + # #1855 P0：在reselect前先为"变体i"准备配置更新——但reselect内部复制的是source.config， + # 所以每个变体独立的节奏模板需要在reselect后单独写入config； + # 但 plan_clip_durations 用的是 source.config.rhythm_template（即源plan的节奏模板）， + # 为了让每个变体在选片阶段就使用自己的节奏模板分配段长，这里采用： + # - reselect 仍使用源 plan 的 rhythm_template（保持片段骨架一致） + # - 选片完成后立即写入该变体自己的 rhythm_template/扰动/BGM 到config， + # 后续不再二次 apply_voice_duration_to_plan（由幂等标记跳过） variant = self.reselect_plan_for_variant( source_plan_id, candidate_asset_ids, @@ -912,73 +1048,26 @@ class EditPlanService: name_suffix=f"变体{i + 1}", voice_duration=voice, rng=rng, + batch_segments=batch_segments_acc, ) + + # 选片完成后写入该变体的独立配置（节奏模板/扰动/BGM） + try: + _cfgi = _build_variant_config_update(i) + if _cfgi: + self.update_plan_config(variant.id, _cfgi) + except Exception: + logger.exception("变体 %d 配置写入失败（不阻断）: plan=%s", i, variant.id) + plan_ids.append(variant.id) - # #1764：为每个变体生成独立节奏模板（让批量视频片段时长分布不同） - from packages.domain.voice_duration_planner import RHYTHM_TEMPLATES, adapt_template_length - - clip_count = 0 - if voice_durations and len(voice_durations) > 0: - # 从源 plan 获取片段数 - source_plan = self.get_plan(source_plan_id) - if source_plan and hasattr(source_plan, "clips"): - clip_count = len(list(source_plan.clips)) if source_plan.clips else 0 - - rhythm_templates_for_variants = [] - if clip_count > 0: - for idx in range(len(plan_ids)): - # 每个变体用不同的 seed 选择节奏模板 - variant_seed = rng.randint(0, 999999) - template = adapt_template_length(RHYTHM_TEMPLATES[variant_seed % len(RHYTHM_TEMPLATES)], clip_count) - rhythm_templates_for_variants.append(template) - logger.info("变体 %d 节奏模板: plan=%s template=%s", idx, plan_ids[idx], template) - - # #1767：BGM 池差异化分配（让批量变体使用不同 BGM / 段落 / 音量） - from packages.domain.bgm_pool import allocate_bgm_pool_for_variants - - source_bgm_config = {} - source_plan = self.get_plan(source_plan_id) - if source_plan and source_plan.config: - source_bgm_config = source_plan.config.get("bgm", {}) or {} - - variant_seeds_for_bgm = [rng.randint(0, 999999) for _ in plan_ids] - bgm_pool_assignments = allocate_bgm_pool_for_variants(source_bgm_config, variant_seeds_for_bgm) - - # 为每个变体生成独立视觉扰动参数（让批量视频画面本身更不同） - from packages.domain.variant_plan_selector import generate_visual_perturbation - - for idx, pid in enumerate(plan_ids): + # #1855 P0：把当前新变体的 clips 区间追加到 batch_segments，供下一变体避让 try: - perturbation = generate_visual_perturbation(rng) - # 变体 0 不做 hflip（保持预览 plan 原始画面方向） - if idx == 0: - perturbation["hflip"] = False - config_update = {"visual_perturbation": perturbation} - # #1764：写入节奏模板 - if idx < len(rhythm_templates_for_variants): - config_update["rhythm_template"] = rhythm_templates_for_variants[idx] - # #1765：写入像素级扰动滤镜 - from packages.domain.variant_plan_selector import generate_pixel_perturbation - - pixel_pert = generate_pixel_perturbation(rng) - config_update["pixel_perturbation"] = pixel_pert - # #1767：写入 BGM 池分配（覆盖 bgm 配置中的 preset_id / audio_offset / volume_adjust_db） - if idx < len(bgm_pool_assignments): - existing_bgm = dict((source_plan.config or {}).get("bgm", {}) or {}) - existing_bgm.update(bgm_pool_assignments[idx]) - config_update["bgm"] = existing_bgm - self.update_plan_config(pid, config_update) - logger.info( - "变体 %d 视觉扰动+像素扰动+BGM池: plan=%s vis=%s pix=%s bgm=%s", - idx, - pid, - perturbation, - pixel_pert, - bgm_pool_assignments[idx] if idx < len(bgm_pool_assignments) else None, - ) + _new_segs = _collect_plan_segments(variant.id) + for _aid, _ivs in _new_segs.items(): + batch_segments_acc.setdefault(_aid, []).extend(_ivs) except Exception: - logger.exception("变体 %d 视觉扰动生成失败（不阻断）: plan=%s", idx, pid) + logger.exception("变体 %d 区间收集失败（不阻断）: plan=%s", i, variant.id) # 标记所有变体 plan 的 clips 为 ready（已分配素材+起点，语义上就是 ready） for pid in plan_ids: -- 2.54.0 From e897a305d2ffe176998fe209c29d13409966061c Mon Sep 17 00:00:00 2001 From: CI Bot Date: Sun, 13 Sep 2026 17:19:00 +0000 Subject: [PATCH 2/2] style: auto-format with black + isort + ruff + prettier [skip ci-format-check] --- apps/api/app/services/edit_plan_service.py | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/apps/api/app/services/edit_plan_service.py b/apps/api/app/services/edit_plan_service.py index 81a962b3b..6f2a4949d 100755 --- a/apps/api/app/services/edit_plan_service.py +++ b/apps/api/app/services/edit_plan_service.py @@ -913,12 +913,12 @@ class EditPlanService: plan_ids: list[str] = [] # #1855 P0：先确定片段数 clip_count（用于节奏模板生成长度匹配） - from packages.domain.voice_duration_planner import RHYTHM_TEMPLATES, adapt_template_length from packages.domain.bgm_pool import allocate_bgm_pool_for_variants from packages.domain.variant_plan_selector import ( - generate_visual_perturbation, generate_pixel_perturbation, + generate_visual_perturbation, ) + from packages.domain.voice_duration_planner import RHYTHM_TEMPLATES, adapt_template_length clip_count = 0 # 从源 plan 获取片段数（分页读，避免关系加载问题） @@ -939,9 +939,7 @@ class EditPlanService: for _idx in range(count): if clip_count > 0: variant_seed = rng.randint(0, 999999) - _tpl = adapt_template_length( - RHYTHM_TEMPLATES[variant_seed % len(RHYTHM_TEMPLATES)], clip_count - ) + _tpl = adapt_template_length(RHYTHM_TEMPLATES[variant_seed % len(RHYTHM_TEMPLATES)], clip_count) rhythm_templates_for_variants.append(_tpl) else: rhythm_templates_for_variants.append(None) -- 2.54.0