数字指纹是错的 抗哈希是错的

This commit is contained in:
jinye_huang 2025-05-06 16:52:03 +08:00
parent 219fcbbd57
commit 5dabb8f807

View File

@ -316,42 +316,42 @@ class PosterNotesCreator:
):
"""处理单张图像 - 此方法可在独立进程中运行"""
try:
# 加载图像
# 加载图像
image = Image.open(image_path)
# 处理图像为3:4比例并添加微小变化
# 处理图像为3:4比例并添加微小变化
processed_image = self.optimized_process_image(
image,
(3, 4),
add_variation=True,
image,
(3, 4),
add_variation=True,
seed=seed,
variation_strength=variation_strength,
extra_effects=extra_effects
)
# 创建元数据
variation_strength=variation_strength,
extra_effects=extra_effects
)
# 创建元数据
additional_metadata = {
"original_image": image_filename,
"original_image": image_filename,
"additional_index": index + 1,
"source_dir": source_dir,
"is_additional_image": True,
"processed": True,
"aspect_ratio": "3:4",
"variation_applied": True,
"variation_strength": variation_strength,
"extra_effects": extra_effects
}
# 使用输出处理器保存图像
"is_additional_image": True,
"processed": True,
"aspect_ratio": "3:4",
"variation_applied": True,
"variation_strength": variation_strength,
"extra_effects": extra_effects
}
# 使用输出处理器保存图像
return self.output_handler.handle_generated_image(
run_id,
topic_index,
variant_index,
'additional', # 图像类型为additional
processed_image,
output_filename,
additional_metadata
)
run_id,
topic_index,
variant_index,
'additional', # 图像类型为additional
processed_image,
output_filename,
additional_metadata
)
except Exception as e:
logger.error(f"处理图像时出错 '{image_filename}': {e}")
logger.error(traceback.format_exc())
@ -376,6 +376,7 @@ class PosterNotesCreator:
return image
try:
logger.debug(f"应用DCT噪声强度: {intensity:.3f}")
# 确保是灰度图或提取亮度通道 (这里以灰度为例)
if image.mode != 'L':
# 如果是彩色图,可以在 Y 通道 (亮度) 操作
@ -436,11 +437,15 @@ class PosterNotesCreator:
r = Image.blend(r, modified_gray, blend_factor)
g = Image.blend(g, modified_gray, blend_factor)
b = Image.blend(b, modified_gray, blend_factor)
return Image.merge('RGB', (r, g, b))
merged_image = Image.merge('RGB', (r, g, b))
else:
# 如果原图是灰度或处理失败,返回修改后的灰度图
return modified_gray
merged_image = modified_gray
# 在函数末尾成功时记录
logger.debug("DCT噪声应用成功。")
return merged_image if 'merged_image' in locals() else modified_gray # 返回最终结果
except Exception as e:
logger.error(f"DCT噪声注入出错: {e}")
return image # 出错时返回原图
@ -457,13 +462,15 @@ class PosterNotesCreator:
Returns:
添加扰动后的图像
"""
logger.debug(f"调用add_dct_noise对抗pHash强度: {intensity:.3f}")
return self.add_dct_noise(image, intensity=intensity)
def optimize_anti_hash_methods(self, image: Image.Image, strength: str = "medium") -> Image.Image:
"""
综合优化的哈希对抗方法强度已增加
"""
# 根据强度设置参数 (显著增加 high 强度)
"""综合优化的哈希对抗方法强度已增加添加日志记录High强度极度强化"""
logger.info(f"--- 开始优化抗哈希方法 (强度: {strength}) ---")
original_image_for_logging = image.copy() # 复制一份用于前后对比日志
# 根据强度设置参数 (极度增加 high 强度)
if strength == "low":
ahash_intensity = 0.03
phash_intensity = 0.05 # 基础DCT噪声强度
@ -471,114 +478,159 @@ class PosterNotesCreator:
region_flip_prob = 0.3
num_ahash_blocks = random.randint(8, 15)
num_dhash_lines = random.randint(6, 10)
ahash_delta_range = (-30, 30)
dhash_delta_range = 30
region_max_factor = 25
gaussian_noise_sigma = 0.0 # Low 不加高斯噪声
gaussian_noise_prob = 0.0
elif strength == "high":
ahash_intensity = 0.18 # 大幅增加
phash_intensity = 0.15 # 大幅增加
dhash_intensity = 0.18 # 大幅增加
region_flip_prob = 0.7 # 更大概率翻转
num_ahash_blocks = random.randint(20, 35) # 更多块
num_dhash_lines = random.randint(15, 25) # 更多线
else: # medium
ahash_intensity = 0.08 # 增加
phash_intensity = 0.08 # 增加
dhash_intensity = 0.08 # 增加
region_flip_prob = 0.5
num_ahash_blocks = random.randint(12, 25)
num_dhash_lines = random.randint(10, 18)
ahash_intensity = 0.40 # 极度增加
phash_intensity = 0.25 # 极度增加
dhash_intensity = 0.40 # 极度增加
region_flip_prob = 0.90 # 很高概率翻转
num_ahash_blocks = random.randint(40, 60) # 很多块
num_dhash_lines = random.randint(30, 45) # 很多线
ahash_delta_range = (-50, 50) # 更大亮度变化
dhash_delta_range = 50 # 更大梯度变化
region_max_factor = 12 # 区域更大
gaussian_noise_sigma = 3.5 # 更强高斯噪声
gaussian_noise_prob = 0.6 # 更高概率
else: # medium (也适度增加)
ahash_intensity = 0.10 # 增加
phash_intensity = 0.10 # 增加
dhash_intensity = 0.10 # 增加
region_flip_prob = 0.6
num_ahash_blocks = random.randint(15, 30)
num_dhash_lines = random.randint(12, 20)
ahash_delta_range = (-40, 40)
dhash_delta_range = 40
region_max_factor = 18
gaussian_noise_sigma = 1.5
gaussian_noise_prob = 0.5
# 1. 针对aHash (平均哈希)的处理 - 强度已增加
img_array = np.array(image, dtype=np.int16)
logger.debug(f"参数: aHash强度={ahash_intensity:.2f}, pHash强度={phash_intensity:.2f}, dHash强度={dhash_intensity:.2f}, 翻转概率={region_flip_prob:.2f}")
# 1. 针对aHash ...
logger.debug(f"应用 aHash 对抗: {num_ahash_blocks} 个亮度块, 强度={ahash_intensity:.2f}, delta范围={ahash_delta_range}")
img_array = np.array(image, dtype=np.int16)
h, w = img_array.shape[0], img_array.shape[1]
# num_ahash_blocks = random.randint(10, 20)
for _ in range(num_ahash_blocks):
block_w = random.randint(w//20, w//10)
block_h = random.randint(h//20, h//10)
x = random.randint(0, w - block_w)
y = random.randint(0, h - block_h)
delta = int(random.uniform(-35, 35) * ahash_intensity) # 增加delta范围
delta = int(random.uniform(ahash_delta_range[0], ahash_delta_range[1]) * ahash_intensity)
block = img_array[y:y+block_h, x:x+block_w]
img_array[y:y+block_h, x:x+block_w] = np.clip(block + delta, 0, 255)
image = Image.fromarray(img_array.astype(np.uint8))
# 2. 调用强化的pHash对抗方法
image = self.add_phash_noise(image, intensity=phash_intensity)
# 3. 针对dHash (差值哈希)的处理 - 强度已增加
img_array = np.array(image, dtype=np.int16)
logger.debug("aHash 对抗完成。")
# 2. 调用强化的pHash对抗方法 (使用 scipy DCT)
logger.debug(f"应用 pHash 对抗 (DCT噪声), 强度={phash_intensity:.2f}")
image = self.add_phash_noise(image, intensity=phash_intensity)
# add_phash_noise 内部已有成功日志
# 3. 针对dHash ...
logger.debug(f"应用 dHash 对抗: {num_dhash_lines} 条梯度线, 强度={dhash_intensity:.2f}, delta范围=+/-({dhash_delta_range})")
img_array = np.array(image, dtype=np.int16)
h, w = img_array.shape[0], img_array.shape[1]
mask = np.zeros_like(img_array, dtype=bool)
# num_dhash_lines = random.randint(8, 12)
for _ in range(num_dhash_lines):
line_width = random.randint(2, 5) # 增加线宽
if random.random() < 0.5:
y = random.randint(0, h - 1)
line_width = random.randint(1, 4) # 增加线宽可能性
if len(mask.shape) == 3:
mask[max(0, y-line_width//2):min(h, y+line_width//2+1), :, :] = True
else:
mask[max(0, y-line_width//2):min(h, y+line_width//2+1), :] = True
else:
x = random.randint(0, w - 1)
line_width = random.randint(1, 4) # 增加线宽可能性
if len(mask.shape) == 3:
mask[:, max(0, x-line_width//2):min(w, x+line_width//2+1), :] = True
else:
mask[:, max(0, x-line_width//2):min(w, x+line_width//2+1)] = True
delta = (np.random.random(img_array.shape) * 2 - 1) * dhash_intensity * 35 # 增加delta范围
delta = (np.random.random(img_array.shape) * 2 - 1) * dhash_intensity * dhash_delta_range
img_array[mask] += delta[mask].astype(np.int16)
img_array = np.clip(img_array, 0, 255)
# 4. 颜色直方图扰动 (强度也略微增加)
image = Image.fromarray(img_array.astype(np.uint8))
color_hist_strength = dhash_intensity * 0.6 # 关联强度
logger.debug("dHash 对抗完成。")
# 4. 颜色直方图扰动 ...
color_hist_strength = dhash_intensity * 0.7 # 关联强度增加
logger.debug(f"应用颜色直方图扰动, 强度={color_hist_strength:.3f}")
image = self.perturb_color_histogram(image, strength=color_hist_strength)
# 5. 区域翻转 - 强度已增加
# perturb_color_histogram 内部已有成功日志
# 5. 区域翻转/旋转 ...
if random.random() < region_flip_prob:
logger.debug(f"应用区域变换 (翻转/旋转), 概率触发成功.")
img_array = np.array(image)
h, w = img_array.shape[0], img_array.shape[1]
region_w = random.randint(w//(region_max_factor+5), w//region_max_factor)
region_h = random.randint(h//(region_max_factor+5), h//region_max_factor)
region_w = max(1, region_w)
region_h = max(1, region_h)
x = random.randint(0, max(0, w - region_w))
y = random.randint(0, max(0, h - region_h))
# 增加区域大小可能性
max_region_factor = 15 if strength == 'high' else 20
region_w = random.randint(w//(max_region_factor+5), w//max_region_factor)
region_h = random.randint(h//(max_region_factor+5), h//max_region_factor)
x = random.randint(0, w - region_w)
y = random.randint(0, h - region_h)
# 加入90度旋转的可能性
action = random.choice(['flip_h', 'flip_v', 'rotate_90']) if strength != 'low' else random.choice(['flip_h', 'flip_v'])
region = img_array[y:y+region_h, x:x+region_w]
if action == 'flip_h':
img_array[y:y+region_h, x:x+region_w] = region[:, ::-1]
elif action == 'flip_v':
img_array[y:y+region_h, x:x+region_w] = region[::-1, :]
elif action == 'rotate_90' and len(img_array.shape) == 3: # 旋转只对原尺寸区域有效
# 注意:旋转可能需要调整区域大小或填充,这里简化处理
# 仅在区域接近正方形时效果较好
if abs(region_w - region_h) < 5:
rotated_region = np.rot90(region)
# 需要确保旋转后尺寸匹配,如果尺寸变化则跳过或填充
if rotated_region.shape[0] == region_h and rotated_region.shape[1] == region_w:
img_array[y:y+region_h, x:x+region_w] = rotated_region
action_applied = "skipped"
try: # 添加 try-except 捕获潜在错误
if y+region_h <= h and x+region_w <= w:
region = img_array[y:y+region_h, x:x+region_w]
if region.size == 0:
action_applied = "empty_region_skipped"
elif action == 'flip_h':
img_array[y:y+region_h, x:x+region_w] = region[:, ::-1]
action_applied = action
elif action == 'flip_v':
img_array[y:y+region_h, x:x+region_w] = region[::-1, :]
action_applied = action
elif action == 'rotate_90' and len(img_array.shape) == 3:
if abs(region_w - region_h) < region_w * 0.3:
rotated_region = np.rot90(region)
if rotated_region.shape[0] == region_h and rotated_region.shape[1] == region_w:
img_array[y:y+region_h, x:x+region_w] = rotated_region
action_applied = action
else:
logger.debug(f" 区域旋转跳过: 尺寸不匹配 ({region.shape} -> {rotated_region.shape})")
action_applied = "rotate_skipped_size_mismatch"
else:
logger.debug(f" 区域旋转跳过: 非方形区域 ({region_w}x{region_h})")
action_applied = "rotate_skipped_not_square"
else:
action_applied = f"{action}_skipped_condition_not_met"
else:
action_applied = "region_bounds_error_skipped"
logger.warning(f"跳过区域变换,切片索引无效: y={y}, h={region_h}, x={x}, w={region_w}, img_shape={img_array.shape}")
except Exception as e_region:
logger.warning(f"应用区域变换 {action} 时出错: {e_region}")
action_applied = f"error: {e_region}"
image = Image.fromarray(img_array)
# 6. (新增可选) 轻微高斯噪声 - 对所有哈希都有轻微普适性干扰
if strength != 'low' and random.random() < 0.4:
if action_applied not in ["skipped", "empty_region_skipped", "rotate_skipped_size_mismatch", "rotate_skipped_not_square", "region_bounds_error_skipped"] and not action_applied.startswith("error"):
image = Image.fromarray(img_array)
logger.debug(f" 执行了区域操作: {action_applied} 在 ({x},{y}) 大小 ({region_w}x{region_h})")
else:
logger.debug(f" 区域变换未执行或跳过: {action_applied}")
else:
logger.debug("跳过区域变换 (概率未触发).")
# 6. 轻微高斯噪声 ...
apply_gaussian_noise = random.random() < gaussian_noise_prob
if gaussian_noise_sigma > 0 and apply_gaussian_noise:
logger.debug(f"应用高斯噪声, sigma={gaussian_noise_sigma:.1f}")
img_array = np.array(image)
noise_sigma = 1.0 if strength == 'medium' else 2.0 # 噪声标准差
noise = np.random.normal(0, noise_sigma, img_array.shape)
noise = np.random.normal(0, gaussian_noise_sigma, img_array.shape)
img_array = np.clip(img_array + noise, 0, 255).astype(np.uint8)
image = Image.fromarray(img_array)
elif gaussian_noise_sigma > 0:
logger.debug("跳过高斯噪声 (概率未触发).")
# ... (对比日志不变) ...
logger.info(f"--- 完成优化抗哈希方法 (强度: {strength}) ---")
return image
def optimized_process_image(
@ -654,6 +706,7 @@ class PosterNotesCreator:
# 如果不需要变化或是低强度且禁用额外效果
if not add_variation:
logger.info("add_variation=False跳过所有变化和抗哈希处理。")
# 重置随机种子
if seed is not None:
random.seed()
@ -662,7 +715,7 @@ class PosterNotesCreator:
# 清除元数据后返回
return self.strip_metadata(result)
# 高效应用基本变化
logger.info(f"应用基础变化和抗哈希处理 (强度: {variation_strength}, 额外效果: {use_extra})")
processed_image = result.convert('RGB')
# 1. 亮度调整
@ -686,14 +739,16 @@ class PosterNotesCreator:
if abs(rotation_angle) > 0.1: # 只有当角度足够大时才旋转
processed_image = processed_image.rotate(rotation_angle, resample=Image.BICUBIC, expand=False)
# 5. 新增 - 应用反查重技术
# 根据变化强度选择性应用
# 5. 应用抗哈希技术
if use_extra:
# 使用综合优化的哈希对抗方法
logger.debug("调用 optimize_anti_hash_methods...")
processed_image = self.optimize_anti_hash_methods(processed_image, variation_strength)
else:
logger.info("use_extra=False跳过 optimize_anti_hash_methods。")
# 应用额外效果 (只在需要时)
# 应用模糊/锐化/边框等额外效果 (如果 use_extra 为 True)
if use_extra:
logger.debug("应用额外效果 (模糊/锐化/边框)...")
# 根据强度决定是否应用特定效果
apply_sharpen = random.random() < 0.4
apply_blur = not apply_sharpen and random.random() < 0.3
@ -719,21 +774,23 @@ class PosterNotesCreator:
w, h = processed_image.size
bordered = Image.new('RGB', (w + border_size*2, h + border_size*2), border_color)
bordered.paste(processed_image, (border_size, border_size))
# 随机裁剪回原尺寸
offset_x = random.randint(0, border_size*2)
offset_y = random.randint(0, border_size*2)
processed_image = bordered.crop((offset_x, offset_y, offset_x + w, offset_y + h))
logger.debug("额外效果应用完成。")
else:
logger.info("use_extra=False跳过额外效果。")
# 6. 始终清除元数据 - 最后一步
processed_image = self.strip_metadata(processed_image)
# **关键:确保在所有修改之后调用修复后的 strip_metadata**
logger.debug("最后调用 strip_metadata 清除元数据。")
final_image = self.strip_metadata(processed_image)
# 重置随机种子
if seed is not None:
random.seed()
np.random.seed()
logger.debug("随机种子已重置。")
return processed_image
logger.info(f"图像处理完成 (强度: {variation_strength})")
return final_image
def perturb_color_histogram(self, image: Image.Image, strength: float = 0.03) -> Image.Image:
"""
@ -746,6 +803,7 @@ class PosterNotesCreator:
Returns:
处理后的图像
"""
logger.debug(f"扰动颜色直方图,强度: {strength:.3f}")
# 确保为RGB模式
if image.mode != 'RGB':
image = image.convert('RGB')
@ -782,11 +840,12 @@ class PosterNotesCreator:
img_array[:,:,channel][mask] + offset, 0, 255).astype(np.uint8)
# 转回PIL图像
logger.debug("颜色直方图扰动成功。")
return Image.fromarray(img_array)
def strip_metadata(self, image: Image.Image) -> Image.Image:
"""
移除图像中的所有元数据
移除图像中的所有元数据 (修复版)
Args:
image: 输入图像
@ -794,10 +853,43 @@ class PosterNotesCreator:
Returns:
无元数据的图像
"""
# 创建无元数据的副本
data = io.BytesIO()
image.save(data, format=image.format if image.format else 'PNG')
return Image.open(data)
logger.debug("移除图像元数据...")
try:
# 从当前图像对象的像素数据创建一个新的Image对象
# 这确保我们使用的是内存中修改后的数据
# 保留原始格式信息,如果可用
img_format = image.format if hasattr(image, 'format') else 'PNG'
# 如果图像有alpha通道需要正确处理
if image.mode == 'RGBA':
# 创建一个白色背景然后粘贴带有alpha的图像
background = Image.new("RGB", image.size, (255, 255, 255))
background.paste(image, mask=image.split()[3]) # 3 is the alpha channel
image_to_save = background
img_format = 'JPEG' # 通常去除alpha后存为JPEG
elif image.mode == 'P':
# 带调色板的图像转换为RGB
image_to_save = image.convert('RGB')
img_format = 'JPEG'
else:
image_to_save = image
# 保存到内存缓冲区
data = io.BytesIO()
# 确保保存时指定质量参数,避免默认压缩导致意外变化
if img_format == 'JPEG':
image_to_save.save(data, format='JPEG', quality=95) # 使用高质量JPEG
else:
# 对于PNG等无损格式不需要质量参数
image_to_save.save(data, format=img_format)
data.seek(0) # 重置缓冲区指针
reloaded_image = Image.open(data)
logger.debug("元数据移除成功。")
return reloaded_image
except Exception as e:
logger.error(f"移除元数据时出错: {e}")
return image # 出错时返回原图
def process_poster_for_notes(
run_id: str,