Compare commits

..

1 Commits

Author SHA1 Message Date
xiaoxia adcfc1e919 perf(ci): ci-base 镜像预装 ffmpeg——消除 unit-tests 每次 24min apt 安装
CI/CD Pipeline / Dedup Check - skip PR tests when covered by push pipeline (pull_request) Successful in 3s
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 1s
CI/CD Pipeline / Check push changed paths (pull_request) Has been skipped
AI Code Review / AI Code Review (pull_request) Successful in 2m27s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 2m41s
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 2m40s
CI/CD Pipeline / Frontend Lint (pull_request) Has been skipped
CI/CD Pipeline / Frontend Unit Tests (pull_request) Has been skipped
CI/CD Pipeline / PR Build Web Image (pull_request) Has been skipped
PR Automation / Auto Approve on CI Green (pull_request) Successful in 4m57s
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / PR Build Worker Image (pull_request) Successful in 1m5s
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 1m42s
CI/CD Pipeline / Retag skipped Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Retag skipped Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Retag skipped Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / ACR Image Cleanup (pull_request) Has been skipped
CI/CD Pipeline / Validate - Python (mypy + alembic) (pull_request) Successful in 4m59s
CI/CD Pipeline / Validate - Style (pull_request) Successful in 5m6s
CI/CD Pipeline / Integration Tests (pull_request) Successful in 7m35s
CI/CD Pipeline / Validate - Security (pull_request) Successful in 8m15s
CI/CD Pipeline / Unit Tests (pull_request) Successful in 16m23s
CI/CD Pipeline / Build Production Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been skipped
CI/CD Pipeline / CI Gate (pull_request) Successful in 1s
CI/CD Pipeline / Deploy Production (pull_request) Has been skipped
CI/CD Pipeline / Canary Release to Production (pull_request) Has been skipped
CI/CD Pipeline / Production Browser E2E (pull_request) Has been skipped
Preview Cleanup / Cleanup Preview Environment (pull_request) Successful in 5m15s
ACR Cleanup / ACR Image Cleanup (pull_request_target) Successful in 5m16s
unit-tests job 跑在全新容器里,ci-base 镜像未预装 ffmpeg,
step_install_ffmpeg.sh 每次都 apt-get update + install ffmpeg,
ffmpeg 拉取 x264/aom/systemd 等数百个编解码依赖,固定耗时约 24 分钟,
pytest 本身 21-23 分钟,导致该 job 每次 45-50 分钟。

把 ffmpeg 预装到 ci-base 镜像后,step_install_ffmpeg.sh 检测到
ffmpeg 已存在会立即退出(exit 0),unit-tests 直接省掉 ~24 分钟。

