"""`peach.scan`：扫描写进账本的行必须能被读取侧翻译回同一个文件。

两种设置文件写法各一条：Windows 上 `[media.locations] local = <真实目录>`、挂载表为空；
macOS 上数据库路径前缀使用 `R:\\media`，`[media.mounts] local = <真实目录>`。两条都在临时目录里跑。

POSIX 形态那条平台无关，哪台机器上都跑。Windows 形态那条只在 Windows 上跑：它的前提是
配置中的路径前缀本身就是一个真实存在的盘符目录，而 POSIX 机器上造不出这种目录——拿 `/var/...` 配
`windows=True` 不是注入平台，是喂一个账本里不可能出现的形态，写入侧门槛会当场拒收。
"""
from __future__ import annotations

import io
import json
import os
import sqlite3
import tempfile
import unittest
from contextlib import closing, redirect_stdout
from pathlib import Path, PureWindowsPath
from unittest.mock import patch

from peach import catalog_rules, scan
from support.ledger import fresh_ledger
from support.web import route_client

class _ScanCase(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.addCleanup(self.tmp.cleanup)
        self.root = Path(self.tmp.name).resolve()
        self.db = fresh_ledger(self.root)
        self.media = self.root / "media"
        (self.media / "创作者" / "sub").mkdir(parents=True)
        (self.media / "创作者" / "sub" / "a.mp4").write_bytes(b"0" * 10)
        (self.media / "cover.jpg").write_bytes(b"1" * 3)
        (self.media / "notes.txt").write_bytes(b"2")

    def _paths(self):
        connection = sqlite3.connect(self.db)
        try:
            return dict(connection.execute(
                "SELECT path,medium FROM asset WHERE location='local'").fetchall())
        finally:
            connection.close()

    def _scan(self, root, **kwargs):
        output = io.StringIO()
        with redirect_stdout(output):
            result = scan.scan_location(self.db, "local", root, **kwargs)
        return result, output.getvalue()


@unittest.skipUnless(os.name == "nt", "本用例需要真实存在的 Windows 盘符目录")
class WindowsShapeTests(_ScanCase):
    """Windows：配置的路径前缀就是真实目录，路径原样保存到数据库。"""

    def test_rows_are_written_under_the_declared_root(self):
        declared = {"local": (str(self.media),)}
        result, output = self._scan(str(self.media), declared_roots=declared, windows=True)
        self.assertEqual(result.files, 3)
        self.assertEqual(result.total_bytes, 14)
        self.assertEqual(result.gone, 0)
        self.assertIn("✓ local: 3 文件", output)
        rows = self._paths()
        self.assertEqual(rows[str(self.media / "创作者" / "sub" / "a.mp4")], "video")
        self.assertEqual(rows[str(self.media / "cover.jpg")], "image")
        self.assertEqual(rows[str(self.media / "notes.txt")], "other")
        # 读取侧：Windows 上账本路径就是本机路径，每一行都指向真实存在的文件。
        for path in rows:
            self.assertTrue(Path(path).is_file(), path)

    def test_scan_isolates_shortcuts_without_removing_files_or_media_sidecars(self):
        ad = self.media / '点击进入.url'
        ad.write_bytes(b'link')
        video = self.media / 'ABC-123.mp4'
        video.write_bytes(b'video')
        (self.media / 'ABC-123.nfo').write_text('<movie/>', encoding='utf-8')
        kwargs = dict(declared_roots={'local': (str(self.media),)}, windows=True)
        first, _ = self._scan(str(self.media), **kwargs)
        second, _ = self._scan(str(self.media), **kwargs)
        self.assertEqual((first.junk_isolated, second.junk_isolated), (1, 0))
        self.assertTrue(ad.is_file())
        with closing(sqlite3.connect(self.db)) as connection:
            self.assertEqual(connection.execute('SELECT disposal,disposal_reason FROM asset WHERE path=?',
                                               (str(ad),)).fetchone(), ('trash', 'junk'))
            self.assertEqual(connection.execute('SELECT disposal FROM asset WHERE path=?',
                                               (str(video),)).fetchone(), (None,))

    def test_scanned_video_is_visible_without_a_thumbnail_filter(self):
        from peach.api import create_app
        from peach.config import PeachSettings
        self._scan(str(self.media), declared_roots={"local": (str(self.media),)}, windows=True)
        with route_client(create_app(PeachSettings(db_path=self.db, configured=True, token=""))) as client:
            self.assertEqual(client.get('/api/items').json()['total'], 1)
            self.assertEqual(client.get('/api/items?thumb=0').json()['total'], 1)
            self.assertEqual(client.get('/api/items?thumb=1').json()['total'], 0)

    def test_a_subdirectory_of_the_declared_root_is_accepted(self):
        declared = {"local": (str(self.media),)}
        result, _ = self._scan(str(self.media / "创作者"), declared_roots=declared, windows=True)
        self.assertEqual(result.files, 1)

    def test_rescanning_refreshes_instead_of_duplicating_and_counts_the_gone(self):
        declared = {"local": (str(self.media),)}
        self._scan(str(self.media), declared_roots=declared, windows=True)
        (self.media / "notes.txt").unlink()
        (self.media / "cover.jpg").write_bytes(b"1" * 8)
        result, _ = self._scan(str(self.media), declared_roots=declared, windows=True)
        self.assertEqual(result.files, 2)
        self.assertEqual(len(self._paths()), 3, "没扫到的行留着，由资源同步对账处理")
        connection = sqlite3.connect(self.db)
        size = connection.execute("SELECT size FROM asset WHERE path=?",
                                  (str(self.media / "cover.jpg"),)).fetchone()[0]
        connection.close()
        self.assertEqual(size, 8)


class PosixShapeTests(_ScanCase):
    """macOS：账本里写 `R:\\media\\...`，遍历的是挂载点，读取侧按挂载表翻回去。"""

    def test_file_identity_preserves_id_and_personal_fields_after_external_rename(self):
        options = dict(declared_roots={'local': ('R:\\media',)}, mounts={'local': (self.media,)}, windows=False)
        first, _ = self._scan('R:\\media', **options)
        old = self.media / '创作者/sub/a.mp4'
        with closing(sqlite3.connect(self.db)) as connection, connection:
            asset_id = connection.execute("SELECT id FROM asset WHERE medium='video'").fetchone()[0]
            connection.execute("UPDATE asset SET rating=5,catalog_title='人工标题' WHERE id=?", (asset_id,))
        old.rename(self.media / 'changed.mp4')
        second, _ = self._scan('R:\\media', **options, new_only=True)
        self.assertEqual(second.new_ids, ())
        with closing(sqlite3.connect(self.db)) as connection:
            self.assertEqual(connection.execute('SELECT id,name,rating,catalog_title FROM asset WHERE medium=\'video\'').fetchall(),
                             [(asset_id,'changed.mp4',5,'人工标题')])

    def test_same_size_copy_and_hardlink_are_separate_files(self):
        options = dict(declared_roots={'local': ('R:\\media',)}, mounts={'local': (self.media,)}, windows=False)
        self._scan('R:\\media', **options)
        old = self.media / '创作者/sub/a.mp4'
        (self.media / 'copy.mp4').write_bytes(old.read_bytes())
        os.link(old, self.media / 'hardlink.mp4')
        result, _ = self._scan('R:\\media', **options, new_only=True)
        self.assertEqual(len(result.new_ids), 2)
        with closing(sqlite3.connect(self.db)) as connection:
            self.assertEqual(connection.execute("SELECT count(*) FROM asset WHERE medium='video'").fetchone()[0], 3)

    def test_same_second_scan_returns_only_exact_insertions(self):
        options = dict(declared_roots={'local': ('R:\\media',)}, mounts={'local': (self.media,)}, windows=False)
        with patch('peach.scan.time.strftime', return_value='2026-10-27 12:00:00'):
            self._scan('R:\\media', **options)
            (self.media / 'new.mp4').write_bytes(b'new-video')
            result, _ = self._scan('R:\\media', **options, new_only=True)
        self.assertEqual(len(result.new_ids), 1)
        with closing(sqlite3.connect(self.db)) as connection:
            self.assertEqual(connection.execute('SELECT name FROM asset WHERE id=?', result.new_ids).fetchone()[0], 'new.mp4')

    def test_repeated_scan_in_one_second_keeps_completed_videos_out_of_the_new_queue(self):
        from peach.library_new_assets import pending, remember
        options = dict(declared_roots={'local': ('R:\\media',)}, mounts={'local': (self.media,)}, windows=False)
        with patch('peach.scan.time.strftime', return_value='2026-10-27 12:00:00'):
            self._scan('R:\\media', **options)
            queued = pending(self.db)
            self.assertEqual(len(queued), 1)
            remember(self.db, completed=queued)
            self._scan('R:\\media', **options)
            self.assertEqual(pending(self.db), [])

    def test_windows_mount_mapping_has_the_same_ledger_path_contract(self):
        options = dict(declared_roots={'local': ('R:\\media',)}, mounts={'local': (self.media,)}, windows=True)
        self.assertEqual(scan.walk_root_for('local', 'R:\\media\\创作者', **options), self.media / '创作者')
        self.assertEqual(scan.ledger_root_for('local', self.media / '创作者', **options), 'R:\\media\\创作者')

    def test_root_case_changes_preserve_asset_identity_and_personal_records(self):
        options = dict(declared_roots={'local': ('R:\\media',)}, mounts={'local': (self.media,)}, windows=False)
        self._scan('R:\\media', **options)
        with closing(sqlite3.connect(self.db)) as connection, connection:
            connection.execute("UPDATE asset SET rating=4,play_count=7")
            before = list(connection.execute("SELECT id,path,rating,play_count FROM asset ORDER BY id"))
        self._scan('r:\\MEDIA', **options)
        with closing(sqlite3.connect(self.db)) as connection, connection:
            self.assertEqual(list(connection.execute("SELECT id,path,rating,play_count FROM asset ORDER BY id")), before)

    def test_conflicting_paths_preserve_each_record_and_allow_other_files_to_register(self):
        paths = ('R:\\media\\copy.mp4', 'r:/MEDIA/COPY.MP4')
        (self.media / 'copy.mp4').write_bytes(b'present')
        with closing(sqlite3.connect(self.db)) as connection, connection:
            for path, rating, plays, disposal in zip(paths, (2, 5), (3, 9), ('vanished', 'trash')):
                connection.execute(
                    'INSERT INTO asset(location,path,name,medium,size,first_seen,last_seen,rating,play_count,disposal) '
                    "VALUES('local',?,'copy.mp4','video',99,'2000-01-01','2000-01-01',?,?,?)",
                    (path, rating, plays, disposal))
            before = list(connection.execute('SELECT id,path,size,rating,play_count,disposal FROM asset ORDER BY id'))
        issues = []
        with patch.object(scan, 'BATCH_SIZE', 1):
            result, output = self._scan('R:\\media', declared_roots={'local': ('R:\\media',)},
                                       mounts={'local': (self.media,)}, windows=False, onerror=issues.append)
        self.assertEqual((result.files, result.path_conflicts, result.gone), (3, 1, 0))
        self.assertEqual(len(result.new_ids), 2)
        self.assertEqual(len(issues), 1)
        self.assertIsInstance(issues[0], scan.ScanPathConflict)
        self.assertEqual(issues[0].paths, paths)
        self.assertEqual(issues[0].asset_ids, (1, 2))
        self.assertIn('路径重复待复核 1 组', output)
        with closing(sqlite3.connect(self.db)) as connection:
            self.assertEqual(list(connection.execute(
                'SELECT id,path,size,rating,play_count,disposal FROM asset WHERE id<=2 ORDER BY id')), before)
            self.assertEqual(connection.execute('SELECT count(*) FROM asset').fetchone()[0], 4)
            self.assertTrue(all(row[0] > '2000-01-01' for row in connection.execute(
                'SELECT last_seen FROM asset WHERE id<=2')))

    def test_missing_conflict_records_remain_unobserved_and_do_not_block_another_root(self):
        (self.root / 'other').mkdir()
        (self.root / 'other' / 'new.mp4').write_bytes(b'new')
        with closing(sqlite3.connect(self.db)) as connection, connection:
            for path in ('R:\\media\\missing.mp4', 'r:\\MEDIA\\MISSING.mp4'):
                connection.execute("INSERT INTO asset(location,path,name,medium,last_seen) "
                                   "VALUES('local',?,'missing.mp4','video','2000-01-01')", (path,))
        options = dict(declared_roots={'local': ('R:\\media', 'S:\\other')},
                       mounts={'local': (self.media, self.root / 'other')}, windows=False)
        first, _ = self._scan('R:\\media', **options)
        second, _ = self._scan('S:\\other', **options)
        self.assertEqual((first.path_conflicts, first.gone), (1, 2))
        self.assertEqual((second.path_conflicts, second.files, second.gone), (0, 1, 0))
        with closing(sqlite3.connect(self.db)) as connection:
            self.assertEqual(list(connection.execute('SELECT last_seen FROM asset WHERE id<=2')),
                             [('2000-01-01',), ('2000-01-01',)])

    def test_subtitles_do_not_choose_between_conflicting_video_records(self):
        (self.media / 'copy.mp4').write_bytes(b'present')
        (self.media / 'copy.srt').write_text('subtitle', encoding='utf-8')
        options = dict(declared_roots={'local': ('R:\\media',)}, mounts={'local': (self.media,)}, windows=False)
        self._scan('R:\\media', **options)
        with closing(sqlite3.connect(self.db)) as connection, connection:
            before = list(connection.execute('SELECT id,asset_id,path,pairing FROM asset_subtitle'))
            self.assertEqual(len(before), 1)
            connection.execute("INSERT INTO asset(location,path,name,medium) "
                               "VALUES('local','r:\\MEDIA\\COPY.mp4','copy.mp4','video')")
        (self.media / 'copy.en.srt').write_text('another subtitle', encoding='utf-8')
        result, _ = self._scan('R:\\media', **options)
        self.assertEqual(result.path_conflicts, 1)
        self._scan('R:\\media', **options, resume_state={'mapping': str(self.media), 'pending': [],
                       'current_directory': str(self.media), 'cursor': 'copy.srt',
                       'here': {'copy.mp4': (7, '2000-01-01'), 'copy.srt': (8, '2000-01-01')}})
        with closing(sqlite3.connect(self.db)) as connection:
            self.assertEqual(list(connection.execute('SELECT id,asset_id,path,pairing FROM asset_subtitle')), before)

    def test_relocated_mount_restarts_traversal_and_preserves_file_id(self):
        saved = []
        options = dict(declared_roots={'local': ('R:\\media',)}, mounts={'local': (self.media,)}, windows=False)
        self._scan('R:\\media', save_checkpoint=lambda value: saved.append(value), **options)
        with closing(sqlite3.connect(self.db)) as connection, connection:
            before = list(connection.execute("SELECT id,path FROM asset ORDER BY id"))
        moved = self.root / 'moved'
        self.media.rename(moved)
        (moved / 'new.mp4').write_bytes(b'new')
        self._scan('R:\\media', resume_state=saved[-1], **{**options, 'mounts': {'local': (moved,)}})
        with closing(sqlite3.connect(self.db)) as connection, connection:
            self.assertEqual(list(connection.execute("SELECT id,path FROM asset ORDER BY id LIMIT 3")), before)
            self.assertEqual(connection.execute('SELECT count(*) FROM asset').fetchone()[0], 4)

    def test_scanning_one_root_does_not_count_other_roots_as_missing(self):
        options = dict(declared_roots={'local': ('R:\\media', 'S:\\other')},
                       mounts={'local': (self.media, self.root / 'other')}, windows=False)
        with closing(sqlite3.connect(self.db)) as connection, connection:
            connection.execute("INSERT INTO asset(location,path,name,medium,last_seen) VALUES('local','S:\\other\\copy.mp4','copy.mp4','video','2000-01-01')")
        result, _ = self._scan('R:\\media', **options)
        self.assertEqual(result.gone, 0)
        self.assertEqual(len(result.new_ids), 3)

    def test_resume_skips_committed_directories_and_retains_subtitles_and_new_ids(self):
        (self.media / '创作者/sub/a.srt').write_text('subtitle', encoding='utf-8')
        saved = {}
        def save(state):
            saved.clear()
            saved.update(json.loads(json.dumps(state)))
            if state.get('files') == 2:
                raise InterruptedError('模拟服务退出')
        options = dict(declared_roots={'local': ('R:\\media',)},
                       mounts={'local': (self.media,)}, windows=False)
        with self.assertRaises(InterruptedError):
            self._scan('R:\\media', save_checkpoint=save, **options)
        visited = []
        scandir = os.scandir
        def walk(path):
            visited.append(Path(path))
            return scandir(path)
        with patch.object(scan.os, 'scandir', side_effect=walk):
            result, _ = self._scan('R:\\media', resume_state=saved, **options)
        self.assertNotIn(self.media, visited)
        self.assertEqual((result.files, result.total_bytes, result.subtitles), (4, 22, 1))
        self.assertEqual(len(result.new_ids), 4)
        self.assertEqual(len(self._paths()), 4)

    def test_finished_root_checkpoint_does_not_enumerate_the_tree(self):
        saved = []
        options = dict(declared_roots={'local': ('R:\\media',)}, mounts={'local': (self.media,)}, windows=False)
        first, _ = self._scan('R:\\media', save_checkpoint=lambda value: saved.append(value), **options)
        with patch.object(scan.os, 'scandir', side_effect=AssertionError('已完成的根不应再次枚举')):
            second, _ = self._scan('R:\\media', resume_state=json.loads(json.dumps(saved[-1])), **options)
        self.assertEqual(second, first)

    def test_release_checkpoint_without_conflict_count_preserves_completed_work(self):
        saved = []
        options = dict(declared_roots={'local': ('R:\\media',)}, mounts={'local': (self.media,)}, windows=False)
        first, _ = self._scan('R:\\media', save_checkpoint=lambda value: saved.append(value), **options)
        released = json.loads(json.dumps(saved[-1]))
        del released['result']['path_conflicts']
        with closing(sqlite3.connect(self.db)) as connection:
            before = list(connection.execute('SELECT id,path,first_seen,last_seen FROM asset ORDER BY id'))
        with patch.object(scan.os, 'scandir', side_effect=AssertionError('已完成目录保留扫描结果')):
            resumed, _ = self._scan('R:\\media', resume_state=released, **options)
        self.assertEqual(resumed, first)
        self.assertEqual(resumed.path_conflicts, 0)
        with closing(sqlite3.connect(self.db)) as connection:
            self.assertEqual(list(connection.execute('SELECT id,path,first_seen,last_seen FROM asset ORDER BY id')), before)

    def test_file_batch_cursor_resumes_inside_a_directory(self):
        saved = {}
        def save(state):
            saved.update(json.loads(json.dumps(state)))
            if state.get('cursor'):
                raise InterruptedError('模拟文件批次提交后退出')
        options = dict(declared_roots={'local': ('R:\\media',)}, mounts={'local': (self.media,)}, windows=False)
        with patch.object(scan, 'BATCH_SIZE', 1), self.assertRaises(InterruptedError):
            self._scan('R:\\media', save_checkpoint=save, **options)
        self.assertEqual(saved['files'], 1)
        (self.media / saved['cursor']).write_bytes(b'changed after checkpoint')
        progress = []
        result, _ = self._scan('R:\\media', resume_state=saved,
                               progress=lambda count, size: progress.append((count, size)), **options)
        self.assertEqual((result.files, result.total_bytes), (3, 14))
        self.assertTrue(all(count >= 1 for count, _ in progress))
        self.assertEqual(len(self._paths()), 3)

    def test_structured_progress_reports_discovered_files_and_bytes(self):
        progress = []
        result, _ = self._scan('R:\\media', declared_roots={'local': ('R:\\media',)},
                               mounts={'local': (str(self.media),)}, windows=False,
                               progress=lambda files, size: progress.append((files, size)))
        self.assertEqual(progress[0][0], 1)
        self.assertEqual(progress[-1], (3, 14))
        self.assertEqual(progress[-1], (result.files, result.total_bytes))

    DECLARED = {"local": (r"R:\media",)}

    def test_rows_keep_the_ledger_shape_and_translate_back_to_real_files(self):
        mounts = {"local": (self.media,)}
        result, _ = self._scan(r"R:\media", declared_roots=self.DECLARED, mounts=mounts,
                               windows=False)
        self.assertEqual(result.files, 3)
        rows = self._paths()
        self.assertEqual(set(rows), {r"R:\media\创作者\sub\a.mp4", r"R:\media\cover.jpg",
                                     r"R:\media\notes.txt"})
        # 与 `platform.translate_ledger_path` 一致：路径前缀后的层级拼到本机媒体目录。
        for path in rows:
            tail = PureWindowsPath(path).relative_to(PureWindowsPath(r"R:\media")).parts
            self.assertTrue(self.media.joinpath(*tail).is_file(), path)

    def test_a_local_directory_is_converted_to_the_ledger_shape(self):
        mounts = {"local": (str(self.media),)}
        root = scan.ledger_root_for("local", self.media / "创作者",
                                    declared_roots=self.DECLARED, mounts=mounts, windows=False)
        self.assertEqual(root, r"R:\media\创作者")
        result, _ = self._scan(root, declared_roots=self.DECLARED, mounts=mounts, windows=False)
        self.assertEqual(set(self._paths()), {r"R:\media\创作者\sub\a.mp4"})

    def test_a_source_without_a_mount_is_refused_before_touching_the_ledger(self):
        with self.assertRaises(scan.ScanTargetError) as caught:
            self._scan(r"R:\media", declared_roots=self.DECLARED, mounts={}, windows=False)
        self.assertIn("media.mounts", str(caught.exception))
        self.assertEqual(self._paths(), {})

    def test_a_second_root_maps_to_its_own_mount(self):
        """两个目录都是 `local`：第二个挂载目录对应第二个数据库路径前缀。"""
        second = self.root / "more"
        (second / "deep").mkdir(parents=True)
        (second / "deep" / "c.mp4").write_bytes(b"2")
        declared = {"local": (r"R:\media", r"R:\media2")}
        mounts = {"local": (str(self.media), str(second))}
        root = scan.ledger_root_for("local", second / "deep", declared_roots=declared,
                                    mounts=mounts, windows=False)
        self.assertEqual(root, r"R:\media2\deep")
        result, _ = self._scan(root, declared_roots=declared, mounts=mounts, windows=False)
        self.assertEqual(result.files, 1)
        self.assertEqual(set(self._paths()), {r"R:\media2\deep\c.mp4"})
        # 第二个路径前缀没有对应的挂载目录时拒绝扫描，不能读取第一个目录。
        with self.assertRaises(scan.ScanTargetError) as caught:
            scan.walk_root_for("local", r"R:\media2", declared_roots=declared,
                               mounts={"local": (str(self.media),)}, windows=False)
        self.assertIn("第 2 个数据库路径前缀", str(caught.exception))

    def test_a_directory_outside_the_mount_is_refused(self):
        with self.assertRaises(scan.ScanTargetError) as caught:
            scan.ledger_root_for("local", "/somewhere/else", declared_roots=self.DECLARED,
                                 mounts={"local": (str(self.media),)}, windows=False)
        self.assertIn("挂载点", str(caught.exception))


class CaseOnlyRenameTests(_ScanCase):
    """只改大小写的文件与目录（#487）：扫描跟随磁盘上的实际写法，资产 id 与个人资料不变。"""

    options = dict(declared_roots={'local': (r'R:\media',)}, windows=False)

    def scan(self):
        return self._scan(r'R:\media', **self.options, mounts={'local': (self.media,)})

    def video(self):
        with closing(sqlite3.connect(self.db)) as connection:
            return connection.execute("SELECT id,path,name,rating FROM asset WHERE medium='video'").fetchall()

    def test_a_file_renamed_only_in_case_keeps_its_id_and_follows_the_new_spelling(self):
        self.scan()
        with closing(sqlite3.connect(self.db)) as connection, connection:
            asset_id = connection.execute("SELECT id FROM asset WHERE medium='video'").fetchone()[0]
            connection.execute('UPDATE asset SET rating=4 WHERE id=?', (asset_id,))
        sub = self.media / '创作者' / 'sub'
        os.rename(sub / 'a.mp4', sub / 'A.mp4')
        self.scan()
        self.assertEqual(self.video(), [(asset_id, r'R:\media\创作者\sub\A.mp4', 'A.mp4', 4)])

    def test_a_directory_renamed_only_in_case_moves_the_files_under_it(self):
        self.scan()
        asset_id = self.video()[0][0]
        os.rename(self.media / '创作者' / 'sub', self.media / '创作者' / 'SUB')
        self.scan()
        self.assertEqual([row[:2] for row in self.video()], [(asset_id, r'R:\media\创作者\SUB\a.mp4')])

    def test_both_spellings_present_on_a_case_sensitive_disk_are_left_alone(self):
        sub = self.media / '创作者' / 'sub'
        (sub / 'A.mp4').write_bytes(b'1' * 10)
        if not (sub / 'a.mp4').exists() or (sub / 'A.mp4').samefile(sub / 'a.mp4'):
            self.skipTest('文件系统不区分大小写，两种写法是同一个文件')
        self.scan()
        paths = sorted(row[1] for row in self.video())
        self.assertIn(r'R:\media\创作者\sub\a.mp4', paths)


class SidecarTests(_ScanCase):
    """刮削器与截图工具写在正片旁边的附属文件不登记，图集里同名的图照常登记。"""

    DECLARED = {"local": (r"R:\media",)}

    def _layout(self, files):
        for name in files:
            path = self.media.joinpath(*name.split("/"))
            path.parent.mkdir(parents=True, exist_ok=True)
            path.write_bytes(b"0")

    def test_a_kodi_movie_folder_registers_only_the_video(self):
        self._layout(["FC2-1/FC2-1-CD1.mp4", "FC2-1/FC2-1-CD1.nfo", "FC2-1/poster.jpg",
                      "FC2-1/fanart.jpg", "FC2-1/thumb.jpg", "FC2-1/extrafanart/1.jpg",
                      "FC2-1/FC2-1-CD1-landscape.jpg",
                      "thumbs_new/detailed/FC2-1_4x4_thumb.jpg", "ABC-1.mp4_thumbs.jpg"])
        result, output = self._scan(r"R:\media", declared_roots=self.DECLARED,
                                    mounts={"local": (self.media,)}, windows=False)
        self.assertEqual(result.sidecars, 8)
        self.assertIn("附属文件未登记 8 个", output)
        self.assertEqual(set(self._paths()), {
            r"R:\media\FC2-1\FC2-1-CD1.mp4", r"R:\media\创作者\sub\a.mp4",
            r"R:\media\cover.jpg", r"R:\media\notes.txt"})

    def test_artwork_names_without_a_video_beside_them_are_photos(self):
        self._layout(["写真/cover.jpg", "写真/extrafanart/1.jpg", "写真/poster-1.jpg"])
        self._scan(r"R:\media\写真", declared_roots=self.DECLARED,
                   mounts={"local": (self.media,)}, windows=False)
        self.assertEqual(set(self._paths()), {
            r"R:\media\写真\cover.jpg", r"R:\media\写真\extrafanart\1.jpg",
            r"R:\media\写真\poster-1.jpg"})

    def test_movie_trailers_are_sidecars_and_short_main_features_are_kept(self):
        self._layout(['ABC-123/ABC-123.mp4', 'ABC-123/ABC-123-trailer.mp4',
                      'ABC-123/ABC-123.nfo', 'ABC-123/ABC-123-poster.jpg',
                      'ABC-123/extrafanart/1.jpg', 'ABC-123/short-film.mp4'])
        result, _ = self._scan(r'R:\media\ABC-123', declared_roots=self.DECLARED,
                               mounts={'local': (self.media,)}, windows=False)
        self.assertEqual((result.files, result.sidecars), (2, 4))
        self.assertEqual(set(self._paths()), {
            r'R:\media\ABC-123\ABC-123.mp4', r'R:\media\ABC-123\short-film.mp4'})
        self.assertTrue((self.media / 'ABC-123/ABC-123-trailer.mp4').is_file())

    def test_trailer_names_use_an_explicit_terminal_marker(self):
        self.assertEqual(scan.video_stems(['ABC-123.mp4', 'ABC-123-trailer.mp4']),
                         frozenset({'abc-123'}))
        for name in ('ABC-123-trailer.mp4', 'ABC-123.TRAILER.MKV', 'ABC-123_trailer.mov',
                     'ABC-123 trailer.m4v', 'trailer.mp4'):
            with self.subTest(name=name):
                self.assertTrue(scan.is_sidecar(name, frozenset()))
        for name in ('ABC-123.mp4', 'Trailer Park.mp4', 'ABC-123-trailer-park.mp4',
                     'ABC-123trailer.mp4', 'ABC-123-trailer.mp4.bak', 'trailer.jpg'):
            with self.subTest(name=name):
                self.assertFalse(scan.is_sidecar(name, frozenset()))

    def test_media_server_extras_and_nested_directories_are_not_registered(self):
        paths = ['Film/Film.mkv', 'Film/Trailer 1-trailer.MP4', 'Film/Interview-interview2.mkv',
                 'Film/trailers/Teaser.mp4', 'Film/Extras/Trailers/Trailer 2.mkv',
                 'Film/behind the scenes/Making of.mp4', 'Film/deleted scenes/Ending.mp4',
                 'Film/shorts/Bonus.mp4', 'Film/specials/Bonus.mp4',
                 'Film/backdrops/Menu.mkv', 'Film/theme-music/Score.flac', 'Film/theme.mp3']
        self._layout(paths)
        result, _ = self._scan(r'R:\media\Film', declared_roots=self.DECLARED,
                               mounts={'local': (self.media,)}, windows=False)
        self.assertEqual((result.files, result.sidecars), (1, len(paths) - 1))
        self.assertEqual(self._paths(), {r'R:\media\Film\Film.mkv': 'video'})
        self.assertTrue(all((self.media / name).is_file() for name in paths))

    def test_a_declared_shorts_root_contains_independent_features(self):
        self._layout(['Shorts/One.mp4', 'Shorts/Film/extras/Making of.mp4'])
        result, _ = self._scan(r'R:\media\Shorts', declared_roots={'local': (r'R:\media\Shorts',)},
                               mounts={'local': (self.media / 'Shorts',)}, windows=False)
        self.assertEqual((result.files, result.sidecars), (1, 1))
        self.assertEqual(self._paths(), {r'R:\media\Shorts\One.mp4': 'video'})


class LocalExtraNamingTests(unittest.TestCase):
    def test_movie_and_episode_suffixes_and_numbered_bare_names(self):
        examples = {
            'trailer.mp4': 'trailer', 'trailer2.MKV': 'trailer', 'sample1.mp4': 'sample',
            'Film-trailer1.mp4': 'trailer', 'Film.trailer2.mkv': 'trailer',
            'Film_trailer3.mov': 'trailer', 'Film trailer4.m4v': 'trailer',
            'Film- trailer5.mp4': 'trailer', 'Film-sample1.mp4': 'sample',
            'Film.sample2.mkv': 'sample', 'Film_sample3.mov': 'sample',
            'Film sample4.mp4': 'sample', 'Film- sample5.mp4': 'sample',
            'Performance Capture-behindthescenes.mkv': 'behindthescenes',
            'Ending-deleted.mp4': 'deleted', 'Ending-deletedscene2.mp4': 'deleted',
            'Special Effects-featurette.mp4': 'featurette', 'Actor-interview.mp4': 'interview',
            'Arrival-scene.mp4': 'scene', 'Moment-clip.mp4': 'clip',
            'Bonus-short.mp4': 'short', 'Bonus-extra.mp4': 'other', 'Bonus-other.mp4': 'other',
            'Show - S01E01 - Episode-deleted2.mkv': 'deleted', 'theme.flac': 'themeaudio',
        }
        for name, expected in examples.items():
            with self.subTest(name=name):
                self.assertEqual(catalog_rules.local_extra_type(name), expected)
                self.assertTrue(scan.is_sidecar(name, frozenset()))

    def test_exact_markers_preserve_titles_and_unsupported_files(self):
        for name in ('Short Film.mp4', 'Trailer Park.mp4', 'Interview.mp4', 'Sample Story.mp4',
                     'Filmtrailer.mp4', 'Film-short-story.mp4', 'Film-deleted-scenes.mp4',
                     'Film-trailerA.mp4', 'Film-trailer.mp4.bak', 'Film-trailer.jpg',
                     'Film-interview.mp3', 'theme.mp4', '', None):
            with self.subTest(name=name):
                self.assertIsNone(catalog_rules.local_extra_type(name))

    def test_dedicated_directories_are_case_insensitive_and_allow_nesting(self):
        directories = {'Trailers': 'trailer', 'Behind The Scenes': 'behindthescenes',
            'Deleted Scenes': 'deleted', 'Interviews': 'interview', 'Scenes': 'scene',
            'Samples': 'sample', 'Sample': 'sample', 'Shorts': 'short',
            'Featurettes': 'featurette', 'Clips': 'clip', 'Other': 'other',
            'Extras': 'other', 'Extra': 'other', 'Specials': 'other', 'Backdrops': 'themevideo'}
        for directory, expected in directories.items():
            for path in (f'R:/Movies/Film/{directory}/Clip.mp4',
                         f'R:/Movies/Film/{directory}/Disc 1/Clip.mp4'):
                with self.subTest(path=path):
                    self.assertEqual(catalog_rules.local_extra_type('Clip.mp4', path,
                                     library_roots=('R:/Movies',)), expected)
        self.assertEqual(catalog_rules.local_extra_type('Score.opus', '/movies/Film/theme-music/Score.opus'),
                         'themeaudio')
        self.assertIsNone(catalog_rules.local_extra_type('Score.opus', '/movies/Film/trailers/Score.opus'))
        self.assertEqual(catalog_rules.local_extra_type('Clip.mp4', '/movies/Film/Extras/theme-music/Clip.mp4'),
                         'other')
        self.assertEqual(catalog_rules.local_extra_type('Score.opus', '/movies/Film/theme-music/clips/Score.opus'),
                         'themeaudio')

    def test_longest_library_root_is_the_directory_boundary(self):
        for path, roots in ((r'R:\Shorts\Film.mp4', ('R:/Shorts',)),
                            ('/Movies/Extras/Film.mp4', ('/Movies', '/Movies/Extras')),
                            (r'\\server\Shorts\Film.mp4', (r'\\server\Shorts',))):
            with self.subTest(path=path):
                self.assertIsNone(catalog_rules.local_extra_type('Film.mp4', path, library_roots=roots))
        self.assertEqual(catalog_rules.local_extra_type('Clip.mp4', 'R:/Shorts/Film/Extras/Clip.mp4',
                         library_roots=('r:/shorts',)), 'other')

    def test_sqlite_classifier_uses_configured_library_boundaries(self):
        from peach.repository import LedgerDatabase
        with tempfile.TemporaryDirectory() as temporary:
            path = Path(temporary).resolve() / 'ledger.db'
            sqlite3.connect(path).close()
            database = LedgerDatabase(path, media_roots=('R:/Movies', 'R:/Movies/Shorts'))
            with database.read_connection() as connection:
                values = [connection.execute("SELECT peach_is_local_extra('Film.mp4',?)", (value,)).fetchone()[0]
                          for value in ('R:/Movies/Shorts/Film.mp4', 'R:/Movies/Shorts/Film/Extras/Film.mp4')]
            self.assertEqual(values, [0, 1])

    def test_the_rule_reads_names_only(self):
        videos = scan.video_stems(["ABC-1.mp4", "notes.txt"])
        self.assertEqual(videos, frozenset({"abc-1"}))
        sidecar = {name: scan.is_sidecar(name, videos) for name in (
            "x.NFO", "Poster.JPG", "ABC-1-fanart.jpg", "ABC-1.jpg", "OTHER-fanart.jpg",
            "a.png.thumb.png", "ABC-1_3x3_thumbs.jpg", "ABC-1.srt", "001.jpg")}
        self.assertEqual({name for name, hit in sidecar.items() if hit}, {
            "x.NFO", "Poster.JPG", "ABC-1-fanart.jpg", "a.png.thumb.png",
            "ABC-1_3x3_thumbs.jpg"})


class ScanTargetTests(unittest.TestCase):
    """写入侧门槛：`location` 与扫描根必须对得上（ADR-0023 第 2 阶段），口径与脚本时期一致。"""

    DECLARED = {"local": (r"R:\media",), "115": ("B:/",)}

    def test_the_declared_root_and_its_subdirectories_are_accepted(self):
        for location, root in (("local", r"R:\media"), ("local", r"R:\media\创作者"),
                               ("115", "B:/x")):
            self.assertIsNone(scan.check_scan_target(location, root,
                                                     declared_roots=self.DECLARED))

    def test_a_root_belonging_to_another_source_is_refused(self):
        with self.assertRaises(scan.ScanTargetError) as caught:
            scan.check_scan_target("115", r"R:\media\创作者", declared_roots=self.DECLARED)
        self.assertIn("115", str(caught.exception))
        self.assertIn("local", str(caught.exception))

    def test_a_root_outside_every_declared_root_is_refused(self):
        with self.assertRaises(scan.ScanTargetError) as caught:
            scan.check_scan_target("local", r"R:\Resources\Intake", declared_roots=self.DECLARED)
        self.assertIn(r"R:\media", str(caught.exception))

    def test_an_undeclared_source_is_refused_and_lists_the_known_ones(self):
        with self.assertRaises(scan.ScanTargetError) as caught:
            scan.check_scan_target("nas", "N:/", declared_roots=self.DECLARED)
        self.assertIn("nas", str(caught.exception))
        self.assertIn("local", str(caught.exception))

    def test_medium_is_decided_by_extension(self):
        self.assertEqual([scan.medium_of(name) for name in
                          ("a.MP4", "b.webp", "c.flac", "d.7z", "e.nfo")],
                         ["video", "image", "audio", "archive", "other"])


class NasDirectoryTests(unittest.TestCase):
    def test_system_directories_are_skipped_and_similarly_named_files_are_kept(self):
        with tempfile.TemporaryDirectory() as temporary:
            root = Path(temporary).resolve()
            for name in ("#recycle", "#RECYCLE", "@eaDir", "@snapshot", "@Recycle",
                         "@Recently-Snapshot", ".@__thumb", ".@__qini", ".@upload_cache", ".streams",
                         ".zfs", ".snapshot", ".snapshots", "~snapshot", ".recycle", ".Trash", ".Trash-1000",
                         ".TemporaryItems", ".AppleDouble", ".AppleDB", ".@#local",
                         ".AppleDesktop", "Network Trash Folder", ".Spotlight-V100", ".Trashes",
                         ".fseventsd", "__MACOSX", "$Recycle.Bin", "System Volume Information",
                         ".DocumentRevisions-V100", ".DocumentRevisions-V100-bad-1",
                         ".wfm_upload123", "normal", "@creator", ".hidden"):
                folder = root / name
                folder.mkdir(exist_ok=True)
                (folder / "sample.mp4").write_bytes(b"x")
            for name in ("@video.mp4", "#recycle.mp4"):
                (root / name).write_bytes(b"x")
            files = [Path(entry.path).relative_to(root).as_posix()
                     for _, entries in scan._walk(root) for entry in entries]
        self.assertEqual(set(files), {"normal/sample.mp4", "@creator/sample.mp4", ".hidden/sample.mp4",
                                      "@video.mp4", "#recycle.mp4"})


class SystemMetadataTests(_ScanCase):
    def test_a_system_directory_scan_root_is_skipped_before_mapping_or_listing(self):
        options = dict(declared_roots={'local': (r'R:\media\.snapshots',)},
                       mounts={'local': (self.media,)}, windows=False)
        progress = []
        with patch.object(scan, 'walk_root_for') as mapped, patch.object(scan, '_walk') as walked:
            result, output = self._scan(r'R:\media\.snapshots', **options,
                                       progress=lambda files, total: progress.append((files, total)))
        self.assertEqual((result.files, result.total_bytes, result.gone), (0, 0, 0))
        self.assertEqual(progress, [(0, 0)])
        self.assertIn('0 文件', output)
        mapped.assert_not_called()
        walked.assert_not_called()
        self.assertEqual(self._paths(), {})
        with self.assertRaises(scan.ScanTargetError):
            self._scan(r'R:\media\.snapshots', declared_roots={})

    def test_system_walk_roots_and_resumed_directories_are_not_listed(self):
        system = self.media / '.Trash-1000'
        system.mkdir()
        (system / 'movie.mp4').write_bytes(b'media')
        with patch.object(scan.os, 'scandir') as listed:
            self.assertEqual(list(scan._walk(system)), [])
            self.assertEqual(list(scan._walk(self.media, [str(system)])), [])
        listed.assert_not_called()

    def test_full_scan_and_direct_ingest_share_metadata_file_rules(self):
        names = ('desktop.ini', 'Thumbs.db', '.DS_Store', '._movie.mp4', '.SYNOPPSDB123',
                 '.wfm_upload123.mp4', 'movie.mp4@SynoEAStream')
        for name in names:
            (self.media / name).write_bytes(b'metadata')
        (self.media / '@creator').mkdir()
        (self.media / '@creator' / '.hidden.mp4').write_bytes(b'media')
        options = dict(declared_roots={'local': (r'R:\media',)},
                       mounts={'local': (self.media,)}, windows=False)
        for name in names:
            result = scan.ingest_path(self.db, 'local', str(PureWindowsPath(r'R:\media', name)), **options)
            self.assertEqual((result.found, result.sidecar, result.asset_id), (True, True, None), name)
        self.assertEqual(self._paths(), {})
        cover = scan.ingest_path(self.db, 'local', r'R:\media\cover.jpg', **options)
        self.assertEqual((cover.found, cover.sidecar), (True, False))
        result, _ = self._scan(r'R:\media', **options)
        self.assertEqual(result.files, 4)
        self.assertIn(r'R:\media\@creator\.hidden.mp4', self._paths())
        self.assertTrue(all((self.media / name).is_file() for name in names))


if __name__ == "__main__":
    unittest.main()