fix(scraping): normalize artwork and reuse downloads for aliases

Add independent alias and Fanart 4K switches, identify exact artwork names, preserve episode and season conventions, and detect missing images across case and format variants. Normalize downloaded bytes to JPEG or PNG and reuse one request with storage hard links or local upload copies.

Refs #7128

app/application/configuration.py

@@ -375,6 +375,8 @@ class ChainRuntimeConfig:
     :param data_cleanup_outbox_dead_days: Outbox 死信记录保留天数，0为不清理
     :param download_subtitle: 下载站点字幕
     :param scrap_season_image_to_root: 季图片是否额外写入剧集根目录；默认开启以兼容已有媒体库
+    :param scrap_image_alias: 是否为背景图和缩略图写入兼容别名文件
+    :param scrap_4k_background: 是否获取 Fanart 提供的 4K 背景图
     :param lyrics_batch_timeout: 单次音乐刮削批次用于在线歌词查询的总预算（秒）
     :param music_metadata_to_simplified: 是否将识别到的音乐标题、艺术家、专辑等标准元数据转换为简体中文
     :param music_lyrics_to_simplified: 是否将写入 LRC/TXT 的歌词正文转为简体中文；保留时间轴、LRC 标签和来源 Lyricsfile
@@ -464,6 +466,8 @@ class ChainRuntimeConfig:
     data_cleanup_outbox_dead_days: Any = 0
     download_subtitle: bool = True
     scrap_season_image_to_root: bool = True
+    scrap_image_alias: bool = True
+    scrap_4k_background: bool = True
     lyrics_batch_timeout: int = 120
     music_metadata_to_simplified: bool = True
     music_lyrics_to_simplified: bool = False

app/application/settings/resources/system_settings_catalog.json

@@ -1321,12 +1321,24 @@
       "source_file": "app/runtime/config.py",
       "source_line": 719
     },
+    "SCRAP_4K_BACKGROUND": {
+      "description": "是否获取 Fanart 的 show4kbackground 和 movie4kbackground 图片",
+      "group": "media_metadata",
+      "source_file": "app/runtime/config.py",
+      "source_line": 349
+    },
     "SCRAP_FOLLOW_TMDB": {
       "description": "新增已入库媒体是否跟随TMDB信息变化",
       "group": "transfer",
       "source_file": "app/runtime/config.py",
       "source_line": 586
     },