注:ci-base 镜像需重建并推送 registry 后生效。
2026-08-31 23:54:22 +08:00
29 changed files with 104 additions and 953 deletions
-5
View File
@@ -150,7 +150,6 @@ jobs:
run: bash scripts/ci/step_timer_start.sh
- name: Cache pip dependencies
uses: actions/cache@v4
continue-on-error: true
with:
path: /root/.cache/pip
key: ${{ runner.os }}-pip-style-${{ hashFiles('requirements*.txt') }}
@@ -253,7 +252,6 @@ jobs:
run: bash scripts/ci/step_timer_start.sh
- name: Cache pip dependencies
uses: actions/cache@v4
continue-on-error: true
with:
path: /root/.cache/pip
key: ${{ runner.os }}-pip-security-${{ hashFiles('requirements*.txt') }}
@@ -354,7 +352,6 @@ jobs:
run: bash scripts/ci/step_timer_start.sh
- name: Cache pip dependencies
uses: actions/cache@v4
continue-on-error: true
with:
path: /root/.cache/pip
key: ${{ runner.os }}-pip-python-${{ hashFiles('requirements*.txt') }}
@@ -459,7 +456,6 @@ jobs:
run: bash scripts/ci/step_install_ffmpeg.sh
- name: Cache pip dependencies
uses: actions/cache@v4
continue-on-error: true
with:
path: /root/.cache/pip
key: ${{ runner.os }}-pip-unittests-${{ hashFiles('requirements*.txt') }}
@@ -668,7 +664,6 @@ jobs:
run: bash scripts/ci/step_timer_start.sh
- name: Cache npm dependencies
uses: actions/cache@v4
continue-on-error: true
with:
path: /root/.npm
key: ${{ runner.os }}-npm-vitest-${{ hashFiles('apps/web/package-lock.json') }}
+1 -1
View File
@@ -3,7 +3,7 @@ name: PR Auto Scan
# 作为短作业模式的兜底,防止事件驱动遗漏
on:
schedule:
- cron: "*/15 * * * *" # 每10分钟扫描一次(脚本自带240s墙钟上限,降频减负)
- cron: "*/10 * * * *" # 每10分钟扫描一次(脚本自带240s墙钟上限,降频减负)
workflow_dispatch:
permissions:
+34 -48
View File
@@ -579,45 +579,39 @@ def smart_match_assets(
filtered_assets = asset_repository.find_by_library(request.library_id, status=["ready"], limit=10000)
total_candidates = len(filtered_assets)
# ── 过滤前置:余量 + 高频使用,过滤在评分/截取 limit 之前完成 ──────────
# 旧实现先 smart_select_assets(limit=N) 再对这 N 条做过滤,过滤后不回补,
# 当排名靠前的素材恰好都被排除时返回空 items(前端回退全选,smart-match 名存实亡)。
# 现在先过滤全量候选,每级过滤后为空/不足则回退上一级,最后才评分截取。
# 调用统一智能选素材算法(kind 已在 DB 层过滤,无需重复过滤)
results = smart_select_assets(
filtered_assets,
limit=request.limit,
kind=None,
)
# 1) 余量过滤:usable=False(零重复可切区间耗尽且历史区间均达复用上限)的素材排除
usable_assets = []
exhausted_assets = []
for a in filtered_assets:
# 结果层过滤:usable=false(零重复可切区间耗尽且历史区间均达复用上限)的素材
# 不返回给前端;不动 smart_select_assets 评分逻辑本身
filtered_results = []
for r in results:
try:
avail = compute_asset_availability(a)
avail = compute_asset_availability(r.asset)
except Exception:
logger.warning(
"smart-match 余量计算失败,按可用处理: asset_id=%s",
getattr(a, "id", "?"),
getattr(r.asset, "id", "?"),
exc_info=True,
)
avail = None
if avail is not None and not avail["usable"]:
exhausted_assets.append(a)
else:
usable_assets.append(a)
logger.info(
"smart-match 排除已用尽素材: asset_id=%s name=%s",
getattr(r.asset, "id", "?"),
getattr(r.asset, "name", ""),
)
continue
filtered_results.append(r)
if exhausted_assets:
logger.info(
"smart-match 余量过滤: 候选 %d,可切区间耗尽 %d",
len(filtered_assets), len(exhausted_assets),
)
# 回退策略:余量过滤后为空(全部耗尽)时,保留全部候选,不返回空结果。
# 宁可让用户在已耗尽素材上复用,也比 smart-match 空结果回退全选更可控
# (全选同样会选到这些素材,且不经过评分排序)。
pool = usable_assets if usable_assets else filtered_assets
# 2) 高频使用排除:同一素材在最近 5 个视频中出现超过 3 次则排除
# 高频使用排除:同一素材在最近 5 个视频中出现超过 3 次则排除
MAX_RECENT_USE_COUNT = 3
high_freq_assets = set()
if pool:
asset_ids = [getattr(a, "id", "") for a in pool if getattr(a, "id", "")]
if filtered_results:
asset_ids = [getattr(r.asset, "id", "") for r in filtered_results if getattr(r.asset, "id", "")]
if asset_ids:
try:
use_counts = get_asset_recent_use_counts(
@@ -625,35 +619,27 @@ def smart_match_assets(
asset_ids=asset_ids,
recent_video_count=5,
)
for a in pool:
aid = getattr(a, "id", "")
high_use_excluded = set()
for r in filtered_results:
aid = getattr(r.asset, "id", "")
count = use_counts.get(aid, 0)
if count > MAX_RECENT_USE_COUNT:
high_freq_assets.add(aid)
logger.info(
"smart-match 排除高频使用素材: asset_id=%s use_count=%d limit=%d",
aid, count, MAX_RECENT_USE_COUNT,
)
# 回退策略:排除后剩余素材不足(为空或不够 limit)时,
# 不再全部排除,保留全部可用素材
if high_freq_assets:
remaining_count = len(pool) - len(high_freq_assets)
enough = request.limit is None or remaining_count >= request.limit
if remaining_count > 0 and enough:
pool = [a for a in pool if getattr(a, "id", "") not in high_freq_assets]
high_use_excluded.add(id(r))
else:
logger.info(
"smart-match 高频排除后素材不足(%d<%s),保留全部 %d",
remaining_count,
request.limit if request.limit is not None else "不限",
len(pool),
)
pass
# 如果排除后不够 limit,放宽到不限制
remaining = [r for r in filtered_results if id(r) not in high_use_excluded]
if len(remaining) >= request.limit:
filtered_results = remaining
else:
logger.info("smart-match 高频排除后素材不足(%d<%d),保留全部", len(remaining), request.limit)
except Exception:
logger.warning("smart-match 高频使用查询失败,跳过排除", exc_info=True)
# 3) 调用统一智能选素材算法(kind 已在 DB 层过滤,无需重复过滤)
results = smart_select_assets(pool, limit=request.limit, kind=None)
# 扁平结构:SmartMatchItem 继承 AssetResponse,素材字段直接在条目顶层,
# 前端无需解析 item.asset 包装层,item.id / item.usable / 余量字段直接可读
items = [
@@ -662,7 +648,7 @@ def smart_match_assets(
score=r.score,
breakdown=r.breakdown,
)
for r in results
for r in filtered_results
]
return SmartMatchResponse(items=items, total_candidates=total_candidates)
@@ -37,9 +37,6 @@ from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query, s
from sqlalchemy.orm import Session
from packages.adapters.sqlalchemy_impl.asset_repository import SQLAlchemyAssetRepository
from packages.adapters.sqlalchemy_impl.template_clip_config_repository import (
SQLAlchemyTemplateClipConfigRepository,
)
from packages.adapters.sqlalchemy_impl.template_repository import (
SQLAlchemyTemplateRepository,
)
@@ -414,25 +411,7 @@ def _get_template_segments(
result.append((cc.order, dur_min, dur_max))
return sorted(result, key=lambda x: x[0])
except Exception:
logger.warning("新模板系统查询clip_configs失败(主表可能不存在),直接查clip_configs表", exc_info=True)
# 兜底:直接查 template_clip_configs 表(片段表有 template_id 外键,不依赖模板主表)
try:
direct_repo = SQLAlchemyTemplateClipConfigRepository(db)
direct_configs = direct_repo.list_by_template(template_id)
if direct_configs:
result = []
for cc in direct_configs:
dur_min = _safe_segment_duration(cc.min_duration, _DEFAULT_EDITOR_CLIP_DURATION)
dur_max = _safe_segment_duration(
cc.max_duration or cc.min_duration,
_DEFAULT_EDITOR_CLIP_DURATION,
)
dur_min, dur_max = min(dur_min, dur_max), max(dur_min, dur_max)
result.append((cc.order, dur_min, dur_max))
return sorted(result, key=lambda x: x[0])
except Exception:
logger.warning("直接查clip_configs表也失败,继续回退旧系统", exc_info=True)
logger.warning("新模板系统查询clip_configs失败,回退到旧系统", exc_info=True)
# 回退到旧模板系统(template_segments表)
try:
File diff suppressed because one or more lines are too long
+1 -4
View File
@@ -338,15 +338,12 @@ const GeneratePage: React.FC = () => {
size: titleSettings.size,
font: titleSettings.font,
color: titleSettings.color,
position: titleSettings.position as "top" | "center" | "bottom" | "custom",
position: titleSettings.position as "top" | "center" | "bottom",
bold: titleSettings.bold,
italic: titleSettings.italic,
stroke: titleSettings.stroke,
shadow: titleSettings.shadow,
posX: titleSettings.posX,
posY: titleSettings.posY,
}}
onTitlePositionChange={styleUpdaters.updateTitlePosition}
/>
)}
{currentStep >= 5 && generated && finalVideo && (
@@ -32,15 +32,12 @@ interface FrontendPreviewPlayerProps {
size: number
font: string
color: string
position: "top" | "center" | "bottom" | "custom"
position: "top" | "center" | "bottom"
bold?: boolean
italic?: boolean
stroke?: boolean
shadow?: boolean
posX?: number | null
posY?: number | null
}
onTitlePositionChange?: (posX: number, posY: number) => void
}
function formatTime(seconds: number): string {
@@ -108,7 +105,6 @@ const FrontendPreviewPlayer: React.FC<FrontendPreviewPlayerProps> = ({
serverClips,
voiceAudioUrl,
titleSettings,
onTitlePositionChange,
}) => {
const segments = useMemo(
() => buildPlaybackSegments(assets, template, serverClips),
@@ -130,55 +126,6 @@ const FrontendPreviewPlayer: React.FC<FrontendPreviewPlayerProps> = ({
return { width: 1080, height: 1920 }
}
})()
const customTitleXPct =
titleSettings?.posX != null && playRes.width > 0
? (titleSettings.posX / playRes.width) * 100
: null
const customTitleYPct =
titleSettings?.posY != null && playRes.height > 0
? (titleSettings.posY / playRes.height) * 100
: null
// ── 拖拽状态(用 ref 避免在每帧渲染中触发重渲染)──
const draggingTitleRef = useRef(false)
const handleTitlePointerDown = useCallback(
(e: React.PointerEvent<HTMLDivElement>) => {
if (!onTitlePositionChange || !playerContainerRef.current) return
e.preventDefault()
e.stopPropagation()
;(e.target as Element).setPointerCapture(e.pointerId)
draggingTitleRef.current = true
;(e.currentTarget as HTMLDivElement).style.cursor = "grabbing"
},
[onTitlePositionChange],
)
const handleTitlePointerMove = useCallback(
(e: React.PointerEvent<HTMLDivElement>) => {
if (!draggingTitleRef.current || !onTitlePositionChange || !playerContainerRef.current) return
e.preventDefault()
e.stopPropagation()
const rect = playerContainerRef.current.getBoundingClientRect()
const relX = Math.max(0, Math.min(rect.width, e.clientX - rect.left))
const relY = Math.max(0, Math.min(rect.height, e.clientY - rect.top))
const posX = Math.round((relX / rect.width) * playRes.width)
const posY = Math.round((relY / rect.height) * playRes.height)
onTitlePositionChange(posX, posY)
},
[onTitlePositionChange, playRes.width, playRes.height],
)
const handleTitlePointerUp = useCallback((e: React.PointerEvent<HTMLDivElement>) => {
if (!draggingTitleRef.current) return
draggingTitleRef.current = false
;(e.currentTarget as HTMLDivElement).style.cursor = "grab"
try {
if ((e.currentTarget as Element).hasPointerCapture(e.pointerId)) {
;(e.currentTarget as Element).releasePointerCapture(e.pointerId)
}
} catch {
/* ignore */
}
}, [])
const playerContainerRef = useRef<HTMLDivElement>(null)
const [containerHeight, setContainerHeight] = useState(0)
useEffect(() => {
@@ -577,34 +524,15 @@ const FrontendPreviewPlayer: React.FC<FrontendPreviewPlayerProps> = ({
<div
style={{
position: "absolute",
...(customTitleXPct != null && customTitleYPct != null
? {
left: `${customTitleXPct}%`,
top: `${customTitleYPct}%`,
transform: "translate(-50%, -50%)",
textAlign: "center" as const,
}
: {
left: `${titleSidePct}%`,
right: `${titleSidePct}%`,
textAlign: "center" as const,
...(titleSettings.position === "top"
? { top: `${titleTopPct}%` }
: titleSettings.position === "center"
? { top: "50%", transform: "translateY(-50%)" }
: { bottom: `${titleBottomPct}%` }),
}),
cursor: onTitlePositionChange ? "grab" : "default",
touchAction: "none",
userSelect: "none",
WebkitUserSelect: "none",
padding: "8px 12px",
boxShadow: "inset 0 0 0 16px transparent",
left: `${titleSidePct}%`,
right: `${titleSidePct}%`,
textAlign: "center",
...(titleSettings.position === "top"
? { top: `${titleTopPct}%` }
: titleSettings.position === "center"
? { top: "50%", transform: "translateY(-50%)" }
: { bottom: `${titleBottomPct}%` }),
}}
onPointerDown={handleTitlePointerDown}
onPointerMove={handleTitlePointerMove}
onPointerUp={handleTitlePointerUp}
onPointerCancel={handleTitlePointerUp}
>
<span
style={{
@@ -23,7 +23,7 @@ const MaterialModeTabs: React.FC<MaterialModeTabsProps> = ({ mode, onModeChange
onClick={() => onModeChange("auto")}
type="button"
>
AI智能匹配
</button>
</div>
)
@@ -38,8 +38,9 @@ const TitlePresetsGrid: React.FC<TitlePresetsGridProps> = ({
className="xx-title-preset-preview-text"
style={{ ...p.previewStyle, fontFamily: getFontFamily(fontFamily || "思源黑体") }}
>
T
</span>
<span className="xx-title-preset-card-label">{p.label}</span>
</button>
)
})}
-1
View File
@@ -42,7 +42,6 @@ export const POSITION_OPTIONS = [
{ value: "top", label: "顶部" },
{ value: "center", label: "居中" },
{ value: "bottom", label: "底部" },
{ value: "custom", label: "自定义" },
]
/* ── 标题字体选项 ── */
+15 -5
View File
@@ -1733,8 +1733,8 @@
/* 标题预设卡片网格 */
.xx-title-presets-grid {
display: grid;
grid-template-columns: repeat(6, 1fr);
gap: 8px;
grid-template-columns: repeat(4, 1fr);
gap: 10px;
}
.xx-title-preset-card {
@@ -1742,8 +1742,7 @@
flex-direction: column;
align-items: center;
justify-content: center;
aspect-ratio: 1 / 1;
padding: 4px;
padding: 14px 8px;
background: var(--bg-secondary);
border: 2px solid transparent;
border-radius: var(--radius-sm);
@@ -1763,10 +1762,21 @@
}
.xx-title-preset-preview-text {
line-height: 1;
line-height: 1.4;
margin-bottom: 6px;
user-select: none;
}
.xx-title-preset-card-label {
font-size: 11px;
color: var(--text-secondary);
}
.xx-title-preset-card.active .xx-title-preset-card-label {
color: var(--primary-color);
font-weight: 500;
}
/* 样式按钮组 */
.xx-style-btns {
display: flex;
@@ -20,14 +20,13 @@ const SECONDS_PER_ASSET = 15
/**
* 根据模板 segments 计算所需素材数量上限。
* 取每个 segment 的 duration_min 之和作为目标视频总时长,
* 再按 15 秒/素材估算需要多少个素材,且保证不少于片段数(每个片段至少 1 个素材);
* 结果钳制到 [1, 200] 区间(后端 limit 上限 200)。
* 再按 15 秒/素材估算需要多少个素材;结果钳制到 [1, 200] 区间(后端 limit 上限 200)。
*/
function computeLimitFromSegments(segments?: TemplateSegment[]): number {
if (!segments || segments.length === 0) return DEFAULT_LIMIT
const totalSeconds = segments.reduce((sum, seg) => sum + (seg.duration_min || 0), 0)
if (totalSeconds <= 0) return DEFAULT_LIMIT
const limit = Math.max(segments.length, Math.ceil(totalSeconds / SECONDS_PER_ASSET))
const limit = Math.ceil(totalSeconds / SECONDS_PER_ASSET)
return Math.max(1, Math.min(limit, 200))
}
@@ -233,7 +233,7 @@ export function useCanvasPlayer(
fontSize: number
fontFamily: string
color: string
position: "top" | "center" | "bottom" | "custom"
position: "top" | "center" | "bottom"
bold?: boolean
stroke?: boolean
shadow?: boolean
@@ -27,8 +27,6 @@ const DEFAULT_TITLE_SETTINGS: TitleSettings = {
stroke: true,
shadow: false,
color: "#ffffff",
posX: null,
posY: null,
}
export interface GenerateFormState {
@@ -117,14 +117,6 @@ export function useGenerateVideo(props: UseGenerateVideoProps) {
font_size: props.titleSettings.size,
font_color: props.titleSettings.color,
position: props.titleSettings.position,
...(props.titleSettings.position === "custom" &&
props.titleSettings.posX != null &&
props.titleSettings.posY != null
? {
pos_x: Math.round(props.titleSettings.posX),
pos_y: Math.round(props.titleSettings.posY),
}
: {}),
bold: props.titleSettings.bold,
stroke: props.titleSettings.stroke,
shadow: props.titleSettings.shadow,
@@ -46,16 +46,7 @@ export function useTitleStyleUpdaters({
const updatePosition = useCallback(
(position: string) => {
// 切回三档预设时清掉自定义坐标
onTitleSettingsChange({ ...titleSettings, position, posX: null, posY: null })
},
[titleSettings, onTitleSettingsChange],
)
/** 拖拽更新自定义位置(由预览播放器调用) */
const updateTitlePosition = useCallback(
(posX: number, posY: number) => {
onTitleSettingsChange({ ...titleSettings, position: "custom", posX, posY })
onTitleSettingsChange({ ...titleSettings, position })
},
[titleSettings, onTitleSettingsChange],
)
@@ -120,7 +111,6 @@ export function useTitleStyleUpdaters({
updateTitle,
toggleAiAutoSelect,
updatePosition,
updateTitlePosition,
updateFont,
updateSize,
updateColor,
-3
View File
@@ -16,9 +16,6 @@ export interface TitleSettings {
stroke: boolean
shadow: boolean
color: string
/** 自由位置坐标(PlayRes 像素),仅当 position="custom" 时有效 */
posX: number | null
posY: number | null
}
/* ── 智能匹配结果 ── */
@@ -11,7 +11,7 @@ import {
} from "@ant-design/icons"
import type { ProductItem } from "../types"
import { statusConfig, reviewStatusConfig } from "../constants"
import { formatTime, formatSize } from "../utils"
import { formatTime } from "../utils"
interface ProductCardProps {
product: ProductItem
@@ -210,20 +210,11 @@ export const ProductCard: React.FC<ProductCardProps> = ({
<span className={`xx-product-status ${st.className}`}>{st.text}</span>
<span className="xx-product-date">{product.date}</span>
</div>
<div className="xx-product-meta-grid">
<span className="xx-product-meta-item">{product.resolution || "-"}</span>
<span className="xx-product-meta-item">
{product.duration > 0 ? formatTime(product.duration) : "-"}
{product.duplicateRate > 0 && (
<span className={`xx-product-dup-rate ${dupClass}`}>
{product.duplicateRate.toFixed(1)}%
</span>
<span className="xx-product-meta-item">{formatSize(product.fileSize)}</span>
<span
className={`xx-product-meta-item xx-product-dup-rate${
product.duplicateRate > 0 ? ` ${dupClass}` : ""
}`}
>
{product.duplicateRate > 0 ? `${product.duplicateRate.toFixed(1)}%` : "-"}
</span>
</div>
)}
</div>
{/* 操作按钮 */}
+2 -17
View File
@@ -276,11 +276,11 @@
background-position: center;
}
/* 内联视频播放器(cover 填满容器,竖屏视频不留左右空白) */
/* 内联视频播放器 */
.xx-product-thumb-video {
width: 100%;
height: 100%;
object-fit: cover;
object-fit: contain;
display: block;
}
@@ -347,21 +347,6 @@
gap: var(--space-xs);
}
/* 卡片信息网格:分辨率/时长 一行,大小/查重率 一行 */
.xx-product-meta-grid {
display: grid;
grid-template-columns: 1fr 1fr;
gap: 4px var(--space-sm);
}
.xx-product-meta-item {
font-size: var(--font-size-xs);
color: var(--text-secondary);
white-space: nowrap;
overflow: hidden;
text-overflow: ellipsis;
}
/* 状态标签 */
.xx-product-status {
padding: 2px 10px;
+1 -50
View File
@@ -256,42 +256,6 @@ def _wrap_title_text(
return "\\N".join(wrapped_segments)
def _parse_title_position(
title_config: dict[str, Any],
video_width: int,
video_height: int,
) -> tuple[int, int] | None:
"""解析标题自由拖拽坐标 pos_x/pos_yPlayRes 像素坐标系)。
要求两个字段同时存在、可转 int,且落在 [0, video_width] × [0, video_height]
闭区间内。任一条件不满足返回 None,调用方回退 position 三档逻辑。
Args:
title_config: 标题配置 dict
video_width: PlayResX(视频宽度像素)
video_height: PlayResY(视频高度像素)
Returns:
(x, y) 整数坐标,或 None 表示不使用自由位置
"""
if "pos_x" not in title_config or "pos_y" not in title_config:
return None
raw_x = title_config["pos_x"]
raw_y = title_config["pos_y"]
# 坐标必须是 PlayRes 像素整数:bool 是 int 子类(isinstance(True,int)=True
# 但 True/False 作坐标无意义;float 静默截断会造成拖拽位置偏差,一律按非法回退
if isinstance(raw_x, bool) or isinstance(raw_y, bool):
return None
if not isinstance(raw_x, int) or not isinstance(raw_y, int):
return None
x, y = raw_x, raw_y
if video_width <= 0 or video_height <= 0:
return None
if not (0 <= x <= video_width and 0 <= y <= video_height):
return None
return (x, y)
def build_ass_content(
*,
video_width: int,
@@ -379,16 +343,7 @@ def build_ass_content(
title_shadow.get("offset_y", 2) if title_shadow.get("enabled", False) else 0,
)
# ── 自由位置拖拽(工单 #1405 方案 B)────────────────────────────
# pos_x/pos_y 为 PlayRes 坐标系像素整数(PlayResX/Y = video_width/height)。
# 合法时:TitleStyle Alignment 固定 5\an5 中对齐,使 \pos 锚点为文本块中心),
# Dialogue 文本前注入 {\pos(x,y)}。字段缺失/非法/越界时一律回退
# position → alignment 三档逻辑,现有输出保持一字节不变。
title_pos = _parse_title_position(title_config, video_width, video_height)
title_alignment = 5 if title_pos is not None else position_to_ass_alignment(
title_config.get("position", "top")
)
title_alignment = position_to_ass_alignment(title_config.get("position", "top"))
styles.append(
build_ass_style(
@@ -415,10 +370,6 @@ def build_ass_content(
safe_title_text_raw = escape_ass_text(title_text)
safe_title_text = _wrap_title_text(safe_title_text_raw, video_width, title_font_size)
# 自由位置:在文本前注入 \pos override tag(锚点为文本块中心,配合 \an5)
if title_pos is not None:
safe_title_text = f"{{\\pos({title_pos[0]},{title_pos[1]})}}{safe_title_text}"
events.append(
"Dialogue: 0,0:00:00.00," f"{format_ass_time(video_duration)}," "TitleStyle,,0,0,0,," f"{safe_title_text}"
)
+9 -6
View File
@@ -28,12 +28,15 @@ if [ "$SKIP_BACKEND" = "true" ]; then
CONTEXTS=("CI/CD Pipeline / Frontend Lint (pull_request)")
else
CONTEXTS=(
# CI 优化后 job 名称(2026-08):Code Quality 拆分为 Style+SecurityType Check+Migration 合并为 Python
# 与 pr_auto_scan.py 的 REQUIRED_CONTEXTS_APPROVE 保持一致
"CI/CD Pipeline / Validate - Style (pull_request)"
"CI/CD Pipeline / Validate - Security (pull_request)"
"CI/CD Pipeline / Validate - Python (mypy + alembic) (pull_request)"
"CI/CD Pipeline / Validate - Code Quality (pull_request)"
"CI/CD Pipeline / Validate - Type Check (mypy) (pull_request)"
"CI/CD Pipeline / Validate - Migration (alembic) (pull_request)"
"CI/CD Pipeline / Frontend Lint (pull_request)"
"CI/CD Pipeline / Unit Tests (pull_request)"
"CI/CD Pipeline / Frontend Unit Tests (pull_request)"
"CI/CD Pipeline / PR Build API Image (pull_request)"
"CI/CD Pipeline / PR Build Web Image (pull_request)"
"CI/CD Pipeline / PR Build Worker Image (pull_request)"
)
fi
@@ -150,4 +153,4 @@ done
echo
echo "⏰ 快速检查超时(2分钟),CI尚未完成,退出等待下次触发(workflow_run事件或5分钟定时扫描)"
exit 0
exit 0
+3 -4
View File
@@ -303,10 +303,9 @@ def main():
"CI/CD Pipeline / CI Gate (pull_request)",
]
REQUIRED_CONTEXTS_APPROVE = [
# CI 优化后 job 名称(2026-08):Code Quality 拆分为 Style+SecurityType Check+Migration 合并为 Python
"CI/CD Pipeline / Validate - Style (pull_request)",
"CI/CD Pipeline / Validate - Security (pull_request)",
"CI/CD Pipeline / Validate - Python (mypy + alembic) (pull_request)",
"CI/CD Pipeline / Validate - Code Quality (pull_request)",
"CI/CD Pipeline / Validate - Type Check (mypy) (pull_request)",
"CI/CD Pipeline / Validate - Migration (alembic) (pull_request)",
"CI/CD Pipeline / Frontend Lint (pull_request)",
]
FRONTEND_ONLY_CONTEXT = [
+5 -5
View File
@@ -197,8 +197,8 @@ if [ "$USE_SHARED_PG" = "true" ]; then
# 创建主数据库(xdist 模式下各 worker 会创建自己的数据库,主库作为 fallback)
echo "创建主测试数据库: $CI_DB_NAME"
PGPASSWORD="$SHARED_PG_PASSWORD" python3 -c "
import psycopg
conn = psycopg.connect(host='$SHARED_PG_HOST', port=$SHARED_PG_PORT, user='$SHARED_PG_USER', password='$SHARED_PG_PASSWORD', dbname='postgres')
import psycopg2
conn = psycopg2.connect(host='$SHARED_PG_HOST', port=$SHARED_PG_PORT, user='$SHARED_PG_USER', password='$SHARED_PG_PASSWORD', dbname='postgres')
conn.autocommit = True
cur = conn.cursor()
cur.execute(f'DROP DATABASE IF EXISTS \"$CI_DB_NAME\" WITH (FORCE)')
@@ -304,8 +304,8 @@ if [ "$USE_SHARED_PG" = "true" ]; then
# 清理所有以 CI_DB_NAME 开头的数据库(主库 + worker 库)
PGPASSWORD="${SHARED_PG_PASSWORD}" python3 -c "
import psycopg
conn = psycopg.connect(host='${SHARED_PG_HOST}', port=${SHARED_PG_PORT}, user='${SHARED_PG_USER}', password='${SHARED_PG_PASSWORD}', dbname='postgres')
import psycopg2
conn = psycopg2.connect(host='${SHARED_PG_HOST}', port=${SHARED_PG_PORT}, user='${SHARED_PG_USER}', password='${SHARED_PG_PASSWORD}', dbname='postgres')
conn.autocommit = True
cur = conn.cursor()
@@ -344,4 +344,4 @@ python3 scripts/ci_coverage_summary.py
set -e
echo ""
echo "=== CI Integration Tests 全部通过 ✅ ==="
echo "=== CI Integration Tests 全部通过 ✅ ==="
+4 -4
View File
@@ -409,8 +409,8 @@ except:
echo "创建测试数据库: $CI_DB_NAME"
PGPASSWORD="$SHARED_PG_PASSWORD" python3 -c "
import psycopg
conn = psycopg.connect(host='$SHARED_PG_HOST', port=$SHARED_PG_PORT, user='$SHARED_PG_USER', password='$SHARED_PG_PASSWORD', dbname='postgres')
import psycopg2
conn = psycopg2.connect(host='$SHARED_PG_HOST', port=$SHARED_PG_PORT, user='$SHARED_PG_USER', password='$SHARED_PG_PASSWORD', dbname='postgres')
conn.autocommit = True
cur = conn.cursor()
cur.execute(f'DROP DATABASE IF EXISTS \"$CI_DB_NAME\" WITH (FORCE)')
@@ -431,8 +431,8 @@ conn.close()
# 清理数据库
echo "清理测试数据库: $CI_DB_NAME"
PGPASSWORD="$SHARED_PG_PASSWORD" python3 -c "
import psycopg
conn = psycopg.connect(host='$SHARED_PG_HOST', port=$SHARED_PG_PORT, user='$SHARED_PG_USER', password='$SHARED_PG_PASSWORD', dbname='postgres')
import psycopg2
conn = psycopg2.connect(host='$SHARED_PG_HOST', port=$SHARED_PG_PORT, user='$SHARED_PG_USER', password='$SHARED_PG_PASSWORD', dbname='postgres')
conn.autocommit = True
cur = conn.cursor()
cur.execute(f'DROP DATABASE IF EXISTS \"$CI_DB_NAME\" WITH (FORCE)')
+4 -4
View File
@@ -169,8 +169,8 @@ if [ "$USE_SHARED_PG" = "true" ]; then
echo "创建测试数据库: $CI_DB_NAME"
PGPASSWORD="$SHARED_PG_PASSWORD" python3 -c "
import psycopg
conn = psycopg.connect(host='$SHARED_PG_HOST', port=$SHARED_PG_PORT, user='$SHARED_PG_USER', password='$SHARED_PG_PASSWORD', dbname='postgres')
import psycopg2
conn = psycopg2.connect(host='$SHARED_PG_HOST', port=$SHARED_PG_PORT, user='$SHARED_PG_USER', password='$SHARED_PG_PASSWORD', dbname='postgres')
conn.autocommit = True
cur = conn.cursor()
cur.execute(f'DROP DATABASE IF EXISTS \"$CI_DB_NAME\" WITH (FORCE)')
@@ -185,8 +185,8 @@ conn.close()
echo ""
echo "清理测试数据库: $CI_DB_NAME"
PGPASSWORD="$SHARED_PG_PASSWORD" python3 -c "
import psycopg
conn = psycopg.connect(host='$SHARED_PG_HOST', port=$SHARED_PG_PORT, user='$SHARED_PG_USER', password='$SHARED_PG_PASSWORD', dbname='postgres')
import psycopg2
conn = psycopg2.connect(host='$SHARED_PG_HOST', port=$SHARED_PG_PORT, user='$SHARED_PG_USER', password='$SHARED_PG_PASSWORD', dbname='postgres')
conn.autocommit = True
cur = conn.cursor()
cur.execute(f'DROP DATABASE IF EXISTS \"$CI_DB_NAME\" WITH (FORCE)')
-145
View File
@@ -602,148 +602,3 @@ class TestFontsizeCompensation:
result = build_ass_style("S", font_size=0)
parts = result.split(",")
assert int(parts[2]) >= 1
# ── 标题自由位置拖拽(工单 #1405 方案 B)──────────────────────────────────────
def _title_style_line(content: str) -> str:
return [line for line in content.splitlines() if line.startswith("Style: TitleStyle")][0]
def _title_dialogue_line(content: str) -> str:
return [line for line in content.splitlines() if line.startswith("Dialogue:") and "TitleStyle" in line][0]
class TestTitleFreePosition:
"""pos_x/pos_y 合法时注入 \\pos 且 Alignment=5;非法/缺失时回退原逻辑。"""
def _base_kwargs(self):
return dict(
video_width=1080,
video_height=1920,
video_duration=8.0,
title_text="测试标题",
)
def test_valid_position_injects_pos_tag_and_alignment_5(self):
content = build_ass_content(
**self._base_kwargs(),
title_config={"position": "top", "size": 36, "pos_x": 540, "pos_y": 300},
)
# Dialogue 文本前注入 {\pos(540,300)}
dialogue = _title_dialogue_line(content)
assert "{\\pos(540,300)}" in dialogue
# TitleStyle Alignment 固定 5\an5 中对齐,\pos 锚点为文本块中心)
fields = [f.strip() for f in _title_style_line(content).split(",")]
assert fields[18] == "5"
def test_boundary_coordinates_zero_and_max_accepted(self):
"""边界值 0 和 video_width/video_height 合法(闭区间)。"""
content = build_ass_content(
**self._base_kwargs(),
title_config={"pos_x": 0, "pos_y": 1920},
)
assert "{\\pos(0,1920)}" in _title_dialogue_line(content)
content2 = build_ass_content(
**self._base_kwargs(),
title_config={"pos_x": 1080, "pos_y": 0},
)
assert "{\\pos(1080,0)}" in _title_dialogue_line(content2)
def test_no_coords_output_identical_to_before(self):
"""不传坐标 → 输出与现有断言完全一致(回归保护)。"""
content = build_ass_content(
**self._base_kwargs(),
title_config={"position": "top", "size": 36},
)
# 无 \pos 注入
assert "\\pos(" not in content
# Alignment 走 position 映射(top → 8
fields = [f.strip() for f in _title_style_line(content).split(",")]
assert fields[18] == "8"
def test_out_of_bounds_falls_back(self):
"""越界坐标 → 回退 position 三档逻辑,输出与无坐标一致。"""
base = build_ass_content(
**self._base_kwargs(),
title_config={"position": "top", "size": 36},
)
for pos_x, pos_y in [(-1, 300), (540, -1), (1081, 300), (540, 1921), (99999, 99999)]:
content = build_ass_content(
**self._base_kwargs(),
title_config={"position": "top", "size": 36, "pos_x": pos_x, "pos_y": pos_y},
)
assert "\\pos(" not in content, f"({pos_x},{pos_y}) should be rejected"
assert content == base, f"({pos_x},{pos_y}) output differs from fallback"
def test_invalid_coords_falls_back(self):
"""非法类型坐标 → 回退原逻辑。"""
base = build_ass_content(
**self._base_kwargs(),
title_config={"position": "top", "size": 36},
)
for pos_x, pos_y in [("abc", 300), (540, None), (None, None), (True, 300), (540, False), (540.5, 300.9)]:
content = build_ass_content(
**self._base_kwargs(),
title_config={"position": "top", "size": 36, "pos_x": pos_x, "pos_y": pos_y},
)
assert "\\pos(" not in content, f"({pos_x!r},{pos_y!r}) should be rejected"
assert content == base, f"({pos_x!r},{pos_y!r}) output differs from fallback"
def test_only_one_coord_falls_back(self):
"""只传 pos_x 或 pos_y → 回退原逻辑。"""
base = build_ass_content(
**self._base_kwargs(),
title_config={"position": "center", "size": 36},
)
content_x = build_ass_content(
**self._base_kwargs(),
title_config={"position": "center", "size": 36, "pos_x": 540},
)
content_y = build_ass_content(
**self._base_kwargs(),
title_config={"position": "center", "size": 36, "pos_y": 300},
)
assert content_x == base
assert content_y == base
assert "\\pos(" not in content_x
assert "\\pos(" not in content_y
def test_position_three_levels_unchanged_without_coords(self):
"""无坐标时 top/center/bottom 三档 Alignment 输出不变。"""
for position, expected_align in [("top", "8"), ("center", "5"), ("bottom", "2")]:
content = build_ass_content(
**self._base_kwargs(),
title_config={"position": position, "size": 36},
)
fields = [f.strip() for f in _title_style_line(content).split(",")]
assert fields[18] == expected_align
def test_pos_overrides_position_alignment(self):
"""有合法坐标时,无论 position 是什么,Alignment 都固定为 5。"""
for position in ["top", "center", "bottom"]:
content = build_ass_content(
**self._base_kwargs(),
title_config={"position": position, "size": 36, "pos_x": 100, "pos_y": 200},
)
fields = [f.strip() for f in _title_style_line(content).split(",")]
assert fields[18] == "5"
assert "{\\pos(100,200)}" in _title_dialogue_line(content)
def test_subtitle_not_affected_by_pos(self):
"""pos_x/pos_y 只影响 TitleSubtitle 输出不变。"""
content = build_ass_content(
**self._base_kwargs(),
title_config={"pos_x": 540, "pos_y": 300},
subtitle_text="配音字幕",
subtitle_config={"position": "bottom", "size": 24},
)
sub_style = [line for line in content.splitlines() if line.startswith("Style: SubtitleStyle")][0]
sub_fields = [f.strip() for f in sub_style.split(",")]
assert sub_fields[18] == "2" # bottom
sub_dialogue = [
line for line in content.splitlines() if line.startswith("Dialogue:") and "SubtitleStyle" in line
][0]
assert "\\pos(" not in sub_dialogue
+3 -5
View File
@@ -395,13 +395,11 @@ class TestSmartMatchFiltersExhausted:
# 返回的素材全部 usable=True
assert all(item.usable for item in resp.items)
def test_all_exhausted_falls_back_to_all(self):
"""全部素材已用尽时回退保留全部(不返回空——空结果会让前端回退全选,
反而绕过评分排序;耗尽素材仍可走复用区间)。"""
def test_all_exhausted_returns_empty(self):
"""全部素材已用尽时返回空列表(不报错,前端显示空结果)。"""
assets = [_exhausted_asset("a-ex-1"), _exhausted_asset("a-ex-2")]
resp = self._call(assets)
returned_ids = {item.id for item in resp.items}
assert returned_ids == {"a-ex-1", "a-ex-2"}
assert resp.items == []
assert resp.total_candidates == 2
def test_fresh_assets_all_returned(self):
@@ -1,243 +0,0 @@
"""_get_template_segments 回退路径测试.
验证三级回退链:
1. 新模板系统(tpl_svc.list_clip_configs)正常 → 直接返回
2. 新模板系统主表不存在(ValueError)→ 直接查 template_clip_configs 表兜底
3. 直接查表也失败 → 回退旧模板系统(template_segments
4. 全部失败 → 返回空列表
覆盖 P0 修复:自建模板在 edit_templates 主表不存在但在 template_clip_configs 有记录时,
from-assets 流程不再 400。
"""
from __future__ import annotations
import os
import sys
from pathlib import Path
from unittest.mock import MagicMock, PropertyMock
os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
from app.api.routes.templates_editor.clips import _get_template_segments
TEST_TEMPLATE_ID = "tmpl-orphan-001"
DEFAULT_DUR = 5.0 # _DEFAULT_EDITOR_CLIP_DURATION
def _make_clip_config(order: int, min_dur: float = 3.0, max_dur: float = 8.0):
"""构造 mock TemplateClipConfig 领域实体."""
cc = MagicMock()
cc.order = order
cc.min_duration = min_dur
cc.max_duration = max_dur
return cc
def _make_old_segment(segment_order: int, dur_min: float = 4.0, dur_max: float = 7.0):
"""构造 mock 旧 TemplateSegment."""
s = MagicMock()
s.segment_order = segment_order
s.duration_min = dur_min
s.duration_max = dur_max
return s
# ---------------------------------------------------------------------------
# 测试
# ---------------------------------------------------------------------------
class TestGetTemplateSegmentsFallback:
"""_get_template_segments 三级回退链."""
def test_new_system_works(self):
"""路径1:新模板系统正常返回 → 直接使用."""
configs = [_make_clip_config(0, 2.0, 6.0), _make_clip_config(1, 3.0, 9.0)]
tpl_svc = MagicMock()
tpl_svc.list_clip_configs.return_value = configs
db = MagicMock()
result = _get_template_segments(TEST_TEMPLATE_ID, tpl_svc, db)
assert len(result) == 2
assert result[0] == (0, 2.0, 6.0)
assert result[1] == (1, 3.0, 9.0)
tpl_svc.list_clip_configs.assert_called_once_with(TEST_TEMPLATE_ID)
def test_main_table_missing_direct_query_succeeds(self):
"""路径2(P0修复):主表不存在 ValueError → 直接查表成功.
模拟自建模板在 edit_templates 主表已删除/不存在,
但 template_clip_configs 表有记录。
"""
tpl_svc = MagicMock()
tpl_svc.list_clip_configs.side_effect = ValueError(f"模板不存在: {TEST_TEMPLATE_ID}")
db = MagicMock()
# Mock SQLAlchemyTemplateClipConfigRepository
direct_configs = [
_make_clip_config(0, 2.0, 5.0),
_make_clip_config(1, 3.0, 7.0),
_make_clip_config(2, 4.0, 8.0),
]
with (
__import__("unittest.mock", fromlist=["patch"]).patch(
"app.api.routes.templates_editor.clips.SQLAlchemyTemplateClipConfigRepository"
) as mock_repo_cls,
):
mock_repo = MagicMock()
mock_repo.list_by_template.return_value = direct_configs
mock_repo_cls.return_value = mock_repo
result = _get_template_segments(TEST_TEMPLATE_ID, tpl_svc, db)
assert len(result) == 3
assert result[0] == (0, 2.0, 5.0)
assert result[1] == (1, 3.0, 7.0)
assert result[2] == (2, 4.0, 8.0)
mock_repo.list_by_template.assert_called_once_with(TEST_TEMPLATE_ID)
def test_main_table_missing_direct_query_empty_falls_to_old(self):
"""路径2→3:主表不存在 + 直接查表为空 → 回退旧系统."""
tpl_svc = MagicMock()
tpl_svc.list_clip_configs.side_effect = ValueError("模板不存在")
db = MagicMock()
old_segments = [_make_old_segment(0, 3.0, 6.0)]
with __import__("unittest.mock", fromlist=["patch"]).patch(
"app.api.routes.templates_editor.clips.SQLAlchemyTemplateClipConfigRepository"
) as mock_repo_cls:
mock_repo = MagicMock()
mock_repo.list_by_template.return_value = [] # 新表也没记录
mock_repo_cls.return_value = mock_repo
with __import__("unittest.mock", fromlist=["patch"]).patch(
"app.api.routes.templates_editor.clips.SQLAlchemyTemplateRepository"
) as mock_old_cls:
mock_old = MagicMock()
mock_old.list_segments.return_value = old_segments
mock_old_cls.return_value = mock_old
result = _get_template_segments(TEST_TEMPLATE_ID, tpl_svc, db)
assert len(result) == 1
assert result[0] == (0, 3.0, 6.0)
def test_all_fail_returns_empty(self):
"""路径4:三级全部失败 → 返回空列表."""
tpl_svc = MagicMock()
tpl_svc.list_clip_configs.side_effect = ValueError("模板不存在")
db = MagicMock()
with __import__("unittest.mock", fromlist=["patch"]).patch(
"app.api.routes.templates_editor.clips.SQLAlchemyTemplateClipConfigRepository"
) as mock_repo_cls:
mock_repo = MagicMock()
mock_repo.list_by_template.side_effect = Exception("DB error")
mock_repo_cls.return_value = mock_repo
with __import__("unittest.mock", fromlist=["patch"]).patch(
"app.api.routes.templates_editor.clips.SQLAlchemyTemplateRepository"
) as mock_old_cls:
mock_old = MagicMock()
mock_old.list_segments.return_value = [] # 旧表也空
mock_old_cls.return_value = mock_old
result = _get_template_segments(TEST_TEMPLATE_ID, tpl_svc, db)
assert result == []
def test_direct_query_sorts_by_order(self):
"""直接查表返回的结果按 order 排序."""
tpl_svc = MagicMock()
tpl_svc.list_clip_configs.side_effect = ValueError("模板不存在")
db = MagicMock()
# 故意乱序
configs = [
_make_clip_config(2, 5.0, 10.0),
_make_clip_config(0, 2.0, 4.0),
_make_clip_config(1, 3.0, 6.0),
]
with __import__("unittest.mock", fromlist=["patch"]).patch(
"app.api.routes.templates_editor.clips.SQLAlchemyTemplateClipConfigRepository"
) as mock_repo_cls:
mock_repo = MagicMock()
mock_repo.list_by_template.return_value = configs
mock_repo_cls.return_value = mock_repo
result = _get_template_segments(TEST_TEMPLATE_ID, tpl_svc, db)
assert [r[0] for r in result] == [0, 1, 2]
assert result[0] == (0, 2.0, 4.0)
assert result[1] == (1, 3.0, 6.0)
assert result[2] == (2, 5.0, 10.0)
def test_direct_query_handles_none_durations(self):
"""直接查表时 min/max_duration 为 None → 使用默认值."""
tpl_svc = MagicMock()
tpl_svc.list_clip_configs.side_effect = ValueError("模板不存在")
db = MagicMock()
cc = MagicMock()
cc.order = 0
cc.min_duration = None
cc.max_duration = None
with __import__("unittest.mock", fromlist=["patch"]).patch(
"app.api.routes.templates_editor.clips.SQLAlchemyTemplateClipConfigRepository"
) as mock_repo_cls:
mock_repo = MagicMock()
mock_repo.list_by_template.return_value = [cc]
mock_repo_cls.return_value = mock_repo
result = _get_template_segments(TEST_TEMPLATE_ID, tpl_svc, db)
assert len(result) == 1
# None → default (5.0), max(None or None) → default (5.0)
assert result[0] == (0, DEFAULT_DUR, DEFAULT_DUR)
def test_new_system_returns_empty_tries_direct(self):
"""新模板系统返回空列表(非异常)→ 继续尝试直接查表."""
tpl_svc = MagicMock()
tpl_svc.list_clip_configs.return_value = [] # 空列表,非异常
db = MagicMock()
direct_configs = [_make_clip_config(0, 3.0, 6.0)]
with __import__("unittest.mock", fromlist=["patch"]).patch(
"app.api.routes.templates_editor.clips.SQLAlchemyTemplateClipConfigRepository"
) as mock_repo_cls:
mock_repo = MagicMock()
mock_repo.list_by_template.return_value = direct_configs
mock_repo_cls.return_value = mock_repo
result = _get_template_segments(TEST_TEMPLATE_ID, tpl_svc, db)
# 新系统返回空 → 不走 except → 但也没 return → 继续往下走
# 直接查表有数据 → 返回
assert len(result) == 1
assert result[0] == (0, 3.0, 6.0)
def test_existing_template_unaffected(self):
"""正常模板(主表存在)行为不变."""
configs = [_make_clip_config(0, 2.0, 5.0)]
tpl_svc = MagicMock()
tpl_svc.list_clip_configs.return_value = configs
db = MagicMock()
with __import__("unittest.mock", fromlist=["patch"]).patch(
"app.api.routes.templates_editor.clips.SQLAlchemyTemplateClipConfigRepository"
) as mock_repo_cls:
result = _get_template_segments(TEST_TEMPLATE_ID, tpl_svc, db)
# 直接查表不应被调用(新系统已返回)
mock_repo_cls.assert_not_called()
assert len(result) == 1
assert result[0] == (0, 2.0, 5.0)
-258
View File
@@ -1,258 +0,0 @@
"""smart-match 过滤回退策略测试:余量过滤/高频排除导致结果集为空时必须回退。
线上事故:路由旧实现先 smart_select_assets(limit=N) 截取,再对这 N 条做
usable / 高频过滤,过滤后不回补——排名靠前素材全部被排除时返回空 items,
前端回退全选。修复后过滤全部前置,且每级过滤后为空/不足时回退保留。
"""
import json
import os
import sys
from pathlib import Path
from unittest.mock import MagicMock
os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT / "apps" / "api"))
sys.path.insert(0, str(ROOT / "packages"))
from app.api.routes.assets import router
from app.auth import AuthenticatedUser, get_current_user
from app.core.storage import get_storage_service
from app.dependencies import (
get_asset_library_repository,
get_asset_repository,
get_project_repository,
)
from packages.domain import (
Asset,
AssetLibrary,
AssetLibraryKind,
AssetStatus,
Project,
User,
)
class _StubProjectRepo:
def __init__(self, projects):
self._projects = projects
def find_by_id(self, pid):
return self._projects.get(pid)
class _StubAssetLibraryRepo:
def __init__(self, libraries):
self._libraries = libraries
def get(self, lid):
return self._libraries.get(lid)
class _StubAssetRepo:
"""模拟仓储;session 属性供 get_asset_recent_use_counts 使用(测试中会被 patch)。"""
def __init__(self, assets):
self._assets = assets
self.session = MagicMock(name="stub-session")
def find_by_library(self, lid, skip=0, limit=100, status=None):
result = [a for a in self._assets if a.library_id == lid]
if status:
result = [a for a in result if (a.status.value if hasattr(a.status, "value") else a.status) in status]
return result[skip : skip + limit]
def find_by_library_and_file_type(self, lid, file_type, skip=0, limit=100, status=None):
result = [a for a in self._assets if a.library_id == lid and a.file_type == file_type]
if status:
result = [a for a in result if (a.status.value if hasattr(a.status, "value") else a.status) in status]
return result[skip : skip + limit]
def _make_app(asset_repo, lib_repo, proj_repo):
app = FastAPI()
app.include_router(router, prefix="/assets")
fake_user = MagicMock()
fake_user.user = User(id="user-1", email="test@test.com", display_name="Test")
app.dependency_overrides[get_current_user] = lambda: AuthenticatedUser(user=fake_user.user)
app.dependency_overrides[get_asset_repository] = lambda: asset_repo
app.dependency_overrides[get_asset_library_repository] = lambda: lib_repo
app.dependency_overrides[get_project_repository] = lambda: proj_repo
app.dependency_overrides[get_storage_service] = lambda: MagicMock()
return app
def _library():
return AssetLibrary(id="lib-1", project_id="proj-1", name="Videos", kind=AssetLibraryKind.VIDEO)
def _exhausted_ranges(duration=15.0):
"""构造 used_time_ranges:整片覆盖 + 每区间 use_count 达上限 3 → usable=False。"""
return [
{"start": 0.0, "end": duration, "use_count": 3, "plan_id": "p1"},
]
def _video_asset(name, duration=15.0, quality=90, used_ranges=None):
meta = {"used_time_ranges": used_ranges} if used_ranges is not None else {}
return Asset.create(
project_id="proj-1",
library_id="lib-1",
name=name,
storage_key=f"key-{name}",
mime_type="video/mp4",
metadata=meta,
quality_score=quality,
duration=duration,
status=AssetStatus.READY,
)
class TestSmartMatchAvailabilityFallback:
"""余量过滤回退:全部素材 usable=False 时不返回空。"""
def test_all_exhausted_returns_assets_instead_of_empty(self):
project = Project(id="proj-1", name="Test", owner_user_id="user-1")
assets = [
_video_asset("v1.mp4", used_ranges=_exhausted_ranges(15)),
_video_asset("v2.mp4", used_ranges=_exhausted_ranges(25)),
]
repo = _StubAssetRepo(assets)
app = _make_app(repo, _StubAssetLibraryRepo({"lib-1": _library()}), _StubProjectRepo({"proj-1": project}))
with TestClient(app) as client:
resp = client.post("/assets/smart-match", json={"library_id": "lib-1"})
assert resp.status_code == 200, resp.text
data = resp.json()
# 回退:保留全部候选,不返回空
assert len(data["items"]) == 2
assert data["total_candidates"] == 2
def test_mixed_exhausted_and_fresh_excludes_exhausted(self):
project = Project(id="proj-1", name="Test", owner_user_id="user-1")
assets = [
_video_asset("exhausted.mp4", quality=99, used_ranges=_exhausted_ranges(15)),
_video_asset("fresh.mp4", quality=50, used_ranges=None),
]
repo = _StubAssetRepo(assets)
app = _make_app(repo, _StubAssetLibraryRepo({"lib-1": _library()}), _StubProjectRepo({"proj-1": project}))
with TestClient(app) as client:
resp = client.post("/assets/smart-match", json={"library_id": "lib-1"})
assert resp.status_code == 200, resp.text
names = [item["name"] for item in resp.json()["items"]]
assert "fresh.mp4" in names
assert "exhausted.mp4" not in names
def test_limit_backfills_from_lower_ranked_when_top_exhausted(self):
"""limit=1 且排名第一的素材耗尽时,必须回补排名靠后的可用素材,不返回空。"""
project = Project(id="proj-1", name="Test", owner_user_id="user-1")
assets = [
_video_asset("top-exhausted.mp4", quality=100, used_ranges=_exhausted_ranges(15)),
_video_asset("second-fresh.mp4", quality=40, used_ranges=None),
_video_asset("third-fresh.mp4", quality=30, used_ranges=None),
]
repo = _StubAssetRepo(assets)
app = _make_app(repo, _StubAssetLibraryRepo({"lib-1": _library()}), _StubProjectRepo({"proj-1": project}))
with TestClient(app) as client:
resp = client.post("/assets/smart-match", json={"library_id": "lib-1", "limit": 1})
assert resp.status_code == 200, resp.text
data = resp.json()
assert len(data["items"]) == 1
# 不能是空,也不能是耗尽的高分素材
assert data["items"][0]["name"] == "second-fresh.mp4"
class TestSmartMatchHighFreqFallback:
"""高频排除回退:排除后为空/不足 limit 时保留全部可用素材。"""
def test_all_high_freq_keeps_all(self, monkeypatch):
import app.api.routes.assets as routes_mod
project = Project(id="proj-1", name="Test", owner_user_id="user-1")
assets = [_video_asset("v1.mp4", quality=90), _video_asset("v2.mp4", quality=80)]
repo = _StubAssetRepo(assets)
# 全部素材在最近 5 个视频中使用 5 次(> 3)
fake_counts = {assets[0].id: 5, assets[1].id: 5}
monkeypatch.setattr(
routes_mod, "get_asset_recent_use_counts", lambda db, asset_ids, recent_video_count=5: fake_counts
)
app = _make_app(repo, _StubAssetLibraryRepo({"lib-1": _library()}), _StubProjectRepo({"proj-1": project}))
with TestClient(app) as client:
resp = client.post("/assets/smart-match", json={"library_id": "lib-1"})
assert resp.status_code == 200, resp.text
data = resp.json()
# 全部高频 → 回退保留全部
assert len(data["items"]) == 2
def test_high_freq_partial_exclusion_with_enough_remaining(self, monkeypatch):
import app.api.routes.assets as routes_mod
project = Project(id="proj-1", name="Test", owner_user_id="user-1")
assets = [
_video_asset("hot.mp4", quality=99),
_video_asset("cool1.mp4", quality=80),
_video_asset("cool2.mp4", quality=70),
]
repo = _StubAssetRepo(assets)
fake_counts = {assets[0].id: 9, assets[1].id: 1, assets[2].id: 0}
monkeypatch.setattr(
routes_mod, "get_asset_recent_use_counts", lambda db, asset_ids, recent_video_count=5: fake_counts
)
app = _make_app(repo, _StubAssetLibraryRepo({"lib-1": _library()}), _StubProjectRepo({"proj-1": project}))
with TestClient(app) as client:
resp = client.post("/assets/smart-match", json={"library_id": "lib-1"})
assert resp.status_code == 200, resp.text
names = [item["name"] for item in resp.json()["items"]]
assert "hot.mp4" not in names
assert set(names) == {"cool1.mp4", "cool2.mp4"}
def test_high_freq_insufficient_for_limit_keeps_all(self, monkeypatch):
"""3 个素材、limit=5、2 个高频 → 剩余 1 < limit → 保留全部。"""
import app.api.routes.assets as routes_mod
project = Project(id="proj-1", name="Test", owner_user_id="user-1")
assets = [
_video_asset("hot1.mp4", quality=99),
_video_asset("hot2.mp4", quality=98),
_video_asset("cool.mp4", quality=50),
]
repo = _StubAssetRepo(assets)
fake_counts = {assets[0].id: 8, assets[1].id: 7, assets[2].id: 0}
monkeypatch.setattr(
routes_mod, "get_asset_recent_use_counts", lambda db, asset_ids, recent_video_count=5: fake_counts
)
app = _make_app(repo, _StubAssetLibraryRepo({"lib-1": _library()}), _StubProjectRepo({"proj-1": project}))
with TestClient(app) as client:
resp = client.post("/assets/smart-match", json={"library_id": "lib-1", "limit": 5})
assert resp.status_code == 200, resp.text
names = [item["name"] for item in resp.json()["items"]]
# 剩余 1 < limit=5 → 回退保留全部 3 条
assert set(names) == {"hot1.mp4", "hot2.mp4", "cool.mp4"}
def test_high_freq_query_failure_skips_exclusion(self, monkeypatch):
import app.api.routes.assets as routes_mod
def _boom(db, asset_ids, recent_video_count=5):
raise RuntimeError("DB down")
project = Project(id="proj-1", name="Test", owner_user_id="user-1")
assets = [_video_asset("v1.mp4", quality=90), _video_asset("v2.mp4", quality=80)]
repo = _StubAssetRepo(assets)
monkeypatch.setattr(routes_mod, "get_asset_recent_use_counts", _boom)
app = _make_app(repo, _StubAssetLibraryRepo({"lib-1": _library()}), _StubProjectRepo({"proj-1": project}))
with TestClient(app) as client:
resp = client.post("/assets/smart-match", json={"library_id": "lib-1"})
assert resp.status_code == 200, resp.text
assert len(resp.json()["items"]) == 2