feat(music): preserve local covers and fall back after online scraping fails

2026-11-01 21:07:01 +08:00 · 2026-10-14 23:15:41 +08:00
parent bb86a6ebea
commit 72c8bff312
17 changed files with 442 additions and 20 deletions
--- a/app/application/settings/resources/system_settings_catalog.json
+++ b/app/application/settings/resources/system_settings_catalog.json
@@ -1075,6 +1075,12 @@
      "source_file": "app/runtime/config.py",
      "source_line": 679
    },
+    "MUSIC_COVER_FILE_PATTERNS": {
+      "description": "同目录音乐封面匹配规则，按顺序选择，逗号分隔、大小写不敏感；留空禁用本地封面",
+      "group": "music",
+      "source_file": "app/runtime/config.py",
+      "source_line": 406
+    },
    "MUSIC_COVER_PROXY": {
      "description": "音乐封面代理地址（用于解决 coverartarchive.org 无法访问导致的封面不显示问题，留空则使用官方地址）",
      "group": "music",
--- a/app/chain/scraping.py
+++ b/app/chain/scraping.py
@@ -28,12 +28,14 @@ from app.domain.context import (
 from app.domain.meta.metabase import MetaBase
 from app.domain.meta.metamusic import MetaMusic
 from app.domain.metainfo import MetaInfo, MetaInfoPath
+from app.domain.music import music_cover_priority
 from app.foundation.singleton import Singleton
 from app.foundation.text import convert
 from app.runtime.cache import cached
 from app.runtime.events import Event, eventmanager
 from app.runtime.log import logger
 from app.runtime.reload import ConfigReloadMixin
+from app.runtime.settings import get_runtime_setting
 from app.schemas.file import FileItem as _SchemaFileItem
 from app.schemas.media import resolve_media_identity
 from app.schemas.types import (
@@ -1421,6 +1423,12 @@ class ScrapingChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
    ) -> _MusicScrapeFileResult:
        """在本地音轨副本上执行刮削，并将变更后的音频和歌词写回目标存储。"""
        scrape_info = self._resolve_music_scrape_info(local_path, mediainfo, media_source=media_source)
+        if with_cover:
+            if cover is None:
+                cover_url = getattr(mediainfo, "cover_url", None) or getattr(scrape_info, "cover_url", None)
+                cover = self._download_music_cover(cover_url)
+            if not cover[0]:
+                cover = self._load_local_music_cover(fileitem)
        metadata_requested = write_tags or with_cover
        metadata_success = True
        if metadata_requested:
@@ -1597,6 +1605,41 @@ class ScrapingChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
            cover_overwrite=cover_overwrite,
        )

+    def _load_local_music_cover(self, fileitem: _SchemaFileItem) -> tuple[Optional[bytes], str]:
+        """在线封面不可用时读取同目录图片；下载经存储端口，坏图不阻断后续候选。"""
+        from PIL import Image
+
+        parent = self.storagechain.get_parent_item(fileitem)
+        if not parent:
+            return None, "image/jpeg"
+        if MetaMusic.parse_disc_dir(Path(str(parent.path)).name) is not None:
+            # 整理把碟片目录内的封面放到专辑根，刮削仍从该发行范围内取图。
+            parent = self.storagechain.get_parent_item(parent) or parent
+        patterns = get_runtime_setting("MUSIC_COVER_FILE_PATTERNS")
+        candidates = []
+        for item in self.storagechain.list_files(parent, recursion=False) or []:
+            priority = music_cover_priority(str(item.path or item.name or ""), patterns)
+            if item.type == "file" and priority is not None:
+                candidates.append((priority, str(item.path).casefold(), item))
+        for _priority, _name, item in sorted(candidates, key=lambda value: value[:2]):
+            try:
+                with TemporaryDirectory(prefix="moviepilot-music-cover-") as directory:
+                    path = self.storagechain.download_file(item, path=Path(directory))
+                    if not path:
+                        continue
+                    with Image.open(path) as picture:
+                        mime = Image.MIME.get(picture.format)
+                        picture.verify()
+                    if mime not in {"image/jpeg", "image/png", "image/webp"}:
+                        continue
+                    data = Path(path).read_bytes()
+                logger.info(f"音乐在线封面不可用，使用同目录封面：{item.path}")
+                return data, mime
+            except (OSError, ValueError) as error:
+                logger.warning(f"读取音乐本地封面失败：{item.path} - {error}")
+        logger.info(f"音乐在线及同目录封面均不可用：{fileitem.path}")
+        return None, "image/jpeg"
+
    def _scrape_music_lyrics(
            self,
            fileitem: _SchemaFileItem,
--- a/app/chain/transfer/contract.py
+++ b/app/chain/transfer/contract.py
@@ -86,6 +86,7 @@ if TYPE_CHECKING:
        _is_media_file: Callable[..., Any]
        _is_movie_year_conflict: Callable[..., Any]
        _is_music_lyrics_file: Callable[..., Any]
+        _is_music_cover_file: Callable[..., Any]
        _is_music_retry_source: Callable[..., Any]
        _is_overwrite_declined: Callable[..., Any]
        _is_primary_media_file: Callable[..., Any]
--- a/app/chain/transfer/filter.py
+++ b/app/chain/transfer/filter.py
@@ -33,12 +33,14 @@ from app.domain.music import (
    MusicDirectoryMatch,
    music_album_matches,
    music_artist_matches,
+    music_cover_priority,
    music_release_year_matches,
    music_title_matches,
    music_track_title_is_weak,
    music_version_matches,
 )
 from app.runtime.log import logger
+from app.runtime.settings import get_runtime_setting
 from app.schemas.transfer import TransferInfo
 from app.schemas.types import (
    MUSIC_ENTITY_ALBUM,
@@ -156,6 +158,13 @@ class _MusicFileFilterBase(_TransferOwnerBase):
        path = str(fileitem.path or fileitem.name or "").casefold()
        return path.endswith((".lrc", ".txt", ".lyricsfile.yaml"))

+    @staticmethod
+    def _is_music_cover_file(fileitem: FileItem) -> bool:
+        """仅接纳配置命中的音乐封面候选，实际归属由音轨批次确认。"""
+        return fileitem.type == "file" and music_cover_priority(
+            str(fileitem.path or fileitem.name or ""), get_runtime_setting("MUSIC_COVER_FILE_PATTERNS")
+        ) is not None
+
    def _is_media_file(
            self,
            fileitem: FileItem,
@@ -367,6 +376,7 @@ class FileFilterMixin(_TransferOwnerBase):
    _is_music_lyrics_file = cast(
        Any, staticmethod(_MusicFileFilterBase._is_music_lyrics_file)
    )
+    _is_music_cover_file = cast(Any, staticmethod(_MusicFileFilterBase._is_music_cover_file))
    _is_media_file = cast(Any, _MusicFileFilterBase._is_media_file)
    _is_primary_media_file = cast(Any, _MusicFileFilterBase._is_primary_media_file)
    _music_info_from_meta = cast(
--- a/app/chain/transfer/music.py
+++ b/app/chain/transfer/music.py
@@ -600,7 +600,7 @@ def prepare_music_batch_context(
        file_items: list[tuple[FileItem, bool]],
        batch_mtype: Optional[MediaType],
 ) -> MusicBatchContext:
-    """建立同目录音轨、歌词和单音轨目录的批次索引。"""
+    """建立同目录音轨、歌词、封面和单音轨目录的批次索引；歧义发行不共享封面。"""
    context = MusicBatchContext()
    if batch_mtype not in (None, MediaType.MUSIC):
        return context
@@ -635,6 +635,7 @@ def prepare_music_batch_context(
                if evidence and len(group.files) > 1 and evidence.album and (evidence.album_artist or evidence.artists):
                    context.album_main_keys.add(item_key)
    for current_item, _current_bluray_dir in file_items:
+        _bind_music_cover_context(owner, context, current_item, main_items_by_dir)
        if str(current_item.extension or "").casefold() == "cue":
            for main_key, cue_meta in context.cue_by_main_key.items():
                if (cue_meta.music_layout == "image_cue" and current_item.storage == main_key[0]
@@ -651,6 +652,19 @@ def prepare_music_batch_context(
    return context


+def _bind_music_cover_context(
+        owner: _TransferOwnerBase, context: MusicBatchContext, item: FileItem,
+        main_items_by_dir: dict[Tuple[str, str], list[FileItem]],
+) -> None:
+    """只把封面绑定到同目录的唯一物理发行，避免多专辑目录共享错图。"""
+    if not owner._is_music_cover_file(item):
+        return
+    main_items = main_items_by_dir.get(owner._get_file_parent_key(item), [])
+    groups = {context.release_by_main_key[owner._get_file_key(main)].preview_id for main in main_items}
+    if len(groups) == 1:
+        context.related_main_keys[owner._get_file_key(item)] = owner._get_file_key(main_items[0])
+
+
 def _finalize_music_group_context(
        owner: _TransferOwnerBase,
        context: MusicBatchContext,
--- a/app/chain/transfer/request.py
+++ b/app/chain/transfer/request.py
@@ -299,6 +299,7 @@ class _TransferCandidatePlanner:
                    self._chain._is_subtitle_file(item)
                    or self._chain._is_audio_file(item)
                    or self._chain._is_music_lyrics_file(item)
+                    or self._is_music_cover_candidate(item)
            ):
                extra_items_by_dir.setdefault(dir_key, []).append((item, is_bluray_dir))
        return main_items_by_dir, extra_items_by_dir
@@ -330,6 +331,7 @@ class _TransferCandidatePlanner:
                    self._chain._is_subtitle_file(item)
                    or self._chain._is_audio_file(item)
                    or self._chain._is_music_lyrics_file(item)
+                    or self._is_music_cover_candidate(item)
            ):
                continue
            if not self._is_allowed_transfer_item(item, False):
@@ -337,6 +339,31 @@ class _TransferCandidatePlanner:
            extra_items.append((item, False))
        return main_fileitems, extra_items

+    def _is_music_cover_candidate(self, item: FileItem) -> bool:
+        """封面只随音乐主文件同步，影视批次及关闭附加文件同步时不接纳图片。"""
+        return (self._sync_extra_files and not self._has_reliable_video_source()
+                and self._chain._is_music_cover_file(item))
+
+    def _infer_music_batch_type(self, items: List[Tuple[FileItem, bool]]) -> None:
+        """未指定类型的纯音频请求仍建立音乐主文件索引，混合影视目录保持原类型。"""
+        if (self._batch_mtype is None and not self._has_reliable_video_source()
+                and any(self._chain._is_audio_file(item) for item, _ in items)
+                and not any(bluray or self._chain._is_media_file(item, None) for item, bluray in items)):
+            self._batch_mtype = MediaType.MUSIC
+
+    def _append_remaining_items(
+            self, items: List[Tuple[FileItem, bool]], planned: List[Tuple[FileItem, bool]],
+            seen: set[Tuple[str, str]], inherited: Dict[Tuple[str, str], MetaBase],
+    ) -> None:
+        """保留未匹配的普通附件，音乐歌词和封面必须已获得主文件归属。"""
+        for item, is_bluray_dir in items:
+            if self._chain._is_music_cover_file(item) and self._chain._get_file_key(item) not in inherited:
+                continue
+            if (self._batch_mtype == MediaType.MUSIC and self._chain._is_music_lyrics_file(item)
+                    and self._chain._get_file_key(item) not in inherited):
+                continue
+            self._append_item(planned, seen, item, is_bluray_dir)
+
    def _plan_file_items(
            self,
            items: List[Tuple[FileItem, bool]]
@@ -347,6 +374,8 @@ class _TransferCandidatePlanner:
        if not items:
            return [], {}

+        self._infer_music_batch_type(items)
+
        download_history_repository = self._chain.download_history_repository
        inherited_map: Dict[Tuple[str, str], MetaBase] = {}
        main_items_by_dir, extra_items_by_dir = self._build_directory_index(items)
@@ -408,7 +437,7 @@ class _TransferCandidatePlanner:
            remaining = [
                item
                for item in items
-                if not (
+                if not self._chain._is_music_cover_file(item[0]) and not (
                    self._batch_mtype == MediaType.MUSIC
                    and self._chain._is_music_lyrics_file(item[0])
                )
@@ -468,6 +497,12 @@ class _TransferCandidatePlanner:
            for extra_item, extra_bluray_dir in extra_items_by_dir.get(dir_key, []):
                if self._chain._get_file_key(extra_item) in seen_file_keys:
                    continue
+                if self._is_music_cover_candidate(extra_item):
+                    # 图片没有曲名，先继承主音轨标签；批次分组再确认唯一发行归属。
+                    if self._chain._is_audio_file(main_item) and self._append_item(
+                            planned_items, seen_file_keys, extra_item, extra_bluray_dir):
+                        inherited_map[self._chain._get_file_key(extra_item)] = deepcopy(main_meta)
+                    continue
                related_main_file_key = self._chain._get_related_main_file_key(
                    extra_fileitem=extra_item,
                    main_fileitems=main_fileitems,
@@ -500,14 +535,7 @@ class _TransferCandidatePlanner:
                ):
                    inherited_map[self._chain._get_file_key(extra_item)] = deepcopy(extra_meta)

-        for item, is_bluray_dir in items:
-            if (
-                    self._batch_mtype == MediaType.MUSIC
-                    and self._chain._is_music_lyrics_file(item)
-                    and self._chain._get_file_key(item) not in inherited_map
-            ):
-                continue
-            self._append_item(planned_items, seen_file_keys, item, is_bluray_dir)
+        self._append_remaining_items(items, planned_items, seen_file_keys, inherited_map)

        return append_cue_companions(self._chain, planned_items, inherited_map, self._has_reliable_video_source(), self._transfer_exclude_words)

--- a/app/chain/transfer/workflow.py
+++ b/app/chain/transfer/workflow.py
@@ -3,7 +3,7 @@
 import re
 import uuid
 from pathlib import Path
-from typing import Any, Callable, Dict, List, Optional, Tuple, Union
+from typing import Any, Callable, Dict, Generator, List, Optional, Tuple, Union

 from app.application.audio import capture_audio_metadata
 from app.application.classification.reference import (
@@ -355,7 +355,7 @@ class TransferWorkflowOwner(_TransferOwnerBase):
                    return False
                matched_template = True
            if batch_mtype == MediaType.MUSIC:
-                if self._is_music_lyrics_file(item):
+                if self._is_music_lyrics_file(item) or self._is_music_cover_file(item):
                    return not self._is_blocked_by_exclude_words(item.path, exclude_words)
                if not self._is_media_file(item, batch_mtype) and not (
                    item.type == "file" and music_package_error(item.path or "")
@@ -363,7 +363,8 @@ class TransferWorkflowOwner(_TransferOwnerBase):
                    return False
                if not self._is_allow_filesize(item, min_filesize):
                    return False
-            elif not is_bluray_dir and not self._is_subtitle_file(item) and not self._is_audio_file(item):
+            elif (not is_bluray_dir and not self._is_subtitle_file(item) and not self._is_audio_file(item)
+                  and not (batch_mtype is None and self._is_music_cover_file(item))):
                if not self._is_media_file(item, batch_mtype):
                    return False
                if not self._is_allow_filesize(item, min_filesize):
@@ -745,7 +746,7 @@ class TransferWorkflowOwner(_TransferOwnerBase):
        )
        music_batch_context = self._prepare_music_batch_context(file_items, batch_mtype)
        try:
-            for file_item, bluray_dir in file_items:
+            for file_item, bluray_dir in self._music_transfer_items(file_items, music_batch_context):
                if runtime_stop_state.is_system_stopped:
                    raise OperationInterrupted()
                if continue_callback and not continue_callback():
@@ -972,6 +973,18 @@ class TransferWorkflowOwner(_TransferOwnerBase):
            skipped_torrents,
        )

+    def _music_transfer_items(
+            self, items: List[Tuple[FileItem, bool]], context: Any,
+    ) -> Generator[Tuple[FileItem, bool], None, None]:
+        """随主音轨逐项解析再放行封面，无法证明唯一音乐归属时不生成图片任务。"""
+        for item, bluray in items:
+            if self._is_music_cover_file(item):
+                related = context.related_main_keys.get(self._get_file_key(item))
+                if related not in context.resolved_contexts:
+                    logger.info(f"{item.name} 没有唯一且已识别的同目录音乐归属，跳过封面整理")
+                    continue
+            yield item, bluray
+
    def _cancel_unexecuted_tasks(self, tasks: list[TransferTask]) -> None:
        """撤销本次同步请求尚未执行的内存占位，让用户之后可以重新提交这些文件。"""
        for task in tasks:
--- a/app/domain/music.py
+++ b/app/domain/music.py
@@ -4,6 +4,8 @@ import re
 from copy import deepcopy
 from dataclasses import dataclass
 from datetime import date
+from fnmatch import fnmatchcase
+from pathlib import Path
 from typing import Any, Generator, Iterable, Literal, Optional
 from unicodedata import combining, normalize

@@ -12,6 +14,15 @@ from app.domain.meta.metamusic import MetaMusic, music_credit_values
 from app.foundation.text import convert as zhconv_convert
 from app.schemas.types import MUSIC_ENTITY_ALBUM, MUSIC_ENTITY_RECORDING, MediaSource, MediaType

+
+def music_cover_priority(path: str, patterns: str) -> Optional[int]:
+    """按配置顺序匹配同目录封面，忽略大小写且只允许可内嵌的常见图片格式。"""
+    name = Path(path).name.casefold()
+    if name.startswith(".") or Path(name).suffix not in {".jpg", ".jpeg", ".png", ".webp"}:
+        return None
+    return next((index for index, pattern in enumerate(patterns.split(","))
+                 if pattern.strip() and fnmatchcase(name, pattern.strip().casefold())), None)
+
 _EDITION = re.compile(
    r"\s*[\[(（【](?:[^\])）】]*\b(?:deluxe|expanded|special|limited|anniversary|remaster(?:ed)?)\b"
    r"[^\])）】]*|[^\])）】]*(?:豪华版|典藏版|纪念版|重制版|周年版)[^\])）】]*)[\])）】]",
--- a/app/modules/filemanager/transhandler.py
+++ b/app/modules/filemanager/transhandler.py
@@ -160,6 +160,8 @@ class TransHandler:
        """比较新旧音乐的实际音质，无法形成可靠结论时交回原覆盖策略。"""
        if getattr(mediainfo, "type", None) != MediaType.MUSIC or not target_item:
            return None
+        if Path(str(target_item.path or "")).suffix.casefold() in {".jpg", ".jpeg", ".png", ".webp"}:
+            return None

        source_score = getattr(meta, "audio_quality_score", 0) or getattr(
            mediainfo, "audio_quality_score", 0
@@ -422,6 +424,10 @@ class TransHandler:
            )
            if not target_directory_path:
                raise ValueError("重命名格式无效")
+            if (mediainfo.type == MediaType.MUSIC
+                    and Path(str(fileitem.path)).suffix.casefold() in {".jpg", ".jpeg", ".png", ".webp"}):
+                # 封面保持原名并放到专辑根，不套用曲序/曲名模板或进入 Disc 子目录。
+                final_target = target_directory_path / Path(str(fileitem.path)).name
            if getattr(planning_meta, "music_layout", None) == "image_cue":
                # 整轨与索引只调整专辑目录，保留原名才能让 CUE FILE 引用继续有效。
                final_target = final_target.parent / Path(str(fileitem.path)).name
@@ -537,7 +543,9 @@ class TransHandler:
                need_notify=need_notify,
            )
        if not target_item:
-            return False, overwrite_mode == "latest", None
+            return False, overwrite_mode == "latest" and target_file.suffix.casefold() not in {
+                ".jpg", ".jpeg", ".png", ".webp"
+            }, None

        checked_target = target_file
        over_flag = False
--- a/app/runtime/config.py
+++ b/app/runtime/config.py
@@ -402,6 +402,8 @@ class ConfigModel(BaseModel):
    # ==================== 音乐配置 ====================
    # 音乐封面代理地址（用于解决 coverartarchive.org 无法访问导致的封面不显示问题，留空则使用官方地址）
    MUSIC_COVER_PROXY: str = ""
+    # 同目录音乐封面匹配规则，按顺序选择，逗号分隔、大小写不敏感；留空禁用本地封面
+    MUSIC_COVER_FILE_PATTERNS: str = "cover.jpg,cover.jpeg,cover.png,folder.jpg,folder.jpeg,folder.png,front.jpg,front.jpeg,front.png,*.png"
    # AcoustID 应用 API Key，用于查询本地音频的 Chromaprint 指纹
    ACOUSTID_API_KEY: str = "b1auxfOzAg"
    # 是否将识别到的音乐标题、艺术家、专辑等标准元数据转换为简体中文
--- a/docs/architecture-overview.md
+++ b/docs/architecture-overview.md
@@ -763,7 +763,7 @@ flowchart LR
 | 指标 | 当前值 |
 |---|---:|
 | Python 模块 | 1119 |
-| 内部导入边 | 9,542 |
+| 内部导入边 | 9,545 |
 | 非平凡 SCC | 1（精确 containment 的 TMDB 移植包环） |
 | Application / Chain 具体 Adapter 直连 | 0 / 0 |
 | Direct egress | 49（债务已清零，49 条精确 containment） |
@@ -773,7 +773,7 @@ flowchart LR
 | Model/Oper 自动事务与自建 Session | 0 |
 | 组合根外 `SystemConfigOper()` | 0 |

-成功通知聚合复用既有 owner 和领域/模板模型，新增 14 条合法内部依赖边（含父包展开），未新增模块或 SCC。整理失败反馈由 `app.application.transfer.feedback` 集中投影；已整理下载的站点字幕复用 `app.chain.transfer` 自动整理队列；网络连通性检测按 `app.application.nettest` 归档，并覆盖 HTTP 与 WebSocket 探测。未启用智能助手的消息音乐交互由 `app.chain.music_interaction` 包中的解析展示与订阅协调模块处理；音乐标签准入与批次共识由 `app.chain.transfer.music` 负责，预览证据由 `app.application.transfer.projection` 只读投影；宿主依赖边现为 9,542；订阅独立日志复用运行时上下文、系统日志端口与资源鉴权，新增 19 条合法依赖边（含父包展开），未新增模块或 SCC；插件实例校验复用 `app.domain.plugin` 的物理身份规则，兼容下划线 ID，新增 2 条 Schema 到 Domain 的合法依赖边（含父包展开）；种子下载限流窗口由 `app.domain.torrent` 定义，失败冷却 owner 复用同一语义，新增 1 条合法 Chain 到 Domain 的导入边；工作流添加下载动作由 `app.workflow.actions.add_download` 转换工作流 Schema 为下载链领域上下文，新增 5 条模块导入边；资源过滤动作 `app.workflow.actions.filter_torrents` 与动作包共用的领域还原函数同样把 Schema 资源还原为领域对象后再交给过滤链，新增 6 条模块导入边。音乐来源链与识别缓存复用领域身份核验，音乐交互包未新增 SCC，插件一次性任务 owner `app.scheduler.oncejob` 带来 1 个模块和 9 条依赖边。Agent 持久回执新增 Application 端口及 DB Model/Oper/Adapter 四个冷导入模块。缓存组合根只在启动时选用 Redis 缓存才导入 Redis 适配器，依赖 langchain_core 的对话记忆模型归属 `app.agent.memory`，文件缓存且未启用智能体的实例不再加载 redis 与 langchain_core。Telegram 渠道改用模块内 `app.modules.telegram.botapi` 精简 Bot API 客户端，经 `RequestUtils` 出站，宿主不再导入 pyTelegramBotAPI，启用 Telegram 的实例也不会因该 SDK 加载 redis；该依赖暂时保留给插件使用。Web Push 协议（aes128gcm 加密、VAPID 签名与经 `RequestUtils` 投递）由网络适配器 `app.adapters.network.webpush` 直接基于 http-ece 与 py-vapid 实现，消息端点与 `WebPushModule` 共用，宿主不再导入 pywebpush 及其连带的 aiohttp；该依赖同样暂时保留给插件使用。飞书渠道改用模块内 `app.modules.feishu.openapi` OpenAPI 客户端（经 `RequestUtils` 出站）与网络适配器 `app.adapters.network.feishu` 事件长连接（websocket-client 同步收发 pbbp2 帧），宿主不再导入 lark-oapi 及其上万个生成模块，长连接也不再依赖 SDK 的模块级事件循环；插件经 `app.sdk.feishu` 复用同一长连接，lark-oapi 暂时保留给尚未迁移的插件使用。API 文档路由由 `app.adapters.web.docs` 按 `API_DOCS_ENABLE` 逐请求开放，默认关闭，未开放时不生成文档。当前 `app.startup.lifecycle` 为 587、`app.factory` 为 600、`app.main` 为 602。专辑订阅入库对账新增 `app.chain.subscribe.music` 一个职责 owner；按注入的下载/整理历史 Port 读取事实，复用音频标签/CUE 读取和领域覆盖规则，依赖图增加 18 条合法边，不引入跨层数据库或 Adapter 直连。订阅业务活动新增 `app.application.subscription.activity` 及对应 DB Model、Oper、Adapter 四个模块，三入口与功能前提交对照均只增加这四个模块；性能基线仅同步结构计数，保留原有历史耗时采样和预算。音乐调用观察 `app.application.music.observation` 新增一个轻量模块，记录来源结果及请求预算；AcoustID及路径识别复用此上下文传递有界指纹候选，保留旧单ID模块合同。站点图片域名快照新增 `app.application.security.image`，各启动入口增加一个模块；性能基线只同步模块数量，原有耗时预算、历史采样和生命周期资源约束保持有效。
+成功通知聚合复用既有 owner 和领域/模板模型，新增 14 条合法内部依赖边（含父包展开），未新增模块或 SCC。整理失败反馈由 `app.application.transfer.feedback` 集中投影；已整理下载的站点字幕复用 `app.chain.transfer` 自动整理队列；网络连通性检测按 `app.application.nettest` 归档，并覆盖 HTTP 与 WebSocket 探测。未启用智能助手的消息音乐交互由 `app.chain.music_interaction` 包中的解析展示与订阅协调模块处理；音乐标签准入与批次共识由 `app.chain.transfer.music` 负责，预览证据由 `app.application.transfer.projection` 只读投影；宿主依赖边现为 9,545；订阅独立日志复用运行时上下文、系统日志端口与资源鉴权，新增 19 条合法依赖边（含父包展开），未新增模块或 SCC；插件实例校验复用 `app.domain.plugin` 的物理身份规则，兼容下划线 ID，新增 2 条 Schema 到 Domain 的合法依赖边（含父包展开）；种子下载限流窗口由 `app.domain.torrent` 定义，失败冷却 owner 复用同一语义，新增 1 条合法 Chain 到 Domain 的导入边；工作流添加下载动作由 `app.workflow.actions.add_download` 转换工作流 Schema 为下载链领域上下文，新增 5 条模块导入边；资源过滤动作 `app.workflow.actions.filter_torrents` 与动作包共用的领域还原函数同样把 Schema 资源还原为领域对象后再交给过滤链，新增 6 条模块导入边。音乐来源链与识别缓存复用领域身份核验，音乐交互包未新增 SCC，插件一次性任务 owner `app.scheduler.oncejob` 带来 1 个模块和 9 条依赖边。Agent 持久回执新增 Application 端口及 DB Model/Oper/Adapter 四个冷导入模块。缓存组合根只在启动时选用 Redis 缓存才导入 Redis 适配器，依赖 langchain_core 的对话记忆模型归属 `app.agent.memory`，文件缓存且未启用智能体的实例不再加载 redis 与 langchain_core。Telegram 渠道改用模块内 `app.modules.telegram.botapi` 精简 Bot API 客户端，经 `RequestUtils` 出站，宿主不再导入 pyTelegramBotAPI，启用 Telegram 的实例也不会因该 SDK 加载 redis；该依赖暂时保留给插件使用。Web Push 协议（aes128gcm 加密、VAPID 签名与经 `RequestUtils` 投递）由网络适配器 `app.adapters.network.webpush` 直接基于 http-ece 与 py-vapid 实现，消息端点与 `WebPushModule` 共用，宿主不再导入 pywebpush 及其连带的 aiohttp；该依赖同样暂时保留给插件使用。飞书渠道改用模块内 `app.modules.feishu.openapi` OpenAPI 客户端（经 `RequestUtils` 出站）与网络适配器 `app.adapters.network.feishu` 事件长连接（websocket-client 同步收发 pbbp2 帧），宿主不再导入 lark-oapi 及其上万个生成模块，长连接也不再依赖 SDK 的模块级事件循环；插件经 `app.sdk.feishu` 复用同一长连接，lark-oapi 暂时保留给尚未迁移的插件使用。API 文档路由由 `app.adapters.web.docs` 按 `API_DOCS_ENABLE` 逐请求开放，默认关闭，未开放时不生成文档。当前 `app.startup.lifecycle` 为 587、`app.factory` 为 600、`app.main` 为 602。专辑订阅入库对账新增 `app.chain.subscribe.music` 一个职责 owner；按注入的下载/整理历史 Port 读取事实，复用音频标签/CUE 读取和领域覆盖规则，依赖图增加 18 条合法边，不引入跨层数据库或 Adapter 直连。订阅业务活动新增 `app.application.subscription.activity` 及对应 DB Model、Oper、Adapter 四个模块，三入口与功能前提交对照均只增加这四个模块；性能基线仅同步结构计数，保留原有历史耗时采样和预算。音乐调用观察 `app.application.music.observation` 新增一个轻量模块，记录来源结果及请求预算；AcoustID及路径识别复用此上下文传递有界指纹候选，保留旧单ID模块合同。站点图片域名快照新增 `app.application.security.image`，各启动入口增加一个模块；性能基线只同步模块数量，原有耗时预算、历史采样和生命周期资源约束保持有效。

 架构专项验证分为两个 CI 投影：`Check event semantic policy` 先运行依赖、Adapter、出口和 Event
 语义门禁，`Check host architecture snapshot` 再执行快照测试及一次
--- a/docs/cli.md
+++ b/docs/cli.md
@@ -507,6 +507,7 @@ moviepilot config set LYRICS_BATCH_TIMEOUT 120
 moviepilot config set LYRICS_PROVIDER_RETRY_MAX_WAIT 5
 moviepilot config set MUSIC_METADATA_TO_SIMPLIFIED true
 moviepilot config set MUSIC_LYRICS_TO_SIMPLIFIED true
+moviepilot config set MUSIC_COVER_FILE_PATTERNS 'cover.jpg,folder.jpg,front.jpg,*.png'
 moviepilot config set MUSIC_RELEASE_REGION_PRIORITY CN,TW,HK
 moviepilot config set MUSIC_RELEASE_SCRIPT_PRIORITY Hans,Hant,Latn
 ```
@@ -531,6 +532,7 @@ moviepilot config describe API_TOKEN --show-secrets
 - `AMLL_BASE_URL` 默认使用无需 API Key 的 AMLL TTML API，也可指向兼容原生接口的自建实例。来源优先按 ISRC 查询，再严格核对歌曲、艺术家及已有专辑信息；下载的 TTML 主唱时间轴转换为 Lyricsfile 和 LRC，限流时短暂冷却并让其他歌词来源继续处理
 - `THEAUDIODB_API_KEY` 用于音乐元数据及纯文本歌词兜底，默认 `123` 为官方公开 V1 Key；额外歌词来源可通过插件提供，接入方式见[歌词插件开发说明](https://github.com/jxxghp/MoviePilot-Plugins/blob/main/docs/faq/21-register-lyrics-provider.md)
 - `MUSIC_METADATA_TO_SIMPLIFIED` 默认开启；开启后会将识别结果中的曲名、艺术家、专辑和分类等标准音乐元数据转换为简体中文，不转换歌词与来源原始响应
+- `MUSIC_COVER_FILE_PATTERNS` 控制同目录音乐封面的文件名匹配与选择顺序，逗号分隔、大小写不敏感；附加文件整理和内嵌封面回退共用此配置，详见[音乐本地封面](music-covers.md)
 - `MUSIC_LYRICS_TO_SIMPLIFIED` 默认关闭，独立控制生成的 `.lrc` / `.txt` 歌词正文转简体；时间轴、LRC 标签及来源 Lyricsfile 保持原样。已有歌词仍遵循原刮削策略，需要转换时选择“总是刮削”重新刮削
 - `MUSIC_RELEASE_REGION_PRIORITY` 与 `MUSIC_RELEASE_SCRIPT_PRIORITY` 分别使用逗号分隔的 ISO 3166-1 地区代码和 ISO 15924 文字代码；仅在 MusicBrainz 候选结构同样可信时作为发行版本排序依据，最多使用前三项
 - `config describe` 显示单个配置项的类型、默认值和当前值
--- a/docs/music-covers.md
+++ b/docs/music-covers.md
@@ -0,0 +1,28 @@
+# 音乐本地封面
+
+开启整理请求的“同步整理附加文件”后，音乐目录或单个音轨整理会携带同目录的封面图片。
+图片沿用当前复制、硬链接、移动和覆盖策略；重命名时保留图片原名，放入目标专辑根目录，
+不套用曲名、曲序或 Disc 子目录。专辑、EP 和单曲共用这一行为。
+
+本地图片作为附件保留，不依赖在线封面是否可下载，也不要求开启刮削。
+同目录包含多个不同发行时不自动分配封面；影视整理和单独选择图片不会触发音乐封面整理。
+后续生成 STRM 时仍需让对应插件同步图片后缀，MoviePilot 不负责该插件的复制配置。
+
+音乐刮削内嵌封面使用“在线优先，下载不到再读取同目录图片”的顺序；本地候选按配置顺序选择，
+损坏或不支持的图片跳过并尝试下一张。目录读取和图片下载均走存储能力，支持本地和远端目标。
+刮削为“跳过”时不写内嵌封面，“缺失时刮削”保留已有内嵌封面，“总是刮削”可替换已有封面。
+写入继续遵守音频源保护，不通过媒体库硬链接或软链接改写下载源文件。
+
+`MUSIC_COVER_FILE_PATTERNS` 为全局封面匹配配置，逗号分隔、忽略大小写，支持 `*` 等文件名通配符。
+默认值为：
+
+```text
+cover.jpg,cover.jpeg,cover.png,folder.jpg,folder.jpeg,folder.png,front.jpg,front.jpeg,front.png,*.png
+```
+
+匹配始终限制为 JPG、JPEG、PNG、WebP 图片，不读取隐藏文件。可以加入 `*.jpg` 或 `artwork.webp`；
+留空关闭本地封面匹配。配置示例：
+
+```shell
+moviepilot config set MUSIC_COVER_FILE_PATTERNS 'cover.jpg,folder.jpg,front.jpg,*.png,artwork.webp'
+```
--- a/docs/refactor/optimization-checklist.md
+++ b/docs/refactor/optimization-checklist.md
@@ -100,7 +100,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁

 | 指标 | 当前值 | 解释 |
 |---|---:|---|
-| 宿主 Python 模块 / 内部依赖边 | 1119 / 9,542 | `dependency-baseline.json` 当前快照；成功通知聚合复用既有 owner 和领域/模板模型，新增 14 条合法内部依赖边（含父包展开），未新增模块或 SCC。专辑订阅入库对账新增 1 个 Chain owner 和 18 条合法依赖边（含父包展开），仍经注入 Port 读取历史并通过既有 mutation 写入；订阅业务活动新增 4 个模块和 50 条合法内部依赖边（含父包展开），Application 仅定义事实与持久化 Port，DB Adapter/Oper 实现存取，组合根注入，未增加 Application/Chain 到具体 Adapter 或数据库的直连；插件实例校验复用 `app.domain.plugin` 的物理身份规则，兼容下划线 ID，新增 2 条合法依赖边（含父包展开）；种子失败冷却复用 `app.domain.torrent` 的限流语义，新增 1 条合法导入边；系统设置合同按 `app.application.settings` 包组织，系统设置 API 端点独立按 `app.api.endpoints.settings` 归档；消息音乐交互由 `app.chain.music_interaction` 包中的音乐解析/展示与订阅协调模块处理；已整理下载的站点字幕经 `app.chain.transfer` 自动整理队列完成命名与归档；网络连通性检测按 `app.application.nettest` 归档；工作流添加下载动作由 `app.workflow.actions.add_download` 转换工作流 Schema 为下载链领域上下文，新增 5 条合法模块导入边，资源过滤动作复用同一还原再新增 6 条；LLM 独占连接关闭复用宿主线程池；歌词字形转换复用 `app.foundation.text`；订阅创建使用 `app.domain.context.MediaInfo` 和运行时配置端口，将非 TMDB 影视身份投影为 TMDB 身份 |
+| 宿主 Python 模块 / 内部依赖边 | 1119 / 9,545 | `dependency-baseline.json` 当前快照；成功通知聚合复用既有 owner 和领域/模板模型，新增 14 条合法内部依赖边（含父包展开），未新增模块或 SCC。专辑订阅入库对账新增 1 个 Chain owner 和 18 条合法依赖边（含父包展开），仍经注入 Port 读取历史并通过既有 mutation 写入；订阅业务活动新增 4 个模块和 50 条合法内部依赖边（含父包展开），Application 仅定义事实与持久化 Port，DB Adapter/Oper 实现存取，组合根注入，未增加 Application/Chain 到具体 Adapter 或数据库的直连；插件实例校验复用 `app.domain.plugin` 的物理身份规则，兼容下划线 ID，新增 2 条合法依赖边（含父包展开）；种子失败冷却复用 `app.domain.torrent` 的限流语义，新增 1 条合法导入边；系统设置合同按 `app.application.settings` 包组织，系统设置 API 端点独立按 `app.api.endpoints.settings` 归档；消息音乐交互由 `app.chain.music_interaction` 包中的音乐解析/展示与订阅协调模块处理；已整理下载的站点字幕经 `app.chain.transfer` 自动整理队列完成命名与归档；网络连通性检测按 `app.application.nettest` 归档；工作流添加下载动作由 `app.workflow.actions.add_download` 转换工作流 Schema 为下载链领域上下文，新增 5 条合法模块导入边，资源过滤动作复用同一还原再新增 6 条；LLM 独占连接关闭复用宿主线程池；歌词字形转换复用 `app.foundation.text`；订阅创建使用 `app.domain.context.MediaInfo` 和运行时配置端口，将非 TMDB 影视身份投影为 TMDB 身份 |
 | 非平凡 SCC | 1 | 仅保留精确 containment 的 29 模块 TMDB 移植包环 |
 | 跨层 DB 边界债务 | 0 | Application、Chain、API、Agent、Runtime、Workflow 到 DB 的受控债务均为零 |
 | Model/Oper 事务债务 | 0 | 自建 Session、自动事务装饰器、直接 commit/rollback 等基线均为零 |
--- a/docs/rules/05-architecture.md
+++ b/docs/rules/05-architecture.md
@@ -600,6 +600,12 @@ architecture snapshot, not through incidental module globals.

 ## Existing Chain, Module and DB Layers

+音乐封面的文件名匹配与优先级是 `app/domain/music.py` 的纯规则；
+`transfer/filter.py` 和 `ScrapingChain` 经 `runtime/settings.py` 读取同一个全局配置。
+候选规划及 `transfer/music.py` 确认同目录唯一发行归属，图片走既有 durable 整理任务；
+`TransHandler` 保留图片名并定位到专辑根。在线失败后的图片读取由 `ScrapingChain` 经
+`StorageChain` 完成，不在领域层引入配置或 I/O，也不恢复未经证明的远端发行标识。
+
 ### Chain layer

 运行时停止信号统一由 `app/runtime/stop.py` 的 `StopState` 持有。业务代码应注入或读取
--- a/tests/fixtures/architecture/dependency-baseline.json
+++ b/tests/fixtures/architecture/dependency-baseline.json
@@ -941,8 +941,8 @@
      "runtime_only": true
    }
  },
-  "edge_count": 9542,
-  "edge_sha256": "74e5a7f8dd6e0529fa7ea16c3b35dcdf363574aaea5ad44551016c89357a3b90",
+  "edge_count": 9545,
+  "edge_sha256": "8784d89756b95ce35702eb7aa43513050e4b181641dad727a39d6a0354ec0eb6",
  "edges": [
    "app -> app.foundation",
    "app -> app.foundation.environment",
@@ -5049,6 +5049,7 @@
    "app.chain.scraping -> app.domain.meta.metabase",
    "app.chain.scraping -> app.domain.meta.metamusic",
    "app.chain.scraping -> app.domain.metainfo",
+    "app.chain.scraping -> app.domain.music",
    "app.chain.scraping -> app.foundation",
    "app.chain.scraping -> app.foundation.singleton",
    "app.chain.scraping -> app.foundation.text",
@@ -5057,6 +5058,7 @@
    "app.chain.scraping -> app.runtime.events",
    "app.chain.scraping -> app.runtime.log",
    "app.chain.scraping -> app.runtime.reload",
+    "app.chain.scraping -> app.runtime.settings",
    "app.chain.scraping -> app.schemas",
    "app.chain.scraping -> app.schemas.file",
    "app.chain.scraping -> app.schemas.media",
@@ -5792,6 +5794,7 @@
    "app.chain.transfer.filter -> app.domain.music",
    "app.chain.transfer.filter -> app.runtime",
    "app.chain.transfer.filter -> app.runtime.log",
+    "app.chain.transfer.filter -> app.runtime.settings",
    "app.chain.transfer.filter -> app.schemas",
    "app.chain.transfer.filter -> app.schemas.transfer",
    "app.chain.transfer.filter -> app.schemas.types",
--- a/tests/test_music_cover_fallback.py
+++ b/tests/test_music_cover_fallback.py
@@ -0,0 +1,247 @@
+"""音乐封面随附加文件整理及在线失败后的本地图片回退回归。"""
+
+import os
+import shutil
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import Mock
+
+import pytest
+from mutagen.flac import FLAC, Picture
+from PIL import Image
+
+from app.application.transfer.workflow import TransferPlanningInput
+from app.chain.media import MediaChain
+from app.chain.scraping import ScrapingChain
+from app.chain.transfer.request import _TransferCandidatePlanner
+from app.domain.context import MusicInfo
+from app.domain.meta.metamusic import MetaMusic
+from app.domain.music import music_cover_priority
+from app.modules.filemanager.storages.local import LocalStorage
+from app.modules.filemanager.transhandler import TransHandler
+from app.runtime.config import settings
+from app.schemas.file import FileItem
+from app.schemas.system import TransferDirectoryConf
+from app.schemas.transfer import TransferInfo
+from app.schemas.types import MediaType
+from tests.test_transfer_sync_extra_files import bind_empty_history_repositories, make_fileitem, make_transfer_chain
+
+
+def _tagged_audio(directory, name="01.flac", album="Album"):
+    """用真实标签建立可离线整理的音轨和唯一发行证据。"""
+    directory.mkdir(parents=True, exist_ok=True)
+    path = directory / name
+    shutil.copyfile(Path(__file__).parent / "fixtures/audio/silence.flac", path)
+    audio = FLAC(path)
+    audio.update(title=[f"Song {path.stem}"], artist=["Artist"], album=[album], albumartist=["Artist"])
+    audio.save()
+    return path
+
+
+def _planner(chain, item, sync=True, mtype=MediaType.MUSIC):
+    """构造真实候选规划器，保留同目录匹配与附加文件开关。"""
+    return _TransferCandidatePlanner(
+        chain, meta=None, season=None, formater=None, batch_mtype=mtype, mediainfo=None,
+        continue_callback=None, has_episode_format_template=False, transfer_exclude_words=[],
+        download_hash=None, sync_extra_files=sync, fileitem=item,
+    )
+
+
+@pytest.mark.parametrize("single", [False, True])
+@pytest.mark.parametrize("sync", [False, True])
+@pytest.mark.parametrize("mtype", [None, MediaType.MUSIC])
+def test_music_cover_enters_directory_and_single_transfer_only_when_sync_enabled(tmp_path, monkeypatch, single, sync, mtype):
+    """目录与单音轨使用相同附加文件开关，图片继承音轨身份并且只规划一次。"""
+    path = _tagged_audio(tmp_path / "source")
+    cover = make_fileitem(str(path.parent / "COVER.JPG"))
+    audio = make_fileitem(str(path))
+    root = audio if single else FileItem(storage="local", path=str(path.parent), type="dir")
+    chain = make_transfer_chain()
+    bind_empty_history_repositories(chain)
+    storage = Mock()
+    storage.get_parent_item.return_value = root
+    storage.list_files.return_value = [audio, cover]
+    monkeypatch.setattr("app.chain.transfer.request.StorageChain", lambda: storage)
+    planner = _planner(chain, root, sync, mtype)
+    items, inherited = planner._plan_file_items([(audio, False)] if single else [(cover, False), (audio, False)])
+
+    assert [item.path for item, _ in items] == ([audio.path, cover.path] if sync else [audio.path])
+    assert (chain._get_file_key(cover) in inherited) is sync
+    if sync:
+        context = chain._prepare_music_batch_context(items, mtype)
+        assert context.related_main_keys[chain._get_file_key(cover)] == chain._get_file_key(audio)
+
+
+def test_music_cover_does_not_borrow_identity_from_different_albums_or_video(tmp_path):
+    """同目录多发行存在歧义时不关联封面，影视批次也不会吸收图片。"""
+    first = make_fileitem(str(_tagged_audio(tmp_path / "source", "01.flac", "First")))
+    second = make_fileitem(str(_tagged_audio(tmp_path / "source", "02.flac", "Second")))
+    cover = make_fileitem(str(tmp_path / "source/cover.jpg"))
+    chain = make_transfer_chain()
+    bind_empty_history_repositories(chain)
+    context = chain._prepare_music_batch_context([(first, False), (second, False), (cover, False)], MediaType.MUSIC)
+    assert chain._get_file_key(cover) not in context.related_main_keys
+    assert not _planner(chain, first, mtype=MediaType.TV)._is_music_cover_candidate(cover)
+    assert _planner(chain, cover)._plan_file_items([(cover, False)])[0] == []
+
+
+@pytest.mark.parametrize("single", [False, True])
+@pytest.mark.parametrize("mtype", [None, MediaType.MUSIC])
+def test_music_cover_reaches_public_transfer_entrypoint(tmp_path, monkeypatch, single, mtype):
+    """从真实目录遍历到公开整理入口，封面只复用主音轨身份而不单独在线识别。"""
+    audio = _tagged_audio(tmp_path / "source")
+    cover = audio.parent / "cover.jpg"
+    Image.new("RGB", (2, 2)).save(cover)
+    storage = LocalStorage()
+    port = SimpleNamespace(
+        get_item=lambda item: storage.get_item(Path(item.path)),
+        get_parent_item=lambda item: storage.get_item(Path(item.path).parent),
+        list_files=lambda item, **_kwargs: [storage.get_item(path) for path in Path(item.path).iterdir()],
+        contains_bluray_subdirectories=lambda _items: False,
+    )
+    monkeypatch.setattr("app.chain.transfer.workflow.StorageChain", lambda: port)
+    monkeypatch.setattr("app.chain.transfer.request.StorageChain", lambda: port)
+    monkeypatch.setattr("app.chain.transfer.workflow.get_configured_system_config", lambda: SimpleNamespace(get=lambda _key: None))
+    chain = make_transfer_chain()
+    bind_empty_history_repositories(chain)
+    chain.eventmanager = SimpleNamespace(send_event=lambda *_args, **_kwargs: None)
+    monkeypatch.setattr(chain, "_finalize_recognition_result", lambda info, **_kwargs: info)
+    monkeypatch.setattr(chain, "_TransferChain__put_to_jobview", lambda _task: True)
+    monkeypatch.setattr(chain, "_register_scrape_batch_task", lambda _task: None)
+    monkeypatch.setattr(chain, "_close_scrape_batch", lambda _batch: None)
+    for name in ("recognize_music_album_directory", "recognize_music_by_path", "recognize_by_meta", "recognize_media"):
+        monkeypatch.setattr(MediaChain, name, Mock(side_effect=AssertionError("完整标签不得在线识别")))
+    tasks = []
+
+    def preview(task, **_kwargs):
+        """检查完整请求给规划执行器传入的主文件和封面上下文。"""
+        tasks.append(task)
+        return TransferInfo(success=True, fileitem=task.fileitem, target_item=task.fileitem)
+
+    monkeypatch.setattr(chain, "_plan_checkpoint_and_execute", preview)
+    success, result = chain.do_transfer(
+        fileitem=storage.get_item(audio if single else audio.parent),
+        target_directory=TransferDirectoryConf(library_storage="local", library_path=str(tmp_path / "library")),
+        mtype=mtype, sync_extra_files=True, force=True, preview=True,
+    )
+    assert success, result
+    assert [task.fileitem.path for task in tasks] == [str(audio), str(cover)]
+    assert all(task.mediainfo.album == "Album" for task in tasks)
+    assert result["summary"] == {"total": 2, "success": 2, "failed": 0}
+
+
+@pytest.mark.parametrize("mode", ["copy", "link", "move"])
+@pytest.mark.parametrize("rename", [False, True])
+def test_music_cover_real_storage_transfer_preserves_filename_and_album_root(tmp_path, monkeypatch, mode, rename):
+    """封面沿用真实复制/硬链接/移动执行及覆盖策略，重命名时不进入碟片子目录。"""
+    source = tmp_path / "source"
+    source.mkdir()
+    path = source / "COVER.JPG"
+    Image.new("RGB", (2, 2), "red").save(path)
+    content = path.read_bytes()
+    storage = LocalStorage()
+    item = storage.get_item(path)
+    info = MusicInfo(title="Song", album="Album", artists=["Artist"], album_artist="Artist", disc_number=2, total_discs=2)
+    meta = MetaMusic(title="Song", album="Album", artists=["Artist"], album_artist="Artist", disc_number=2, total_discs=2)
+    planning = TransferPlanningInput(source_fileitem=item.model_dump(mode="json"), target_storage="local",
+                                     target_path=str(tmp_path / "library"), requested_transfer_type=mode, need_rename=rename)
+    monkeypatch.setattr("app.modules.filemanager.transhandler.eventmanager.send_event", Mock(return_value=None))
+    handler = TransHandler()
+    checkpoint = handler.plan_transfer(planning, meta=meta, mediainfo=info, source_oper=storage, target_storage="local",
+                                       target_path=tmp_path / "library", transfer_type=mode, need_scrape=False,
+                                       need_rename=rename, need_notify=False, overwrite_mode="never", episodes_info=None, preview=False)
+    target = Path(checkpoint.final_target_path)
+    assert target.name == path.name and "Disc 2" not in target.parts
+    result = handler.execute_transfer_plan(checkpoint, meta=meta, mediainfo=info, source_oper=storage, target_oper=storage)
+    assert result.success and target.read_bytes() == content
+    assert path.exists() is (mode != "move")
+    if mode == "link":
+        assert os.path.samefile(path, target)
+    refused = handler._TransHandler__resolve_overwrite(
+        fileitem=item, meta=meta, mediainfo=info, target_oper=storage, target_storage="local", target_file=target,
+        transfer_type=mode, overwrite_mode="never", need_notify=False,
+    )
+    assert refused[2].overwrite_skipped
+
+
+@pytest.mark.parametrize("online", ["missing", "failed", "available"])
+@pytest.mark.parametrize("policy", ["missing", "skip", "overwrite"])
+@pytest.mark.parametrize("existing", [False, True])
+def test_music_scrape_uses_online_first_then_valid_local_cover_without_changing_seed(tmp_path, online, policy, existing):
+    """真实 FLAC 验证下载失败回退、坏图跳过、刮削策略和硬链接源保护。"""
+    source = _tagged_audio(tmp_path / "source")
+    existing_path = tmp_path / "existing.png"
+    Image.new("RGB", (2, 2), "green").save(existing_path)
+    existing_bytes = existing_path.read_bytes()
+    if existing:
+        audio = FLAC(source)
+        picture = Picture()
+        picture.type, picture.mime, picture.data = 3, "image/png", existing_bytes
+        audio.add_picture(picture)
+        audio.save()
+    target_dir = tmp_path / "library"
+    target_dir.mkdir()
+    target = target_dir / "01.flac"
+    os.link(source, target)
+    before = source.read_bytes()
+    (target_dir / "cover.jpg").write_bytes(b"invalid image")
+    Image.new("RGB", (2, 2), "blue").save(target_dir / "front.png")
+    local_bytes = (target_dir / "front.png").read_bytes()
+    online_path = tmp_path / "online.png"
+    Image.new("RGB", (2, 2), "red").save(online_path)
+    online_bytes = online_path.read_bytes()
+    info = MusicInfo(title="Song", artists=["Artist"], album="Album", cover_url=None if online == "missing" else "https://example.invalid/cover")
+    storage = LocalStorage()
+    chain = object.__new__(ScrapingChain)
+    chain.storagechain = SimpleNamespace(
+        get_parent_item=lambda item: storage.get_item(Path(item.path).parent),
+        list_files=lambda item, **_kwargs: [storage.get_item(path) for path in Path(item.path).iterdir()],
+        download_file=lambda item, **_kwargs: Path(item.path),
+    )
+    chain._download_music_cover = Mock(return_value=(online_bytes, "image/png") if online == "available" else (None, "image/jpeg"))
+    result = chain._apply_music_file_scrape(
+        fileitem=storage.get_item(target), local_path=target, mediainfo=info, write_tags=False, tag_overwrite=False,
+        with_cover=policy != "skip", cover_overwrite=policy == "overwrite", cover=None, lyrics_option=None,
+        lyrics_overwrite=False, lyrics_chain=None, album_info=None, media_source=None,
+    )
+    assert result.metadata_success
+    assert source.read_bytes() == before
+    pictures = FLAC(target).pictures
+    if policy == "skip":
+        assert [picture.data for picture in pictures] == ([existing_bytes] if existing else [])
+        chain._download_music_cover.assert_not_called()
+    else:
+        expected = existing_bytes if existing and policy == "missing" else online_bytes if online == "available" else local_bytes
+        assert pictures[0].data == expected
+
+
+@pytest.mark.parametrize("remote", [False, True])
+def test_local_cover_reader_supports_storage_paths_and_album_root(tmp_path, remote):
+    """碟片音轨从专辑根读取归档封面，远端路径只通过存储下载读取。"""
+    picture_path = tmp_path / "cover.png"
+    Image.new("RGB", (2, 2), "blue").save(picture_path)
+    storage = "cloud" if remote else "local"
+    audio = FileItem(storage=storage, path="/library/Album/Disc 2/01.flac", type="file")
+    disc = FileItem(storage=storage, path="/library/Album/Disc 2", type="dir")
+    album = FileItem(storage=storage, path="/library/Album", type="dir")
+    cover = FileItem(storage=storage, path="/library/Album/cover.png", type="file")
+    port = Mock()
+    port.get_parent_item.side_effect = [disc, album]
+    port.list_files.return_value = [cover]
+    port.download_file.return_value = picture_path
+    chain = object.__new__(ScrapingChain)
+    chain.storagechain = port
+    assert chain._load_local_music_cover(audio) == (picture_path.read_bytes(), "image/png")
+    port.list_files.assert_called_once_with(album, recursion=False)
+    assert port.download_file.call_args.args[0] == cover
+
+
+def test_music_cover_patterns_are_case_insensitive_and_reject_non_images(monkeypatch):
+    """通配配置可扩展文件名，但不能把文本或隐藏文件当作音乐封面。"""
+    monkeypatch.setattr(settings, "MUSIC_COVER_FILE_PATTERNS", "Artwork.*,*.PNG")
+    chain = make_transfer_chain()
+    assert chain._is_music_cover_file(make_fileitem("/music/ARTWORK.JPEG"))
+    assert chain._is_music_cover_file(make_fileitem("/music/custom.png"))
+    assert not chain._is_music_cover_file(make_fileitem("/music/Artwork.txt"))
+    assert music_cover_priority("/music/.cover.png", "*.png") is None
+    assert music_cover_priority("/music/cover.jpg", "") is None