Compare commits
9 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 xiaoxia
|
807606d5f0 | ||
|
xiaoxia
|
71ac73a889 | ||
|
xiaoxia
|
dd669f6ad9 | ||
|
xiaoxia
|
c34b3b2c5c | ||
|
xiaoxia
|
3283c3e01c | ||
|
xiaoxia
|
a6ebb762e6 | ||
|
xiaoxia
|
b99a760bb1 | ||
|
xiaoxia
|
0b998421d8 | ||
|
xiaoxia
|
018cf59018 |
@@ -4,731 +4,132 @@
|
||||
* 支持：标题卡片展示、AI 生成标题、复制/编辑/删除、收藏、分类筛选、搜索
|
||||
* 对接后端真实 API（GET/POST/PUT/DELETE /titles）
|
||||
*/
|
||||
import React, { useMemo, useState, useCallback } from "react"
|
||||
import { Modal as AntModal, message, Popconfirm } from "antd"
|
||||
import { useQuery, useMutation, useQueryClient } from "@tanstack/react-query"
|
||||
import {
|
||||
PlusOutlined,
|
||||
SearchOutlined,
|
||||
DeleteOutlined,
|
||||
EditOutlined,
|
||||
CopyOutlined,
|
||||
CheckOutlined,
|
||||
RobotOutlined,
|
||||
StarOutlined,
|
||||
StarFilled,
|
||||
FileTextOutlined,
|
||||
} from "@ant-design/icons"
|
||||
import { Button, Input, Select } from "@/components/ui"
|
||||
import { getTitles, createTitle, updateTitle, deleteTitle, type TitleItem } from "@/api/titles"
|
||||
import React from "react"
|
||||
import { useTitleLibrary } from "./hooks/useTitleLibrary"
|
||||
import { useTitleEdit } from "./hooks/useTitleEdit"
|
||||
import { useTitleAI } from "./hooks/useTitleAI"
|
||||
import { CategorySidebar } from "./components/title-library/CategorySidebar"
|
||||
import { FilterBar } from "./components/title-library/FilterBar"
|
||||
import { TitleGrid } from "./components/title-library/TitleGrid"
|
||||
import { CreateTitleModal } from "./components/title-library/CreateTitleModal"
|
||||
import { AIGenerateModal } from "./components/title-library/AIGenerateModal"
|
||||
import "./titles.css"
|
||||
|
||||
/* ============================================================
|
||||
* 类型
|
||||
* ============================================================ */
|
||||
type TitleType = "hot" | "normal" | "creative"
|
||||
type Industry = "general" | "food" | "tech" | "beauty" | "education" | "travel"
|
||||
type Frequency = "all" | "high" | "medium" | "low"
|
||||
|
||||
interface TitleData {
|
||||
id: string
|
||||
content: string
|
||||
type: TitleType
|
||||
industry: Industry
|
||||
category: string
|
||||
usageCount: number
|
||||
isFavorited: boolean
|
||||
createdAt: string
|
||||
}
|
||||
|
||||
/** 后端 TitleItem → 前端 TitleData 映射 */
|
||||
const toTitleData = (item: TitleItem): TitleData => ({
|
||||
id: item.id,
|
||||
content: item.content,
|
||||
type: (item.category as TitleType) || "normal",
|
||||
industry: "general",
|
||||
category: item.category || "未分类",
|
||||
usageCount: 0,
|
||||
isFavorited: false,
|
||||
createdAt: item.created_at?.slice(0, 10) || "",
|
||||
})
|
||||
|
||||
/* ============================================================
|
||||
* 工具函数
|
||||
* ============================================================ */
|
||||
const typeLabel = (type: TitleType): string => {
|
||||
switch (type) {
|
||||
case "hot":
|
||||
return "爆款"
|
||||
case "normal":
|
||||
return "常规"
|
||||
case "creative":
|
||||
return "创意"
|
||||
}
|
||||
}
|
||||
|
||||
/** 复制文本到剪贴板 */
|
||||
const copyToClipboard = async (text: string): Promise<boolean> => {
|
||||
try {
|
||||
await navigator.clipboard.writeText(text)
|
||||
return true
|
||||
} catch {
|
||||
/* 降级方案 */
|
||||
const textarea = document.createElement("textarea")
|
||||
textarea.value = text
|
||||
textarea.style.position = "fixed"
|
||||
textarea.style.opacity = "0"
|
||||
document.body.appendChild(textarea)
|
||||
textarea.select()
|
||||
try {
|
||||
document.execCommand("copy")
|
||||
return true
|
||||
} catch {
|
||||
return false
|
||||
} finally {
|
||||
document.body.removeChild(textarea)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/* ============================================================
|
||||
* TitleCard 组件
|
||||
* ============================================================ */
|
||||
const TitleCard: React.FC<{
|
||||
title: TitleData
|
||||
isEditing: boolean
|
||||
editText: string
|
||||
onEditChange: (text: string) => void
|
||||
onStartEdit: () => void
|
||||
onSaveEdit: () => void
|
||||
onCancelEdit: () => void
|
||||
onCopy: () => void
|
||||
onDelete: () => void
|
||||
onToggleFavorite: () => void
|
||||
}> = ({
|
||||
title,
|
||||
isEditing,
|
||||
editText,
|
||||
onEditChange,
|
||||
onStartEdit,
|
||||
onSaveEdit,
|
||||
onCancelEdit,
|
||||
onCopy,
|
||||
onDelete,
|
||||
onToggleFavorite,
|
||||
}) => {
|
||||
return (
|
||||
<div className="xx-title-card">
|
||||
{/* 收藏按钮 */}
|
||||
<button
|
||||
className="xx-title-fav-btn"
|
||||
onClick={onToggleFavorite}
|
||||
title={title.isFavorited ? "取消收藏" : "收藏"}
|
||||
style={{
|
||||
position: "absolute",
|
||||
top: 12,
|
||||
right: 12,
|
||||
color: title.isFavorited ? "#f59e0b" : "var(--text-tertiary)",
|
||||
}}
|
||||
>
|
||||
{title.isFavorited ? <StarFilled /> : <StarOutlined />}
|
||||
</button>
|
||||
|
||||
{/* 标题文本 / 编辑区 */}
|
||||
{isEditing ? (
|
||||
<textarea
|
||||
className="xx-title-card-edit"
|
||||
value={editText}
|
||||
onChange={(e) => onEditChange(e.target.value)}
|
||||
autoFocus
|
||||
onKeyDown={(e) => {
|
||||
if (e.key === "Enter" && !e.shiftKey) {
|
||||
e.preventDefault()
|
||||
onSaveEdit()
|
||||
}
|
||||
if (e.key === "Escape") {
|
||||
onCancelEdit()
|
||||
}
|
||||
}}
|
||||
/>
|
||||
) : (
|
||||
<div className="xx-title-card-text" style={{ paddingRight: 24 }}>
|
||||
{title.content}
|
||||
</div>
|
||||
)}
|
||||
|
||||
{/* 底部元信息 */}
|
||||
<div className="xx-title-card-meta">
|
||||
<div className="xx-title-card-meta-left">
|
||||
<span className={`xx-title-type-tag ${title.type}`}>{typeLabel(title.type)}</span>
|
||||
<span className="xx-title-card-stat">使用 {title.usageCount} 次</span>
|
||||
<span className="xx-title-card-stat">{title.createdAt}</span>
|
||||
</div>
|
||||
|
||||
<div className="xx-title-card-actions">
|
||||
{isEditing ? (
|
||||
<>
|
||||
<button className="xx-title-card-action-btn" onClick={onSaveEdit} title="保存">
|
||||
<CheckOutlined />
|
||||
</button>
|
||||
<button className="xx-title-card-action-btn" onClick={onCancelEdit} title="取消">
|
||||
✕
|
||||
</button>
|
||||
</>
|
||||
) : (
|
||||
<>
|
||||
<button className="xx-title-card-action-btn" onClick={onCopy} title="复制">
|
||||
<CopyOutlined />
|
||||
</button>
|
||||
<button className="xx-title-card-action-btn" onClick={onStartEdit} title="编辑">
|
||||
<EditOutlined />
|
||||
</button>
|
||||
<Popconfirm
|
||||
title="确定删除此标题？"
|
||||
onConfirm={onDelete}
|
||||
okText="删除"
|
||||
cancelText="取消"
|
||||
>
|
||||
<button className="xx-title-card-action-btn danger" title="删除">
|
||||
<DeleteOutlined />
|
||||
</button>
|
||||
</Popconfirm>
|
||||
</>
|
||||
)}
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
)
|
||||
}
|
||||
|
||||
/* ============================================================
|
||||
* 主组件
|
||||
* ============================================================ */
|
||||
const TitleLibrary: React.FC = () => {
|
||||
const queryClient = useQueryClient()
|
||||
const {
|
||||
categories,
|
||||
activeCatId,
|
||||
filteredTitles,
|
||||
searchText,
|
||||
filterType,
|
||||
filterIndustry,
|
||||
filterFrequency,
|
||||
createMutation,
|
||||
updateMutation,
|
||||
setActiveCatId,
|
||||
setSearchText,
|
||||
setFilterType,
|
||||
setFilterIndustry,
|
||||
setFilterFrequency,
|
||||
handleToggleFavorite,
|
||||
handleCopy,
|
||||
handleDelete,
|
||||
} = useTitleLibrary()
|
||||
|
||||
/* 分类数据 — 从真实标题数据动态派生 */
|
||||
const [activeCatId, setActiveCatId] = useState<string>("cat-all")
|
||||
const {
|
||||
editingId,
|
||||
editText,
|
||||
setEditText,
|
||||
createTitleModalOpen,
|
||||
setCreateTitleModalOpen,
|
||||
newTitleContent,
|
||||
setNewTitleContent,
|
||||
newTitleType,
|
||||
setNewTitleType,
|
||||
handleStartEdit,
|
||||
handleSaveEdit,
|
||||
handleCancelEdit,
|
||||
handleCreateTitle,
|
||||
handleCloseCreateModal,
|
||||
} = useTitleEdit({ updateMutation, createMutation })
|
||||
|
||||
/* 标题数据 — 真实 API */
|
||||
const { data: apiTitles = [] } = useQuery({
|
||||
queryKey: ["titles"],
|
||||
queryFn: getTitles,
|
||||
staleTime: 30_000,
|
||||
})
|
||||
const titles: TitleData[] = useMemo(() => apiTitles.map(toTitleData), [apiTitles])
|
||||
|
||||
/* 从真实标题数据动态派生分类（无需后端分类 API） */
|
||||
const categories = useMemo(() => {
|
||||
const cats = new Map<string, number>()
|
||||
apiTitles.forEach((t) => {
|
||||
const cat = t.category || "未分类"
|
||||
cats.set(cat, (cats.get(cat) || 0) + 1)
|
||||
})
|
||||
return [
|
||||
{ id: "cat-all", name: "全部标题", count: apiTitles.length },
|
||||
...Array.from(cats.entries()).map(([name, count]) => ({
|
||||
id: `cat-${name}`,
|
||||
name,
|
||||
count,
|
||||
})),
|
||||
]
|
||||
}, [apiTitles])
|
||||
|
||||
/* CRUD mutations */
|
||||
const createMutation = useMutation({
|
||||
mutationFn: (content: string) => createTitle({ content }),
|
||||
onSuccess: () => {
|
||||
queryClient.invalidateQueries({ queryKey: ["titles"] })
|
||||
},
|
||||
onError: () => message.error("创建标题失败"),
|
||||
})
|
||||
|
||||
const updateMutation = useMutation({
|
||||
mutationFn: ({ id, content }: { id: string; content: string }) => updateTitle(id, { content }),
|
||||
onSuccess: () => {
|
||||
queryClient.invalidateQueries({ queryKey: ["titles"] })
|
||||
},
|
||||
onError: () => message.error("更新标题失败"),
|
||||
})
|
||||
|
||||
const deleteMutation = useMutation({
|
||||
mutationFn: (id: string) => deleteTitle(id),
|
||||
onSuccess: () => {
|
||||
queryClient.invalidateQueries({ queryKey: ["titles"] })
|
||||
},
|
||||
onError: () => message.error("删除标题失败"),
|
||||
})
|
||||
|
||||
/* 筛选 */
|
||||
const [searchText, setSearchText] = useState("")
|
||||
const [filterType, setFilterType] = useState<string>("all")
|
||||
const [filterIndustry, setFilterIndustry] = useState<string>("all")
|
||||
const [filterFrequency, setFilterFrequency] = useState<Frequency>("all")
|
||||
|
||||
/* 编辑状态 */
|
||||
const [editingId, setEditingId] = useState<string | null>(null)
|
||||
const [editText, setEditText] = useState("")
|
||||
|
||||
/* 新建标题 */
|
||||
const [createTitleModalOpen, setCreateTitleModalOpen] = useState(false)
|
||||
const [newTitleContent, setNewTitleContent] = useState("")
|
||||
const [newTitleType, setNewTitleType] = useState<TitleType>("normal")
|
||||
|
||||
/* AI 生成 */
|
||||
const [aiModalOpen, setAiModalOpen] = useState(false)
|
||||
const [aiKeyword, setAiKeyword] = useState("")
|
||||
const [aiLoading, setAiLoading] = useState(false)
|
||||
const [aiResults, setAiResults] = useState<string[]>([])
|
||||
|
||||
/* 派生数据 */
|
||||
const activeCategory = categories.find((c) => c.id === activeCatId)
|
||||
|
||||
const filteredTitles = useMemo(() => {
|
||||
let list = titles
|
||||
|
||||
/* 按分类过滤（"全部标题" 不过滤）— 直接匹配后端 category 字段 */
|
||||
if (activeCatId !== "cat-all") {
|
||||
const catName = activeCategory?.name || ""
|
||||
if (catName) {
|
||||
list = list.filter((t) => t.category === catName)
|
||||
}
|
||||
}
|
||||
|
||||
/* 按类型筛选 */
|
||||
if (filterType !== "all") {
|
||||
list = list.filter((t) => t.type === filterType)
|
||||
}
|
||||
|
||||
/* 按行业筛选 */
|
||||
if (filterIndustry !== "all") {
|
||||
list = list.filter((t) => t.industry === filterIndustry)
|
||||
}
|
||||
|
||||
/* 按使用频率筛选 */
|
||||
if (filterFrequency !== "all") {
|
||||
switch (filterFrequency) {
|
||||
case "high":
|
||||
list = list.filter((t) => t.usageCount >= 100)
|
||||
break
|
||||
case "medium":
|
||||
list = list.filter((t) => t.usageCount >= 30 && t.usageCount < 100)
|
||||
break
|
||||
case "low":
|
||||
list = list.filter((t) => t.usageCount < 30)
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
/* 搜索 */
|
||||
if (searchText.trim()) {
|
||||
const q = searchText.trim().toLowerCase()
|
||||
list = list.filter((t) => t.content.toLowerCase().includes(q))
|
||||
}
|
||||
|
||||
return list
|
||||
}, [titles, activeCatId, activeCategory, filterType, filterIndustry, filterFrequency, searchText])
|
||||
|
||||
/* 收藏切换（暂不支持，待后端 API） */
|
||||
const handleToggleFavorite = useCallback((_id: string) => {
|
||||
message.info("收藏功能即将上线")
|
||||
}, [])
|
||||
|
||||
/* 复制 */
|
||||
const handleCopy = useCallback(async (title: TitleData) => {
|
||||
const ok = await copyToClipboard(title.content)
|
||||
if (ok) {
|
||||
message.success("已复制到剪贴板")
|
||||
} else {
|
||||
message.error("复制失败")
|
||||
}
|
||||
}, [])
|
||||
|
||||
/* 编辑 */
|
||||
const handleStartEdit = useCallback((title: TitleData) => {
|
||||
setEditingId(title.id)
|
||||
setEditText(title.content)
|
||||
}, [])
|
||||
|
||||
const handleSaveEdit = useCallback(() => {
|
||||
if (!editText.trim()) {
|
||||
message.warning("标题内容不能为空")
|
||||
return
|
||||
}
|
||||
if (editingId) {
|
||||
updateMutation.mutate({ id: editingId, content: editText.trim() })
|
||||
}
|
||||
setEditingId(null)
|
||||
setEditText("")
|
||||
message.success("标题已更新")
|
||||
}, [editingId, editText, updateMutation])
|
||||
|
||||
const handleCancelEdit = useCallback(() => {
|
||||
setEditingId(null)
|
||||
setEditText("")
|
||||
}, [])
|
||||
|
||||
/* 删除 */
|
||||
const handleDelete = useCallback(
|
||||
(id: string) => {
|
||||
deleteMutation.mutate(id)
|
||||
message.success("标题已删除")
|
||||
},
|
||||
[deleteMutation],
|
||||
)
|
||||
|
||||
/* 新建标题 */
|
||||
const handleCreateTitle = () => {
|
||||
if (!newTitleContent.trim()) {
|
||||
message.warning("请输入标题内容")
|
||||
return
|
||||
}
|
||||
createMutation.mutate(newTitleContent.trim(), {
|
||||
onSuccess: () => {
|
||||
setCreateTitleModalOpen(false)
|
||||
setNewTitleContent("")
|
||||
setNewTitleType("normal")
|
||||
message.success("标题创建成功")
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
/* AI 生成标题 */
|
||||
const handleAIGenerate = () => {
|
||||
if (!aiKeyword.trim()) {
|
||||
message.warning("请输入关键词或主题")
|
||||
return
|
||||
}
|
||||
setAiLoading(true)
|
||||
setAiResults([])
|
||||
|
||||
/* Mock AI 生成延迟 */
|
||||
setTimeout(() => {
|
||||
const keyword = aiKeyword.trim()
|
||||
const results = [
|
||||
`${keyword}：这个方法让我事半功倍！`,
|
||||
`关于${keyword}，99%的人都不知道的事`,
|
||||
`${keyword}全攻略，看完这篇就够了`,
|
||||
`我花了 3 个月研究${keyword}，总结出这些经验`,
|
||||
`${keyword}避坑指南，帮你省下 1000 块`,
|
||||
]
|
||||
setAiResults(results)
|
||||
setAiLoading(false)
|
||||
}, 2000)
|
||||
}
|
||||
|
||||
/* 采纳 AI 生成的标题 */
|
||||
const handleAdoptAITitle = (text: string) => {
|
||||
createMutation.mutate(text, {
|
||||
onSuccess: () => {
|
||||
message.success("标题已采纳并添加到标题库")
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
/* 复制 AI 生成的标题 */
|
||||
const handleCopyAI = async (text: string) => {
|
||||
const ok = await copyToClipboard(text)
|
||||
if (ok) {
|
||||
message.success("已复制到剪贴板")
|
||||
} else {
|
||||
message.error("复制失败")
|
||||
}
|
||||
}
|
||||
const {
|
||||
aiModalOpen,
|
||||
setAiModalOpen,
|
||||
aiKeyword,
|
||||
setAiKeyword,
|
||||
aiLoading,
|
||||
aiResults,
|
||||
handleAIGenerate,
|
||||
handleAdoptAITitle,
|
||||
handleCopyAI,
|
||||
handleCloseAIModal,
|
||||
} = useTitleAI({ createMutation })
|
||||
|
||||
return (
|
||||
<div className="xx-titles-page">
|
||||
{/* 两栏布局 */}
|
||||
<div className="xx-titles-layout">
|
||||
{/* ─── 左侧：分类列表 ─── */}
|
||||
<div className="xx-title-category-list">
|
||||
{categories.map((cat) => (
|
||||
<div
|
||||
key={cat.id}
|
||||
className={`xx-title-category-item${cat.id === activeCatId ? " active" : ""}`}
|
||||
onClick={() => setActiveCatId(cat.id)}
|
||||
>
|
||||
<div
|
||||
style={{
|
||||
display: "flex",
|
||||
alignItems: "center",
|
||||
justifyContent: "space-between",
|
||||
flex: 1,
|
||||
minWidth: 0,
|
||||
}}
|
||||
>
|
||||
<div style={{ minWidth: 0 }}>
|
||||
<h4 style={{ margin: 0 }}>
|
||||
<FileTextOutlined /> {cat.name}
|
||||
</h4>
|
||||
<span>{cat.count} 条</span>
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
))}
|
||||
{/* 左侧：分类列表 */}
|
||||
<CategorySidebar
|
||||
categories={categories}
|
||||
activeCatId={activeCatId}
|
||||
onSelect={setActiveCatId}
|
||||
/>
|
||||
|
||||
{/* TODO: 新建分类功能待后端分类 API 就绪后启用 */}
|
||||
</div>
|
||||
|
||||
{/* ─── 右侧：内容区 ─── */}
|
||||
{/* 右侧：内容区 */}
|
||||
<div className="xx-titles-content">
|
||||
{/* 筛选栏 */}
|
||||
<div className="xx-titles-filters">
|
||||
<div className="xx-titles-filters-left">
|
||||
<Input
|
||||
placeholder="搜索标题关键词..."
|
||||
prefix={<SearchOutlined />}
|
||||
value={searchText}
|
||||
onChange={(e) => setSearchText(e.target.value)}
|
||||
allowClear
|
||||
style={{ width: 220 }}
|
||||
/>
|
||||
<Select
|
||||
value={filterType}
|
||||
onChange={setFilterType}
|
||||
style={{ width: 110 }}
|
||||
options={[
|
||||
{ value: "all", label: "全部类型" },
|
||||
{ value: "hot", label: "爆款" },
|
||||
{ value: "normal", label: "常规" },
|
||||
{ value: "creative", label: "创意" },
|
||||
]}
|
||||
/>
|
||||
<Select
|
||||
value={filterIndustry}
|
||||
onChange={setFilterIndustry}
|
||||
style={{ width: 110 }}
|
||||
options={[
|
||||
{ value: "all", label: "全部行业" },
|
||||
{ value: "food", label: "美食" },
|
||||
{ value: "tech", label: "科技" },
|
||||
{ value: "beauty", label: "美妆" },
|
||||
{ value: "education", label: "教育" },
|
||||
{ value: "travel", label: "旅行" },
|
||||
]}
|
||||
/>
|
||||
<Select
|
||||
value={filterFrequency}
|
||||
onChange={(v) => setFilterFrequency(v as Frequency)}
|
||||
style={{ width: 120 }}
|
||||
options={[
|
||||
{ value: "all", label: "全部频率" },
|
||||
{ value: "high", label: "高频使用" },
|
||||
{ value: "medium", label: "中频使用" },
|
||||
{ value: "low", label: "低频使用" },
|
||||
]}
|
||||
/>
|
||||
</div>
|
||||
<div className="xx-titles-filters-right">
|
||||
<Button
|
||||
buttonType="ghost"
|
||||
buttonSize="sm"
|
||||
icon={<PlusOutlined />}
|
||||
onClick={() => setCreateTitleModalOpen(true)}
|
||||
>
|
||||
新建标题
|
||||
</Button>
|
||||
<Button
|
||||
buttonType="primary"
|
||||
buttonSize="sm"
|
||||
icon={<RobotOutlined />}
|
||||
onClick={() => setAiModalOpen(true)}
|
||||
>
|
||||
AI 生成标题
|
||||
</Button>
|
||||
</div>
|
||||
</div>
|
||||
<FilterBar
|
||||
searchText={searchText}
|
||||
onSearchChange={setSearchText}
|
||||
filterType={filterType}
|
||||
onFilterTypeChange={setFilterType}
|
||||
filterIndustry={filterIndustry}
|
||||
onFilterIndustryChange={setFilterIndustry}
|
||||
filterFrequency={filterFrequency}
|
||||
onFilterFrequencyChange={setFilterFrequency}
|
||||
onCreateClick={() => setCreateTitleModalOpen(true)}
|
||||
onAIClick={() => setAiModalOpen(true)}
|
||||
/>
|
||||
|
||||
{/* 标题卡片网格 */}
|
||||
{filteredTitles.length > 0 ? (
|
||||
<div className="xx-title-grid">
|
||||
{filteredTitles.map((title) => (
|
||||
<TitleCard
|
||||
key={title.id}
|
||||
title={title}
|
||||
isEditing={editingId === title.id}
|
||||
editText={editingId === title.id ? editText : ""}
|
||||
onEditChange={setEditText}
|
||||
onStartEdit={() => handleStartEdit(title)}
|
||||
onSaveEdit={handleSaveEdit}
|
||||
onCancelEdit={handleCancelEdit}
|
||||
onCopy={() => handleCopy(title)}
|
||||
onDelete={() => handleDelete(title.id)}
|
||||
onToggleFavorite={() => handleToggleFavorite(title.id)}
|
||||
/>
|
||||
))}
|
||||
</div>
|
||||
) : (
|
||||
<div className="xx-titles-empty">
|
||||
<div className="xx-titles-empty-icon">
|
||||
<FileTextOutlined />
|
||||
</div>
|
||||
<p>
|
||||
{searchText
|
||||
? "未找到匹配的标题"
|
||||
: "暂无标题，点击「新建标题」或「AI 生成标题」开始"}
|
||||
</p>
|
||||
</div>
|
||||
)}
|
||||
<TitleGrid
|
||||
titles={filteredTitles}
|
||||
editingId={editingId}
|
||||
editText={editText}
|
||||
searchText={searchText}
|
||||
onEditChange={setEditText}
|
||||
onStartEdit={handleStartEdit}
|
||||
onSaveEdit={handleSaveEdit}
|
||||
onCancelEdit={handleCancelEdit}
|
||||
onCopy={handleCopy}
|
||||
onDelete={handleDelete}
|
||||
onToggleFavorite={handleToggleFavorite}
|
||||
/>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
{/* ─── 新建标题弹窗 ─── */}
|
||||
<AntModal
|
||||
title="新建标题"
|
||||
{/* 新建标题弹窗 */}
|
||||
<CreateTitleModal
|
||||
open={createTitleModalOpen}
|
||||
onCancel={() => setCreateTitleModalOpen(false)}
|
||||
onOk={handleCreateTitle}
|
||||
okText="创建"
|
||||
cancelText="取消"
|
||||
destroyOnClose
|
||||
>
|
||||
<div
|
||||
style={{
|
||||
display: "flex",
|
||||
flexDirection: "column",
|
||||
gap: 16,
|
||||
padding: "8px 0",
|
||||
}}
|
||||
>
|
||||
<div>
|
||||
<div
|
||||
style={{
|
||||
marginBottom: 6,
|
||||
fontSize: "var(--font-size-sm)",
|
||||
color: "var(--text-secondary)",
|
||||
}}
|
||||
>
|
||||
标题内容
|
||||
</div>
|
||||
<Input.TextArea
|
||||
placeholder="请输入标题内容"
|
||||
value={newTitleContent}
|
||||
onChange={(e) => setNewTitleContent(e.target.value)}
|
||||
rows={3}
|
||||
maxLength={200}
|
||||
showCount
|
||||
/>
|
||||
</div>
|
||||
<div>
|
||||
<div
|
||||
style={{
|
||||
marginBottom: 6,
|
||||
fontSize: "var(--font-size-sm)",
|
||||
color: "var(--text-secondary)",
|
||||
}}
|
||||
>
|
||||
标题类型
|
||||
</div>
|
||||
<Select
|
||||
value={newTitleType}
|
||||
onChange={(v) => setNewTitleType(v)}
|
||||
style={{ width: "100%" }}
|
||||
options={[
|
||||
{ value: "hot", label: "爆款" },
|
||||
{ value: "normal", label: "常规" },
|
||||
{ value: "creative", label: "创意" },
|
||||
]}
|
||||
/>
|
||||
</div>
|
||||
</div>
|
||||
</AntModal>
|
||||
newTitleContent={newTitleContent}
|
||||
newTitleType={newTitleType}
|
||||
onContentChange={setNewTitleContent}
|
||||
onTypeChange={setNewTitleType}
|
||||
onCancel={handleCloseCreateModal}
|
||||
onSubmit={handleCreateTitle}
|
||||
/>
|
||||
|
||||
{/* ─── AI 生成标题弹窗 ─── */}
|
||||
<AntModal
|
||||
title="AI 生成标题"
|
||||
{/* AI 生成标题弹窗 */}
|
||||
<AIGenerateModal
|
||||
open={aiModalOpen}
|
||||
onCancel={() => {
|
||||
setAiModalOpen(false)
|
||||
setAiLoading(false)
|
||||
setAiResults([])
|
||||
setAiKeyword("")
|
||||
}}
|
||||
onOk={handleAIGenerate}
|
||||
okText={aiLoading ? "生成中..." : "生成"}
|
||||
cancelText="关闭"
|
||||
okButtonProps={{ disabled: aiLoading }}
|
||||
destroyOnClose
|
||||
width={640}
|
||||
>
|
||||
<div
|
||||
style={{
|
||||
display: "flex",
|
||||
flexDirection: "column",
|
||||
gap: 16,
|
||||
padding: "8px 0",
|
||||
}}
|
||||
>
|
||||
<div>
|
||||
<div
|
||||
style={{
|
||||
marginBottom: 6,
|
||||
fontSize: "var(--font-size-sm)",
|
||||
color: "var(--text-secondary)",
|
||||
}}
|
||||
>
|
||||
输入关键词或主题
|
||||
</div>
|
||||
<Input
|
||||
placeholder="例如：美食探店、科技评测、旅行攻略..."
|
||||
value={aiKeyword}
|
||||
onChange={(e) => setAiKeyword(e.target.value)}
|
||||
maxLength={100}
|
||||
onPressEnter={handleAIGenerate}
|
||||
/>
|
||||
</div>
|
||||
|
||||
{/* AI 加载动画 */}
|
||||
{aiLoading && (
|
||||
<div className="xx-ai-loading">
|
||||
<div className="xx-ai-loading-dots">
|
||||
<div className="xx-ai-loading-dot" />
|
||||
<div className="xx-ai-loading-dot" />
|
||||
<div className="xx-ai-loading-dot" />
|
||||
</div>
|
||||
<span>AI 正在生成标题候选...</span>
|
||||
</div>
|
||||
)}
|
||||
|
||||
{/* AI 生成结果列表 */}
|
||||
{aiResults.length > 0 && (
|
||||
<div className="xx-ai-results">
|
||||
<div
|
||||
style={{
|
||||
fontSize: "var(--font-size-sm)",
|
||||
color: "var(--text-secondary)",
|
||||
marginBottom: 4,
|
||||
}}
|
||||
>
|
||||
已生成 {aiResults.length} 个候选标题，点击采纳或复制：
|
||||
</div>
|
||||
{aiResults.map((text, idx) => (
|
||||
<div key={idx} className="xx-ai-result-item">
|
||||
<span className="xx-ai-result-text">{text}</span>
|
||||
<div className="xx-ai-result-actions">
|
||||
<Button
|
||||
buttonType="ghost"
|
||||
buttonSize="sm"
|
||||
icon={<CopyOutlined />}
|
||||
onClick={() => handleCopyAI(text)}
|
||||
>
|
||||
复制
|
||||
</Button>
|
||||
<Button
|
||||
buttonType="primary"
|
||||
buttonSize="sm"
|
||||
icon={<CheckOutlined />}
|
||||
onClick={() => handleAdoptAITitle(text)}
|
||||
>
|
||||
采纳
|
||||
</Button>
|
||||
</div>
|
||||
</div>
|
||||
))}
|
||||
</div>
|
||||
)}
|
||||
</div>
|
||||
</AntModal>
|
||||
aiKeyword={aiKeyword}
|
||||
aiLoading={aiLoading}
|
||||
aiResults={aiResults}
|
||||
onKeywordChange={setAiKeyword}
|
||||
onGenerate={handleAIGenerate}
|
||||
onCancel={handleCloseAIModal}
|
||||
onCopy={handleCopyAI}
|
||||
onAdopt={handleAdoptAITitle}
|
||||
/>
|
||||
</div>
|
||||
)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,121 @@
|
||||
import React from "react"
|
||||
import { Modal as AntModal } from "antd"
|
||||
import { CopyOutlined, CheckOutlined } from "@ant-design/icons"
|
||||
import { Button, Input } from "@/components/ui"
|
||||
import { AI_KEYWORD_MAX_LENGTH } from "../../constants/titleLibrary"
|
||||
|
||||
interface AIGenerateModalProps {
|
||||
open: boolean
|
||||
aiKeyword: string
|
||||
aiLoading: boolean
|
||||
aiResults: string[]
|
||||
onKeywordChange: (keyword: string) => void
|
||||
onGenerate: () => void
|
||||
onCancel: () => void
|
||||
onCopy: (text: string) => void
|
||||
onAdopt: (text: string) => void
|
||||
}
|
||||
|
||||
export const AIGenerateModal: React.FC<AIGenerateModalProps> = ({
|
||||
open,
|
||||
aiKeyword,
|
||||
aiLoading,
|
||||
aiResults,
|
||||
onKeywordChange,
|
||||
onGenerate,
|
||||
onCancel,
|
||||
onCopy,
|
||||
onAdopt,
|
||||
}) => {
|
||||
return (
|
||||
<AntModal
|
||||
title="AI 生成标题"
|
||||
open={open}
|
||||
onCancel={onCancel}
|
||||
onOk={onGenerate}
|
||||
okText={aiLoading ? "生成中..." : "生成"}
|
||||
cancelText="关闭"
|
||||
okButtonProps={{ disabled: aiLoading }}
|
||||
destroyOnClose
|
||||
width={640}
|
||||
>
|
||||
<div
|
||||
style={{
|
||||
display: "flex",
|
||||
flexDirection: "column",
|
||||
gap: 16,
|
||||
padding: "8px 0",
|
||||
}}
|
||||
>
|
||||
<div>
|
||||
<div
|
||||
style={{
|
||||
marginBottom: 6,
|
||||
fontSize: "var(--font-size-sm)",
|
||||
color: "var(--text-secondary)",
|
||||
}}
|
||||
>
|
||||
输入关键词或主题
|
||||
</div>
|
||||
<Input
|
||||
placeholder="例如：美食探店、科技评测、旅行攻略..."
|
||||
value={aiKeyword}
|
||||
onChange={(e) => onKeywordChange(e.target.value)}
|
||||
maxLength={AI_KEYWORD_MAX_LENGTH}
|
||||
onPressEnter={onGenerate}
|
||||
/>
|
||||
</div>
|
||||
|
||||
{/* AI 加载动画 */}
|
||||
{aiLoading && (
|
||||
<div className="xx-ai-loading">
|
||||
<div className="xx-ai-loading-dots">
|
||||
<div className="xx-ai-loading-dot" />
|
||||
<div className="xx-ai-loading-dot" />
|
||||
<div className="xx-ai-loading-dot" />
|
||||
</div>
|
||||
<span>AI 正在生成标题候选...</span>
|
||||
</div>
|
||||
)}
|
||||
|
||||
{/* AI 生成结果列表 */}
|
||||
{aiResults.length > 0 && (
|
||||
<div className="xx-ai-results">
|
||||
<div
|
||||
style={{
|
||||
fontSize: "var(--font-size-sm)",
|
||||
color: "var(--text-secondary)",
|
||||
marginBottom: 4,
|
||||
}}
|
||||
>
|
||||
已生成 {aiResults.length} 个候选标题，点击采纳或复制：
|
||||
</div>
|
||||
{aiResults.map((text, idx) => (
|
||||
<div key={idx} className="xx-ai-result-item">
|
||||
<span className="xx-ai-result-text">{text}</span>
|
||||
<div className="xx-ai-result-actions">
|
||||
<Button
|
||||
buttonType="ghost"
|
||||
buttonSize="sm"
|
||||
icon={<CopyOutlined />}
|
||||
onClick={() => onCopy(text)}
|
||||
>
|
||||
复制
|
||||
</Button>
|
||||
<Button
|
||||
buttonType="primary"
|
||||
buttonSize="sm"
|
||||
icon={<CheckOutlined />}
|
||||
onClick={() => onAdopt(text)}
|
||||
>
|
||||
采纳
|
||||
</Button>
|
||||
</div>
|
||||
</div>
|
||||
))}
|
||||
</div>
|
||||
)}
|
||||
</div>
|
||||
</AntModal>
|
||||
)
|
||||
}
|
||||
@@ -0,0 +1,44 @@
|
||||
import React from "react"
|
||||
import { FileTextOutlined } from "@ant-design/icons"
|
||||
import type { CategoryItem } from "../../types/titleLibrary"
|
||||
|
||||
interface CategorySidebarProps {
|
||||
categories: CategoryItem[]
|
||||
activeCatId: string
|
||||
onSelect: (catId: string) => void
|
||||
}
|
||||
|
||||
export const CategorySidebar: React.FC<CategorySidebarProps> = ({
|
||||
categories,
|
||||
activeCatId,
|
||||
onSelect,
|
||||
}) => {
|
||||
return (
|
||||
<div className="xx-title-category-list">
|
||||
{categories.map((cat) => (
|
||||
<div
|
||||
key={cat.id}
|
||||
className={`xx-title-category-item${cat.id === activeCatId ? " active" : ""}`}
|
||||
onClick={() => onSelect(cat.id)}
|
||||
>
|
||||
<div
|
||||
style={{
|
||||
display: "flex",
|
||||
alignItems: "center",
|
||||
justifyContent: "space-between",
|
||||
flex: 1,
|
||||
minWidth: 0,
|
||||
}}
|
||||
>
|
||||
<div style={{ minWidth: 0 }}>
|
||||
<h4 style={{ margin: 0 }}>
|
||||
<FileTextOutlined /> {cat.name}
|
||||
</h4>
|
||||
<span>{cat.count} 条</span>
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
))}
|
||||
</div>
|
||||
)
|
||||
}
|
||||
@@ -0,0 +1,89 @@
|
||||
import React from "react"
|
||||
import { Modal as AntModal } from "antd"
|
||||
import { Input, Select } from "@/components/ui"
|
||||
import type { TitleType } from "../../types/titleLibrary"
|
||||
import { TITLE_MAX_LENGTH } from "../../constants/titleLibrary"
|
||||
|
||||
const TITLE_TYPE_CREATE_OPTIONS: Array<{ value: TitleType; label: string }> = [
|
||||
{ value: "hot", label: "爆款" },
|
||||
{ value: "normal", label: "常规" },
|
||||
{ value: "creative", label: "创意" },
|
||||
]
|
||||
|
||||
interface CreateTitleModalProps {
|
||||
open: boolean
|
||||
newTitleContent: string
|
||||
newTitleType: TitleType
|
||||
onContentChange: (content: string) => void
|
||||
onTypeChange: (type: TitleType) => void
|
||||
onCancel: () => void
|
||||
onSubmit: () => void
|
||||
}
|
||||
|
||||
export const CreateTitleModal: React.FC<CreateTitleModalProps> = ({
|
||||
open,
|
||||
newTitleContent,
|
||||
newTitleType,
|
||||
onContentChange,
|
||||
onTypeChange,
|
||||
onCancel,
|
||||
onSubmit,
|
||||
}) => {
|
||||
return (
|
||||
<AntModal
|
||||
title="新建标题"
|
||||
open={open}
|
||||
onCancel={onCancel}
|
||||
onOk={onSubmit}
|
||||
okText="创建"
|
||||
cancelText="取消"
|
||||
destroyOnClose
|
||||
>
|
||||
<div
|
||||
style={{
|
||||
display: "flex",
|
||||
flexDirection: "column",
|
||||
gap: 16,
|
||||
padding: "8px 0",
|
||||
}}
|
||||
>
|
||||
<div>
|
||||
<div
|
||||
style={{
|
||||
marginBottom: 6,
|
||||
fontSize: "var(--font-size-sm)",
|
||||
color: "var(--text-secondary)",
|
||||
}}
|
||||
>
|
||||
标题内容
|
||||
</div>
|
||||
<Input.TextArea
|
||||
placeholder="请输入标题内容"
|
||||
value={newTitleContent}
|
||||
onChange={(e) => onContentChange(e.target.value)}
|
||||
rows={3}
|
||||
maxLength={TITLE_MAX_LENGTH}
|
||||
showCount
|
||||
/>
|
||||
</div>
|
||||
<div>
|
||||
<div
|
||||
style={{
|
||||
marginBottom: 6,
|
||||
fontSize: "var(--font-size-sm)",
|
||||
color: "var(--text-secondary)",
|
||||
}}
|
||||
>
|
||||
标题类型
|
||||
</div>
|
||||
<Select
|
||||
value={newTitleType}
|
||||
onChange={(v) => onTypeChange(v as TitleType)}
|
||||
style={{ width: "100%" }}
|
||||
options={TITLE_TYPE_CREATE_OPTIONS}
|
||||
/>
|
||||
</div>
|
||||
</div>
|
||||
</AntModal>
|
||||
)
|
||||
}
|
||||
@@ -0,0 +1,76 @@
|
||||
import React from "react"
|
||||
import { SearchOutlined, PlusOutlined, RobotOutlined } from "@ant-design/icons"
|
||||
import { Button, Input, Select } from "@/components/ui"
|
||||
import type { Frequency } from "../../types/titleLibrary"
|
||||
import {
|
||||
TITLE_TYPE_OPTIONS,
|
||||
INDUSTRY_OPTIONS,
|
||||
FREQUENCY_OPTIONS,
|
||||
} from "../../constants/titleLibrary"
|
||||
|
||||
interface FilterBarProps {
|
||||
searchText: string
|
||||
onSearchChange: (text: string) => void
|
||||
filterType: string
|
||||
onFilterTypeChange: (value: string) => void
|
||||
filterIndustry: string
|
||||
onFilterIndustryChange: (value: string) => void
|
||||
filterFrequency: Frequency
|
||||
onFilterFrequencyChange: (value: Frequency) => void
|
||||
onCreateClick: () => void
|
||||
onAIClick: () => void
|
||||
}
|
||||
|
||||
export const FilterBar: React.FC<FilterBarProps> = ({
|
||||
searchText,
|
||||
onSearchChange,
|
||||
filterType,
|
||||
onFilterTypeChange,
|
||||
filterIndustry,
|
||||
onFilterIndustryChange,
|
||||
filterFrequency,
|
||||
onFilterFrequencyChange,
|
||||
onCreateClick,
|
||||
onAIClick,
|
||||
}) => {
|
||||
return (
|
||||
<div className="xx-titles-filters">
|
||||
<div className="xx-titles-filters-left">
|
||||
<Input
|
||||
placeholder="搜索标题关键词..."
|
||||
prefix={<SearchOutlined />}
|
||||
value={searchText}
|
||||
onChange={(e) => onSearchChange(e.target.value)}
|
||||
allowClear
|
||||
style={{ width: 220 }}
|
||||
/>
|
||||
<Select
|
||||
value={filterType}
|
||||
onChange={onFilterTypeChange}
|
||||
style={{ width: 110 }}
|
||||
options={TITLE_TYPE_OPTIONS}
|
||||
/>
|
||||
<Select
|
||||
value={filterIndustry}
|
||||
onChange={onFilterIndustryChange}
|
||||
style={{ width: 110 }}
|
||||
options={INDUSTRY_OPTIONS}
|
||||
/>
|
||||
<Select
|
||||
value={filterFrequency}
|
||||
onChange={(v) => onFilterFrequencyChange(v as Frequency)}
|
||||
style={{ width: 120 }}
|
||||
options={FREQUENCY_OPTIONS}
|
||||
/>
|
||||
</div>
|
||||
<div className="xx-titles-filters-right">
|
||||
<Button buttonType="ghost" buttonSize="sm" icon={<PlusOutlined />} onClick={onCreateClick}>
|
||||
新建标题
|
||||
</Button>
|
||||
<Button buttonType="primary" buttonSize="sm" icon={<RobotOutlined />} onClick={onAIClick}>
|
||||
AI 生成标题
|
||||
</Button>
|
||||
</div>
|
||||
</div>
|
||||
)
|
||||
}
|
||||
@@ -0,0 +1,121 @@
|
||||
import React from "react"
|
||||
import { Popconfirm } from "antd"
|
||||
import {
|
||||
StarOutlined,
|
||||
StarFilled,
|
||||
EditOutlined,
|
||||
CopyOutlined,
|
||||
DeleteOutlined,
|
||||
CheckOutlined,
|
||||
} from "@ant-design/icons"
|
||||
import type { TitleData } from "../../types/titleLibrary"
|
||||
import { typeLabel } from "../../utils/titleLibrary"
|
||||
|
||||
interface TitleCardProps {
|
||||
title: TitleData
|
||||
isEditing: boolean
|
||||
editText: string
|
||||
onEditChange: (text: string) => void
|
||||
onStartEdit: () => void
|
||||
onSaveEdit: () => void
|
||||
onCancelEdit: () => void
|
||||
onCopy: () => void
|
||||
onDelete: () => void
|
||||
onToggleFavorite: () => void
|
||||
}
|
||||
|
||||
export const TitleCard: React.FC<TitleCardProps> = ({
|
||||
title,
|
||||
isEditing,
|
||||
editText,
|
||||
onEditChange,
|
||||
onStartEdit,
|
||||
onSaveEdit,
|
||||
onCancelEdit,
|
||||
onCopy,
|
||||
onDelete,
|
||||
onToggleFavorite,
|
||||
}) => {
|
||||
return (
|
||||
<div className="xx-title-card">
|
||||
{/* 收藏按钮 */}
|
||||
<button
|
||||
className="xx-title-fav-btn"
|
||||
onClick={onToggleFavorite}
|
||||
title={title.isFavorited ? "取消收藏" : "收藏"}
|
||||
style={{
|
||||
position: "absolute",
|
||||
top: 12,
|
||||
right: 12,
|
||||
color: title.isFavorited ? "#f59e0b" : "var(--text-tertiary)",
|
||||
}}
|
||||
>
|
||||
{title.isFavorited ? <StarFilled /> : <StarOutlined />}
|
||||
</button>
|
||||
|
||||
{/* 标题文本 / 编辑区 */}
|
||||
{isEditing ? (
|
||||
<textarea
|
||||
className="xx-title-card-edit"
|
||||
value={editText}
|
||||
onChange={(e) => onEditChange(e.target.value)}
|
||||
autoFocus
|
||||
onKeyDown={(e) => {
|
||||
if (e.key === "Enter" && !e.shiftKey) {
|
||||
e.preventDefault()
|
||||
onSaveEdit()
|
||||
}
|
||||
if (e.key === "Escape") {
|
||||
onCancelEdit()
|
||||
}
|
||||
}}
|
||||
/>
|
||||
) : (
|
||||
<div className="xx-title-card-text" style={{ paddingRight: 24 }}>
|
||||
{title.content}
|
||||
</div>
|
||||
)}
|
||||
|
||||
{/* 底部元信息 */}
|
||||
<div className="xx-title-card-meta">
|
||||
<div className="xx-title-card-meta-left">
|
||||
<span className={`xx-title-type-tag ${title.type}`}>{typeLabel(title.type)}</span>
|
||||
<span className="xx-title-card-stat">使用 {title.usageCount} 次</span>
|
||||
<span className="xx-title-card-stat">{title.createdAt}</span>
|
||||
</div>
|
||||
|
||||
<div className="xx-title-card-actions">
|
||||
{isEditing ? (
|
||||
<>
|
||||
<button className="xx-title-card-action-btn" onClick={onSaveEdit} title="保存">
|
||||
<CheckOutlined />
|
||||
</button>
|
||||
<button className="xx-title-card-action-btn" onClick={onCancelEdit} title="取消">
|
||||
✕
|
||||
</button>
|
||||
</>
|
||||
) : (
|
||||
<>
|
||||
<button className="xx-title-card-action-btn" onClick={onCopy} title="复制">
|
||||
<CopyOutlined />
|
||||
</button>
|
||||
<button className="xx-title-card-action-btn" onClick={onStartEdit} title="编辑">
|
||||
<EditOutlined />
|
||||
</button>
|
||||
<Popconfirm
|
||||
title="确定删除此标题？"
|
||||
onConfirm={onDelete}
|
||||
okText="删除"
|
||||
cancelText="取消"
|
||||
>
|
||||
<button className="xx-title-card-action-btn danger" title="删除">
|
||||
<DeleteOutlined />
|
||||
</button>
|
||||
</Popconfirm>
|
||||
</>
|
||||
)}
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
)
|
||||
}
|
||||
@@ -0,0 +1,63 @@
|
||||
import React from "react"
|
||||
import { FileTextOutlined } from "@ant-design/icons"
|
||||
import { TitleCard } from "./TitleCard"
|
||||
import type { TitleData } from "../../types/titleLibrary"
|
||||
|
||||
interface TitleGridProps {
|
||||
titles: TitleData[]
|
||||
editingId: string | null
|
||||
editText: string
|
||||
searchText: string
|
||||
onEditChange: (text: string) => void
|
||||
onStartEdit: (title: TitleData) => void
|
||||
onSaveEdit: () => void
|
||||
onCancelEdit: () => void
|
||||
onCopy: (title: TitleData) => void
|
||||
onDelete: (id: string) => void
|
||||
onToggleFavorite: (id: string) => void
|
||||
}
|
||||
|
||||
export const TitleGrid: React.FC<TitleGridProps> = ({
|
||||
titles,
|
||||
editingId,
|
||||
editText,
|
||||
searchText,
|
||||
onEditChange,
|
||||
onStartEdit,
|
||||
onSaveEdit,
|
||||
onCancelEdit,
|
||||
onCopy,
|
||||
onDelete,
|
||||
onToggleFavorite,
|
||||
}) => {
|
||||
if (titles.length > 0) {
|
||||
return (
|
||||
<div className="xx-title-grid">
|
||||
{titles.map((title) => (
|
||||
<TitleCard
|
||||
key={title.id}
|
||||
title={title}
|
||||
isEditing={editingId === title.id}
|
||||
editText={editingId === title.id ? editText : ""}
|
||||
onEditChange={onEditChange}
|
||||
onStartEdit={() => onStartEdit(title)}
|
||||
onSaveEdit={onSaveEdit}
|
||||
onCancelEdit={onCancelEdit}
|
||||
onCopy={() => onCopy(title)}
|
||||
onDelete={() => onDelete(title.id)}
|
||||
onToggleFavorite={() => onToggleFavorite(title.id)}
|
||||
/>
|
||||
))}
|
||||
</div>
|
||||
)
|
||||
}
|
||||
|
||||
return (
|
||||
<div className="xx-titles-empty">
|
||||
<div className="xx-titles-empty-icon">
|
||||
<FileTextOutlined />
|
||||
</div>
|
||||
<p>{searchText ? "未找到匹配的标题" : "暂无标题，点击「新建标题」或「AI 生成标题」开始"}</p>
|
||||
</div>
|
||||
)
|
||||
}
|
||||
@@ -0,0 +1,34 @@
|
||||
import type { TitleType, Industry, Frequency } from "../types/titleLibrary"
|
||||
|
||||
export const TITLE_TYPE_OPTIONS: Array<{ value: TitleType | "all"; label: string }> = [
|
||||
{ value: "all", label: "全部类型" },
|
||||
{ value: "hot", label: "爆款" },
|
||||
{ value: "normal", label: "常规" },
|
||||
{ value: "creative", label: "创意" },
|
||||
]
|
||||
|
||||
export const INDUSTRY_OPTIONS: Array<{ value: Industry | "all"; label: string }> = [
|
||||
{ value: "all", label: "全部行业" },
|
||||
{ value: "food", label: "美食" },
|
||||
{ value: "tech", label: "科技" },
|
||||
{ value: "beauty", label: "美妆" },
|
||||
{ value: "education", label: "教育" },
|
||||
{ value: "travel", label: "旅行" },
|
||||
]
|
||||
|
||||
export const FREQUENCY_OPTIONS: Array<{ value: Frequency; label: string }> = [
|
||||
{ value: "all", label: "全部频率" },
|
||||
{ value: "high", label: "高频使用" },
|
||||
{ value: "medium", label: "中频使用" },
|
||||
{ value: "low", label: "低频使用" },
|
||||
]
|
||||
|
||||
export const FREQUENCY_THRESHOLDS = {
|
||||
high: 100,
|
||||
medium: 30,
|
||||
} as const
|
||||
|
||||
export const AI_GENERATE_DELAY = 2000
|
||||
export const TITLE_MAX_LENGTH = 200
|
||||
export const AI_KEYWORD_MAX_LENGTH = 100
|
||||
export const ALL_CATEGORY_ID = "cat-all"
|
||||
@@ -0,0 +1,84 @@
|
||||
import { useState, useCallback } from "react"
|
||||
import { message } from "antd"
|
||||
import type { UseMutationResult } from "@tanstack/react-query"
|
||||
import type { TitleItem } from "@/api/titles"
|
||||
import { copyToClipboard } from "../utils/titleLibrary"
|
||||
import { AI_GENERATE_DELAY } from "../constants/titleLibrary"
|
||||
|
||||
interface UseTitleAIProps {
|
||||
createMutation: UseMutationResult<TitleItem, Error, string, unknown>
|
||||
}
|
||||
|
||||
const generateMockTitles = (keyword: string): string[] => [
|
||||
`${keyword}：这个方法让我事半功倍！`,
|
||||
`关于${keyword}，99%的人都不知道的事`,
|
||||
`${keyword}全攻略，看完这篇就够了`,
|
||||
`我花了 3 个月研究${keyword}，总结出这些经验`,
|
||||
`${keyword}避坑指南，帮你省下 1000 块`,
|
||||
]
|
||||
|
||||
export const useTitleAI = ({ createMutation }: UseTitleAIProps) => {
|
||||
const [aiModalOpen, setAiModalOpen] = useState(false)
|
||||
const [aiKeyword, setAiKeyword] = useState("")
|
||||
const [aiLoading, setAiLoading] = useState(false)
|
||||
const [aiResults, setAiResults] = useState<string[]>([])
|
||||
|
||||
/* AI 生成标题 */
|
||||
const handleAIGenerate = useCallback(() => {
|
||||
if (!aiKeyword.trim()) {
|
||||
message.warning("请输入关键词或主题")
|
||||
return
|
||||
}
|
||||
setAiLoading(true)
|
||||
setAiResults([])
|
||||
|
||||
setTimeout(() => {
|
||||
const results = generateMockTitles(aiKeyword.trim())
|
||||
setAiResults(results)
|
||||
setAiLoading(false)
|
||||
}, AI_GENERATE_DELAY)
|
||||
}, [aiKeyword])
|
||||
|
||||
/* 采纳 AI 生成的标题 */
|
||||
const handleAdoptAITitle = useCallback(
|
||||
(text: string) => {
|
||||
createMutation.mutate(text, {
|
||||
onSuccess: () => {
|
||||
message.success("标题已采纳并添加到标题库")
|
||||
},
|
||||
})
|
||||
},
|
||||
[createMutation],
|
||||
)
|
||||
|
||||
/* 复制 AI 生成的标题 */
|
||||
const handleCopyAI = useCallback(async (text: string) => {
|
||||
const ok = await copyToClipboard(text)
|
||||
if (ok) {
|
||||
message.success("已复制到剪贴板")
|
||||
} else {
|
||||
message.error("复制失败")
|
||||
}
|
||||
}, [])
|
||||
|
||||
/* 关闭 AI 弹窗 */
|
||||
const handleCloseAIModal = useCallback(() => {
|
||||
setAiModalOpen(false)
|
||||
setAiLoading(false)
|
||||
setAiResults([])
|
||||
setAiKeyword("")
|
||||
}, [])
|
||||
|
||||
return {
|
||||
aiModalOpen,
|
||||
setAiModalOpen,
|
||||
aiKeyword,
|
||||
setAiKeyword,
|
||||
aiLoading,
|
||||
aiResults,
|
||||
handleAIGenerate,
|
||||
handleAdoptAITitle,
|
||||
handleCopyAI,
|
||||
handleCloseAIModal,
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,87 @@
|
||||
import { useState, useCallback } from "react"
|
||||
import { message } from "antd"
|
||||
import type { TitleData, TitleType } from "../types/titleLibrary"
|
||||
import type { UseMutationResult } from "@tanstack/react-query"
|
||||
import type { TitleItem } from "@/api/titles"
|
||||
|
||||
interface UseTitleEditProps {
|
||||
updateMutation: UseMutationResult<TitleItem, Error, { id: string; content: string }, unknown>
|
||||
createMutation: UseMutationResult<TitleItem, Error, string, unknown>
|
||||
}
|
||||
|
||||
export const useTitleEdit = ({ updateMutation, createMutation }: UseTitleEditProps) => {
|
||||
/* 编辑状态 */
|
||||
const [editingId, setEditingId] = useState<string | null>(null)
|
||||
const [editText, setEditText] = useState("")
|
||||
|
||||
/* 新建标题弹窗 */
|
||||
const [createTitleModalOpen, setCreateTitleModalOpen] = useState(false)
|
||||
const [newTitleContent, setNewTitleContent] = useState("")
|
||||
const [newTitleType, setNewTitleType] = useState<TitleType>("normal")
|
||||
|
||||
/* 开始编辑 */
|
||||
const handleStartEdit = useCallback((title: TitleData) => {
|
||||
setEditingId(title.id)
|
||||
setEditText(title.content)
|
||||
}, [])
|
||||
|
||||
/* 保存编辑 */
|
||||
const handleSaveEdit = useCallback(() => {
|
||||
if (!editText.trim()) {
|
||||
message.warning("标题内容不能为空")
|
||||
return
|
||||
}
|
||||
if (editingId) {
|
||||
updateMutation.mutate({ id: editingId, content: editText.trim() })
|
||||
}
|
||||
setEditingId(null)
|
||||
setEditText("")
|
||||
message.success("标题已更新")
|
||||
}, [editingId, editText, updateMutation])
|
||||
|
||||
/* 取消编辑 */
|
||||
const handleCancelEdit = useCallback(() => {
|
||||
setEditingId(null)
|
||||
setEditText("")
|
||||
}, [])
|
||||
|
||||
/* 新建标题提交 */
|
||||
const handleCreateTitle = useCallback(() => {
|
||||
if (!newTitleContent.trim()) {
|
||||
message.warning("请输入标题内容")
|
||||
return
|
||||
}
|
||||
createMutation.mutate(newTitleContent.trim(), {
|
||||
onSuccess: () => {
|
||||
setCreateTitleModalOpen(false)
|
||||
setNewTitleContent("")
|
||||
setNewTitleType("normal")
|
||||
message.success("标题创建成功")
|
||||
},
|
||||
})
|
||||
}, [newTitleContent, newTitleType, createMutation])
|
||||
|
||||
/* 关闭新建弹窗 */
|
||||
const handleCloseCreateModal = useCallback(() => {
|
||||
setCreateTitleModalOpen(false)
|
||||
setNewTitleContent("")
|
||||
setNewTitleType("normal")
|
||||
}, [])
|
||||
|
||||
return {
|
||||
editingId,
|
||||
editText,
|
||||
setEditText,
|
||||
createTitleModalOpen,
|
||||
setCreateTitleModalOpen,
|
||||
newTitleContent,
|
||||
setNewTitleContent,
|
||||
newTitleType,
|
||||
setNewTitleType,
|
||||
handleStartEdit,
|
||||
handleSaveEdit,
|
||||
handleCancelEdit,
|
||||
handleCreateTitle,
|
||||
handleCloseCreateModal,
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,174 @@
|
||||
import { useMemo, useState, useCallback } from "react"
|
||||
import { message } from "antd"
|
||||
import { useQuery, useMutation, useQueryClient } from "@tanstack/react-query"
|
||||
import { getTitles, createTitle, updateTitle, deleteTitle } from "@/api/titles"
|
||||
import type { TitleData, CategoryItem, Frequency } from "../types/titleLibrary"
|
||||
import { toTitleData, copyToClipboard } from "../utils/titleLibrary"
|
||||
import { ALL_CATEGORY_ID, FREQUENCY_THRESHOLDS } from "../constants/titleLibrary"
|
||||
|
||||
export const useTitleLibrary = () => {
|
||||
const queryClient = useQueryClient()
|
||||
|
||||
/* 分类 */
|
||||
const [activeCatId, setActiveCatId] = useState<string>(ALL_CATEGORY_ID)
|
||||
|
||||
/* 数据获取 */
|
||||
const { data: apiTitles = [] } = useQuery({
|
||||
queryKey: ["titles"],
|
||||
queryFn: getTitles,
|
||||
staleTime: 30_000,
|
||||
})
|
||||
|
||||
const titles: TitleData[] = useMemo(() => apiTitles.map(toTitleData), [apiTitles])
|
||||
|
||||
/* 动态派生分类 */
|
||||
const categories: CategoryItem[] = useMemo(() => {
|
||||
const cats = new Map<string, number>()
|
||||
apiTitles.forEach((t) => {
|
||||
const cat = t.category || "未分类"
|
||||
cats.set(cat, (cats.get(cat) || 0) + 1)
|
||||
})
|
||||
return [
|
||||
{ id: ALL_CATEGORY_ID, name: "全部标题", count: apiTitles.length },
|
||||
...Array.from(cats.entries()).map(([name, count]) => ({
|
||||
id: `cat-${name}`,
|
||||
name,
|
||||
count,
|
||||
})),
|
||||
]
|
||||
}, [apiTitles])
|
||||
|
||||
/* CRUD mutations */
|
||||
const createMutation = useMutation({
|
||||
mutationFn: (content: string) => createTitle({ content }),
|
||||
onSuccess: () => {
|
||||
queryClient.invalidateQueries({ queryKey: ["titles"] })
|
||||
},
|
||||
onError: () => message.error("创建标题失败"),
|
||||
})
|
||||
|
||||
const updateMutation = useMutation({
|
||||
mutationFn: ({ id, content }: { id: string; content: string }) => updateTitle(id, { content }),
|
||||
onSuccess: () => {
|
||||
queryClient.invalidateQueries({ queryKey: ["titles"] })
|
||||
},
|
||||
onError: () => message.error("更新标题失败"),
|
||||
})
|
||||
|
||||
const deleteMutation = useMutation({
|
||||
mutationFn: (id: string) => deleteTitle(id),
|
||||
onSuccess: () => {
|
||||
queryClient.invalidateQueries({ queryKey: ["titles"] })
|
||||
},
|
||||
onError: () => message.error("删除标题失败"),
|
||||
})
|
||||
|
||||
/* 筛选状态 */
|
||||
const [searchText, setSearchText] = useState("")
|
||||
const [filterType, setFilterType] = useState<string>("all")
|
||||
const [filterIndustry, setFilterIndustry] = useState<string>("all")
|
||||
const [filterFrequency, setFilterFrequency] = useState<Frequency>("all")
|
||||
|
||||
/* 派生：筛选后的标题列表 */
|
||||
const activeCategory = categories.find((c) => c.id === activeCatId)
|
||||
|
||||
const filteredTitles = useMemo(() => {
|
||||
let list = titles
|
||||
|
||||
/* 按分类过滤 */
|
||||
if (activeCatId !== ALL_CATEGORY_ID) {
|
||||
const catName = activeCategory?.name || ""
|
||||
if (catName) {
|
||||
list = list.filter((t) => t.category === catName)
|
||||
}
|
||||
}
|
||||
|
||||
/* 按类型筛选 */
|
||||
if (filterType !== "all") {
|
||||
list = list.filter((t) => t.type === filterType)
|
||||
}
|
||||
|
||||
/* 按行业筛选 */
|
||||
if (filterIndustry !== "all") {
|
||||
list = list.filter((t) => t.industry === filterIndustry)
|
||||
}
|
||||
|
||||
/* 按使用频率筛选 */
|
||||
if (filterFrequency !== "all") {
|
||||
switch (filterFrequency) {
|
||||
case "high":
|
||||
list = list.filter((t) => t.usageCount >= FREQUENCY_THRESHOLDS.high)
|
||||
break
|
||||
case "medium":
|
||||
list = list.filter(
|
||||
(t) =>
|
||||
t.usageCount >= FREQUENCY_THRESHOLDS.medium &&
|
||||
t.usageCount < FREQUENCY_THRESHOLDS.high,
|
||||
)
|
||||
break
|
||||
case "low":
|
||||
list = list.filter((t) => t.usageCount < FREQUENCY_THRESHOLDS.medium)
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
/* 搜索 */
|
||||
if (searchText.trim()) {
|
||||
const q = searchText.trim().toLowerCase()
|
||||
list = list.filter((t) => t.content.toLowerCase().includes(q))
|
||||
}
|
||||
|
||||
return list
|
||||
}, [titles, activeCatId, activeCategory, filterType, filterIndustry, filterFrequency, searchText])
|
||||
|
||||
/* 操作：收藏 */
|
||||
const handleToggleFavorite = useCallback((_id: string) => {
|
||||
message.info("收藏功能即将上线")
|
||||
}, [])
|
||||
|
||||
/* 操作：复制 */
|
||||
const handleCopy = useCallback(async (title: TitleData) => {
|
||||
const ok = await copyToClipboard(title.content)
|
||||
if (ok) {
|
||||
message.success("已复制到剪贴板")
|
||||
} else {
|
||||
message.error("复制失败")
|
||||
}
|
||||
}, [])
|
||||
|
||||
/* 操作：删除 */
|
||||
const handleDelete = useCallback(
|
||||
(id: string) => {
|
||||
deleteMutation.mutate(id)
|
||||
message.success("标题已删除")
|
||||
},
|
||||
[deleteMutation],
|
||||
)
|
||||
|
||||
return {
|
||||
/* 状态 */
|
||||
titles,
|
||||
categories,
|
||||
activeCatId,
|
||||
activeCategory,
|
||||
filteredTitles,
|
||||
searchText,
|
||||
filterType,
|
||||
filterIndustry,
|
||||
filterFrequency,
|
||||
/* mutations */
|
||||
createMutation,
|
||||
updateMutation,
|
||||
deleteMutation,
|
||||
/* setters */
|
||||
setActiveCatId,
|
||||
setSearchText,
|
||||
setFilterType,
|
||||
setFilterIndustry,
|
||||
setFilterFrequency,
|
||||
/* handlers */
|
||||
handleToggleFavorite,
|
||||
handleCopy,
|
||||
handleDelete,
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,20 @@
|
||||
export type TitleType = "hot" | "normal" | "creative"
|
||||
export type Industry = "general" | "food" | "tech" | "beauty" | "education" | "travel"
|
||||
export type Frequency = "all" | "high" | "medium" | "low"
|
||||
|
||||
export interface TitleData {
|
||||
id: string
|
||||
content: string
|
||||
type: TitleType
|
||||
industry: Industry
|
||||
category: string
|
||||
usageCount: number
|
||||
isFavorited: boolean
|
||||
createdAt: string
|
||||
}
|
||||
|
||||
export interface CategoryItem {
|
||||
id: string
|
||||
name: string
|
||||
count: number
|
||||
}
|
||||
@@ -0,0 +1,49 @@
|
||||
import type { TitleData, TitleType } from "../types/titleLibrary"
|
||||
import type { TitleItem } from "@/api/titles"
|
||||
|
||||
export const typeLabel = (type: TitleType): string => {
|
||||
switch (type) {
|
||||
case "hot":
|
||||
return "爆款"
|
||||
case "normal":
|
||||
return "常规"
|
||||
case "creative":
|
||||
return "创意"
|
||||
}
|
||||
}
|
||||
|
||||
/** 后端 TitleItem → 前端 TitleData 映射 */
|
||||
export const toTitleData = (item: TitleItem): TitleData => ({
|
||||
id: item.id,
|
||||
content: item.content,
|
||||
type: (item.category as TitleType) || "normal",
|
||||
industry: "general",
|
||||
category: item.category || "未分类",
|
||||
usageCount: 0,
|
||||
isFavorited: false,
|
||||
createdAt: item.created_at?.slice(0, 10) || "",
|
||||
})
|
||||
|
||||
/** 复制文本到剪贴板 */
|
||||
export const copyToClipboard = async (text: string): Promise<boolean> => {
|
||||
try {
|
||||
await navigator.clipboard.writeText(text)
|
||||
return true
|
||||
} catch {
|
||||
/* 降级方案 */
|
||||
const textarea = document.createElement("textarea")
|
||||
textarea.value = text
|
||||
textarea.style.position = "fixed"
|
||||
textarea.style.opacity = "0"
|
||||
document.body.appendChild(textarea)
|
||||
textarea.select()
|
||||
try {
|
||||
document.execCommand("copy")
|
||||
return true
|
||||
} catch {
|
||||
return false
|
||||
} finally {
|
||||
document.body.removeChild(textarea)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -65,6 +65,18 @@ vi.mock("@/store/authStore", () => ({
|
||||
vi.mock("@/pages/titles/titles.css", () => ({}))
|
||||
|
||||
import TitleLibrary from "@/pages/titles/TitleLibrary"
|
||||
import "@/pages/titles/types/titleLibrary"
|
||||
import "@/pages/titles/constants/titleLibrary"
|
||||
import "@/pages/titles/utils/titleLibrary"
|
||||
import "@/pages/titles/hooks/useTitleLibrary"
|
||||
import "@/pages/titles/hooks/useTitleEdit"
|
||||
import "@/pages/titles/hooks/useTitleAI"
|
||||
import "@/pages/titles/components/title-library/TitleCard"
|
||||
import "@/pages/titles/components/title-library/CategorySidebar"
|
||||
import "@/pages/titles/components/title-library/FilterBar"
|
||||
import "@/pages/titles/components/title-library/TitleGrid"
|
||||
import "@/pages/titles/components/title-library/CreateTitleModal"
|
||||
import "@/pages/titles/components/title-library/AIGenerateModal"
|
||||
|
||||
describe("TitleLibrary Page", () => {
|
||||
it("should render without crashing", () => {
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
"""视频调速引擎 — 基于 FFmpeg setpts + atempo 的速度调整能力。
|
||||
"""视频调速引擎 — 基于 FFmpeg setpts + atempo 的速度调整能力.
|
||||
|
||||
支持：
|
||||
- 0.25x ~ 4x 变速范围
|
||||
@@ -6,147 +6,57 @@
|
||||
- 音频调速（atempo，多级串联处理超范围值）
|
||||
- 音调修正（pitch_correct，默认开启）
|
||||
- 边界自动钳制，不阻断渲染
|
||||
|
||||
注：核心领域模型已抽离到 packages/domain/speed_config.py，
|
||||
本模块保留薄包装层，确保向后兼容。
|
||||
"""
|
||||
|
||||
from dataclasses import dataclass
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Optional
|
||||
|
||||
# ─── 常量 ───────────────────────────────────────────────
|
||||
MIN_SPEED = 0.25
|
||||
MAX_SPEED = 4.0
|
||||
DEFAULT_SPEED = 1.0
|
||||
|
||||
# atempo 单级有效范围
|
||||
_ATEMPO_MIN = 0.5
|
||||
_ATEMPO_MAX = 2.0
|
||||
|
||||
|
||||
@dataclass
|
||||
class SpeedConfig:
|
||||
"""调速配置。
|
||||
|
||||
Attributes:
|
||||
speed: 播放速度，0.25~4.0，1.0 为原速
|
||||
pitch_correct: 是否保持音调（默认 True，用 atempo 时间拉伸算法）
|
||||
"""
|
||||
|
||||
speed: float = DEFAULT_SPEED
|
||||
pitch_correct: bool = True
|
||||
|
||||
@classmethod
|
||||
def parse(cls, data: Optional[dict]) -> "SpeedConfig":
|
||||
"""从 dict 解析配置，无效值回退到默认。"""
|
||||
if not data or not isinstance(data, dict):
|
||||
return cls()
|
||||
|
||||
speed = data.get("speed", DEFAULT_SPEED)
|
||||
if not isinstance(speed, (int, float)):
|
||||
speed = DEFAULT_SPEED
|
||||
|
||||
pitch_correct = data.get("pitch_correct", True)
|
||||
if not isinstance(pitch_correct, bool):
|
||||
pitch_correct = True
|
||||
|
||||
config = cls(speed=float(speed), pitch_correct=pitch_correct)
|
||||
config.clamp()
|
||||
return config
|
||||
|
||||
def clamp(self) -> None:
|
||||
"""将速度钳制到合法范围。"""
|
||||
if self.speed <= 0:
|
||||
self.speed = DEFAULT_SPEED
|
||||
elif self.speed < MIN_SPEED:
|
||||
self.speed = MIN_SPEED
|
||||
elif self.speed > MAX_SPEED:
|
||||
self.speed = MAX_SPEED
|
||||
|
||||
@property
|
||||
def is_original(self) -> bool:
|
||||
"""是否原速（无需调速）。"""
|
||||
return abs(self.speed - 1.0) < 1e-6
|
||||
from packages.domain.speed_config import ( # noqa: F401 — 向后兼容
|
||||
DEFAULT_SPEED,
|
||||
MAX_SPEED,
|
||||
MIN_SPEED,
|
||||
SpeedConfig,
|
||||
_split_atempo_stages,
|
||||
adjust_duration as _adjust_duration_base,
|
||||
build_audio_filter as _build_audio_filter_base,
|
||||
build_video_filter as _build_video_filter_base,
|
||||
resolve_clip_speed as _resolve_clip_speed_base,
|
||||
)
|
||||
|
||||
|
||||
class SpeedEngine:
|
||||
"""调速引擎 — 生成 FFmpeg 调速滤镜链。
|
||||
"""调速引擎 — 生成 FFmpeg 调速滤镜链.
|
||||
|
||||
用法：
|
||||
engine = SpeedEngine()
|
||||
video_filter = engine.build_video_filter(config)
|
||||
audio_filter = engine.build_audio_filter(config)
|
||||
new_duration = engine.adjust_duration(duration, config)
|
||||
薄包装层，实际逻辑委托给 packages.domain.speed_config。
|
||||
"""
|
||||
|
||||
def build_video_filter(self, config: SpeedConfig) -> str:
|
||||
"""生成视频调速滤镜字符串。
|
||||
|
||||
返回 setpts 滤镜表达式，原速时返回空字符串。
|
||||
"""
|
||||
if config.is_original:
|
||||
return ""
|
||||
# setpts=PTS/speed — speed>1 加速，speed<1 减速
|
||||
return f"setpts=PTS/{config.speed:.4f}"
|
||||
"""生成视频调速滤镜字符串."""
|
||||
return _build_video_filter_base(config)
|
||||
|
||||
def build_audio_filter(self, config: SpeedConfig) -> str:
|
||||
"""生成音频调速滤镜字符串。
|
||||
|
||||
atempo 单级范围 0.5~2.0，超出范围时自动多级串联：
|
||||
- 0.25x → atempo=0.5,atempo=0.5
|
||||
- 4x → atempo=2.0,atempo=2.0
|
||||
- 0.3x → atempo=0.5,atempo=0.6
|
||||
- 3x → atempo=2.0,atempo=1.5
|
||||
|
||||
原速时返回空字符串。
|
||||
"""
|
||||
if config.is_original:
|
||||
return ""
|
||||
|
||||
speed = config.speed
|
||||
stages: list[float] = self._split_atempo_stages(speed)
|
||||
return ",".join(f"atempo={s:.4f}" for s in stages)
|
||||
"""生成音频调速滤镜字符串."""
|
||||
return _build_audio_filter_base(config)
|
||||
|
||||
@staticmethod
|
||||
def _split_atempo_stages(speed: float) -> list[float]:
|
||||
"""将速度拆分为多级 atempo 串联，每级都在 [0.5, 2.0] 范围内。"""
|
||||
if _ATEMPO_MIN <= speed <= _ATEMPO_MAX:
|
||||
return [speed]
|
||||
|
||||
stages: list[float] = []
|
||||
remaining = speed
|
||||
|
||||
# 加速场景（speed > 2.0）
|
||||
if speed > _ATEMPO_MAX:
|
||||
while remaining > _ATEMPO_MAX:
|
||||
stages.append(_ATEMPO_MAX)
|
||||
remaining /= _ATEMPO_MAX
|
||||
stages.append(remaining)
|
||||
|
||||
# 减速场景（speed < 0.5）
|
||||
else:
|
||||
while remaining < _ATEMPO_MIN:
|
||||
stages.append(_ATEMPO_MIN)
|
||||
remaining /= _ATEMPO_MIN
|
||||
stages.append(remaining)
|
||||
|
||||
return stages
|
||||
"""将速度拆分为多级 atempo 串联（内部方法，向后兼容）."""
|
||||
return _split_atempo_stages(speed)
|
||||
|
||||
def adjust_duration(self, original_duration: float, config: SpeedConfig) -> float:
|
||||
"""计算调速后的时长。
|
||||
|
||||
加速 → 时长变短；减速 → 时长变长。
|
||||
"""
|
||||
if config.is_original or original_duration <= 0:
|
||||
return original_duration
|
||||
return original_duration / config.speed
|
||||
"""计算调速后的时长."""
|
||||
return _adjust_duration_base(original_duration, config)
|
||||
|
||||
def build_clip_speed_filter(
|
||||
self,
|
||||
speed: float,
|
||||
pitch_correct: bool = True,
|
||||
) -> tuple[str, str, SpeedConfig]:
|
||||
"""便捷方法：从单一 speed 值生成视频+音频滤镜。
|
||||
|
||||
返回 (video_filter, audio_filter, config)。
|
||||
"""
|
||||
"""便捷方法：从单一 speed 值生成视频+音频滤镜."""
|
||||
config = SpeedConfig(speed=speed, pitch_correct=pitch_correct)
|
||||
config.clamp()
|
||||
return (
|
||||
@@ -160,8 +70,5 @@ class SpeedEngine:
|
||||
clip_config: dict,
|
||||
global_speed: float = DEFAULT_SPEED,
|
||||
) -> float:
|
||||
"""从 clip config 中解析 playback_speed，0 或缺失则使用全局速度。"""
|
||||
speed = clip_config.get("playback_speed", 0) if clip_config else 0
|
||||
if not isinstance(speed, (int, float)) or speed <= 0:
|
||||
return global_speed
|
||||
return float(speed)
|
||||
"""从 clip config 中解析 playback_speed，0 或缺失则使用全局速度."""
|
||||
return _resolve_clip_speed_base(clip_config, global_speed)
|
||||
|
||||
@@ -0,0 +1,179 @@
|
||||
"""调速配置领域模型 — 纯逻辑，无FFmpeg依赖.
|
||||
|
||||
抽离自 speed_engine.py，包含：
|
||||
- SpeedConfig 数据类（解析/钳制/原速判断）
|
||||
- 视频/音频调速滤镜构建
|
||||
- atempo 多级拆分算法
|
||||
- 时长计算
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import Any
|
||||
|
||||
# ─── 常量 ───────────────────────────────────────────────
|
||||
MIN_SPEED = 0.25
|
||||
MAX_SPEED = 4.0
|
||||
DEFAULT_SPEED = 1.0
|
||||
|
||||
# atempo 单级有效范围
|
||||
_ATEMPO_MIN = 0.5
|
||||
_ATEMPO_MAX = 2.0
|
||||
|
||||
|
||||
@dataclass
|
||||
class SpeedConfig:
|
||||
"""调速配置.
|
||||
|
||||
Attributes:
|
||||
speed: 播放速度，0.25~4.0，1.0 为原速
|
||||
pitch_correct: 是否保持音调（默认 True，用 atempo 时间拉伸算法）
|
||||
"""
|
||||
|
||||
speed: float = DEFAULT_SPEED
|
||||
pitch_correct: bool = True
|
||||
|
||||
@classmethod
|
||||
def parse(cls, data: dict[str, Any] | None) -> SpeedConfig:
|
||||
"""从 dict 解析配置，无效值回退到默认."""
|
||||
if not data or not isinstance(data, dict):
|
||||
return cls()
|
||||
|
||||
speed = data.get("speed", DEFAULT_SPEED)
|
||||
if not isinstance(speed, (int, float)):
|
||||
speed = DEFAULT_SPEED
|
||||
|
||||
pitch_correct = data.get("pitch_correct", True)
|
||||
if not isinstance(pitch_correct, bool):
|
||||
pitch_correct = True
|
||||
|
||||
config = cls(speed=float(speed), pitch_correct=pitch_correct)
|
||||
config.clamp()
|
||||
return config
|
||||
|
||||
def clamp(self) -> None:
|
||||
"""将速度钳制到合法范围."""
|
||||
if self.speed <= 0:
|
||||
self.speed = DEFAULT_SPEED
|
||||
elif self.speed < MIN_SPEED:
|
||||
self.speed = MIN_SPEED
|
||||
elif self.speed > MAX_SPEED:
|
||||
self.speed = MAX_SPEED
|
||||
|
||||
@property
|
||||
def is_original(self) -> bool:
|
||||
"""是否原速（无需调速）."""
|
||||
return abs(self.speed - 1.0) < 1e-6
|
||||
|
||||
@property
|
||||
def is_fast(self) -> bool:
|
||||
"""是否加速播放."""
|
||||
return self.speed > 1.0
|
||||
|
||||
@property
|
||||
def is_slow(self) -> bool:
|
||||
"""是否减速播放."""
|
||||
return self.speed < 1.0
|
||||
|
||||
|
||||
# ── 滤镜构建 ────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def build_video_filter(config: SpeedConfig) -> str:
|
||||
"""生成视频调速滤镜字符串.
|
||||
|
||||
返回 setpts 滤镜表达式，原速时返回空字符串。
|
||||
"""
|
||||
if config.is_original:
|
||||
return ""
|
||||
# setpts=PTS/speed — speed>1 加速，speed<1 减速
|
||||
return f"setpts=PTS/{config.speed:.4f}"
|
||||
|
||||
|
||||
def build_audio_filter(config: SpeedConfig) -> str:
|
||||
"""生成音频调速滤镜字符串.
|
||||
|
||||
atempo 单级范围 0.5~2.0，超出范围时自动多级串联：
|
||||
- 0.25x → atempo=0.5,atempo=0.5
|
||||
- 4x → atempo=2.0,atempo=2.0
|
||||
- 0.3x → atempo=0.5,atempo=0.6
|
||||
- 3x → atempo=2.0,atempo=1.5
|
||||
|
||||
原速时返回空字符串。
|
||||
"""
|
||||
if config.is_original:
|
||||
return ""
|
||||
|
||||
speed = config.speed
|
||||
stages: list[float] = _split_atempo_stages(speed)
|
||||
return ",".join(f"atempo={s:.4f}" for s in stages)
|
||||
|
||||
|
||||
def _split_atempo_stages(speed: float) -> list[float]:
|
||||
"""将速度拆分为多级 atempo 串联，每级都在 [0.5, 2.0] 范围内."""
|
||||
if _ATEMPO_MIN <= speed <= _ATEMPO_MAX:
|
||||
return [speed]
|
||||
|
||||
stages: list[float] = []
|
||||
remaining = speed
|
||||
|
||||
# 加速场景（speed > 2.0）
|
||||
if speed > _ATEMPO_MAX:
|
||||
while remaining > _ATEMPO_MAX:
|
||||
stages.append(_ATEMPO_MAX)
|
||||
remaining /= _ATEMPO_MAX
|
||||
stages.append(remaining)
|
||||
|
||||
# 减速场景（speed < 0.5）
|
||||
else:
|
||||
while remaining < _ATEMPO_MIN:
|
||||
stages.append(_ATEMPO_MIN)
|
||||
remaining /= _ATEMPO_MIN
|
||||
stages.append(remaining)
|
||||
|
||||
return stages
|
||||
|
||||
|
||||
# ── 时长计算 ────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def adjust_duration(original_duration: float, config: SpeedConfig) -> float:
|
||||
"""计算调速后的时长.
|
||||
|
||||
加速 → 时长变短；减速 → 时长变长。
|
||||
"""
|
||||
if config.is_original or original_duration <= 0:
|
||||
return original_duration
|
||||
return original_duration / config.speed
|
||||
|
||||
|
||||
# ── 便捷方法 ────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def build_clip_speed_filter(
|
||||
speed: float,
|
||||
pitch_correct: bool = True,
|
||||
) -> tuple[str, str, SpeedConfig]:
|
||||
"""便捷方法：从单一 speed 值生成视频+音频滤镜.
|
||||
|
||||
返回 (video_filter, audio_filter, config)。
|
||||
"""
|
||||
config = SpeedConfig(speed=speed, pitch_correct=pitch_correct)
|
||||
config.clamp()
|
||||
return (
|
||||
build_video_filter(config),
|
||||
build_audio_filter(config),
|
||||
config,
|
||||
)
|
||||
|
||||
|
||||
def resolve_clip_speed(
|
||||
clip_config: dict[str, Any] | None,
|
||||
global_speed: float = DEFAULT_SPEED,
|
||||
) -> float:
|
||||
"""从 clip config 中解析 playback_speed，0 或缺失则使用全局速度."""
|
||||
speed = clip_config.get("playback_speed", 0) if clip_config else 0
|
||||
if not isinstance(speed, (int, float)) or speed <= 0:
|
||||
return global_speed
|
||||
return float(speed)
|
||||
@@ -1,338 +0,0 @@
|
||||
"""URL 安全校验纯逻辑 — SSRF 防护.
|
||||
|
||||
纯函数模块，无网络/文件 IO，无环境变量依赖，所有配置通过参数传入。
|
||||
供 packages/shared/url_security.py 作为薄包装调用，也可直接用于单测。
|
||||
|
||||
防护要点（纯逻辑部分）：
|
||||
1. Scheme 白名单校验
|
||||
2. 内部主机名拦截（字符串匹配）
|
||||
3. 端口白名单校验
|
||||
4. IP 格式 SSRF 检查（回环/私有/链路本地/组播/未指定/保留）
|
||||
5. 可信域名匹配（支持子域名）
|
||||
6. 文件头魔数校验（接收 bytes）
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import ipaddress
|
||||
from urllib.parse import urlparse
|
||||
|
||||
|
||||
class UrlSecurityError(ValueError):
|
||||
"""URL 安全校验失败."""
|
||||
|
||||
pass
|
||||
|
||||
|
||||
# ── 常量 ────────────────────────────────────────────────────────────────────
|
||||
|
||||
ALLOWED_SCHEMES = frozenset({"http", "https"})
|
||||
ALLOWED_PORTS = frozenset({80, 443})
|
||||
MAX_URL_LENGTH = 2048
|
||||
|
||||
# 已知内部/敏感主机名集合
|
||||
INTERNAL_HOSTNAMES = frozenset(
|
||||
{
|
||||
"localhost",
|
||||
"localhost.localdomain",
|
||||
"ip6-localhost",
|
||||
"ip6-loopback",
|
||||
"metadata",
|
||||
"metadata.google.internal",
|
||||
"169.254.169.254",
|
||||
}
|
||||
)
|
||||
|
||||
# 内网域名后缀
|
||||
INTERNAL_DOMAIN_SUFFIXES = (".local", ".internal", ".localdomain")
|
||||
|
||||
# 文件魔数表 — key: MIME, value: list of 签名组，每组内所有 (offset, bytes) 都匹配才算命中
|
||||
MAGIC_NUMBERS: dict[str, list[list[tuple[int, bytes]]]] = {
|
||||
# 音频
|
||||
"audio/mpeg": [
|
||||
[(0, b"ID3")],
|
||||
[(0, b"\xff\xfb")],
|
||||
[(0, b"\xff\xf3")],
|
||||
[(0, b"\xff\xf2")],
|
||||
[(0, b"\xff\xfa")],
|
||||
[(0, b"\xff\xf9")],
|
||||
],
|
||||
"audio/wav": [[(0, b"RIFF"), (8, b"WAVE")]],
|
||||
"audio/x-wav": [[(0, b"RIFF"), (8, b"WAVE")]],
|
||||
"audio/ogg": [[(0, b"OggS")]],
|
||||
"application/ogg": [[(0, b"OggS")]],
|
||||
"audio/flac": [[(0, b"fLaC")]],
|
||||
"audio/aac": [[(0, b"\xff\xf1")], [(0, b"\xff\xf9")]],
|
||||
"audio/aacp": [[(0, b"\xff\xf1")], [(0, b"\xff\xf9")]],
|
||||
"audio/mp4": [[(4, b"ftyp")]],
|
||||
"audio/x-m4a": [[(4, b"ftyp")]],
|
||||
# 视频
|
||||
"video/mp4": [[(4, b"ftyp")]],
|
||||
"video/quicktime": [[(4, b"ftyp")]],
|
||||
"video/x-matroska": [[(0, b"\x1a\x45\xdf\xa3")]],
|
||||
"video/webm": [[(0, b"\x1a\x45\xdf\xa3")]],
|
||||
"video/x-msvideo": [[(0, b"RIFF"), (8, b"AVI ")]],
|
||||
# 图片
|
||||
"image/jpeg": [[(0, b"\xff\xd8\xff")]],
|
||||
"image/png": [[(0, b"\x89PNG\r\n\x1a\n")]],
|
||||
"image/gif": [[(0, b"GIF87a")], [(0, b"GIF89a")]],
|
||||
"image/webp": [[(0, b"RIFF"), (8, b"WEBP")]],
|
||||
"image/bmp": [[(0, b"BM")]],
|
||||
}
|
||||
|
||||
ALLOWED_AUDIO_MIME_TYPES = frozenset(
|
||||
{
|
||||
"audio/mpeg",
|
||||
"audio/mp3",
|
||||
"audio/wav",
|
||||
"audio/x-wav",
|
||||
"audio/pcm",
|
||||
"audio/ogg",
|
||||
"audio/opus",
|
||||
"audio/flac",
|
||||
"audio/aac",
|
||||
"audio/m4a",
|
||||
"audio/x-m4a",
|
||||
"audio/mp4",
|
||||
"application/octet-stream",
|
||||
}
|
||||
)
|
||||
|
||||
ALLOWED_VIDEO_MIME_TYPES = frozenset(
|
||||
{
|
||||
"video/mp4",
|
||||
"video/quicktime",
|
||||
"video/x-matroska",
|
||||
"video/webm",
|
||||
"video/avi",
|
||||
"video/x-msvideo",
|
||||
"video/mpeg",
|
||||
"application/octet-stream",
|
||||
}
|
||||
)
|
||||
|
||||
ALLOWED_IMAGE_MIME_TYPES = frozenset(
|
||||
{
|
||||
"image/jpeg",
|
||||
"image/png",
|
||||
"image/gif",
|
||||
"image/webp",
|
||||
"image/bmp",
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
# ── 主机名 / 域名校验 ──────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def check_internal_hostname(hostname: str) -> None:
|
||||
"""检查主机名是否为内部/敏感主机名，是则抛出 UrlSecurityError.
|
||||
|
||||
检查内容：
|
||||
- 精确匹配 INTERNAL_HOSTNAMES 集合
|
||||
- 后缀匹配 INTERNAL_DOMAIN_SUFFIXES（.local/.internal/.localdomain）
|
||||
"""
|
||||
hostname_lower = hostname.lower()
|
||||
if hostname_lower in INTERNAL_HOSTNAMES:
|
||||
raise UrlSecurityError(f"禁止访问内部主机名: {hostname}")
|
||||
if hostname_lower.endswith(INTERNAL_DOMAIN_SUFFIXES):
|
||||
raise UrlSecurityError(f"禁止访问内网域名: {hostname}")
|
||||
|
||||
|
||||
def is_trusted_domain(hostname: str, trusted_domains: set[str]) -> bool:
|
||||
"""检查域名是否在可信白名单中（支持子域名匹配）.
|
||||
|
||||
匹配规则：
|
||||
- 精确匹配
|
||||
- 子域名匹配（hostname 以 .domain 结尾）
|
||||
|
||||
Args:
|
||||
hostname: 待检查的主机名
|
||||
trusted_domains: 可信域名集合，为空表示不限制
|
||||
"""
|
||||
if not trusted_domains:
|
||||
return True
|
||||
hostname_lower = hostname.lower()
|
||||
if hostname_lower in {d.lower() for d in trusted_domains}:
|
||||
return True
|
||||
for domain in trusted_domains:
|
||||
if hostname_lower.endswith("." + domain.lower()):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
# ── IP SSRF 检查 ───────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def check_ssrf_ip(ip_str: str) -> None:
|
||||
"""检查 IP 地址是否存在 SSRF 风险，有风险则抛出 UrlSecurityError.
|
||||
|
||||
检查项：回环、私有、链路本地、组播、未指定、保留地址。
|
||||
|
||||
Args:
|
||||
ip_str: IP 地址字符串（IPv4 或 IPv6）
|
||||
|
||||
Raises:
|
||||
UrlSecurityError: IP 属于 SSRF 风险范围
|
||||
ValueError: ip_str 不是合法 IP 地址（调用方应自行捕获处理）
|
||||
"""
|
||||
ip_obj = ipaddress.ip_address(ip_str)
|
||||
if ip_obj.is_loopback:
|
||||
raise UrlSecurityError(f"禁止访问回环地址: {ip_obj}")
|
||||
if ip_obj.is_link_local:
|
||||
raise UrlSecurityError(f"禁止访问链路本地地址: {ip_obj}")
|
||||
if ip_obj.is_unspecified:
|
||||
raise UrlSecurityError(f"禁止访问未指定地址: {ip_obj}")
|
||||
if ip_obj.is_multicast:
|
||||
raise UrlSecurityError(f"禁止访问组播地址: {ip_obj}")
|
||||
if ip_obj.is_reserved:
|
||||
raise UrlSecurityError(f"禁止访问保留地址: {ip_obj}")
|
||||
if ip_obj.is_private:
|
||||
raise UrlSecurityError(f"禁止访问内网地址: {ip_obj}")
|
||||
|
||||
|
||||
def is_ip_address(hostname: str) -> bool:
|
||||
"""判断主机名是否为 IP 地址格式（IPv4 或 IPv6）."""
|
||||
try:
|
||||
ipaddress.ip_address(hostname)
|
||||
return True
|
||||
except ValueError:
|
||||
return False
|
||||
|
||||
|
||||
# ── URL 基础校验 ───────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def validate_url_basic(
|
||||
url: str,
|
||||
*,
|
||||
trusted_domains: set[str] | None = None,
|
||||
allow_direct_ip: bool = False,
|
||||
) -> str:
|
||||
"""URL 基础安全校验（纯逻辑，不含 DNS 解析）.
|
||||
|
||||
校验项：
|
||||
1. URL 非空 & 长度限制
|
||||
2. Scheme 白名单
|
||||
3. 主机名存在性
|
||||
4. 内部主机名拦截
|
||||
5. 端口白名单
|
||||
6. 直接 IP 访问限制
|
||||
7. IP 格式 SSRF 检查（如果 hostname 是 IP）
|
||||
8. 可信域名白名单（如果配置了）
|
||||
|
||||
注意：域名格式的 SSRF 检查需要 DNS 解析，不在本函数范围内。
|
||||
|
||||
Args:
|
||||
url: 待校验 URL
|
||||
trusted_domains: 可信域名白名单，None/空表示不限制
|
||||
allow_direct_ip: 是否允许直接 IP 访问
|
||||
|
||||
Returns:
|
||||
原始 URL（校验通过）
|
||||
|
||||
Raises:
|
||||
UrlSecurityError: 校验失败
|
||||
"""
|
||||
if not url:
|
||||
raise UrlSecurityError("URL 为空")
|
||||
if len(url) > MAX_URL_LENGTH:
|
||||
raise UrlSecurityError(f"URL 过长 ({len(url)} > {MAX_URL_LENGTH})")
|
||||
|
||||
try:
|
||||
parsed = urlparse(url)
|
||||
except Exception as e:
|
||||
raise UrlSecurityError(f"URL 解析失败: {e}") from e
|
||||
|
||||
# Scheme
|
||||
if not parsed.scheme or parsed.scheme.lower() not in ALLOWED_SCHEMES:
|
||||
raise UrlSecurityError(f"不允许的 URL scheme: {parsed.scheme}")
|
||||
|
||||
# Hostname
|
||||
hostname = parsed.hostname
|
||||
if not hostname:
|
||||
raise UrlSecurityError("URL 缺少主机名")
|
||||
|
||||
# 内部主机名前置拦截
|
||||
check_internal_hostname(hostname)
|
||||
|
||||
# 端口
|
||||
port = parsed.port
|
||||
if port is not None and port not in ALLOWED_PORTS:
|
||||
raise UrlSecurityError(f"不允许的端口: {port}")
|
||||
|
||||
# IP 格式检查 & SSRF
|
||||
if is_ip_address(hostname):
|
||||
if not allow_direct_ip:
|
||||
raise UrlSecurityError(f"禁止直接 IP 访问: {hostname}")
|
||||
check_ssrf_ip(hostname)
|
||||
|
||||
# 可信域名白名单
|
||||
if trusted_domains and not is_trusted_domain(hostname, trusted_domains):
|
||||
raise UrlSecurityError(f"域名不在可信白名单中: {hostname}")
|
||||
|
||||
return url
|
||||
|
||||
|
||||
def is_url_basic_safe(
|
||||
url: str,
|
||||
*,
|
||||
trusted_domains: set[str] | None = None,
|
||||
allow_direct_ip: bool = False,
|
||||
) -> bool:
|
||||
"""便捷函数：基础安全检查，不抛异常，返回 bool."""
|
||||
try:
|
||||
validate_url_basic(url, trusted_domains=trusted_domains, allow_direct_ip=allow_direct_ip)
|
||||
return True
|
||||
except UrlSecurityError:
|
||||
return False
|
||||
|
||||
|
||||
# ── 魔数校验 ───────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def validate_magic_number(header_bytes: bytes, allowed_mime_types: set[str]) -> None:
|
||||
"""校验文件头魔数是否与允许的 MIME 类型匹配（纯函数）.
|
||||
|
||||
读取 header_bytes，与 allowed_mime_types 对应格式的魔数逐一比对，
|
||||
任一类型匹配即通过；全部不匹配则抛出 UrlSecurityError。
|
||||
|
||||
Args:
|
||||
header_bytes: 文件头字节（建议至少 256 字节）
|
||||
allowed_mime_types: 允许的 MIME 类型集合
|
||||
|
||||
Raises:
|
||||
UrlSecurityError: 文件魔数与所有允许类型均不匹配
|
||||
"""
|
||||
# 收集所有允许类型对应的魔数签名
|
||||
signatures: list[list[tuple[int, bytes]]] = []
|
||||
for mime in allowed_mime_types:
|
||||
sigs = MAGIC_NUMBERS.get(mime)
|
||||
if sigs:
|
||||
signatures.extend(sigs)
|
||||
|
||||
# 没有已知魔数的 MIME，跳过不阻断
|
||||
if not signatures:
|
||||
return
|
||||
|
||||
if not header_bytes:
|
||||
raise UrlSecurityError("文件为空，无法校验格式")
|
||||
|
||||
# 任一签名匹配即通过
|
||||
for sig in signatures:
|
||||
match = True
|
||||
for offset, expected in sig:
|
||||
if offset + len(expected) > len(header_bytes):
|
||||
match = False
|
||||
break
|
||||
if header_bytes[offset : offset + len(expected)] != expected:
|
||||
match = False
|
||||
break
|
||||
if match:
|
||||
return
|
||||
|
||||
raise UrlSecurityError(
|
||||
f"文件魔数与允许的 MIME 类型不匹配，"
|
||||
f"允许类型: {sorted(allowed_mime_types)}，"
|
||||
f"文件头前16字节: {header_bytes[:16].hex()}"
|
||||
)
|
||||
+362
-87
@@ -1,73 +1,239 @@
|
||||
"""URL 安全校验工具 — SSRF 防护（薄包装层）.
|
||||
"""URL 安全校验工具 — SSRF 防护.
|
||||
|
||||
本文件保留原有对外 API，纯逻辑部分委托给 packages/domain/url_security.py。
|
||||
新增了 DNS 解析、文件下载、环境变量配置等有副作用的逻辑。
|
||||
统一的外部 URL 安全校验方案，覆盖所有渲染管线和 TTS 中的外部下载场景。
|
||||
放在 packages/shared/ 作为单一来源，worker 和 application 层都可引用。
|
||||
|
||||
防护要点：
|
||||
1. Scheme 白名单：仅允许 http/https
|
||||
2. 主机 SSRF 防护：禁止内网 IP、回环地址、链路本地地址、元数据服务
|
||||
3. 端口白名单：仅允许 80/443
|
||||
3. 端口白名单：仅允许 80/443（标准 HTTP/HTTPS）
|
||||
4. 域名校验：禁止 IP 直接访问（除非在白名单中）
|
||||
5. 重定向防护：手动跟随重定向，每次跳转前重新校验
|
||||
5. 重定向防护：手动跟随重定向，每次跳转前重新校验目标 URL
|
||||
6. 文件大小限制：流式下载，超过上限立即中断
|
||||
7. MIME 类型白名单 + 魔数二次校验
|
||||
7. MIME 类型白名单：可选的内容类型校验
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import ipaddress
|
||||
import logging
|
||||
import os
|
||||
import socket
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from urllib.parse import urljoin
|
||||
|
||||
from packages.domain.url_security import (
|
||||
ALLOWED_AUDIO_MIME_TYPES,
|
||||
ALLOWED_IMAGE_MIME_TYPES,
|
||||
ALLOWED_PORTS as _allowed_ports_base,
|
||||
ALLOWED_SCHEMES as _allowed_schemes_base,
|
||||
ALLOWED_VIDEO_MIME_TYPES,
|
||||
MAX_URL_LENGTH,
|
||||
MAGIC_NUMBERS,
|
||||
UrlSecurityError as _UrlSecurityError_base,
|
||||
check_internal_hostname as _check_internal_hostname_base,
|
||||
check_ssrf_ip as _check_ssrf_ip_base,
|
||||
is_ip_address as _is_ip_address_base,
|
||||
is_trusted_domain as _is_trusted_domain_base,
|
||||
validate_magic_number as _validate_magic_number_base,
|
||||
validate_url_basic as _validate_url_basic_base,
|
||||
)
|
||||
from urllib.parse import urljoin, urlparse
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# ── 兼容导出（保持原有变量名供外部引用） ──────────────────────────────────
|
||||
ALLOWED_SCHEMES = set(_allowed_schemes_base)
|
||||
ALLOWED_PORTS = set(_allowed_ports_base)
|
||||
UrlSecurityError = _UrlSecurityError_base
|
||||
# 允许的 URL scheme
|
||||
ALLOWED_SCHEMES = {"http", "https"}
|
||||
|
||||
# 可信域名白名单（从环境变量读取）
|
||||
# 允许的端口（标准 HTTP/HTTPS）
|
||||
ALLOWED_PORTS = {80, 443}
|
||||
|
||||
# 可信域名白名单（可根据实际 OSS/CDN 域名配置）
|
||||
# 从环境变量读取，格式："oss-cn-hangzhou.aliyuncs.com,cdn.example.com"
|
||||
# 默认空表示所有公网域名都允许，但仍会做 SSRF 检查
|
||||
TRUSTED_DOMAINS: set[str] = set()
|
||||
_env_trusted = os.environ.get("URL_SECURITY_TRUSTED_DOMAINS", "")
|
||||
if _env_trusted:
|
||||
TRUSTED_DOMAINS = {d.strip() for d in _env_trusted.split(",") if d.strip()}
|
||||
|
||||
# 是否允许 IP 直接访问
|
||||
# 是否允许 IP 直接访问（默认禁止，防止绕过 DNS 校验）
|
||||
ALLOW_DIRECT_IP = os.environ.get("URL_SECURITY_ALLOW_DIRECT_IP", "false").lower() == "true"
|
||||
|
||||
# 最大 URL 长度
|
||||
MAX_URL_LENGTH = 2048
|
||||
|
||||
# 单次下载最大文件大小（默认 200MB）
|
||||
DEFAULT_MAX_DOWNLOAD_SIZE = int(os.environ.get("URL_SECURITY_MAX_DOWNLOAD_MB", "200")) * 1024 * 1024
|
||||
|
||||
# 允许的音频 MIME 类型白名单
|
||||
ALLOWED_AUDIO_MIME_TYPES = {
|
||||
"audio/mpeg",
|
||||
"audio/mp3",
|
||||
"audio/wav",
|
||||
"audio/x-wav",
|
||||
"audio/pcm",
|
||||
"audio/ogg",
|
||||
"audio/opus",
|
||||
"audio/flac",
|
||||
"audio/aac",
|
||||
"audio/m4a",
|
||||
"audio/x-m4a",
|
||||
"audio/mp4",
|
||||
"application/octet-stream", # 兼容一些 CDN 返回通用类型
|
||||
}
|
||||
|
||||
# 允许的视频 MIME 类型白名单
|
||||
ALLOWED_VIDEO_MIME_TYPES = {
|
||||
"video/mp4",
|
||||
"video/quicktime",
|
||||
"video/x-matroska",
|
||||
"video/webm",
|
||||
"video/avi",
|
||||
"video/x-msvideo",
|
||||
"video/mpeg",
|
||||
"application/octet-stream",
|
||||
}
|
||||
|
||||
# 允许的图片 MIME 类型白名单
|
||||
ALLOWED_IMAGE_MIME_TYPES = {
|
||||
"image/jpeg",
|
||||
"image/png",
|
||||
"image/gif",
|
||||
"image/webp",
|
||||
"image/bmp",
|
||||
}
|
||||
|
||||
# 下载块大小
|
||||
_DOWNLOAD_CHUNK_SIZE = 8192
|
||||
|
||||
# 最大重定向次数
|
||||
_MAX_REDIRECTS = 5
|
||||
|
||||
# 魔数校验最大读取字节数
|
||||
# 文件魔数（文件头签名）表 — 用于 MIME 白名单校验后的二次真实性校验
|
||||
# key: MIME 类型，value: 签名列表，任一签名匹配即通过
|
||||
# 每条签名: list of (offset, bytes)，所有条目都匹配才算该签名命中（支持多处联合匹配如 RIFF+WAVE）
|
||||
_MAGIC_NUMBERS: dict[str, list[list[tuple[int, bytes]]]] = {
|
||||
# ── 音频 ──
|
||||
"audio/mpeg": [
|
||||
[(0, b"ID3")], # ID3v2 标签
|
||||
[(0, b"\xff\xfb")], # MPEG1 Layer3
|
||||
[(0, b"\xff\xf3")], # MPEG2 Layer3
|
||||
[(0, b"\xff\xf2")], # MPEG2.5 Layer3
|
||||
[(0, b"\xff\xfa")], # MPEG1 Layer2
|
||||
[(0, b"\xff\xf9")], # 其他 MPEG ADTS
|
||||
],
|
||||
"audio/wav": [
|
||||
[(0, b"RIFF"), (8, b"WAVE")], # RIFF + WAVE
|
||||
],
|
||||
"audio/x-wav": [
|
||||
[(0, b"RIFF"), (8, b"WAVE")],
|
||||
],
|
||||
"audio/ogg": [
|
||||
[(0, b"OggS")],
|
||||
],
|
||||
"application/ogg": [
|
||||
[(0, b"OggS")],
|
||||
],
|
||||
"audio/flac": [
|
||||
[(0, b"fLaC")],
|
||||
],
|
||||
"audio/aac": [
|
||||
[(0, b"\xff\xf1")], # ADTS MPEG-4
|
||||
[(0, b"\xff\xf9")], # ADTS MPEG-2
|
||||
],
|
||||
"audio/aacp": [
|
||||
[(0, b"\xff\xf1")],
|
||||
[(0, b"\xff\xf9")],
|
||||
],
|
||||
"audio/mp4": [
|
||||
[(4, b"ftyp")], # ISO Base Media (M4A)
|
||||
],
|
||||
"audio/x-m4a": [
|
||||
[(4, b"ftyp")],
|
||||
],
|
||||
# ── 视频 ──
|
||||
"video/mp4": [
|
||||
[(4, b"ftyp")], # ISO Base Media (MP4)
|
||||
],
|
||||
"video/quicktime": [
|
||||
[(4, b"ftyp")],
|
||||
],
|
||||
"video/x-matroska": [
|
||||
[(0, b"\x1a\x45\xdf\xa3")], # EBML header
|
||||
],
|
||||
"video/webm": [
|
||||
[(0, b"\x1a\x45\xdf\xa3")],
|
||||
],
|
||||
"video/x-msvideo": [
|
||||
[(0, b"RIFF"), (8, b"AVI ")],
|
||||
],
|
||||
# ── 图片 ──
|
||||
"image/jpeg": [
|
||||
[(0, b"\xff\xd8\xff")],
|
||||
],
|
||||
"image/png": [
|
||||
[(0, b"\x89PNG\r\n\x1a\n")],
|
||||
],
|
||||
"image/gif": [
|
||||
[(0, b"GIF87a")],
|
||||
[(0, b"GIF89a")],
|
||||
],
|
||||
"image/webp": [
|
||||
[(0, b"RIFF"), (8, b"WEBP")],
|
||||
],
|
||||
"image/bmp": [
|
||||
[(0, b"BM")],
|
||||
],
|
||||
}
|
||||
|
||||
# 魔数校验最大读取字节数（文件头）
|
||||
_MAGIC_CHECK_READ_SIZE = 256
|
||||
|
||||
|
||||
def _validate_magic_number(file_path: str, allowed_mime_types: set[str]) -> None:
|
||||
"""校验文件头魔数是否与允许的 MIME 类型匹配.
|
||||
|
||||
读取文件前 256 字节，与 allowed_mime_types 对应格式的魔数逐一比对，
|
||||
任一类型匹配即通过；全部不匹配则抛出 UrlSecurityError。
|
||||
|
||||
仅当 allowed_mime_types 非空时执行；空文件视为不匹配。
|
||||
|
||||
Args:
|
||||
file_path: 本地文件路径
|
||||
allowed_mime_types: 允许的 MIME 类型集合
|
||||
|
||||
Raises:
|
||||
UrlSecurityError: 文件魔数与所有允许类型均不匹配
|
||||
"""
|
||||
# 收集所有允许类型对应的魔数签名
|
||||
signatures: list[list[tuple[int, bytes]]] = []
|
||||
for mime in allowed_mime_types:
|
||||
sigs = _MAGIC_NUMBERS.get(mime)
|
||||
if sigs:
|
||||
signatures.extend(sigs)
|
||||
|
||||
# 如果没有已知魔数（比如自定义 MIME），跳过校验不阻断
|
||||
if not signatures:
|
||||
return
|
||||
|
||||
try:
|
||||
with open(file_path, "rb") as f:
|
||||
header = f.read(_MAGIC_CHECK_READ_SIZE)
|
||||
except OSError as e:
|
||||
raise UrlSecurityError(f"读取文件头失败: {e}") from e
|
||||
|
||||
if not header:
|
||||
raise UrlSecurityError("文件为空，无法校验格式")
|
||||
|
||||
# 任一签名匹配即通过
|
||||
for sig in signatures:
|
||||
match = True
|
||||
for offset, expected in sig:
|
||||
if offset + len(expected) > len(header):
|
||||
match = False
|
||||
break
|
||||
if header[offset : offset + len(expected)] != expected:
|
||||
match = False
|
||||
break
|
||||
if match:
|
||||
return
|
||||
|
||||
raise UrlSecurityError(
|
||||
f"文件魔数与允许的 MIME 类型不匹配，"
|
||||
f"允许类型: {sorted(allowed_mime_types)}，"
|
||||
f"文件头前16字节: {header[:16].hex()}"
|
||||
)
|
||||
|
||||
|
||||
class UrlSecurityError(ValueError):
|
||||
"""URL 安全校验失败."""
|
||||
|
||||
pass
|
||||
|
||||
|
||||
class NoRedirectHandler(urllib.request.HTTPRedirectHandler):
|
||||
"""禁止自动重定向的 handler，用于手动控制重定向以做安全校验."""
|
||||
|
||||
@@ -75,7 +241,124 @@ class NoRedirectHandler(urllib.request.HTTPRedirectHandler):
|
||||
return None
|
||||
|
||||
|
||||
# ── DNS 解析 SSRF 检查（有副作用） ─────────────────────────────────────────
|
||||
def validate_url_safety(url: str, *, purpose: str = "download") -> str:
|
||||
"""校验 URL 安全性，返回标准化后的 URL（供下游使用）.
|
||||
|
||||
Args:
|
||||
url: 待校验的 URL
|
||||
purpose: 用途描述（用于日志），如 "bgm_download"、"tts_download"
|
||||
|
||||
Returns:
|
||||
标准化后的 URL
|
||||
|
||||
Raises:
|
||||
UrlSecurityError: URL 不安全
|
||||
"""
|
||||
if not url:
|
||||
raise UrlSecurityError("URL 为空")
|
||||
|
||||
if len(url) > MAX_URL_LENGTH:
|
||||
raise UrlSecurityError(f"URL 过长 ({len(url)} > {MAX_URL_LENGTH})")
|
||||
|
||||
# 解析 URL
|
||||
try:
|
||||
parsed = urlparse(url)
|
||||
except Exception as e:
|
||||
raise UrlSecurityError(f"URL 解析失败: {e}") from e
|
||||
|
||||
# 1. Scheme 校验
|
||||
if not parsed.scheme or parsed.scheme.lower() not in ALLOWED_SCHEMES:
|
||||
raise UrlSecurityError(f"不允许的 URL scheme: {parsed.scheme}")
|
||||
|
||||
# 2. 主机名校验
|
||||
hostname = parsed.hostname
|
||||
if not hostname:
|
||||
raise UrlSecurityError("URL 缺少主机名")
|
||||
|
||||
# 2.1 常见内网主机名前置拦截（防止 DNS rebinding 绕过）
|
||||
_check_internal_hostnames(hostname)
|
||||
|
||||
# 3. 端口校验
|
||||
port = parsed.port
|
||||
if port is not None and port not in ALLOWED_PORTS:
|
||||
raise UrlSecurityError(f"不允许的端口: {port}")
|
||||
|
||||
# 4. SSRF 防护 - 解析 IP 并检查
|
||||
try:
|
||||
# 先判断是否是 IP 地址
|
||||
ip_obj = None
|
||||
try:
|
||||
ip_obj = ipaddress.ip_address(hostname)
|
||||
except ValueError:
|
||||
pass # 不是 IP，继续走域名解析
|
||||
|
||||
if ip_obj is not None:
|
||||
# 是直接 IP 访问
|
||||
if not ALLOW_DIRECT_IP and not _is_trusted_ip(ip_obj):
|
||||
raise UrlSecurityError(f"禁止直接 IP 访问: {hostname}")
|
||||
_check_ssrf_ip(ip_obj)
|
||||
else:
|
||||
# 域名 — 解析 DNS 检查 SSRF
|
||||
_check_ssrf_domain(hostname)
|
||||
except UrlSecurityError:
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.warning("URL 安全校验异常: url=%s purpose=%s error=%s", url[:80], purpose, e)
|
||||
raise UrlSecurityError(f"URL 安全校验异常: {e}") from e
|
||||
|
||||
# 5. 可信域名校验（如果配置了白名单）
|
||||
if TRUSTED_DOMAINS and not _is_trusted_domain(hostname):
|
||||
raise UrlSecurityError(f"域名不在可信白名单中: {hostname}")
|
||||
|
||||
logger.debug("URL 安全校验通过: url=%s purpose=%s", url[:80], purpose)
|
||||
return url
|
||||
|
||||
|
||||
def _check_internal_hostnames(hostname: str) -> None:
|
||||
"""前置检查常见内网/敏感主机名，防止 DNS 解析层绕过."""
|
||||
hostname_lower = hostname.lower()
|
||||
internal_hostnames = {
|
||||
"localhost",
|
||||
"localhost.localdomain",
|
||||
"ip6-localhost",
|
||||
"ip6-loopback",
|
||||
"metadata",
|
||||
"metadata.google.internal",
|
||||
"169.254.169.254", # 云元数据服务
|
||||
}
|
||||
if hostname_lower in internal_hostnames:
|
||||
raise UrlSecurityError(f"禁止访问内部主机名: {hostname}")
|
||||
|
||||
# 检查以 .local / .internal 结尾的主机名
|
||||
if hostname_lower.endswith((".local", ".internal", ".localdomain")):
|
||||
raise UrlSecurityError(f"禁止访问内网域名: {hostname}")
|
||||
|
||||
|
||||
def _check_ssrf_ip(ip_obj: ipaddress.IPv4Address | ipaddress.IPv6Address) -> None:
|
||||
"""检查 IP 是否属于 SSRF 风险范围."""
|
||||
# 回环地址
|
||||
if ip_obj.is_loopback:
|
||||
raise UrlSecurityError(f"禁止访问回环地址: {ip_obj}")
|
||||
|
||||
# 私有地址（内网）
|
||||
if ip_obj.is_private:
|
||||
raise UrlSecurityError(f"禁止访问内网地址: {ip_obj}")
|
||||
|
||||
# 链路本地地址
|
||||
if ip_obj.is_link_local:
|
||||
raise UrlSecurityError(f"禁止访问链路本地地址: {ip_obj}")
|
||||
|
||||
# 组播地址
|
||||
if ip_obj.is_multicast:
|
||||
raise UrlSecurityError(f"禁止访问组播地址: {ip_obj}")
|
||||
|
||||
# 未指定地址（0.0.0.0 / ::）
|
||||
if ip_obj.is_unspecified:
|
||||
raise UrlSecurityError(f"禁止访问未指定地址: {ip_obj}")
|
||||
|
||||
# 保留地址
|
||||
if ip_obj.is_reserved:
|
||||
raise UrlSecurityError(f"禁止访问保留地址: {ip_obj}")
|
||||
|
||||
|
||||
def _check_ssrf_domain(hostname: str) -> None:
|
||||
@@ -84,6 +367,7 @@ def _check_ssrf_domain(hostname: str) -> None:
|
||||
注意：这不能完全防止 DNS rebinding，但能防御大部分 SSRF 场景。
|
||||
"""
|
||||
try:
|
||||
# 解析所有地址
|
||||
infos = socket.getaddrinfo(hostname, None)
|
||||
if not infos:
|
||||
raise UrlSecurityError(f"域名解析失败: {hostname}")
|
||||
@@ -91,63 +375,30 @@ def _check_ssrf_domain(hostname: str) -> None:
|
||||
for info in infos:
|
||||
ip_str = info[4][0]
|
||||
try:
|
||||
_check_ssrf_ip_base(ip_str)
|
||||
ip_obj = ipaddress.ip_address(ip_str)
|
||||
_check_ssrf_ip(ip_obj)
|
||||
except ValueError:
|
||||
# 无法解析为 IP，跳过（不应该发生）
|
||||
continue
|
||||
except socket.gaierror as e:
|
||||
raise UrlSecurityError(f"域名解析失败: {hostname} ({e})") from e
|
||||
|
||||
|
||||
def _validate_magic_number(file_path: str, allowed_mime_types: set[str]) -> None:
|
||||
"""校验文件头魔数（从文件读取后委托给 domain 纯逻辑）."""
|
||||
try:
|
||||
with open(file_path, "rb") as f:
|
||||
header = f.read(_MAGIC_CHECK_READ_SIZE)
|
||||
except OSError as e:
|
||||
raise UrlSecurityError(f"读取文件头失败: {e}") from e
|
||||
|
||||
_validate_magic_number_base(header, allowed_mime_types)
|
||||
def _is_trusted_ip(ip_obj: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool:
|
||||
"""检查 IP 是否在可信列表中（目前通过环境变量配置域名，IP 级信任暂不开放）."""
|
||||
return False
|
||||
|
||||
|
||||
# ── 对外 API ────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def validate_url_safety(url: str, *, purpose: str = "download") -> str:
|
||||
"""校验 URL 安全性，返回标准化后的 URL（含 DNS 解析 SSRF 检查）.
|
||||
|
||||
Args:
|
||||
url: 待校验的 URL
|
||||
purpose: 用途描述（用于日志）
|
||||
|
||||
Returns:
|
||||
标准化后的 URL
|
||||
|
||||
Raises:
|
||||
UrlSecurityError: URL 不安全
|
||||
"""
|
||||
# 基础校验（纯逻辑，不含 DNS）
|
||||
_validate_url_basic_base(
|
||||
url,
|
||||
trusted_domains=TRUSTED_DOMAINS,
|
||||
allow_direct_ip=ALLOW_DIRECT_IP,
|
||||
)
|
||||
|
||||
# 如果 hostname 是域名（不是 IP），做 DNS 解析 SSRF 检查
|
||||
from urllib.parse import urlparse
|
||||
|
||||
parsed = urlparse(url)
|
||||
hostname = parsed.hostname
|
||||
if hostname and not _is_ip_address_base(hostname):
|
||||
try:
|
||||
_check_ssrf_domain(hostname)
|
||||
except UrlSecurityError:
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.warning("URL 安全校验异常: url=%s purpose=%s error=%s", url[:80], purpose, e)
|
||||
raise UrlSecurityError(f"URL 安全校验异常: {e}") from e
|
||||
|
||||
logger.debug("URL 安全校验通过: url=%s purpose=%s", url[:80], purpose)
|
||||
return url
|
||||
def _is_trusted_domain(hostname: str) -> bool:
|
||||
"""检查域名是否在可信白名单中（支持子域名匹配）."""
|
||||
hostname_lower = hostname.lower()
|
||||
if hostname_lower in TRUSTED_DOMAINS:
|
||||
return True
|
||||
# 检查子域名
|
||||
for domain in TRUSTED_DOMAINS:
|
||||
if hostname_lower.endswith("." + domain.lower()):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def is_url_safe(url: str, *, purpose: str = "download") -> bool:
|
||||
@@ -159,7 +410,7 @@ def is_url_safe(url: str, *, purpose: str = "download") -> bool:
|
||||
return False
|
||||
|
||||
|
||||
# ── 安全下载 ────────────────────────────────────────────────────────────────
|
||||
# ── 安全下载 ───────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def safe_download_file(
|
||||
@@ -175,20 +426,34 @@ def safe_download_file(
|
||||
|
||||
包含防护：
|
||||
- SSRF 校验（初始 URL + 每次重定向后都校验）
|
||||
- 重定向次数限制 + 手动跟随
|
||||
- 文件大小限制（流式读取）
|
||||
- MIME 类型白名单 + 魔数二次校验
|
||||
- 重定向次数限制 + 手动跟随（避免重定向绕过 SSRF）
|
||||
- 文件大小限制（流式读取，超过立即中断）
|
||||
- MIME 类型白名单（可选）
|
||||
- 文件头魔数校验（配合 MIME 白名单做二次真实性校验）
|
||||
|
||||
Args:
|
||||
url: 下载 URL
|
||||
dest_path: 目标文件路径
|
||||
purpose: 用途描述（日志用）
|
||||
max_size: 最大下载字节数，超过则中断并抛出 UrlSecurityError
|
||||
allowed_mime_types: 允许的 Content-Type 集合，None 表示不校验
|
||||
timeout: 单次请求超时（秒）
|
||||
|
||||
Returns:
|
||||
实际下载的字节数
|
||||
|
||||
Raises:
|
||||
UrlSecurityError: 安全校验失败
|
||||
"""
|
||||
current_url = url
|
||||
redirect_count = 0
|
||||
total_bytes = 0
|
||||
|
||||
# 使用不自动跟随重定向的 opener
|
||||
no_redirect_opener = urllib.request.build_opener(NoRedirectHandler())
|
||||
|
||||
while True:
|
||||
# 每次请求前都做 SSRF 校验（重定向目标也会校验）
|
||||
validate_url_safety(current_url, purpose=purpose)
|
||||
|
||||
req = urllib.request.Request(current_url, method="GET")
|
||||
@@ -197,6 +462,7 @@ def safe_download_file(
|
||||
try:
|
||||
resp = no_redirect_opener.open(req, timeout=timeout) # nosec B310
|
||||
except urllib.error.HTTPError as e:
|
||||
# 3xx 重定向
|
||||
if 300 <= e.code < 400 and e.headers.get("Location"):
|
||||
if redirect_count >= _MAX_REDIRECTS:
|
||||
raise UrlSecurityError(f"重定向次数超过限制 ({_MAX_REDIRECTS})") from e
|
||||
@@ -208,15 +474,20 @@ def safe_download_file(
|
||||
raise UrlSecurityError(f"URL 错误: {e.reason}") from e
|
||||
|
||||
try:
|
||||
# Content-Type 校验
|
||||
if allowed_mime_types is not None:
|
||||
content_type = resp.headers.get("Content-Type", "").split(";")[0].strip().lower()
|
||||
if content_type and content_type not in allowed_mime_types:
|
||||
raise UrlSecurityError(f"不允许的 Content-Type: {content_type}, 允许: {sorted(allowed_mime_types)}")
|
||||
raise UrlSecurityError(
|
||||
f"不允许的 Content-Type: {content_type}, " f"允许: {sorted(allowed_mime_types)}"
|
||||
)
|
||||
|
||||
# Content-Length 预检
|
||||
content_length = resp.headers.get("Content-Length")
|
||||
if content_length and int(content_length) > max_size:
|
||||
raise UrlSecurityError(f"文件过大: {content_length} bytes > {max_size} bytes 上限")
|
||||
|
||||
# 流式下载，实时检查大小
|
||||
with open(dest_path, "wb") as f:
|
||||
while True:
|
||||
chunk = resp.read(_DOWNLOAD_CHUNK_SIZE)
|
||||
@@ -227,6 +498,7 @@ def safe_download_file(
|
||||
raise UrlSecurityError(f"下载超过大小限制: {total_bytes} bytes > {max_size} bytes")
|
||||
f.write(chunk)
|
||||
|
||||
# 文件头魔数校验（MIME 白名单基础上的二次真实性校验）
|
||||
if allowed_mime_types is not None:
|
||||
_validate_magic_number(dest_path, allowed_mime_types)
|
||||
|
||||
@@ -243,7 +515,10 @@ def safe_download_bytes(
|
||||
allowed_mime_types: set[str] | None = None,
|
||||
timeout: float = 60.0,
|
||||
) -> bytes:
|
||||
"""安全下载 URL 并返回字节内容（适合小文件）。"""
|
||||
"""安全下载 URL 并返回字节内容。
|
||||
|
||||
防护同 safe_download_file，但结果返回在内存中（适合小文件）。
|
||||
"""
|
||||
import tempfile
|
||||
|
||||
fd, tmp_path = tempfile.mkstemp()
|
||||
|
||||
@@ -0,0 +1,313 @@
|
||||
"""speed_config 领域模型单测."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from packages.domain.speed_config import (
|
||||
DEFAULT_SPEED,
|
||||
MAX_SPEED,
|
||||
MIN_SPEED,
|
||||
SpeedConfig,
|
||||
adjust_duration,
|
||||
build_audio_filter,
|
||||
build_video_filter,
|
||||
build_clip_speed_filter,
|
||||
resolve_clip_speed,
|
||||
)
|
||||
|
||||
# ── 常量测试 ────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestConstants:
|
||||
def test_min_speed(self):
|
||||
assert MIN_SPEED == 0.25
|
||||
|
||||
def test_max_speed(self):
|
||||
assert MAX_SPEED == 4.0
|
||||
|
||||
def test_default_speed(self):
|
||||
assert DEFAULT_SPEED == 1.0
|
||||
|
||||
|
||||
# ── SpeedConfig.parse 测试 ────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestSpeedConfigParse:
|
||||
def test_none_returns_default(self):
|
||||
cfg = SpeedConfig.parse(None)
|
||||
assert cfg.speed == DEFAULT_SPEED
|
||||
assert cfg.pitch_correct is True
|
||||
|
||||
def test_empty_dict_returns_default(self):
|
||||
cfg = SpeedConfig.parse({})
|
||||
assert cfg.speed == DEFAULT_SPEED
|
||||
|
||||
def test_invalid_type_returns_default(self):
|
||||
cfg = SpeedConfig.parse("not_a_dict")
|
||||
assert cfg.speed == DEFAULT_SPEED
|
||||
|
||||
def test_valid_speed(self):
|
||||
cfg = SpeedConfig.parse({"speed": 2.0})
|
||||
assert cfg.speed == 2.0
|
||||
|
||||
def test_speed_clamped_low(self):
|
||||
cfg = SpeedConfig.parse({"speed": 0.1})
|
||||
assert cfg.speed == MIN_SPEED
|
||||
|
||||
def test_speed_clamped_high(self):
|
||||
cfg = SpeedConfig.parse({"speed": 5.0})
|
||||
assert cfg.speed == MAX_SPEED
|
||||
|
||||
def test_zero_speed_returns_default(self):
|
||||
cfg = SpeedConfig.parse({"speed": 0})
|
||||
assert cfg.speed == DEFAULT_SPEED
|
||||
|
||||
def test_negative_speed_returns_default(self):
|
||||
cfg = SpeedConfig.parse({"speed": -1.0})
|
||||
assert cfg.speed == DEFAULT_SPEED
|
||||
|
||||
def test_pitch_correct_false(self):
|
||||
cfg = SpeedConfig.parse({"pitch_correct": False})
|
||||
assert cfg.pitch_correct is False
|
||||
|
||||
def test_pitch_correct_invalid_type_defaults_true(self):
|
||||
cfg = SpeedConfig.parse({"pitch_correct": "yes"})
|
||||
assert cfg.pitch_correct is True
|
||||
|
||||
def test_string_speed_invalid_uses_default(self):
|
||||
cfg = SpeedConfig.parse({"speed": "fast"})
|
||||
assert cfg.speed == DEFAULT_SPEED
|
||||
|
||||
|
||||
# ── SpeedConfig.clamp 测试 ────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestClamp:
|
||||
def test_already_valid_unchanged(self):
|
||||
cfg = SpeedConfig(speed=1.5)
|
||||
cfg.clamp()
|
||||
assert cfg.speed == 1.5
|
||||
|
||||
def test_below_min_clamped(self):
|
||||
cfg = SpeedConfig(speed=0.1)
|
||||
cfg.clamp()
|
||||
assert cfg.speed == MIN_SPEED
|
||||
|
||||
def test_above_max_clamped(self):
|
||||
cfg = SpeedConfig(speed=10.0)
|
||||
cfg.clamp()
|
||||
assert cfg.speed == MAX_SPEED
|
||||
|
||||
def test_zero_defaults(self):
|
||||
cfg = SpeedConfig(speed=0.0)
|
||||
cfg.clamp()
|
||||
assert cfg.speed == DEFAULT_SPEED
|
||||
|
||||
def test_negative_defaults(self):
|
||||
cfg = SpeedConfig(speed=-2.0)
|
||||
cfg.clamp()
|
||||
assert cfg.speed == DEFAULT_SPEED
|
||||
|
||||
def test_exact_min_stays(self):
|
||||
cfg = SpeedConfig(speed=MIN_SPEED)
|
||||
cfg.clamp()
|
||||
assert cfg.speed == MIN_SPEED
|
||||
|
||||
def test_exact_max_stays(self):
|
||||
cfg = SpeedConfig(speed=MAX_SPEED)
|
||||
cfg.clamp()
|
||||
assert cfg.speed == MAX_SPEED
|
||||
|
||||
|
||||
# ── is_original / is_fast / is_slow 测试 ─────────────────────────────────
|
||||
|
||||
|
||||
class TestSpeedProperties:
|
||||
def test_is_original_true(self):
|
||||
cfg = SpeedConfig(speed=1.0)
|
||||
assert cfg.is_original is True
|
||||
|
||||
def test_is_original_false_fast(self):
|
||||
cfg = SpeedConfig(speed=2.0)
|
||||
assert cfg.is_original is False
|
||||
|
||||
def test_is_original_false_slow(self):
|
||||
cfg = SpeedConfig(speed=0.5)
|
||||
assert cfg.is_original is False
|
||||
|
||||
def test_is_original_near_one(self):
|
||||
cfg = SpeedConfig(speed=1.0000001)
|
||||
assert cfg.is_original is True
|
||||
|
||||
def test_is_fast_true(self):
|
||||
cfg = SpeedConfig(speed=2.0)
|
||||
assert cfg.is_fast is True
|
||||
|
||||
def test_is_fast_false(self):
|
||||
cfg = SpeedConfig(speed=0.5)
|
||||
assert cfg.is_fast is False
|
||||
|
||||
def test_is_false_for_original(self):
|
||||
cfg = SpeedConfig(speed=1.0)
|
||||
assert cfg.is_fast is False
|
||||
assert cfg.is_slow is False
|
||||
|
||||
def test_is_slow_true(self):
|
||||
cfg = SpeedConfig(speed=0.5)
|
||||
assert cfg.is_slow is True
|
||||
|
||||
def test_is_slow_false(self):
|
||||
cfg = SpeedConfig(speed=2.0)
|
||||
assert cfg.is_slow is False
|
||||
|
||||
|
||||
# ── build_video_filter 测试 ───────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestBuildVideoFilter:
|
||||
def test_original_speed_empty(self):
|
||||
cfg = SpeedConfig(speed=1.0)
|
||||
assert build_video_filter(cfg) == ""
|
||||
|
||||
def test_fast_speed_setpts(self):
|
||||
cfg = SpeedConfig(speed=2.0)
|
||||
result = build_video_filter(cfg)
|
||||
assert "setpts=PTS/2.0000" in result
|
||||
|
||||
def test_slow_speed_setpts(self):
|
||||
cfg = SpeedConfig(speed=0.5)
|
||||
result = build_video_filter(cfg)
|
||||
assert "setpts=PTS/0.5000" in result
|
||||
|
||||
def test_format_four_decimals(self):
|
||||
cfg = SpeedConfig(speed=1.5)
|
||||
result = build_video_filter(cfg)
|
||||
assert "1.5000" in result
|
||||
|
||||
|
||||
# ── build_audio_filter 测试 ───────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestBuildAudioFilter:
|
||||
def test_original_speed_empty(self):
|
||||
cfg = SpeedConfig(speed=1.0)
|
||||
assert build_audio_filter(cfg) == ""
|
||||
|
||||
def test_single_stage_within_range(self):
|
||||
cfg = SpeedConfig(speed=1.5)
|
||||
result = build_audio_filter(cfg)
|
||||
assert result == "atempo=1.5000"
|
||||
assert result.count("atempo") == 1
|
||||
|
||||
def test_fast_two_stages(self):
|
||||
cfg = SpeedConfig(speed=3.0)
|
||||
result = build_audio_filter(cfg)
|
||||
assert result.count("atempo") == 2
|
||||
# 2.0 * 1.5 = 3.0
|
||||
assert "atempo=2.0000" in result
|
||||
assert "atempo=1.5000" in result
|
||||
|
||||
def test_max_speed_two_stages(self):
|
||||
cfg = SpeedConfig(speed=4.0)
|
||||
result = build_audio_filter(cfg)
|
||||
assert result.count("atempo") == 2
|
||||
# 2.0 * 2.0 = 4.0
|
||||
assert result == "atempo=2.0000,atempo=2.0000"
|
||||
|
||||
def test_slow_two_stages(self):
|
||||
cfg = SpeedConfig(speed=0.25)
|
||||
result = build_audio_filter(cfg)
|
||||
assert result.count("atempo") == 2
|
||||
# 0.5 * 0.5 = 0.25
|
||||
assert result == "atempo=0.5000,atempo=0.5000"
|
||||
|
||||
def test_slow_single_stage(self):
|
||||
cfg = SpeedConfig(speed=0.8)
|
||||
result = build_audio_filter(cfg)
|
||||
assert result == "atempo=0.8000"
|
||||
assert result.count("atempo") == 1
|
||||
|
||||
def test_exactly_two_point_zero_single(self):
|
||||
cfg = SpeedConfig(speed=2.0)
|
||||
result = build_audio_filter(cfg)
|
||||
assert result.count("atempo") == 1
|
||||
assert "atempo=2.0000" in result
|
||||
|
||||
def test_exactly_half_single(self):
|
||||
cfg = SpeedConfig(speed=0.5)
|
||||
result = build_audio_filter(cfg)
|
||||
assert result.count("atempo") == 1
|
||||
assert "atempo=0.5000" in result
|
||||
|
||||
|
||||
# ── adjust_duration 测试 ──────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestAdjustDuration:
|
||||
def test_original_speed_unchanged(self):
|
||||
cfg = SpeedConfig(speed=1.0)
|
||||
assert adjust_duration(10.0, cfg) == 10.0
|
||||
|
||||
def test_double_speed_halved(self):
|
||||
cfg = SpeedConfig(speed=2.0)
|
||||
assert adjust_duration(10.0, cfg) == 5.0
|
||||
|
||||
def test_half_speed_doubled(self):
|
||||
cfg = SpeedConfig(speed=0.5)
|
||||
assert adjust_duration(10.0, cfg) == 20.0
|
||||
|
||||
def test_zero_duration_unchanged(self):
|
||||
cfg = SpeedConfig(speed=2.0)
|
||||
assert adjust_duration(0.0, cfg) == 0.0
|
||||
|
||||
def test_negative_duration_unchanged(self):
|
||||
cfg = SpeedConfig(speed=2.0)
|
||||
assert adjust_duration(-5.0, cfg) == -5.0
|
||||
|
||||
|
||||
# ── build_clip_speed_filter 测试 ─────────────────────────────────────────
|
||||
|
||||
|
||||
class TestBuildClipSpeedFilter:
|
||||
def test_normal_speed(self):
|
||||
vf, af, cfg = build_clip_speed_filter(2.0)
|
||||
assert vf == "setpts=PTS/2.0000"
|
||||
assert "atempo=2.0000" in af
|
||||
assert cfg.speed == 2.0
|
||||
|
||||
def test_clamped_speed(self):
|
||||
vf, af, cfg = build_clip_speed_filter(10.0)
|
||||
assert cfg.speed == MAX_SPEED
|
||||
|
||||
def test_pitch_correct_param(self):
|
||||
vf, af, cfg = build_clip_speed_filter(1.5, pitch_correct=False)
|
||||
assert cfg.pitch_correct is False
|
||||
|
||||
def test_original_speed_empty_filters(self):
|
||||
vf, af, cfg = build_clip_speed_filter(1.0)
|
||||
assert vf == ""
|
||||
assert af == ""
|
||||
|
||||
|
||||
# ── resolve_clip_speed 测试 ──────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestResolveClipSpeed:
|
||||
def test_none_config_uses_global(self):
|
||||
assert resolve_clip_speed(None, 1.5) == 1.5
|
||||
|
||||
def test_no_playback_speed_uses_global(self):
|
||||
assert resolve_clip_speed({}, 1.5) == 1.5
|
||||
|
||||
def test_zero_speed_uses_global(self):
|
||||
assert resolve_clip_speed({"playback_speed": 0}, 1.5) == 1.5
|
||||
|
||||
def test_valid_speed_returns_speed(self):
|
||||
assert resolve_clip_speed({"playback_speed": 2.0}, 1.0) == 2.0
|
||||
|
||||
def test_invalid_type_uses_global(self):
|
||||
assert resolve_clip_speed({"playback_speed": "fast"}, 1.0) == 1.0
|
||||
|
||||
def test_default_global_speed(self):
|
||||
assert resolve_clip_speed({}) == DEFAULT_SPEED
|
||||
@@ -1,423 +0,0 @@
|
||||
"""URL 安全校验纯逻辑单元测试 — wave128."""
|
||||
|
||||
import pytest
|
||||
|
||||
from packages.domain.url_security import (
|
||||
ALLOWED_PORTS,
|
||||
ALLOWED_SCHEMES,
|
||||
MAGIC_NUMBERS,
|
||||
MAX_URL_LENGTH,
|
||||
UrlSecurityError,
|
||||
check_internal_hostname,
|
||||
check_ssrf_ip,
|
||||
is_ip_address,
|
||||
is_trusted_domain,
|
||||
is_url_basic_safe,
|
||||
validate_magic_number,
|
||||
validate_url_basic,
|
||||
)
|
||||
|
||||
# ── 常量校验 ────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestConstants:
|
||||
def test_allowed_schemes(self):
|
||||
assert "http" in ALLOWED_SCHEMES
|
||||
assert "https" in ALLOWED_SCHEMES
|
||||
|
||||
def test_allowed_ports(self):
|
||||
assert 80 in ALLOWED_PORTS
|
||||
assert 443 in ALLOWED_PORTS
|
||||
|
||||
def test_max_url_length(self):
|
||||
assert MAX_URL_LENGTH == 2048
|
||||
|
||||
def test_magic_numbers_has_common_formats(self):
|
||||
assert "audio/mpeg" in MAGIC_NUMBERS
|
||||
assert "image/png" in MAGIC_NUMBERS
|
||||
assert "video/mp4" in MAGIC_NUMBERS
|
||||
|
||||
|
||||
# ── 内部主机名检查 ──────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestCheckInternalHostname:
|
||||
@pytest.mark.parametrize(
|
||||
"hostname",
|
||||
[
|
||||
"localhost",
|
||||
"LOCALHOST",
|
||||
"LocalHost",
|
||||
"localhost.localdomain",
|
||||
"ip6-localhost",
|
||||
"ip6-loopback",
|
||||
"metadata",
|
||||
"metadata.google.internal",
|
||||
"169.254.169.254",
|
||||
],
|
||||
)
|
||||
def test_internal_hostnames_rejected(self, hostname):
|
||||
with pytest.raises(UrlSecurityError, match="禁止访问内部主机名"):
|
||||
check_internal_hostname(hostname)
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"hostname",
|
||||
[
|
||||
"foo.local",
|
||||
"bar.internal",
|
||||
"baz.localdomain",
|
||||
"sub.foo.local",
|
||||
],
|
||||
)
|
||||
def test_internal_domain_suffixes_rejected(self, hostname):
|
||||
with pytest.raises(UrlSecurityError, match="禁止访问内网域名"):
|
||||
check_internal_hostname(hostname)
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"hostname",
|
||||
[
|
||||
"example.com",
|
||||
"www.google.com",
|
||||
"oss-cn-hangzhou.aliyuncs.com",
|
||||
"123.45.67.89",
|
||||
],
|
||||
)
|
||||
def test_normal_hostnames_allowed(self, hostname):
|
||||
check_internal_hostname("example.com") # 不抛异常即通过
|
||||
|
||||
|
||||
# ── 可信域名匹配 ────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestIsTrustedDomain:
|
||||
def test_empty_trusted_always_true(self):
|
||||
assert is_trusted_domain("anything.com", set()) is True
|
||||
|
||||
def test_exact_match(self):
|
||||
trusted = {"example.com", "foo.bar"}
|
||||
assert is_trusted_domain("example.com", trusted) is True
|
||||
assert is_trusted_domain("foo.bar", trusted) is True
|
||||
|
||||
def test_exact_no_match(self):
|
||||
trusted = {"example.com"}
|
||||
assert is_trusted_domain("other.com", trusted) is False
|
||||
|
||||
def test_subdomain_match(self):
|
||||
trusted = {"example.com"}
|
||||
assert is_trusted_domain("sub.example.com", trusted) is True
|
||||
assert is_trusted_domain("a.b.example.com", trusted) is True
|
||||
|
||||
def test_subdomain_partial_no_match(self):
|
||||
trusted = {"example.com"}
|
||||
# fakeexample.com 不是 example.com 的子域名
|
||||
assert is_trusted_domain("fakeexample.com", trusted) is False
|
||||
|
||||
def test_case_insensitive(self):
|
||||
trusted = {"Example.COM"}
|
||||
assert is_trusted_domain("example.com", trusted) is True
|
||||
assert is_trusted_domain("SUB.Example.COM", trusted) is True
|
||||
|
||||
|
||||
# ── IP SSRF 检查 ────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestCheckSrfIp:
|
||||
@pytest.mark.parametrize("ip", ["127.0.0.1", "127.1.2.3", "::1"])
|
||||
def test_loopback_rejected(self, ip):
|
||||
with pytest.raises(UrlSecurityError, match="回环"):
|
||||
check_ssrf_ip(ip)
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"ip",
|
||||
[
|
||||
"10.0.0.1",
|
||||
"10.255.255.255",
|
||||
"172.16.0.1",
|
||||
"172.31.255.255",
|
||||
"192.168.1.1",
|
||||
"192.168.0.1",
|
||||
"fd00::1", # IPv6 unique local
|
||||
],
|
||||
)
|
||||
def test_private_rejected(self, ip):
|
||||
with pytest.raises(UrlSecurityError, match="内网"):
|
||||
check_ssrf_ip(ip)
|
||||
|
||||
@pytest.mark.parametrize("ip", ["169.254.1.1", "169.254.169.254", "fe80::1"])
|
||||
def test_link_local_rejected(self, ip):
|
||||
with pytest.raises(UrlSecurityError, match="链路本地"):
|
||||
check_ssrf_ip(ip)
|
||||
|
||||
@pytest.mark.parametrize("ip", ["224.0.0.1", "239.255.255.255", "ff00::1"])
|
||||
def test_multicast_rejected(self, ip):
|
||||
with pytest.raises(UrlSecurityError, match="组播"):
|
||||
check_ssrf_ip(ip)
|
||||
|
||||
@pytest.mark.parametrize("ip", ["0.0.0.0", "::"])
|
||||
def test_unspecified_rejected(self, ip):
|
||||
with pytest.raises(UrlSecurityError, match="未指定"):
|
||||
check_ssrf_ip(ip)
|
||||
|
||||
def test_reserved_rejected(self):
|
||||
with pytest.raises(UrlSecurityError):
|
||||
check_ssrf_ip("240.0.0.1") # 保留地址段
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"ip",
|
||||
[
|
||||
"8.8.8.8",
|
||||
"1.1.1.1",
|
||||
"223.5.5.5",
|
||||
"2001:4860:4860::8888",
|
||||
],
|
||||
)
|
||||
def test_public_ip_allowed(self, ip):
|
||||
check_ssrf_ip(ip) # 不抛异常即通过
|
||||
|
||||
def test_invalid_ip_raises_value_error(self):
|
||||
with pytest.raises(ValueError):
|
||||
check_ssrf_ip("not-an-ip")
|
||||
|
||||
|
||||
# ── IP 地址判断 ─────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestIsIpAddress:
|
||||
@pytest.mark.parametrize(
|
||||
"host",
|
||||
[
|
||||
"127.0.0.1",
|
||||
"8.8.8.8",
|
||||
"192.168.1.1",
|
||||
"::1",
|
||||
"2001:db8::1",
|
||||
"fe80::1",
|
||||
],
|
||||
)
|
||||
def test_ip_addresses(self, host):
|
||||
assert is_ip_address(host) is True
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"host",
|
||||
[
|
||||
"example.com",
|
||||
"www.google.com",
|
||||
"localhost",
|
||||
"not-an-ip",
|
||||
"",
|
||||
],
|
||||
)
|
||||
def test_not_ip_addresses(self, host):
|
||||
assert is_ip_address(host) is False
|
||||
|
||||
|
||||
# ── URL 基础校验 ────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestValidateUrlBasic:
|
||||
def test_normal_http_url_passes(self):
|
||||
result = validate_url_basic("http://example.com/file.txt")
|
||||
assert result == "http://example.com/file.txt"
|
||||
|
||||
def test_normal_https_url_passes(self):
|
||||
result = validate_url_basic("https://www.example.com/path?q=1")
|
||||
assert result == "https://www.example.com/path?q=1"
|
||||
|
||||
def test_standard_port_80_passes(self):
|
||||
validate_url_basic("http://example.com:80/file")
|
||||
|
||||
def test_standard_port_443_passes(self):
|
||||
validate_url_basic("https://example.com:443/file")
|
||||
|
||||
def test_empty_url_rejected(self):
|
||||
with pytest.raises(UrlSecurityError, match="URL 为空"):
|
||||
validate_url_basic("")
|
||||
|
||||
def test_none_url_rejected(self):
|
||||
with pytest.raises(UrlSecurityError, match="URL 为空"):
|
||||
validate_url_basic(None) # type: ignore
|
||||
|
||||
def test_too_long_url_rejected(self):
|
||||
long_url = "https://example.com/" + "a" * 2100
|
||||
with pytest.raises(UrlSecurityError, match="URL 过长"):
|
||||
validate_url_basic(long_url)
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"url",
|
||||
[
|
||||
"ftp://example.com/file",
|
||||
"file:///etc/passwd",
|
||||
"javascript:alert(1)",
|
||||
"data:text/html,<h1>hi</h1>",
|
||||
],
|
||||
)
|
||||
def test_bad_scheme_rejected(self, url):
|
||||
with pytest.raises(UrlSecurityError, match="不允许的 URL scheme"):
|
||||
validate_url_basic(url)
|
||||
|
||||
def test_missing_hostname_rejected(self):
|
||||
with pytest.raises(UrlSecurityError, match="URL 缺少主机名"):
|
||||
validate_url_basic("http:///path")
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"url",
|
||||
[
|
||||
"http://localhost/test",
|
||||
"http://metadata/test",
|
||||
"http://foo.local/test",
|
||||
],
|
||||
)
|
||||
def test_internal_hostname_rejected(self, url):
|
||||
with pytest.raises(UrlSecurityError):
|
||||
validate_url_basic(url)
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"url",
|
||||
[
|
||||
"http://example.com:8080/file",
|
||||
"http://example.com:22/file",
|
||||
"http://example.com:3306/file",
|
||||
],
|
||||
)
|
||||
def test_non_standard_port_rejected(self, url):
|
||||
with pytest.raises(UrlSecurityError, match="不允许的端口"):
|
||||
validate_url_basic(url)
|
||||
|
||||
def test_direct_ip_rejected_by_default(self):
|
||||
with pytest.raises(UrlSecurityError, match="禁止直接 IP 访问"):
|
||||
validate_url_basic("http://8.8.8.8/file")
|
||||
|
||||
def test_direct_ip_allowed_with_flag_public(self):
|
||||
result = validate_url_basic("http://8.8.8.8/file", allow_direct_ip=True)
|
||||
assert result == "http://8.8.8.8/file"
|
||||
|
||||
def test_direct_ip_allowed_flag_but_private_still_rejected(self):
|
||||
with pytest.raises(UrlSecurityError, match="内网地址"):
|
||||
validate_url_basic("http://10.0.0.1/file", allow_direct_ip=True)
|
||||
|
||||
def test_direct_ip_loopback_rejected(self):
|
||||
with pytest.raises(UrlSecurityError):
|
||||
validate_url_basic("http://127.0.0.1/test", allow_direct_ip=True)
|
||||
|
||||
def test_trusted_domains_whitelist_pass(self):
|
||||
trusted = {"example.com"}
|
||||
result = validate_url_basic("https://example.com/file", trusted_domains=trusted)
|
||||
assert result == "https://example.com/file"
|
||||
|
||||
def test_trusted_domains_subdomain_pass(self):
|
||||
trusted = {"example.com"}
|
||||
result = validate_url_basic("https://cdn.example.com/file", trusted_domains=trusted)
|
||||
assert result == "https://cdn.example.com/file"
|
||||
|
||||
def test_trusted_domains_not_in_list_rejected(self):
|
||||
trusted = {"example.com"}
|
||||
with pytest.raises(UrlSecurityError, match="不在可信白名单"):
|
||||
validate_url_basic("https://other.com/file", trusted_domains=trusted)
|
||||
|
||||
def test_case_insensitive_scheme(self):
|
||||
# 大写 HTTP 也应该通过（我们用 .lower() 检查）
|
||||
result = validate_url_basic("HTTP://example.com/file")
|
||||
assert "HTTP://example.com/file" == result
|
||||
|
||||
|
||||
# ── is_url_basic_safe 便捷函数 ──────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestIsUrlBasicSafe:
|
||||
def test_safe_url_returns_true(self):
|
||||
assert is_url_basic_safe("https://example.com/file") is True
|
||||
|
||||
def test_unsafe_url_returns_false(self):
|
||||
assert is_url_basic_safe("http://localhost/test") is False
|
||||
|
||||
def test_empty_returns_false(self):
|
||||
assert is_url_basic_safe("") is False
|
||||
|
||||
def test_with_trusted_domains(self):
|
||||
trusted = {"allowed.com"}
|
||||
assert is_url_basic_safe("https://allowed.com/x", trusted_domains=trusted) is True
|
||||
assert is_url_basic_safe("https://other.com/x", trusted_domains=trusted) is False
|
||||
|
||||
|
||||
# ── 魔数校验 ────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestValidateMagicNumber:
|
||||
def test_mp3_id3_header(self):
|
||||
data = b"ID3" + b"\x00" * 100
|
||||
validate_magic_number(data, {"audio/mpeg"}) # 不抛异常
|
||||
|
||||
def test_mp3_frame_sync(self):
|
||||
data = b"\xff\xfb" + b"\x00" * 100
|
||||
validate_magic_number(data, {"audio/mpeg"})
|
||||
|
||||
def test_wav_header(self):
|
||||
data = b"RIFF" + b"\x00" * 4 + b"WAVE" + b"\x00" * 100
|
||||
validate_magic_number(data, {"audio/wav"})
|
||||
|
||||
def test_png_header(self):
|
||||
data = b"\x89PNG\r\n\x1a\n" + b"\x00" * 100
|
||||
validate_magic_number(data, {"image/png"})
|
||||
|
||||
def test_jpeg_header(self):
|
||||
data = b"\xff\xd8\xff" + b"\x00" * 100
|
||||
validate_magic_number(data, {"image/jpeg"})
|
||||
|
||||
def test_gif87a_header(self):
|
||||
data = b"GIF87a" + b"\x00" * 100
|
||||
validate_magic_number(data, {"image/gif"})
|
||||
|
||||
def test_gif89a_header(self):
|
||||
data = b"GIF89a" + b"\x00" * 100
|
||||
validate_magic_number(data, {"image/gif"})
|
||||
|
||||
def test_mp4_ftyp_header(self):
|
||||
data = b"\x00\x00\x00\x20ftypisom" + b"\x00" * 100
|
||||
validate_magic_number(data, {"video/mp4"})
|
||||
|
||||
def test_ogg_header(self):
|
||||
data = b"OggS" + b"\x00" * 100
|
||||
validate_magic_number(data, {"audio/ogg"})
|
||||
|
||||
def test_flac_header(self):
|
||||
data = b"fLaC" + b"\x00" * 100
|
||||
validate_magic_number(data, {"audio/flac"})
|
||||
|
||||
def test_webp_header(self):
|
||||
data = b"RIFF" + b"\x00" * 4 + b"WEBP" + b"\x00" * 100
|
||||
validate_magic_number(data, {"image/webp"})
|
||||
|
||||
def test_bmp_header(self):
|
||||
data = b"BM" + b"\x00" * 100
|
||||
validate_magic_number(data, {"image/bmp"})
|
||||
|
||||
def test_empty_file_rejected(self):
|
||||
with pytest.raises(UrlSecurityError, match="文件为空"):
|
||||
validate_magic_number(b"", {"image/png"})
|
||||
|
||||
def test_mismatched_magic_rejected(self):
|
||||
data = b"NOTAPNG" + b"\x00" * 100
|
||||
with pytest.raises(UrlSecurityError, match="魔数与允许的 MIME 类型不匹配"):
|
||||
validate_magic_number(data, {"image/png"})
|
||||
|
||||
def test_multiple_allowed_types_one_matches(self):
|
||||
data = b"\x89PNG\r\n\x1a\n" + b"\x00" * 100
|
||||
# 多个允许类型，只要有一个匹配就通过
|
||||
validate_magic_number(data, {"image/png", "image/jpeg", "image/gif"})
|
||||
|
||||
def test_no_known_magic_skips_validation(self):
|
||||
# 自定义 MIME 类型没有已知魔数，跳过校验不阻断
|
||||
validate_magic_number(b"random data", {"application/x-custom"})
|
||||
|
||||
def test_too_short_header_no_match(self):
|
||||
# 文件头太短，无法匹配需要 8 字节偏移的格式
|
||||
data = b"RIFF" # 只有 4 字节，不够 offset 8 的 WAVE 匹配
|
||||
with pytest.raises(UrlSecurityError, match="魔数"):
|
||||
validate_magic_number(data, {"audio/wav"})
|
||||
|
||||
def test_error_message_contains_mime_and_header(self):
|
||||
with pytest.raises(UrlSecurityError) as exc_info:
|
||||
validate_magic_number(b"XXXXYYY", {"image/png"})
|
||||
msg = str(exc_info.value)
|
||||
assert "image/png" in msg
|
||||
assert "文件头前16字节" in msg
|
||||
Reference in New Issue
Block a user