"""改字工具保存源文案与动态模板，冲突与代码路径不能误写。"""
import json
from pathlib import Path
import tempfile
import unittest
from unittest.mock import patch
from types import SimpleNamespace
import threading
import time
import os
from peach import dev_copy


class DevCopyTests(unittest.TestCase):
    def test_python_copy_sources_can_be_saved_with_digest_protection(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp).resolve()
            (root / 'src/peach').mkdir(parents=True)
            for name in ('routes_pages.py', 'onboarding.py', 'media_sources.py', 'release_updates.py'):
                with self.subTest(name=name):
                    path = root / 'src/peach' / name
                    path.write_text("LABEL = '设置说明'", encoding='utf-8')
                    candidate = next(row for row in dev_copy.candidates(root, '设置说明') if row['file'].endswith(name))
                    dev_copy.save(root, {'original': '设置说明', 'replacement': '媒体设置说明', 'candidate': candidate})
                    self.assertEqual(path.read_text(encoding='utf-8'), "LABEL = '媒体设置说明'")
                    with self.assertRaises(ValueError):
                        dev_copy.save(root, {'original': '设置说明', 'replacement': '无效修改', 'candidate': candidate})

    def test_extracted_copy_is_saved_into_the_source_language_file_only(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp).resolve()
            for language, text in (('zh-CN', '首页'), ('en', 'Home')):
                (root / 'frontend/src/locales' / language).mkdir(parents=True)
                (root / 'frontend/src/locales' / language / 'navigation.json').write_text(
                    '{\n  "sidebar": { "home": "%s" }\n}\n' % text, encoding='utf-8')
            candidate, = dev_copy.candidates(root, '首页')
            self.assertEqual(candidate['file'], 'frontend/src/locales/zh-CN/navigation.json')
            self.assertEqual(dev_copy.candidates(root, 'Home'), [])
            dev_copy.save(root, {'original': '首页', 'replacement': '主页 "精选"', 'candidate': candidate})
            saved = json.loads((root / candidate['file']).read_text(encoding='utf-8'))
            self.assertEqual(saved, {'sidebar': {'home': '主页 "精选"'}})

    def test_interpolated_locale_copy_is_found_by_its_rendered_text_and_keeps_placeholders(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp).resolve()
            (root / 'frontend/src/locales/zh-CN').mkdir(parents=True)
            path = root / 'frontend/src/locales/zh-CN/cleanup.json'
            path.write_text('{\n  "trash": { "count": "{{count}} 项在回收站" }\n}\n', encoding='utf-8')
            candidate, = dev_copy.candidates(root, '12 项在回收站')
            self.assertEqual(candidate['template'], '{count} 项在回收站')
            self.assertEqual(candidate['values'], {'count': '12'})
            with self.assertRaises(ValueError):
                dev_copy.save(root, {'original': '12 项在回收站', 'replacement': '回收站里有若干项', 'candidate': candidate})
            result = dev_copy.save(root, {'original': '12 项在回收站', 'replacement': '回收站里有 {count} 项', 'candidate': candidate})
            self.assertEqual(result['text'], '回收站里有 12 项')
            self.assertEqual(json.loads(path.read_text(encoding='utf-8')), {'trash': {'count': '回收站里有 {{count}} 项'}})

    def test_status_recovers_from_a_short_read_lock_and_preserves_persistent_denial(self):
        from peach import dev_copy_runtime as runtime
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp).resolve()
            with patch.object(Path, 'read_text', side_effect=[PermissionError('sharing'), '{"state":"ready"}']), \
                    patch.object(runtime.time, 'sleep') as sleep:
                self.assertEqual(runtime.status(root)['state'], 'ready')
                sleep.assert_called_once()
            with patch.object(Path, 'read_text', side_effect=PermissionError('denied')) as read, \
                    patch.object(runtime.time, 'sleep') as sleep, self.assertRaises(PermissionError):
                runtime.status(root)
            self.assertEqual(read.call_count, 5)
            self.assertEqual(sleep.call_count, 4)

    def test_static_text_can_be_deleted_and_dynamic_placeholders_are_required(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp).resolve()
            (root / 'frontend/src').mkdir(parents=True)
            path = root / 'frontend/src/copy.tsx'
            path.write_text("const a = '说明'; const b = <p>正文</p>;", encoding='utf-8')
            for original in ('说明', '正文'):
                candidate, = dev_copy.candidates(root, original)
                result = dev_copy.save(root, {'original': original, 'replacement': '', 'candidate': candidate})
                self.assertEqual(result['text'], '')
            self.assertEqual(path.read_text(encoding='utf-8'), "const a = ''; const b = <p></p>;")
            (root / 'src/peach').mkdir(parents=True)
            dynamic = root / 'src/peach/library_processing.py'
            dynamic.write_text('message = f"待处理 {count} 项"', encoding='utf-8')
            candidate, = dev_copy.candidates(root, '待处理 5 项')
            with self.assertRaises(ValueError):
                dev_copy.save(root, {'original': '待处理 5 项', 'replacement': '', 'candidate': candidate})
            self.assertEqual(dynamic.read_text(encoding='utf-8'), 'message = f"待处理 {count} 项"')

    def test_build_failure_preserves_served_files_and_can_be_retried(self):
        from peach import dev_copy_runtime as runtime
        from subprocess import CompletedProcess
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp).resolve()
            (root / 'frontend/src').mkdir(parents=True)
            path = root / 'frontend/src/copy.ts'
            path.write_text("const title = '说明';", encoding='utf-8')
            (root / 'web/dist').mkdir(parents=True)
            served = root / 'web/dist/peach-app.js'
            served.write_text('有效产物', encoding='utf-8')
            candidate, = dev_copy.candidates(root, '说明')
            with patch('peach.dev_copy_runtime.subprocess.run', return_value=CompletedProcess([], 1, '', '语法错误')):
                result = runtime.submit(root, {'original': '说明', 'replacement': '新说明', 'candidate': candidate})
                self._wait_for_state(runtime, root, 'error')
            self.assertEqual(served.read_text(encoding='utf-8'), '有效产物')
            self.assertEqual(path.read_text(encoding='utf-8'), "const title = '新说明';")
            self.assertIn('语法错误', runtime.status(root)['message'])
            with patch('peach.dev_copy_runtime.build_frontend'):
                retry = runtime.submit(root, {'id': result['apply']['id']}, retry=True)
                self._wait_for_state(runtime, root, 'ready')
            self.assertNotEqual(result['apply']['id'], retry['apply']['id'])

    def test_tray_reload_obeys_task_gate_and_releases_ownership_lock(self):
        from peach import dev_copy_runtime as runtime
        from peach.runtime_prepare import TaskGate, ActiveRun, Step
        from unittest.mock import Mock
        for blocked in (True, False):
            with self.subTest(blocked=blocked), tempfile.TemporaryDirectory() as tmp:
                root = Path(tmp).resolve()
                (root / 'src/peach').mkdir(parents=True)
                path = root / 'src/peach/library_processing.py'
                path.write_text("TITLE = '说明'", encoding='utf-8')
                candidate, = dev_copy.candidates(root, '说明')
                runtime.submit(root, {'original': '说明', 'replacement': '新说明', 'candidate': candidate})
                with self.assertRaises(ValueError):
                    runtime.submit(root, {})
                tray = SimpleNamespace(_action_lock=threading.Lock(), manager=Mock(),
                                       _readiness=lambda: Step('readiness', 'current', '服务已就绪'))
                gate = TaskGate(blocking=(ActiveRun(1, 'test', 'manual', 'running', os.getpid(), None),)) if blocked else TaskGate()
                locked_at_terminal = []
                def recording_write(target, data, write=runtime._write, lock=tray._action_lock, seen=locked_at_terminal):
                    if data.get('state') in ('ready', 'error'):
                        seen.append(lock.locked())
                    return write(target, data)
                with patch('peach.runtime_prepare.task_gate', return_value=gate), \
                        patch.object(runtime, '_write', side_effect=recording_write):
                    runtime.poll(tray, root)
                    self._wait_for_state(runtime, root, 'error' if blocked else 'ready')
                self.assertEqual(tray.manager.restart.call_count, 0 if blocked else 1)
                if not blocked:
                    tray.manager.restart.assert_called_once_with(reason='源码改字构建完成，重载服务')
                # 终态写盘时托盘动作锁已经释放，看到终态的调用方不会抢在释放之前。
                self.assertEqual(locked_at_terminal, [False])
                self.assertFalse(tray._action_lock.locked())

    @staticmethod
    def _wait_for_state(runtime, root, state):
        deadline = time.monotonic() + 5
        while time.monotonic() < deadline:
            if runtime.status(root).get('state') == state:
                return
            time.sleep(.01)
        raise AssertionError(runtime.status(root))

    def test_saved_frontend_copy_is_built_and_published(self):
        from peach import dev_copy_runtime
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp).resolve()
            (root / 'web/dist').mkdir(parents=True)
            (root / 'web/dist/peach-app.js').write_text('现有产物', encoding='utf-8')
            def build(argv, **kwargs):
                from subprocess import CompletedProcess
                stage = Path(kwargs['env']['PEACH_FRONTEND_OUT_DIR'])
                for name in dev_copy_runtime.BUNDLES:
                    (stage / name).write_text('有效产物', encoding='utf-8')
                self.assertEqual((root / 'web/dist/peach-app.js').read_text(encoding='utf-8'), '现有产物')
                return CompletedProcess(argv, 0, stdout='构建完成', stderr='')
            with patch('peach.dev_copy_runtime.subprocess.run', side_effect=build):
                dev_copy_runtime.build_frontend(root)
            self.assertEqual((root / 'web/dist/peach-app.js').read_text(encoding='utf-8'), '有效产物')

    def test_frontend_copy_publishes_new_chunks_and_drops_the_previous_ones(self):
        """主包按页拆出的 chunk 名字带哈希（ADR-0118）：新的一批跟入口一起发出，上一批随后删掉。"""
        from peach import dev_copy_runtime
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp).resolve()
            dist = root / 'web/dist'
            dist.mkdir(parents=True)
            (dist / 'peach-app.stats-page-Old00000.js').write_text('旧 chunk', encoding='utf-8')
            def build(argv, **kwargs):
                from subprocess import CompletedProcess
                stage = Path(kwargs['env']['PEACH_FRONTEND_OUT_DIR'])
                for name in (*dev_copy_runtime.BUNDLES, 'peach-app.stats-page-New11111.js'):
                    (stage / name).write_text(name, encoding='utf-8')
                (stage / 'peach-app.stats-page-New11111.js.map').write_text('{}', encoding='utf-8')
                return CompletedProcess(argv, 0, stdout='构建完成', stderr='')
            with patch('peach.dev_copy_runtime.subprocess.run', side_effect=build):
                dev_copy_runtime.build_frontend(root)
            self.assertEqual(sorted(path.name for path in dist.iterdir()),
                             sorted([*dev_copy_runtime.BUNDLES, 'peach-app.stats-page-New11111.js']))
            self.assertEqual((dist / 'peach-app.stats-page-New11111.js').read_text(encoding='utf-8'),
                             'peach-app.stats-page-New11111.js')

    def test_processing_template_keeps_live_values_and_validates_placeholders(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp).resolve()
            package = root / "src/peach"
            package.mkdir(parents=True)
            path = package / "library_processing.py"
            source = ("# 中文说明\n"
                      "def summary(problems, paused):\n"
                      "    return (f'{problems} 项需要处理，其中 {paused} 项是来源限流；'\n"
                      "            '其余可重试。')\n")
            path.write_text(source, encoding="utf-8")
            original = "768 项需要处理，其中 533 项是来源限流；其余可重试。"
            candidate, = dev_copy.candidates(root, original)
            self.assertEqual(candidate["template"], "{problems} 项需要处理，其中 {paused} 项是来源限流；其余可重试。")
            for replacement in ["处理 {problems} 项", "处理 {unknown} 项", "处理 {problems!r}、{paused}", "{problems} {paused} {paused}"]:
                with self.assertRaises(ValueError):
                    dev_copy.save(root, {"original": original, "replacement": replacement, "candidate": candidate})
                self.assertEqual(path.read_text(encoding="utf-8"), source)
            result = dev_copy.save(root, {"original": original,
                "replacement": "来源限流 {paused} 项，共 {problems} 项待处理。", "candidate": candidate})
            self.assertEqual(result["text"], "来源限流 533 项，共 768 项待处理。")
            namespace = {}
            exec(compile(path.read_text(encoding="utf-8"), str(path), "exec"), namespace)
            self.assertEqual(namespace["summary"](10, 3), "来源限流 3 项，共 10 项待处理。")
            self.assertEqual(len(list((root / "build/copy-editor/backups").rglob(path.name))), 1)
            with self.assertRaises(ValueError):
                dev_copy.save(root, {"original": original, "replacement": "{problems} {paused}", "candidate": candidate})

    def test_template_preserves_expressions_formatting_and_literal_braces(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp).resolve()
            package = root / "src/peach"
            package.mkdir(parents=True)
            path = package / "library_processing.py"
            path.write_text("def summary(total):\n    return f'共 {{数量}} {total + 1:04d} 项'\n", encoding="utf-8")
            candidate, = dev_copy.candidates(root, "共 {数量} 0003 项")
            self.assertEqual(candidate["template"], "共 {{数量}} {value1} 项")
            dev_copy.save(root, {"original": "共 {数量} 0003 项", "replacement": "{{总数}} '{value1}' 项", "candidate": candidate})
            namespace = {}
            exec(compile(path.read_text(encoding="utf-8"), str(path), "exec"), namespace)
            self.assertEqual(namespace["summary"](4), "{总数} '0005' 项")

    def test_library_notice_is_editable_and_other_python_files_are_excluded(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp).resolve()
            package = root / "src/peach"
            package.mkdir(parents=True)
            notice = package / "library_sharing.py"
            notice.write_text('NOTICE = "上传说明"\n', encoding="utf-8")
            (package / "config.py").write_text('SECRET = "上传说明"\n', encoding="utf-8")
            candidate, = dev_copy.candidates(root, "上传说明")
            self.assertEqual(candidate["file"], "src/peach/library_sharing.py")
            dev_copy.save(root, {"original": "上传说明", "replacement": "新的说明", "candidate": candidate})
            self.assertEqual(notice.read_text(encoding="utf-8"), 'NOTICE = "新的说明"\n')
            self.assertEqual((package / "config.py").read_text(encoding="utf-8"), 'SECRET = "上传说明"\n')

    def test_jsx_text_keeps_braces_literal_and_invalid_values_are_rejected(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            (root / "frontend/src").mkdir(parents=True)
            path = root / "frontend/src/example.tsx"
            path.write_text('<p>示例</p>', encoding="utf-8")
            candidate, = dev_copy.candidates(root, "示例")
            dev_copy.save(root, {"original": "示例", "replacement": "文字{原样}", "candidate": candidate})
            self.assertEqual(path.read_text(encoding="utf-8"), '<p>文字&#123;原样&#125;</p>')
            for body in [[], {"original": None, "replacement": "文字"}]:
                with self.assertRaises(ValueError):
                    dev_copy.save(root, body)

    def test_save_preserves_code_and_records_the_original_file(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp).resolve()
            (root / "frontend/src/application").mkdir(parents=True)
            path = root / "frontend/src/application/tutorial.js"
            path.write_text("const label='示例文字';", encoding="utf-8")
            candidate, = dev_copy.candidates(root, "示例文字")
            result = dev_copy.save(root, {"original": "示例文字", "replacement": "新的'文字", "candidate": candidate})
            self.assertTrue(result["ok"])
            self.assertTrue(result["needs_build"])
            self.assertEqual(path.read_text(encoding="utf-8"), "const label='新的\\'文字';")
            self.assertEqual(len(list((root / "build/copy-editor/backups").rglob("tutorial.js"))), 1)
            with self.assertRaises(ValueError):
                dev_copy.save(root, {"original": "示例文字", "replacement": "错误", "candidate": candidate})

    def test_ambiguous_labels_require_a_specific_unchanged_candidate(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            (root / "web").mkdir()
            path = root / "web/index.html"
            path.write_text('<p>示例</p><button>示例</button>', encoding="utf-8")
            choices = dev_copy.candidates(root, "示例")
            self.assertEqual(len(choices), 2)
            dev_copy.save(root, {"original": "示例", "replacement": "按钮", "candidate": choices[1]})
            self.assertEqual(path.read_text(encoding="utf-8"), '<p>示例</p><button>按钮</button>')
            with self.assertRaises(ValueError):
                dev_copy.save(root, {"original": "示例", "replacement": "错误", "candidate": {"file": "../secret"}})