+    "SCRAP_IMAGE_ALIAS": {
+      "description": "是否为背景图和缩略图写入兼容别名文件；关闭后每张图片只写入来源名称",
+      "group": "media_metadata",
+      "source_file": "app/runtime/config.py",
+      "source_line": 345
+    },
     "SCRAP_SEASON_IMAGE_TO_ROOT": {
       "description": "季图片是否同时写入剧集根目录；关闭后仅写季目录，不清理已有图片",
       "group": "media_metadata",

app/chain/scraping.py

@@ -4,6 +4,7 @@
 
 import os
 import re
+import shutil
 import threading
 import time
 from collections.abc import Iterator, Mapping
@@ -13,6 +14,7 @@ from pathlib import Path
 from tempfile import NamedTemporaryFile, TemporaryDirectory
 from threading import Lock
 from typing import Any, Iterable, List, Optional, Protocol, Self, Tuple, Union, cast
+from urllib.parse import urlsplit
 
 from app.application.audio import AudioMetadataHelper
 from app.application.configuration import (
@@ -23,6 +25,7 @@ from app.chain.base import ChainBase
 from app.chain.lyrics import LyricsChain
 from app.chain.media import MediaChain
 from app.chain.storage import StorageChain
+from app.domain.artwork import normalize_artwork
 from app.domain.context import (
     MediaInfo,
     MusicAlbumInfo,
@@ -368,11 +371,15 @@ class ScrapingChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
         "backdrop": ScrapingMetadata.BACKDROP,
         "fanart": ScrapingMetadata.BACKDROP,
         "background": ScrapingMetadata.BACKDROP,
+        "show4kbackground": ScrapingMetadata.BACKDROP,
+        "movie4kbackground": ScrapingMetadata.BACKDROP,
         "logo": ScrapingMetadata.LOGO,
+        "clearlogo": ScrapingMetadata.LOGO,
         "disc": ScrapingMetadata.DISC,
         "cdart": ScrapingMetadata.DISC,
         "banner": ScrapingMetadata.BANNER,
         "thumb": ScrapingMetadata.THUMB,
+        "episode-thumb": ScrapingMetadata.THUMB,
         "landscape": ScrapingMetadata.LANDSCAPE,
         "clearart": ScrapingMetadata.CLEARART,
     }
@@ -385,6 +392,7 @@ class ScrapingChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
     }
 
     MUSIC_LYRICS_EXTENSIONS = (".lyricsfile.yaml", ".lrc", ".txt")
+    IMAGE_EXTENSIONS = frozenset({".jpg", ".jpeg", ".png", ".webp"})
     _music_track_prefix_pattern = re.compile(
         r"^\s*(?:(?:cd|disc)\s*\d+\s*[-_. ]+)?(?:\d+\s*[-_. ]+)+",
         flags=re.IGNORECASE,
@@ -490,53 +498,336 @@ class ScrapingChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
             self._cleanup_temp_file(tmp_file_path)
 
     def _download_and_save_image(
-            self, fileitem: _SchemaFileItem, path: Path, url: str
+            self,
+            fileitem: _SchemaFileItem,
+            path: Path,
+            url: str,
+            reuse_targets: Optional[List[Tuple[_SchemaFileItem, Path]]] = None,
     ) -> None:
         """
-        流式下载图片并保存到文件
+        下载一次图片并写入一个或多个目标文件。
 
-        :param fileitem: 关联的媒体文件项
-        :param path: 图片文件路径
-        :param url: 图片下载URL
+        首个目标上传成功后，后续目标优先尝试同存储硬链接；不支持或链接失败时，
+        从同一份下载临时文件复制后上传，避免重复请求远端地址。
+
+        :param fileitem: 首个图片目标所在的目录项
+        :param path: 首个图片目标路径
+        :param url: 图片下载 URL
+        :param reuse_targets: 复用本次下载的其它目录项和路径
         """
         if not fileitem or not url or not path:
             return
+
+        targets = [(fileitem, path)] + list(reuse_targets or [])
+        downloaded_path: Optional[Path] = None
         try:
             logger.info(f"正在下载图片：{url} ...")
-            http = _scraping_http_snapshot()
-            response = http.stream(
-                url,
-                proxies=self.runtime_config.proxy,
-                ua=self.runtime_config.normal_user_agent,
-            )
-            with response as r:
-                if r and r.status_code == 200:
-                    tmp_file_path = None
-                    try:
-                        # delete_on_close 是 Python 3.12 才支持的参数，使用 delete=False 后手动清理以兼容低版本。
-                        with NamedTemporaryFile(delete=False, suffix=path.suffix) as tmp_file:
-                            tmp_file_path = Path(tmp_file.name)
-                            for chunk in r.iter_content(chunk_size=8192):
-                                if chunk:
-                                    tmp_file.write(chunk)
-                            tmp_file.flush()
+            downloaded = self._download_image_to_temp(url=url, suffix=path.suffix)
+            if not downloaded:
+                return
+            downloaded_path, image_suffix = downloaded
 
-                        # 刮削的图片只需要读写权限
-                        tmp_file_path.chmod(0o666 & ~current_umask)
+            source_item: Optional[_SchemaFileItem] = None
+            saved_targets: set[tuple[Optional[str], str]] = set()
+            for target_item, target_path in targets:
+                if not target_item or not target_path:
+                    continue
+                target_path = target_path.with_suffix(image_suffix)
+                target_key = (target_item.storage, str(target_path))
+                if target_key in saved_targets:
+                    continue
+                saved_targets.add(target_key)
 
-                        item = self.storagechain.upload_file(
-                            fileitem=fileitem, path=tmp_file_path, new_name=path.name
-                        )
-                        if item:
-                            logger.info(f"已保存图片：{item.path}")
-                        else:
-                            logger.warn(f"图片保存失败：{path}")
-                    finally:
-                        self._cleanup_temp_file(tmp_file_path)
+                if self._reuse_image_file(source_item, target_item, target_path):
+                    continue
+
+                item = self._upload_downloaded_image(
+                    fileitem=target_item,
+                    path=target_path,
+                    downloaded_path=downloaded_path,
+                )
+                if item and source_item is None:
+                    source_item = item
+                if item:
+                    logger.info(f"已保存图片：{item.path}")
                 else:
-                    logger.info(f"{url} 图片下载失败")
+                    logger.warn(f"图片保存失败：{target_path}")
         except Exception as err:
             logger.error(f"{url} 图片下载失败：{str(err)}！")
+        finally:
+            self._cleanup_temp_file(downloaded_path)
+
+    def _reuse_image_file(
+            self,
+            source_item: Optional[_SchemaFileItem],
+            target_item: _SchemaFileItem,
+            target_path: Path,
+    ) -> bool:
+        """
+        在同一存储中尝试复用已保存图片；失败交由调用方从本地副本上传。
+
+        :param source_item: 本次下载已写入的图片文件项；尚未保存时为 None
+        :param target_item: 后续图片目标所在的目录项
+        :param target_path: 后续图片完整保存路径
+        :return: 已通过硬链接保存目标图片时返回 True
+        """
+        if not source_item or source_item.storage != target_item.storage:
+            return False
+        try:
+            if self.storagechain.link_file(source_item, target_path):
+                logger.info(f"已通过硬链接保存图片：{target_path}")
+                return True
+        except Exception as err:
+            logger.debug(f"图片硬链接不可用，改用下载副本上传：{target_path} - {err}")
+        return False
+
+    def _download_image_to_temp(self, url: str, suffix: str) -> Optional[Tuple[Path, str]]:
+        """
+        将远程图片流式下载到本地临时文件。
+
+        :param url: 图片下载 URL
+        :param suffix: 临时文件后缀，用于保留图片格式提示
+        :return: 下载完成的临时文件及按实际图片内容确定的后缀；请求失败时返回 None
+        """
+        http = _scraping_http_snapshot()
+        response = http.stream(
+            url,
+            proxies=self.runtime_config.proxy,
+            ua=self.runtime_config.normal_user_agent,
+        )
+        with response as result:
+            if not result or result.status_code != 200:
+                logger.info(f"{url} 图片下载失败")
+                return None
+
+            tmp_file_path: Optional[Path] = None
+            try:
+                # delete_on_close 是 Python 3.12 才支持的参数，使用 delete=False 后手动清理以兼容低版本。
+                with NamedTemporaryFile(delete=False, suffix=suffix) as tmp_file:
+                    tmp_file_path = Path(tmp_file.name)
+                    for chunk in result.iter_content(chunk_size=8192):
+                        if chunk:
+                            tmp_file.write(chunk)
+                    tmp_file.flush()
+
+                original_content = tmp_file_path.read_bytes()
+                image_content, image_suffix = normalize_artwork(original_content)
+                if image_content != original_content:
+                    tmp_file_path.write_bytes(image_content)
+                # 刮削的图片只需要读写权限
+                tmp_file_path.chmod(0o666 & ~current_umask)
+                return tmp_file_path, image_suffix
+            except Exception:
+                self._cleanup_temp_file(tmp_file_path)
+                raise
+
+    def _upload_downloaded_image(
+            self,
+            fileitem: _SchemaFileItem,
+            path: Path,
+            downloaded_path: Path,
+    ) -> Optional[_SchemaFileItem]:
+        """
+        从已下载的主临时文件复制一份并上传到目标路径。
+
+        :param fileitem: 目标目录项
+        :param path: 目标图片路径
+        :param downloaded_path: 本次请求产生的主临时文件
+        :return: 上传后的文件项；上传失败时返回 None
+        """
+        upload_path: Optional[Path] = None
+        try:
+            # 每次上传使用独立副本，因为存储适配器成功后会删除本地源文件。
+            with NamedTemporaryFile(delete=False, suffix=path.suffix) as upload_file:
+                upload_path = Path(upload_file.name)
+            shutil.copyfile(downloaded_path, upload_path)
+            upload_path.chmod(0o666 & ~current_umask)
+            return self.storagechain.upload_file(
+                fileitem=fileitem,
+                path=upload_path,
+                new_name=path.name,
+            )
+        except Exception as err:
+            logger.warn(f"图片上传失败：{path} - {err}")
+            return None
+        finally:
+            self._cleanup_temp_file(upload_path)
+
+    @classmethod
+    def _normalize_image_suffix(cls, value: Optional[str]) -> str:
+        """
+        将图片地址或文件名的扩展名归一为受支持的小写后缀。
+
+        :param value: 图片 URL 或来源文件名
+        :return: ``.jpg``、``.png`` 或 ``.webp`` 等归一后的后缀
+        """
+        suffix = Path(urlsplit(value or "").path).suffix.lower()
+        if suffix == ".jpeg":
+            return ".jpg"
+        if suffix in cls.IMAGE_EXTENSIONS:
+            return suffix
+        return ".jpg"
+
+    @classmethod
+    def _normalize_image_filename(cls, filename: Optional[str]) -> Optional[str]:
+        """
+        归一图片文件名大小写与扩展名，避免同类型文件因 URL 写法不同重复落盘。
+
+        :param filename: 刮削源提供的图片文件名
+        :return: 归一后的文件名；输入为空时返回 None
+        """
+        if not filename:
+            return None
+        path = Path(urlsplit(filename).path)
+        stem = path.stem.lower()
+        canonical_names = {
+            "show4kbackground": "background",
+            "movie4kbackground": "background",
+            "clearlogo": "logo",
+        }
+        stem = canonical_names.get(stem, stem)
+        return f"{stem}{cls._normalize_image_suffix(filename)}"
+
+    @classmethod
+    def _get_image_metadata_type(
+            cls,
+            image_name: str,
+            item_type: ScrapingTarget,
+            current_fileitem: Optional[_SchemaFileItem] = None,
+    ) -> Tuple[Optional[ScrapingMetadata], bool]:
+        """
+        按去扩展名后的图片名精确解析元数据类型。
+
+        季图片保留 ``seasonNN-`` 和 ``season-specials-`` 前缀；单集媒体文件同名
+        图片及 ``*-thumb`` 是既有来源约定，仍按缩略图处理。返回值第二项标记
+        Fanart 的 4K 背景来源，供独立开关过滤。
+
+        :param image_name: 刮削源图片文件名
+        :param item_type: 当前刮削目标层级
+        :param current_fileitem: 当前媒体文件，用于识别单集同名图片
+        :return: 元数据类型与是否为 4K 背景的标记
+        """
+        stem = Path(image_name).stem.lower()
+        lookup_stem = stem
+
+        season_match = re.match(r"^season(?:\d+|-specials)-(.+)$", stem)
+        if season_match:
+            lookup_stem = season_match.group(1)
+
+        is_4k_background = lookup_stem in {"show4kbackground", "movie4kbackground"}
+
+        if lookup_stem == "episode-thumb" and item_type != ScrapingTarget.EPISODE:
+            return None, is_4k_background
+        metadata_type = cls.IMAGE_METADATA_MAP.get(lookup_stem)
+        if metadata_type:
+            return metadata_type, is_4k_background
+
+        if item_type == ScrapingTarget.EPISODE:
+            media_stem = Path(current_fileitem.path).stem.lower() if current_fileitem and current_fileitem.path else ""
+            if media_stem and stem in {media_stem, f"{media_stem}-thumb"}:
+                return ScrapingMetadata.THUMB, is_4k_background
+
+        return None, is_4k_background
+
+    @staticmethod
+    def _matches_image_season(
+            image_name: str,
+            item_type: ScrapingTarget,
+            season_number: Optional[int],
+    ) -> bool:
+        """
+        限定季前缀图片只用于对应季目录，整剧根目录刮削不消费季图片。
+
+        :param image_name: 来源图片文件名
+        :param item_type: 当前刮削层级
+        :param season_number: 当前季号；None 表示不按季号过滤
+        :return: 图片符合当前刮削上下文时返回 True
+        """
+        stem = Path(image_name).stem.lower()
+        match = re.match(r"^season(\d+|-specials)-", stem)
+        if not match:
+            return True
+        if item_type == ScrapingTarget.TV:
+            logger.info(f"当前为电视剧根目录刮削，跳过季图片：{image_name}")
+            return False
+        if item_type != ScrapingTarget.SEASON or season_number is None:
+            return True
+        image_season = 0 if match.group(1) == "-specials" else int(match.group(1))
+        if image_season != season_number:
+            logger.info(f"当前刮削季为：{season_number}，跳过非本季图片：{image_name}")
+            return False
+        return True
+
+    @classmethod
+    def _image_type_stems(
+            cls,
+            image_path: Path,
+            metadata_type: ScrapingMetadata,
+    ) -> set[str]:
+        """
+        返回同一图片类型可接受的文件名主体。
+
+        :param image_path: 当前目标图片路径
+        :param metadata_type: 当前图片元数据类型
+        :return: 用于大小写不敏感存在性判断的文件名主体集合
+        """
+        stem = image_path.stem.lower()
+        aliases = {
+            key
+            for key, value in cls.IMAGE_METADATA_MAP.items()
+            if value == metadata_type and not key.startswith("season")
+        }
+        if metadata_type in {ScrapingMetadata.THUMB, ScrapingMetadata.LANDSCAPE}:
+            aliases.update({"thumb", "landscape"})
+        if stem in aliases:
+            return aliases
+        return {stem}
+
+    def _image_exists_for_type(
+            self,
+            fileitem: _SchemaFileItem,
+            image_path: Path,
+            metadata_type: ScrapingMetadata,
+            directory_cache: dict[tuple[str, str], set[str]],
+    ) -> bool:
+        """
+        按图片类型和受支持扩展名判断目标目录是否已有图片。
+
+        精确查询优先，未命中时只列举一次目标目录，并忽略文件名及扩展名大小写。
+
+        :param fileitem: 目标目录项
+        :param image_path: 期望图片路径
+        :param metadata_type: 当前图片元数据类型
+        :param directory_cache: 本次刮削的目录文件名缓存
+        :return: 已存在同类型图片时返回 True
+        """
+        if not fileitem.storage or not fileitem.path:
+            return False
+        existing_item = self.storagechain.get_file_item(storage=fileitem.storage, path=image_path)
+        if existing_item and existing_item.type == "file":
+            return True
+
+        cache_key = (fileitem.storage, str(fileitem.path))
+        names = directory_cache.get(cache_key)
+        if names is None:
+            names = set()
+            children = self.storagechain.list_files(fileitem, recursion=False)
+            if isinstance(children, list):
+                names = {
+                    child.name.lower()
+                    for child in children
+                    if child.type == "file" and child.name
+                }
+            directory_cache[cache_key] = names
+
+        stems = self._image_type_stems(image_path=image_path, metadata_type=metadata_type)
+        return any(
+            Path(name).stem in stems
+            and (
+                ".jpg" if Path(name).suffix.lower() == ".jpeg" else Path(name).suffix.lower()
+            ) in self.IMAGE_EXTENSIONS
+            for name in names
+        )
 
     def _get_target_fileitem_and_path(
             self,
@@ -552,7 +843,8 @@ class ScrapingChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
         """
         target_dir_item = current_fileitem
         target_dir_path = Path(cast(str, current_fileitem.path))
-        final_filename = filename_hint
+        normalized_hint = self._normalize_image_filename(filename_hint)
+        final_filename = filename_hint if metadata_type == ScrapingMetadata.NFO else normalized_hint
 
         # 针对 NFO 文件的特殊命名和存储逻辑
         if metadata_type == ScrapingMetadata.NFO:
@@ -605,7 +897,7 @@ class ScrapingChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
                 ScrapingMetadata.LANDSCAPE: "landscape",
             }
             if season_image_name := season_image_name_map.get(metadata_type):
-                hint_ext = Path(filename_hint).suffix if filename_hint else ".jpg"
+                hint_ext = self._normalize_image_suffix(filename_hint)
                 final_filename = f"{season_image_name}{hint_ext}"
         elif item_type == ScrapingTarget.MOVIE and current_fileitem.type == "file":
             # 电影文件的图片应与视频文件同级保存，避免把图片路径拼到文件名下面。
@@ -622,7 +914,7 @@ class ScrapingChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
                 metadata_type in [ScrapingMetadata.THUMB]
                 and item_type == ScrapingTarget.EPISODE
         ):
-            hint_ext = Path(filename_hint).suffix if filename_hint else ".jpg"
+            hint_ext = self._normalize_image_suffix(filename_hint)
             final_filename = f"{target_dir_path.stem}{hint_ext}"
             parent_item = parent_fileitem or self.storagechain.get_parent_item(
                 current_fileitem
@@ -696,7 +988,9 @@ class ScrapingChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
             logger.warn(f"无法获取季目录 {current_fileitem.path} 的父目录项，跳过根目录季图片")
             return targets
 
-        season_root_path = Path(current_fileitem.path).with_name(filename_hint)
+        season_root_path = Path(current_fileitem.path).with_name(
+            self._normalize_image_filename(filename_hint) or filename_hint
+        )
         root_target = (season_parent_item, season_root_path)
         if root_target not in targets:
             targets.insert(0, root_target)
@@ -712,6 +1006,8 @@ class ScrapingChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
         例如 backdrop.jpg 同时保存为 fanart.jpg，thumb.jpg 同时保存为 landscape.jpg。
         """
         expanded = list(targets)
+        if not self.runtime_config.scrap_image_alias:
+            return expanded
         for base_item, image_path in list(targets):
             if not image_path:
                 continue
@@ -1013,74 +1309,81 @@ class ScrapingChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
             logger.info(f"未获取到 {item_type.value} 的图片信息，跳过图片刮削。")
             return
 
+        pending_targets: dict[str, list[Tuple[_SchemaFileItem, Path]]] = {}
+        directory_cache: dict[tuple[str, str], set[str]] = {}
+        seen_targets: set[tuple[str, str, str]] = set()
+
         for image_name, image_url in image_dict.items():
-            metadata_type = None
-            for keyword, meta_type in self.IMAGE_METADATA_MAP.items():
-                if keyword in image_name.lower():
-                    metadata_type = meta_type
-                    break
-
-            if metadata_type:
-                option = self.scraping_policies.option(item_type, metadata_type)
-
-                if option.is_skip:
-                    logger.info(
-                        f"{item_type.value} {option.metadata.value} 刮削策略 {option.policy.value}"
-                    )
-                    continue
-
-                # 判断是否匹配当前刮削的季号
-                if item_type == ScrapingTarget.TV and image_name.lower().startswith(
-                        "season"
-                ):
-                    logger.info(f"当前为电视剧根目录刮削，跳过季图片：{image_name}")
-                    continue
-                if (
-                        item_type == ScrapingTarget.SEASON
-                        and season_number is not None
-                        and image_name.lower().startswith("season")
-                ):
-                    # 检查是否只下载当前刮削季的图片
-                    image_season_str = (
-                        "00" if "specials" in image_name.lower() else image_name[6:8]
-                    )
-
-                    if image_season_str is not None and image_season_str != str(
-                            season_number
-                    ).rjust(2, "0"):
-                        logger.info(
-                            f"当前刮削季为：{season_number}，跳过非本季图片：{image_name}"
-                        )
-                        continue
-
-                # 按配置生成季图片的根目录和季目录目标，再逐个应用刮削策略。
-                image_targets = self._get_target_fileitems_and_paths(
-                    current_fileitem=current_fileitem,
-                    item_type=item_type,
-                    metadata_type=metadata_type,
-                    filename_hint=image_name,
-                    parent_fileitem=parent_fileitem,
-                )
-
-                # 扩展别名目标（如 backdrop→fanart, thumb→landscape）
-                image_targets = self._expand_with_aliases(image_targets, item_type)
-
-                for base_item, image_path in image_targets:
-                    if not image_path or not base_item.storage:
-                        continue
-
-                    file_exists = self.storagechain.get_file_item(
-                        storage=base_item.storage, path=image_path
-                    )
-
-                    if self._should_scrape(option, bool(file_exists), overwrite):
-                        self._download_and_save_image(
-                            fileitem=base_item, path=image_path, url=image_url
-                        )
-            else:
+            metadata_type, is_4k_background = self._get_image_metadata_type(
+                image_name=image_name,
+                item_type=item_type,
+                current_fileitem=current_fileitem,
+            )
+            if not metadata_type:
                 logger.debug(
                     f"未找到图片类型 {image_name} 对应的 ScrapingMetadata，跳过。"
                 )
+                continue
+            if is_4k_background and not self.runtime_config.scrap_4k_background:
+                logger.info(f"已关闭 4K 背景图刮削，跳过：{image_name}")
+                continue
+
+            option = self.scraping_policies.option(item_type, metadata_type)
+            if option.is_skip:
+                logger.info(
+                    f"{item_type.value} {option.metadata.value} 刮削策略 {option.policy.value}"
+                )
+                continue
+
+            if not self._matches_image_season(image_name, item_type, season_number):
+                continue
+
+            # 按配置生成季图片的根目录和季目录目标，再逐个应用刮削策略。
+            image_targets = self._get_target_fileitems_and_paths(
+                current_fileitem=current_fileitem,
+                item_type=item_type,
+                metadata_type=metadata_type,
+                filename_hint=image_name,
+                parent_fileitem=parent_fileitem,
+            )
+
+            # 扩展别名目标（如 backdrop→fanart, thumb→landscape）。
+            image_targets = self._expand_with_aliases(image_targets, item_type)
+
+            for base_item, image_path in image_targets:
+                if not image_path or not base_item.storage:
+                    continue
+                target_key = (base_item.storage, str(image_path.parent), image_path.stem.lower())
+                if target_key in seen_targets:
+                    continue
+                seen_targets.add(target_key)
+
+                file_exists = self._image_exists_for_type(
+                    fileitem=base_item,
+                    image_path=image_path,
+                    metadata_type=metadata_type,
+                    directory_cache=directory_cache,
+                )
+                if self._should_scrape(option, file_exists, overwrite):
+                    pending_targets.setdefault(str(image_url), []).append(
+                        (base_item, image_path)
+                    )
+
+        for image_url, targets in pending_targets.items():
+            first_item, first_path = targets[0]
+            if len(targets) == 1:
+                self._download_and_save_image(
+                    fileitem=first_item,
+                    path=first_path,
+                    url=image_url,
+                )
+            else:
+                self._download_and_save_image(
+                    fileitem=first_item,
+                    path=first_path,
+                    url=image_url,
+                    reuse_targets=targets[1:],
+                )
 
     def scrape_metadata(
             self,

app/chain/storage.py

@@ -75,6 +75,16 @@ class StorageChain(ChainBase):
         """
         return cast(Optional[_SchemaFileItem], self.run_module("upload_file", fileitem=fileitem, path=path, new_name=new_name))
 
+    def link_file(self, fileitem: _SchemaFileItem, target_file: Path) -> Optional[bool]:
+        """
+        尝试在同一存储中为文件创建硬链接。
+
+        :param fileitem: 已存在的源文件项
+        :param target_file: 目标存储中的完整文件路径
+        :return: 存储实现报告的成功标记；未提供硬链接能力时返回 None
+        """
+        return cast(Optional[bool], self.run_module("link_file", fileitem=fileitem, target_file=target_file))
+
     def delete_file(self, fileitem: _SchemaFileItem) -> Optional[bool]:
         """
         删除文件或目录

app/domain/artwork.py

@@ -0,0 +1,33 @@
+"""
+媒体图片内容与落盘格式的纯转换规则。
+"""
+
+from io import BytesIO
+
+from PIL import Image
+
+
+def normalize_artwork(content: bytes) -> tuple[bytes, str]:
+    """
+    验证图片并按实际内容选择媒体服务器通用的文件后缀。
+
+    JPEG 和 PNG 保留原始字节，WebP 无损转为 PNG，以保留透明度并避免远程
+    URL 后缀决定本地格式。本函数只处理内存数据，不访问文件或网络。
+
+    :param content: 完整下载的图片字节
+    :return: 规范图片字节与小写后缀；JPEG 统一使用 ``.jpg``
+    :raises ValueError: 图片无效或格式不是 JPEG、PNG、WebP
+    """
+    try:
+        with Image.open(BytesIO(content)) as image:
+            image_format = image.format
+            if image_format in {"JPEG", "PNG"}:
+                image.verify()
+                return content, ".jpg" if image_format == "JPEG" else ".png"
+            if image_format != "WEBP":
+                raise ValueError(f"不支持的媒体图片格式：{image_format}")
+            with BytesIO() as output:
+                image.save(output, format="PNG")
+                return output.getvalue(), ".png"
+    except (OSError, SyntaxError) as err:
+        raise ValueError("下载内容不是有效的媒体图片") from err

app/modules/fanart/__init__.py

@@ -6,6 +6,7 @@ import asyncio
 import hashlib
 from pathlib import Path
 from typing import Any, Optional, Tuple, Union, cast
+from urllib.parse import urlsplit
 
 from app.adapters.network.http import AsyncRequestUtils, RequestUtils
 from app.domain.context import MediaInfo
@@ -217,6 +218,12 @@ class FanartModule(_ModuleBase):
             if not isinstance(images, list):
                 continue
 
+            if (
+                name.lower() in {"show4kbackground", "movie4kbackground"}
+                and not get_runtime_setting("SCRAP_4K_BACKGROUND", True)
+            ):
+                continue
+
             image_name = self.__name(name)
             if image_name.startswith("season"):
                 image_type = image_name[6:]
@@ -234,14 +241,14 @@ class FanartModule(_ModuleBase):
                             )
                         if image_url := image_obj.get("url"):
                             ret.setdefault(
-                                f"{season_image}{Path(image_url).suffix}", image_url
+                                f"{season_image}{self.__image_suffix(image_url)}", image_url
                             )
             else:
                 if season is not None:
                     continue
                 image_obj = self.__pick_best_image(images)
                 if image_url := image_obj.get("url"):
-                    ret[f"{image_name}{Path(image_url).suffix}"] = image_url
+                    ret[f"{image_name}{self.__image_suffix(image_url)}"] = image_url
 
         return ret or None
 
@@ -293,9 +300,12 @@ class FanartModule(_ModuleBase):
     _FANART_NAME_MAP = {
         "showbackground": "fanart",
         "moviebackground": "fanart",
+        "show4kbackground": "background",
+        "movie4kbackground": "background",
         "hdtvlogo": "logo",
         "hdmovielogo": "logo",
         "movielogo": "logo",
+        "clearlogo": "logo",
         "tvposter": "poster",
         "movieposter": "poster",
         "tvthumb": "thumb",
@@ -319,6 +329,24 @@ class FanartModule(_ModuleBase):
         """
         return cls._FANART_NAME_MAP.get(fanart_name.lower(), fanart_name)
 
+    @staticmethod
+    def __image_suffix(image_url: str) -> str:
+        """
+        将 Fanart URL 的后缀归一为小写媒体图片后缀。
+
+        JPEG 的两种常见写法共享 ``.jpg`` 文件名；无法从 URL 判断格式时沿用
+        ``.jpg``，避免把查询参数或未知后缀写入媒体目录。
+
+        :param image_url: Fanart 图片地址
+        :return: 归一后的图片后缀
+        """
+        suffix = Path(urlsplit(image_url).path).suffix.lower()
+        if suffix == ".jpeg":
+            return ".jpg"
+        if suffix in {".jpg", ".png", ".webp"}:
+            return suffix
+        return ".jpg"
+
     @classmethod
     @cached(maxsize=get_runtime_setting('CONF').fanart, ttl=get_runtime_setting('CONF').meta, shared_key="get")
     def __request_fanart(

app/modules/filemanager/module.py

@@ -444,6 +444,21 @@ class FileManagerModule(_ModuleBase):
             return None
         return storage_oper.upload(fileitem, path, new_name)
 
+    def link_file(self, fileitem: FileItem, target_file: Path) -> bool:
+        """
+        在源文件所属存储内创建硬链接。
+
+        :param fileitem: 已存在的源文件项
+        :param target_file: 目标存储中的完整文件路径
+        :return: 存储实现完成链接时返回 True，否则返回 False
+        """
+        if not fileitem.storage or fileitem.storage not in self._support_storages:
+            return False
+        storage_oper = self.__get_storage_oper(fileitem.storage, "link")
+        if not storage_oper:
+            return False
+        return bool(storage_oper.link(fileitem, target_file))
+
     def get_file_item(self, storage: str, path: Path) -> Optional[FileItem]:
         """
         根据路径严格查询文件或目录。

app/runtime/config.py

@@ -342,6 +342,14 @@ class ConfigModel(BaseModel):
         default=True,
         description='季图片是否同时写入剧集根目录；关闭后仅写季目录，不清理已有图片',
     )
+    SCRAP_IMAGE_ALIAS: bool = Field(
+        default=True,
+        description='是否为背景图和缩略图写入兼容别名文件；关闭后每张图片只写入来源名称',
+    )
+    SCRAP_4K_BACKGROUND: bool = Field(
+        default=True,
+        description='是否获取 Fanart 的 show4kbackground 和 movie4kbackground 图片',
+    )
     ANIME_GENREIDS: List[int] = Field(default=[16], description='电视剧动漫的分类genre_ids')
 
     # TMDB配置

app/runtime/extensions/module/contracts.py

@@ -341,6 +341,15 @@ _METHOD_CONTRACTS = {
         aggregation=ModuleResultAggregation.FIRST_NON_EMPTY,
         required_parameters=("fileitem", "path", "new_name"),
     ),
+    "link_file": ModuleMethodContract(
+        family="storage",
+        input_contract="StorageLinkRequest",
+        result_contract="bool | None",
+        result_shape=ModuleResultShape.BOOLEAN,
+        aggregation=ModuleResultAggregation.FIRST_NON_EMPTY,
+        required_parameters=("fileitem", "target_file"),
+        public_to_plugins=False,
+    ),
     "list_files": ModuleMethodContract(
         family="storage",
         input_contract="StorageListRequest",

app/startup/composition/configuration.py

@@ -277,6 +277,8 @@ def build_chain_runtime_config(settings: Settings) -> ChainRuntimeConfig:
         data_cleanup_outbox_dead_days=settings.DATA_CLEANUP_OUTBOX_DEAD_DAYS,
         download_subtitle=settings.DOWNLOAD_SUBTITLE,
         scrap_season_image_to_root=settings.SCRAP_SEASON_IMAGE_TO_ROOT,
+        scrap_image_alias=settings.SCRAP_IMAGE_ALIAS,
+        scrap_4k_background=settings.SCRAP_4K_BACKGROUND,
         music_metadata_to_simplified=settings.MUSIC_METADATA_TO_SIMPLIFIED,
         music_lyrics_to_simplified=settings.MUSIC_LYRICS_TO_SIMPLIFIED,
         music_cue_enable=settings.MUSIC_CUE_ENABLE,

docs/architecture-overview.md

@@ -193,7 +193,7 @@ SDK/Compat。领域查询与写入继续使用 Application 所属的冻结 DTO/P
 | 包 | 职责 | 代表性文件 |
 |---|---|---|
 | `app/foundation/` | 无状态、无配置、无 I/O 的底层原语：反射/动态导入、加密、DOM、单例、文本、URL、版本比较 | `reflection.py`、`crypto.py`、`singleton.py` |
-| `app/domain/` | 纯 MoviePilot 业务语义：媒体上下文、识别解析、站点状态解释、磁力语义、NFO 刮削 | `context.py`、`metainfo.py`、`meta/`、`scraper.py` |
+| `app/domain/` | 纯 MoviePilot 业务语义：媒体上下文、识别解析、站点状态解释、磁力语义、NFO 刮削、图片内容与格式归一 | `context.py`、`metainfo.py`、`meta/`、`scraper.py`、`artwork.py` |
 | `app/runtime/` | 进程级运行机制：配置、进程拓扑、事件、完整日志、缓存契约与内存后端、运行依赖 profile 与原生载荷激活检测、任务所有权、执行/关联上下文、并发、调度、限流、本地化、GC、重启状态 | `config.py`、`events.py`、`event/`、`dependencies/`、`tasks.py`、`execution.py`、`correlation.py`、`log.py`、`cache.py` |
 | `app/runtime/extensions/` | 模块 / 插件 / 配置化服务 / 托管资源的发现、注册与生命周期适配；管理器归入对应主题包，旧插件路径只由 Compat 精确映射 | `module/manager.py`、`plugin/manager.py`、`service.py` |
 | `app/runtime/compat/` | 仅标准库的精确旧模块、包与符号导入路由；不是业务实现，也不是通用 re-export 层 | `manifest.py`、`imports.py` |
@@ -817,7 +817,7 @@ flowchart LR
 | 指标 | 当前值 |
 |---|---:|
 | Python 模块 | 1127 |
-| 内部导入边 | 9,640 |
+| 内部导入边 | 9,641 |
 | 非平凡 SCC | 1（29 个模块的 TMDB vendor 精确 containment） |
 | Application / Chain 具体 Adapter 直连 | 0 / 0 |
 | Direct egress | 51（精确 policy 准入，Application / Chain 均为 0） |

tests/fixtures/architecture/complexity-baseline.json

@@ -137,7 +137,6 @@
       "app/agent/policy/sanitizer.py:sanitize_for_host": 6,
       "app/api/endpoints/transfer.py:_execute_manual_transfer": 6,
       "app/application/rss.py:RssHelper._parse_impl": 7,
-      "app/chain/scraping.py:ScrapingChain._download_and_save_image": 7,
       "app/chain/scraping.py:ScrapingChain.scrape_metadata_event": 6,
       "app/chain/subscribe/match.py:SubscribeMatchOwner._execute_match": 7,
       "app/chain/subscribe/query.py:SubscribeQueryOwner.subscribe_files_info": 6,

tests/fixtures/architecture/dependency-baseline.json

@@ -961,8 +961,8 @@
       "runtime_only": true
     }
   },
-  "edge_count": 9640,
-  "edge_sha256": "db43a5eb49af3c4c08dff6b6238b1c1a9ec0a408d1c79315a0ac0e720857722d",
+  "edge_count": 9641,
+  "edge_sha256": "c8b2f2c58802b0729444bbb2b8f373722179116733442c513270661a61d8af88",
   "edges": [
     "app -> app.foundation",
     "app -> app.foundation.environment",
@@ -5111,6 +5111,7 @@
     "app.chain.scraping -> app.chain.media",
     "app.chain.scraping -> app.chain.storage",
     "app.chain.scraping -> app.domain",
+    "app.chain.scraping -> app.domain.artwork",
     "app.chain.scraping -> app.domain.context",
     "app.chain.scraping -> app.domain.meta",
     "app.chain.scraping -> app.domain.meta.metabase",
@@ -10605,7 +10606,7 @@
     "app.workflow.actions.transfer_file -> app.workflow",
     "app.workflow.actions.transfer_file -> app.workflow.actions"
   ],
-  "module_count": 1127,
+  "module_count": 1128,
   "modules": [
     "app",
     "app.adapters",
@@ -11270,6 +11271,7 @@
     "app.doctor.models",
     "app.doctor.runner",
     "app.domain",
+    "app.domain.artwork",
     "app.domain.classification",
     "app.domain.classification.conditions",
     "app.domain.classification.evaluator",

tests/fixtures/architecture/runtime-contract-baseline.json

@@ -6374,6 +6374,26 @@
       "timeout_policy": "caller_budget",
       "version": 1
     },
+    "link_file": {
+      "aggregation": "first_non_empty",
+      "error_policy": "isolate_provider",
+      "execution": "sync_or_async",
+      "family": "storage",
+      "input_contract": "StorageLinkRequest",
+      "optional_parameters": [],
+      "plugin_short_circuit": true,
+      "public_to_plugins": false,
+      "required_parameters": [
+        "fileitem",
+        "target_file"
+      ],
+      "result_contract": "bool | None",
+      "result_shape": "boolean",
+      "supports_async": true,
+      "supports_sync": true,
+      "timeout_policy": "caller_budget",
+      "version": 1
+    },
     "list_files": {
       "aggregation": "ordered_list_merge",
       "error_policy": "isolate_provider",
@@ -8443,10 +8463,10 @@
     }
   },
   "run_module": {
-    "call_count": 272,
+    "call_count": 273,
     "dynamic_call_count": 0,
     "dynamic_calls": [],
-    "method_count": 220,
+    "method_count": 221,
     "methods": {
       "anilist_credits": [
         {
@@ -9396,6 +9416,13 @@
           "mode": "sync"
         }
       ],
+      "link_file": [
+        {
+          "caller": "app.chain.storage",
+          "count": 1,
+          "mode": "sync"
+        }
+      ],
       "list_files": [
         {
           "caller": "app.chain.storage",

tests/fixtures/architecture/startup-performance-baseline.json

@@ -6,7 +6,7 @@
   "repeat": 3,
   "targets": {
     "app.startup.lifecycle": {
-      "loaded_app_module_count": 590,
+      "loaded_app_module_count": 591,
       "max_ms": 1293.338,
       "median_ms": 1156.239,
       "min_ms": 1102.806,
@@ -17,7 +17,7 @@
       ]
     },
     "app.factory": {
-      "loaded_app_module_count": 604,
+      "loaded_app_module_count": 605,
       "max_ms": 1127.911,
       "median_ms": 1122.382,
       "min_ms": 1119.221,
@@ -28,7 +28,7 @@
       ]
     },
     "app.main": {
-      "loaded_app_module_count": 606,
+      "loaded_app_module_count": 607,
       "max_ms": 1188.652,
       "median_ms": 1183.509,
       "min_ms": 1174.522,

tests/test_artwork.py

@@ -0,0 +1,49 @@
+"""
+验证媒体图片格式识别、WebP 转换和无效内容拒绝。
+"""
+
+from io import BytesIO
+
+import pytest
+from PIL import Image
+
+from app.domain.artwork import normalize_artwork
+
+
+@pytest.mark.parametrize("image_format,expected_suffix", [("JPEG", ".jpg"), ("PNG", ".png"), ("WEBP", ".png")])
+def test_normalize_artwork_uses_actual_format(image_format, expected_suffix) -> None:
+    """
+    扩展名由实际图片内容决定；JPEG 和 PNG 不重编码，WebP 转成可读取的 PNG。
+    """
+    source = BytesIO()
+    Image.new("RGB", (2, 2), color="blue").save(source, format=image_format)
+    normalized, suffix = normalize_artwork(source.getvalue())
+
+    assert suffix == expected_suffix
+    if image_format != "WEBP":
+        assert normalized == source.getvalue()
+    with Image.open(BytesIO(normalized)) as image:
+        assert image.format == ("JPEG" if suffix == ".jpg" else "PNG")
+        assert image.size == (2, 2)
+
+
+def test_webp_conversion_preserves_transparency() -> None:
+    """
+    透明 WebP 转成 PNG 后保留透明度，避免 Logo 出现不透明底色。
+    """
+    source = BytesIO()
+    Image.new("RGBA", (2, 2), color=(10, 20, 30, 0)).save(source, format="WEBP", lossless=True)
+    normalized, suffix = normalize_artwork(source.getvalue())
+
+    assert suffix == ".png"
+    with Image.open(BytesIO(normalized)) as image:
+        assert image.convert("RGBA").getpixel((0, 0))[3] == 0
+
+
+@pytest.mark.parametrize("content", [b"", b"<html>not artwork</html>", b"RIFF0000WEBPbroken"])
+def test_normalize_artwork_rejects_invalid_content(content) -> None:
+    """
+    无效响应不能以图片扩展名写入媒体目录。
+    """
+    with pytest.raises(ValueError):
+        normalize_artwork(content)

tests/test_fanart_module.py

@@ -6,7 +6,11 @@ import asyncio
 from types import SimpleNamespace
 from unittest.mock import Mock, patch
 
+import pytest
+
+from app.domain.context import MediaInfo
 from app.modules.fanart import FanartModule
+from app.runtime.config import settings
 from app.runtime.tasks import TaskRegistry
 
 
@@ -78,8 +82,6 @@ def test_fanart_cache_key_follows_api_key(monkeypatch) -> None:
     """
     同一媒体在 API Key 不变时命中缓存，换 Key 后重新请求而不依赖清空缓存。
     """
-    from app.domain.context import MediaInfo
-    from app.runtime.config import settings
     from app.schemas.types import MediaType
 
     requested_urls: list[str] = []
@@ -90,7 +92,9 @@ def test_fanart_cache_key_follows_api_key(monkeypatch) -> None:
         """
 
         def __init__(self, **_kwargs) -> None:
-            pass
+            """
+            接受与真实客户端相同的配置关键字，不创建网络资源。
+            """
 
         def get_res(self, url: str, **_kwargs) -> _FakeFanartResponse:
             """
@@ -125,3 +129,42 @@ def test_fanart_init_module_keeps_request_cache() -> None:
         module.init_module()
 
     clear_cache.assert_not_called()
+
+
+@pytest.mark.parametrize("enabled", [True, False])
+@pytest.mark.parametrize("name", ["show4kbackground", "movie4kbackground"])
+def test_fanart_4k_background_has_standard_name_and_independent_switch(monkeypatch, enabled, name) -> None:
+    """
+    4K 背景使用可读取的 background 名称，独立开关不影响普通背景及其它艺术图。
+    """
+    monkeypatch.setattr(settings, "SCRAP_4K_BACKGROUND", enabled)
+    module = FanartModule()
+    result = {
+        "showbackground": [{"url": "https://artwork.invalid/ordinary.jpeg?token=value", "lang": "en"}],
+        name: [{"url": "https://artwork.invalid/4k.JPG", "lang": "en"}],
+        "clearlogo": [{"url": "https://artwork.invalid/logo.PNG", "lang": "en"}],
+    }
+
+    images = module._FanartModule__extract_images(MediaInfo(), result)
+
+    assert images["fanart.jpg"] == "https://artwork.invalid/ordinary.jpeg?token=value"
+    assert images["logo.png"] == "https://artwork.invalid/logo.PNG"
+    if enabled:
+        assert images["background.jpg"] == "https://artwork.invalid/4k.JPG"
+    else:
+        assert "background.jpg" not in images
+    assert not any(key.startswith(("show4k", "movie4k", "clearlogo")) for key in images)
+
+
+@pytest.mark.parametrize("season,prefix", [(1, "season01"), (0, "season-specials")])
+def test_fanart_season_image_suffix_is_normalized(season, prefix) -> None:
+    """
+    普通季与特别篇继续保留前缀约定，并剥离 URL 查询参数及 JPEG 后缀差异。
+    """
+    images = FanartModule()._FanartModule__extract_images(
+        MediaInfo(),
+        {"seasonposter": [{"url": "https://artwork.invalid/season.JPEG?key=value", "season": str(season)}]},
+        season=season,
+    )
+
+    assert images == {f"{prefix}-poster.jpg": "https://artwork.invalid/season.JPEG?key=value"}

tests/test_mediascrape.py

@@ -3,11 +3,13 @@
 """
 
 from dataclasses import replace
+from io import BytesIO
 from pathlib import Path
 from types import SimpleNamespace
 from unittest.mock import MagicMock, patch
 
 import pytest
+from PIL import Image
 
 # 使用导入期替身隔离站点与配置依赖，因此导入位置受控。
 # 测试先准备环境与依赖替身，再导入刮削模块。
@@ -395,7 +397,7 @@ def test_scrape_images_mapping(image_context) -> None:
 
     image_context.media_chain._scrape_images_generic(fileitem, mediainfo, ScrapingTarget.MOVIE)
 
-    # 图片类型映射与别名都应触发下载。
+    # 图片类型映射触发下载，别名目标复用同一个下载调用。
     calls = image_context.media_chain._download_and_save_image.call_args_list
     urls = [call.kwargs["url"] for call in calls]
     paths = [call.kwargs["path"] for call in calls]
@@ -404,7 +406,10 @@ def test_scrape_images_mapping(image_context) -> None:
     assert "http://logo" in urls
     # fanart.jpg 同时生成 backdrop.jpg 别名。
     assert Path("/movies/Avatar/fanart.jpg") in paths
-    assert Path("/movies/Avatar/backdrop.jpg") in paths
+    fanart_call = next(call for call in calls if call.kwargs["url"] == "http://fanart")
+    assert fanart_call.kwargs["reuse_targets"] == [
+        (fileitem, Path("/movies/Avatar/backdrop.jpg"))
+    ]
 
 
 def test_scrape_images_season_filter(image_context) -> None:
@@ -425,11 +430,11 @@ def test_scrape_images_season_filter(image_context) -> None:
     image_context.media_chain._scrape_images_generic(fileitem, mediainfo, ScrapingTarget.SEASON, season_number=1)
 
     calls = image_context.media_chain._download_and_save_image.call_args_list
-    assert len(calls) == 2
-    assert all((call.kwargs["url"] == "http://season01" for call in calls))
-    assert [call.kwargs["path"] for call in calls] == [
-        Path("/tv/Show/season01-poster.jpg"),
-        Path("/tv/Show/Season 1/poster.jpg"),
+    assert len(calls) == 1
+    assert calls[0].kwargs["url"] == "http://season01"
+    assert calls[0].kwargs["path"] == Path("/tv/Show/season01-poster.jpg")
+    assert calls[0].kwargs["reuse_targets"] == [
+        (fileitem, Path("/tv/Show/Season 1/poster.jpg"))
     ]
 
 
@@ -448,7 +453,9 @@ def test_season_image_missing_only_checks_each_target(image_context, write_root,
     existing = {root_path: root_exists, season_path: season_exists}
     chain.metadata_img.return_value = {"season01-poster.jpg": "http://season01"}
     chain.scraping_policies = ScrapingConfig({})
-    chain.storagechain.get_file_item.side_effect = lambda *, storage, path: existing[path]
+    chain.storagechain.get_file_item.side_effect = lambda *, storage, path: (
+        FileItem(path=str(path), name=path.name, storage=storage, type="file") if existing[path] else None
+    )
 
     chain._scrape_images_generic(
         season, MediaInfo(), ScrapingTarget.SEASON, parent_fileitem=parent, season_number=1
@@ -456,9 +463,11 @@ def test_season_image_missing_only_checks_each_target(image_context, write_root,
 
     selected = [root_path, season_path] if write_root else [season_path]
     assert [call.kwargs["path"] for call in chain.storagechain.get_file_item.call_args_list] == selected
-    assert [call.kwargs["path"] for call in chain._download_and_save_image.call_args_list] == [
-        path for path in selected if not existing[path]
-    ]
+    actual_targets = []
+    for call in chain._download_and_save_image.call_args_list:
+        actual_targets.append(call.kwargs["path"])
+        actual_targets.extend(path for _, path in call.kwargs.get("reuse_targets", []))
+    assert actual_targets == [path for path in selected if not existing[path]]
 
 
 @pytest.mark.parametrize("season_number,filename", [(1, "season01-thumb.jpg"), (0, "season-specials-thumb.jpg")])
@@ -472,11 +481,14 @@ def test_season_image_root_disabled_preserves_policy_and_aliases(image_context,
     season = FileItem(path="/tv/Show/Season 1", name="Season 1", type="dir", storage="local")
     chain.metadata_img.return_value = {filename: "http://thumb"}
     chain.scraping_policies = ScrapingConfig({"season_thumb": policy})
-    chain.storagechain.get_file_item.return_value = True
+    chain.storagechain.get_file_item.return_value = FileItem(path="/tv/Show/Season 1/thumb.jpg", type="file")
 
     chain._scrape_images_generic(season, MediaInfo(), ScrapingTarget.SEASON, season_number=season_number)
 
-    paths = [call.kwargs["path"] for call in chain._download_and_save_image.call_args_list]
+    paths = []
+    for call in chain._download_and_save_image.call_args_list:
+        paths.append(call.kwargs["path"])
+        paths.extend(path for _, path in call.kwargs.get("reuse_targets", []))
     assert paths == ([] if policy == ScrapingPolicy.SKIP else [
         Path("/tv/Show/Season 1/thumb.jpg"),
         Path("/tv/Show/Season 1/landscape.jpg"),
@@ -510,7 +522,14 @@ def test_scrape_movie_file_images_when_initialized_directly(image_context) -> No
     # 海报不生成别名，背景图生成 fanart 别名。
     assert Path("/movies/Avatar/poster.jpg") in paths
     assert Path("/movies/Avatar/backdrop.jpg") in paths
-    assert Path("/movies/Avatar/fanart.jpg") in paths
+    backdrop_call = next(
+        call
+        for call in image_context.media_chain._download_and_save_image.call_args_list
+        if call.kwargs["path"] == Path("/movies/Avatar/backdrop.jpg")
+    )
+    assert backdrop_call.kwargs["reuse_targets"] == [
+        (parent_item, Path("/movies/Avatar/fanart.jpg"))
+    ]
 
 
 def test_scrape_episode_thumb_image_path(image_context) -> None:
@@ -629,6 +648,136 @@ def test_expand_with_aliases_respects_skip_policy(image_context) -> None:
     assert len(expanded) == 1
 
 
+@pytest.mark.parametrize("stem", ["backdrop", "fanart", "thumb", "landscape"])
+def test_image_alias_switch_preserves_only_source_target(image_context, stem) -> None:
+    """
+    关闭别名后保留本体文件，避免跳过背景图或缩略图本身。
+    """
+    chain = image_context.media_chain
+    chain.runtime_config = replace(chain.runtime_config, scrap_image_alias=False)
+    directory = FileItem(path="/movies/Avatar", name="Avatar", type="dir", storage="local")
+    targets = [(directory, Path(f"/movies/Avatar/{stem}.jpg"))]
+
+    assert chain._expand_with_aliases(targets, ScrapingTarget.MOVIE) == targets
+
+
+@pytest.mark.parametrize("image_name,metadata_type", [
+    ("poster.jpg", ScrapingMetadata.POSTER),
+    ("clearlogo.PNG", ScrapingMetadata.LOGO),
+    ("show4kbackground.jpeg", ScrapingMetadata.BACKDROP),
+    ("season01-poster.jpg", ScrapingMetadata.POSTER),
+    ("season-specials-thumb.jpg", ScrapingMetadata.THUMB),
+    ("unregistered-poster.jpg", None),
+    ("background-copy.jpg", None),
+    ("logo.backup.png", None),
+])
+def test_image_type_matching_is_exact(image_context, image_name, metadata_type) -> None:
+    """
+    只识别精确类型及已登记的季前缀，未知名称不能因包含关键字而写入媒体目录。
+    """
+    actual_type, _ = image_context.media_chain._get_image_metadata_type(image_name, ScrapingTarget.MOVIE)
+    assert actual_type is metadata_type
+
+
+@pytest.mark.parametrize("image_name", ["episode-thumb.jpg", "S01E01.jpg", "S01E01-thumb.jpg"])
+def test_episode_thumbnail_special_names_are_preserved(image_context, image_name) -> None:
+    """
+    精确类型匹配保留来源的 episode-thumb 和单集媒体同名缩略图约定。
+    """
+    fileitem = FileItem(path="/tv/Show/S01E01.mkv", name="S01E01.mkv", type="file", storage="local")
+    actual_type, _ = image_context.media_chain._get_image_metadata_type(
+        image_name, ScrapingTarget.EPISODE, fileitem
+    )
+    assert actual_type is ScrapingMetadata.THUMB
+
+
+@pytest.mark.parametrize("existing_name", ["poster.JPG", "poster.jpeg", "POSTER.PNG", "poster.webp"])
+def test_missing_only_recognizes_other_extensions_and_case(image_context, existing_name) -> None:
+    """
+    已有同类型图片时仅缺失策略不因后缀、大小写或远程 URL 变化重复下载。
+    """
+    chain = image_context.media_chain
+    directory = FileItem(path="/movies/Avatar", name="Avatar", type="dir", storage="local")
+    chain.metadata_img.return_value = {"poster.JPEG": "http://poster"}
+    chain.scraping_policies = ScrapingConfig({})
+    chain.storagechain.get_file_item.return_value = None
+    chain.storagechain.list_files.return_value = [
+        FileItem(path=f"/movies/Avatar/{existing_name}", name=existing_name, type="file", storage="local")
+    ]
+
+    chain._scrape_images_generic(directory, MediaInfo(), ScrapingTarget.MOVIE)
+
+    chain._download_and_save_image.assert_not_called()
+    chain.storagechain.list_files.assert_called_once_with(directory, recursion=False)
+
+
+@pytest.mark.parametrize("metadata_name,existing_name", [
+    ("backdrop", "fanart.PNG"),
+    ("thumb", "LANDSCAPE.jpeg"),
+    ("logo", "clearlogo.PNG"),
+    ("background", "backdrop.jpg"),
+    ("backdrop", "show4kbackground.jpeg"),
+])
+def test_missing_only_recognizes_same_type_aliases(image_context, metadata_name, existing_name) -> None:
+    """
+    同一图片类型的其它合法名称也满足补缺条件，避免跨别名反复获取图片。
+    """
+    chain = image_context.media_chain
+    directory = FileItem(path="/movies/Avatar", name="Avatar", type="dir", storage="local")
+    chain.metadata_img.return_value = {f"{metadata_name}.jpg": "http://artwork"}
+    chain.scraping_policies = ScrapingConfig({})
+    chain.storagechain.get_file_item.return_value = None
+    chain.storagechain.list_files.return_value = [
+        FileItem(path=f"/movies/Avatar/{existing_name}", name=existing_name, type="file", storage="local")
+    ]
+
+    chain._scrape_images_generic(directory, MediaInfo(), ScrapingTarget.MOVIE)
+
+    chain._download_and_save_image.assert_not_called()
+
+
+def test_unrelated_file_suffix_does_not_satisfy_missing_image(image_context) -> None:
+    """
+    同名文本和目录不应被错误视作已经刮削的图片。
+    """
+    chain = image_context.media_chain
+    directory = FileItem(path="/movies/Avatar", name="Avatar", type="dir", storage="local")
+    chain.metadata_img.return_value = {"poster.jpeg": "http://poster"}
+    chain.scraping_policies = ScrapingConfig({})
+    chain.storagechain.get_file_item.return_value = FileItem(name="poster.jpg", type="dir", storage="local")
+    chain.storagechain.list_files.return_value = [
+        FileItem(name="poster.txt", type="file", storage="local"),
+        FileItem(name="poster.png", type="dir", storage="local"),
+    ]
+
+    chain._scrape_images_generic(directory, MediaInfo(), ScrapingTarget.MOVIE)
+
+    chain._download_and_save_image.assert_called_once_with(
+        fileitem=directory, path=Path("/movies/Avatar/poster.jpg"), url="http://poster"
+    )
+
+
+def test_same_url_targets_share_one_download_plan(image_context) -> None:
+    """
+    不同来源名称中的相同 URL 也只下载一次，并合并重复别名目标。
+    """
+    chain = image_context.media_chain
+    directory = FileItem(path="/movies/Avatar", name="Avatar", type="dir", storage="local")
+    chain.metadata_img.return_value = {"backdrop.jpeg": "http://shared", "fanart.JPG": "http://shared"}
+    chain.scraping_policies = ScrapingConfig({})
+    chain.storagechain.get_file_item.return_value = None
+    chain.storagechain.list_files.return_value = []
+
+    chain._scrape_images_generic(directory, MediaInfo(), ScrapingTarget.MOVIE)
+
+    chain._download_and_save_image.assert_called_once_with(
+        fileitem=directory,
+        path=Path("/movies/Avatar/backdrop.jpg"),
+        url="http://shared",
+        reuse_targets=[(directory, Path("/movies/Avatar/fanart.jpg"))],
+    )
+
+
 def test_season_backdrop_path(image_context) -> None:
     """
     季背景图保存到季目录。
@@ -644,53 +793,98 @@ def test_season_backdrop_path(image_context) -> None:
     assert target_path == Path("/tv/Show/Season 1/backdrop.jpg")
 
 
+@pytest.mark.parametrize("hardlink", [True, False])
+@pytest.mark.parametrize("image_format,expected_suffix", [("JPEG", ".jpg"), ("PNG", ".png"), ("WEBP", ".png")])
 @patch("app.chain.scraping._scraping_http_snapshot")
-@patch("app.chain.scraping.NamedTemporaryFile")
-@patch("app.chain.scraping.Path.chmod")
-def test_download_and_save_image(mock_chmod, mock_temp_file, mock_request_utils, image_context) -> None:
-    # 恢复真实下载方法，只替换网络和文件系统边界。
+def test_download_and_save_image_reuses_one_download(
+        mock_http, image_context, tmp_path, hardlink, image_format, expected_suffix,
+) -> None:
     """
-    下载图片时分块写入临时文件，再通过存储接口上传。
+    别名仅请求一次远端，支持时共享 inode，不支持时从同一下载副本上传并清理临时文件。
     """
-    image_context.media_chain = ScrapingChain()
-    image_context.media_chain._download_and_save_image = image_context.original_download
-    image_context.media_chain.storagechain = MagicMock()
+    chain = image_context.media_chain
+    chain._download_and_save_image = image_context.original_download
+    fileitem = FileItem(path=str(tmp_path), name=tmp_path.name, type="dir", storage="local")
+    content_buffer = BytesIO()
+    Image.new("RGB", (2, 2), color="blue").save(content_buffer, format=image_format)
+    content = content_buffer.getvalue()
+    stream = mock_http.return_value.stream.return_value.__enter__.return_value
+    stream.status_code = 200
+    stream.iter_content.return_value = [content[:20], b"", content[20:]]
+    uploaded_paths = []
 
-    fileitem = FileItem(path="/movies/Avatar", name="Avatar", type="dir", storage="local")
-    target_path = Path("/movies/Avatar/poster.jpg")
-    url = "http://poster"
+    def upload(*, fileitem, path, new_name):
+        """
+        模拟会消费本地源文件的存储上传，并保留真实目标文件用于内容检查。
+        """
+        uploaded_paths.append(path)
+        destination = Path(fileitem.path) / new_name
+        destination.write_bytes(path.read_bytes())
+        path.unlink()
+        return FileItem(path=str(destination), name=new_name, type="file", storage="local")
 
-    # 模拟临时文件，避免写入真实目标路径。
-    tmp_mock = MagicMock()
-    tmp_mock.name = "/tmp/mockfile"
-    mock_temp_file.return_value.__enter__.return_value = tmp_mock
+    def link(source_item, target_path):
+        """
+        使用临时目录中的真实硬链接，或模拟跨设备失败。
+        """
+        if not hardlink:
+            raise OSError("cross-device link")
+        target_path.hardlink_to(source_item.path)
+        return True
 
-    # 模拟分块响应，不发起真实网络请求。
-    mock_stream = MagicMock()
-    mock_stream.status_code = 200
-    mock_stream.iter_content.return_value = [b"data1", b"data2"]
-
-    mock_instance = mock_request_utils.return_value
-    mock_instance.get_stream.return_value.__enter__.return_value = mock_stream
-    mock_instance.stream.return_value.__enter__.return_value = mock_stream
-
-    image_context.media_chain.storagechain.upload_file.return_value = fileitem
-
-    image_context.media_chain._download_and_save_image(fileitem, target_path, url)
-
-    mock_instance.stream.assert_called_with(
-        url,
-        proxies=image_context.media_chain.runtime_config.proxy,
-        ua=image_context.media_chain.runtime_config.normal_user_agent,
+    chain.storagechain.upload_file.side_effect = upload
+    chain.storagechain.link_file.side_effect = link
+    chain._download_and_save_image(
+        fileitem,
+        tmp_path / "backdrop.jpg",
+        "https://artwork.invalid/image.jpeg",
+        [(fileitem, tmp_path / "fanart.jpg")],
     )
-    mock_temp_file.assert_called_once_with(delete=False, suffix=".jpg")
-    tmp_mock.write.assert_any_call(b"data1")
-    tmp_mock.write.assert_any_call(b"data2")
-    mock_chmod.assert_called()
-    image_context.media_chain.storagechain.upload_file.assert_called_once()
-    call_args = image_context.media_chain.storagechain.upload_file.call_args.kwargs
-    assert call_args["fileitem"] == fileitem
-    assert call_args["new_name"] == "poster.jpg"
+
+    mock_http.return_value.stream.assert_called_once()
+    first = tmp_path / f"backdrop{expected_suffix}"
+    alias = tmp_path / f"fanart{expected_suffix}"
+    assert first.read_bytes() == alias.read_bytes()
+    if image_format != "WEBP":
+        assert first.read_bytes() == content
+    with Image.open(first) as image:
+        assert image.format == ("JPEG" if expected_suffix == ".jpg" else "PNG")
+    assert (first.stat().st_ino == alias.stat().st_ino) is hardlink
+    assert chain.storagechain.upload_file.call_count == (1 if hardlink else 2)
+    assert all(not path.exists() for path in uploaded_paths)
+
+
+@patch("app.chain.scraping._scraping_http_snapshot")
+def test_image_download_failure_cleans_temp_files(mock_http, image_context, tmp_path) -> None:
+    """
+    错误响应内容不会上传，下载临时文件在格式验证失败后仍会清理。
+    """
+    chain = image_context.media_chain
+    chain._download_and_save_image = image_context.original_download
+    cleanup = MagicMock(wraps=chain._cleanup_temp_file)
+    chain._cleanup_temp_file = cleanup
+    stream = mock_http.return_value.stream.return_value.__enter__.return_value
+    stream.status_code = 200
+    stream.iter_content.return_value = [b"not an image"]
+    fileitem = FileItem(path=str(tmp_path), name=tmp_path.name, type="dir", storage="local")
+
+    chain._download_and_save_image(fileitem, tmp_path / "poster.jpg", "https://artwork.invalid/error")
+
+    chain.storagechain.upload_file.assert_not_called()
+    cleaned_paths = [call.args[0] for call in cleanup.call_args_list if call.args[0]]
+    assert cleaned_paths
+    assert all(not path.exists() for path in cleaned_paths)
+
+
+def test_image_reuse_never_links_across_storages(image_context) -> None:
+    """
+    不同存储中的相同路径不能作为硬链接目标，必须由上传边界接收本地副本。
+    """
+    source = FileItem(path="/movies/backdrop.jpg", name="backdrop.jpg", type="file", storage="local")
+    target = FileItem(path="/movies", name="movies", type="dir", storage="smb")
+
+    assert not image_context.media_chain._reuse_image_file(source, target, Path("/movies/fanart.jpg"))
+    image_context.media_chain.storagechain.link_file.assert_not_called()
 
 
 @patch("app.chain.scraping.NamedTemporaryFile")

tests/test_module_method_contracts.py

@@ -77,7 +77,7 @@ def test_contract_v2_freezes_every_observed_host_method() -> None:
     contracts = list_explicit_module_contracts()
 
     assert len(contracts) >= 211
-    host_internal_methods = {"plan_transfer", "execute_transfer_plan", "restore_search_torrent"}
+    host_internal_methods = {"plan_transfer", "execute_transfer_plan", "restore_search_torrent", "link_file"}
     for method_name, contract in contracts.items():
         assert contract.version == 1
         assert contract.input_contract != "legacy_args"

tests/test_scraping_configuration.py

@@ -20,3 +20,21 @@ def test_season_image_root_setting_reaches_chain_snapshot(monkeypatch, tmp_path,
 
     assert settings.SCRAP_SEASON_IMAGE_TO_ROOT is expected
     assert build_chain_runtime_config(settings).scrap_season_image_to_root is expected
+
+
+@pytest.mark.parametrize("env_value,expected", [(None, True), ("true", True), ("false", False)])
+@pytest.mark.parametrize("key,attribute", [
+    ("SCRAP_IMAGE_ALIAS", "scrap_image_alias"),
+    ("SCRAP_4K_BACKGROUND", "scrap_4k_background"),
+])
+def test_image_settings_reach_chain_snapshot(monkeypatch, tmp_path, key, attribute, env_value, expected) -> None:
+    """
+    图片别名和 4K 背景开关默认兼容现状，并由部署配置传入刮削链。
+    """
+    monkeypatch.delenv(key, raising=False)
+    if env_value is not None:
+        monkeypatch.setenv(key, env_value)
+    settings = Settings(_env_file=None, CONFIG_DIR=str(tmp_path), API_TOKEN="0123456789abcdef")
+
+    assert getattr(settings, key) is expected
+    assert getattr(build_chain_runtime_config(settings), attribute) is expected

tests/test_storage_chain_strict.py

@@ -9,7 +9,9 @@ import pytest
 
 from app.chain.storage import StorageChain
 from app.modules.filemanager.module import FileManagerModule
+from app.modules.filemanager.storages.local import LocalStorage
 from app.schemas.exception import StorageQueryError
+from app.schemas.file import FileItem
 
 
 def test_storage_chain_strict_query_distinguishes_absent_and_failure() -> None:
@@ -47,3 +49,42 @@ def test_filemanager_storage_query_uses_strict_storage_adapter(monkeypatch) -> N
         module.get_file_item("local", Path("/library/old.mkv"))
 
     storage.get_item.assert_not_called()
+
+
+def test_filemanager_image_link_uses_local_storage_adapter(monkeypatch, tmp_path) -> None:
+    """
+    图片复用接口通过存储适配器建立真实硬链接，避免重新上传同一份图片。
+    """
+    module = object.__new__(FileManagerModule)
+    module._support_storages = ["local"]
+    storage = object.__new__(LocalStorage)
+    lookup = Mock(return_value=storage)
+    monkeypatch.setattr(module, "_FileManagerModule__get_storage_oper", lookup)
+    source = tmp_path / "backdrop.jpg"
+    source.write_bytes(b"artwork")
+    target = tmp_path / "fanart.jpg"
+    fileitem = FileItem(storage="local", path=str(source), name=source.name, type="file")
+
+    assert module.link_file(fileitem, target) is True
+
+    lookup.assert_called_once_with("local", "link")
+    assert target.read_bytes() == source.read_bytes()
+    assert target.stat().st_ino == source.stat().st_ino
+
+
+def test_filemanager_image_link_rejects_unavailable_storage(monkeypatch) -> None:
+    """
+    存储没有注册硬链接能力时返回 False，让刮削链从下载副本回退上传。
+    """
+    module = object.__new__(FileManagerModule)
+    module._support_storages = ["local"]
+    lookup = Mock(return_value=None)
+    monkeypatch.setattr(module, "_FileManagerModule__get_storage_oper", lookup)
+    target = Path("/library/fanart.jpg")
+    source = FileItem(storage="unregistered", path="/library/backdrop.jpg", type="file")
+
+    assert module.link_file(source, target) is False
+    lookup.assert_not_called()
+
+    assert module.link_file(source.model_copy(update={"storage": "local"}), target) is False
+    lookup.assert_called_once_with("local", "link")