diff --git a/src/wechat_decrypt_tool/chat_export_service.py b/src/wechat_decrypt_tool/chat_export_service.py index bdae2c3..46a11f0 100644 --- a/src/wechat_decrypt_tool/chat_export_service.py +++ b/src/wechat_decrypt_tool/chat_export_service.py @@ -53,7 +53,7 @@ from .chat_helpers import ( ) from .logging_config import get_logger from .media_helpers import ( - _convert_silk_to_wav, + _convert_silk_to_browser_audio, _detect_image_media_type, _fallback_search_media_by_file_id, _read_and_maybe_decrypt_media, @@ -121,9 +121,10 @@ def _resolve_ui_public_dir() -> Optional[Path]: if ui_dir_env: candidates.append(Path(ui_dir_env)) - # Repo default: `frontend/.output/public` after `npm --prefix frontend run generate`. + # Repo defaults: generated Nuxt output or checked-in desktop UI assets. repo_root = Path(__file__).resolve().parents[2] candidates.append(repo_root / "frontend" / ".output" / "public") + candidates.append(repo_root / "desktop" / "resources" / "ui") for p in candidates: try: @@ -622,6 +623,68 @@ body { background: #EDEDED; } .wce-audio-actions a { font-size: 0.75rem; color: #07c160; text-decoration: none; } .wce-audio-actions a:hover { text-decoration: underline; } +/* Voice message fallback styles (keep close to `frontend/pages/chat/[[username]].vue`). */ +.wechat-voice-wrapper { display: flex; width: 100%; position: relative; } +.wechat-voice-bubble { + border-radius: var(--message-radius); + position: relative; + transition: opacity 0.15s ease; + min-width: 80px; + max-width: 200px; + cursor: pointer; +} +.wechat-voice-bubble:hover { opacity: 0.85; } +.wechat-voice-bubble:active { opacity: 0.7; } +.wechat-voice-sent { background: #95EC69; } +.wechat-voice-sent::after { + content: ''; + position: absolute; + top: 50%; + right: -4px; + transform: translateY(-50%) rotate(45deg); + width: 10px; + height: 10px; + background: #95EC69; + border-radius: 2px; +} +.wechat-voice-received { background: #fff; } +.wechat-voice-received::before { + content: ''; + position: absolute; + top: 50%; + left: -4px; + transform: translateY(-50%) rotate(45deg); + width: 10px; + height: 10px; + background: #fff; + border-radius: 2px; +} +.wechat-voice-content { display: flex; align-items: center; padding: 8px 12px; gap: 8px; } +.wechat-voice-icon { width: 18px; height: 18px; flex-shrink: 0; color: #1a1a1a; } +.wechat-quote-voice-icon { width: 14px; height: 14px; color: inherit; } +.voice-icon-sent { transform: scaleX(-1); } +.wechat-voice-icon.voice-playing .voice-wave-2 { animation: voice-wave-2 1s infinite; } +.wechat-voice-icon.voice-playing .voice-wave-3 { animation: voice-wave-3 1s infinite; } +@keyframes voice-wave-2 { + 0%, 33% { opacity: 0; } + 34%, 100% { opacity: 1; } +} +@keyframes voice-wave-3 { + 0%, 66% { opacity: 0; } + 67%, 100% { opacity: 1; } +} +.wechat-voice-duration { font-size: 14px; color: #1a1a1a; } +.wechat-voice-unread { + position: absolute; + top: 50%; + right: -20px; + transform: translateY(-50%); + width: 8px; + height: 8px; + border-radius: 50%; + background: #e75e58; +} + /* Index page helpers. */ .wce-index { min-height: 100vh; background: #EDEDED; } .wce-index-container { max-width: 880px; margin: 0 auto; padding: 24px; } @@ -4958,40 +5021,38 @@ def _write_conversation_html( tw.write(f'
{render_text_with_emojis(msg.get("content") or "")}
\n') elif rt == "voice": voice = offline_path(msg, "voice") - if voice: - duration_ms = msg.get("voiceLength") - width = get_voice_width(duration_ms) - seconds = get_voice_duration_in_seconds(duration_ms) - voice_dir_cls = "wechat-voice-sent" if is_sent else "wechat-voice-received" - content_dir_cls = " flex-row-reverse" if is_sent else "" - icon_dir_cls = "voice-icon-sent" if is_sent else "voice-icon-received" - voice_id = str(msg.get("id") or "").strip() + duration_ms = msg.get("voiceLength") + width = get_voice_width(duration_ms) + seconds = get_voice_duration_in_seconds(duration_ms) + voice_dir_cls = "wechat-voice-sent" if is_sent else "wechat-voice-received" + content_dir_cls = " flex-row-reverse" if is_sent else "" + icon_dir_cls = "voice-icon-sent" if is_sent else "voice-icon-received" + voice_id = str(msg.get("id") or "").strip() - tw.write('
\n') - tw.write( - f'
\n' - ) - tw.write(f'
\n') - tw.write( - f' \n' - ) - tw.write( - ' \n' - ) - tw.write( - ' \n' - ) - tw.write( - ' \n' - ) - tw.write(" \n") - tw.write(f' {esc_text(seconds)}"\n') - tw.write("
\n") - tw.write("
\n") + tw.write('
\n') + tw.write( + f'
\n' + ) + tw.write(f'
\n') + tw.write( + f' \n' + ) + tw.write( + ' \n' + ) + tw.write( + ' \n' + ) + tw.write( + ' \n' + ) + tw.write(" \n") + tw.write(f' {esc_text(seconds)}"\n') + tw.write("
\n") + tw.write("
\n") + if voice: tw.write(f' \n') - tw.write("
\n") - else: - tw.write(f'
{render_text_with_emojis(msg.get("content") or "")}
\n') + tw.write("
\n") elif rt == "file": fsrc = offline_path(msg, "file") title = str(msg.get("title") or msg.get("content") or "文件").strip() @@ -5982,13 +6043,9 @@ def _materialize_voice( if not isinstance(data, (bytes, bytearray)): data = bytes(data) - wav = _convert_silk_to_wav(data) - if wav != data and wav[:4] == b"RIFF": - ext = "wav" - payload = wav - else: - ext = "silk" - payload = data + payload, ext, _media_type = _convert_silk_to_browser_audio(data, preferred_format="mp3") + if not payload: + return "", False arc = f"media/voices/voice_{int(server_id)}.{ext}" zf.writestr(arc, payload) diff --git a/src/wechat_decrypt_tool/media_helpers.py b/src/wechat_decrypt_tool/media_helpers.py index 92a9b20..3b247b9 100644 --- a/src/wechat_decrypt_tool/media_helpers.py +++ b/src/wechat_decrypt_tool/media_helpers.py @@ -1964,6 +1964,114 @@ def _convert_silk_to_wav(silk_data: bytes) -> bytes: return silk_data +def _looks_like_mp3(data: bytes) -> bool: + if not data: + return False + if data.startswith(b"ID3"): + return True + return len(data) >= 2 and data[0] == 0xFF and (data[1] & 0xE0) == 0xE0 + + +@lru_cache(maxsize=1) +def _find_ffmpeg_executable() -> str: + import shutil + + env_value = str(os.environ.get("WECHAT_TOOL_FFMPEG") or "").strip() + if env_value: + resolved = shutil.which(env_value) + if resolved: + return resolved + candidate = Path(env_value).expanduser() + if candidate.is_file(): + return str(candidate) + + return shutil.which("ffmpeg") or "" + + +def _convert_wav_to_mp3(wav_data: bytes) -> bytes: + import subprocess + import tempfile + + if not wav_data or not wav_data.startswith(b"RIFF"): + return b"" + + ffmpeg_exe = _find_ffmpeg_executable() + if not ffmpeg_exe: + return b"" + + try: + with tempfile.TemporaryDirectory() as tmp_dir: + tmp_path = Path(tmp_dir) + wav_path = tmp_path / "voice.wav" + mp3_path = tmp_path / "voice.mp3" + wav_path.write_bytes(wav_data) + + proc = subprocess.run( + [ + ffmpeg_exe, + "-y", + "-hide_banner", + "-loglevel", + "error", + "-i", + str(wav_path), + "-vn", + "-codec:a", + "libmp3lame", + "-q:a", + "4", + str(mp3_path), + ], + check=False, + capture_output=True, + ) + if proc.returncode != 0 or not mp3_path.exists(): + err = proc.stderr.decode("utf-8", errors="ignore").strip() + if err: + logger.warning(f"WAV to MP3 conversion failed: {err}") + return b"" + + mp3_data = mp3_path.read_bytes() + if _looks_like_mp3(mp3_data): + return mp3_data + except Exception as e: + logger.warning(f"WAV to MP3 conversion failed: {e}") + + return b"" + + +def _convert_silk_to_browser_audio( + silk_data: bytes, + *, + preferred_format: str = "mp3", +) -> tuple[bytes, str, str]: + """Convert SILK audio to a browser-friendly format. + + Returns `(payload, ext, media_type)`. + Preference order: + 1) MP3 if ffmpeg is available + 2) WAV if SILK decoding succeeds + 3) original SILK bytes as a last-resort fallback + """ + + data = bytes(silk_data or b"") + if not data: + return b"", "silk", "audio/silk" + + if _looks_like_mp3(data): + return data, "mp3", "audio/mpeg" + + wav_data = data if data.startswith(b"RIFF") else _convert_silk_to_wav(data) + if wav_data.startswith(b"RIFF"): + if str(preferred_format or "").strip().lower() == "mp3": + mp3_data = _convert_wav_to_mp3(wav_data) + if mp3_data: + return mp3_data, "mp3", "audio/mpeg" + return wav_data, "wav", "audio/wav" + + return data, "silk", "audio/silk" + + def _resolve_media_path_for_kind( account_dir: Path, kind: str, diff --git a/src/wechat_decrypt_tool/routers/chat_media.py b/src/wechat_decrypt_tool/routers/chat_media.py index a514b43..70b7290 100644 --- a/src/wechat_decrypt_tool/routers/chat_media.py +++ b/src/wechat_decrypt_tool/routers/chat_media.py @@ -33,7 +33,7 @@ from ..avatar_cache import ( ) from ..logging_config import get_logger from ..media_helpers import ( - _convert_silk_to_wav, + _convert_silk_to_browser_audio, _decrypt_emoticon_aes_cbc, _detect_image_extension, _detect_image_media_type, @@ -1762,12 +1762,12 @@ async def get_chat_voice(server_id: int, account: Optional[str] = None): if not isinstance(data, (bytes, bytearray)): data = bytes(data) - # Try to convert SILK to WAV for browser playback - wav_data = _convert_silk_to_wav(data) - if wav_data != data: + payload, ext, media_type = _convert_silk_to_browser_audio(data, preferred_format="mp3") + if payload and ext != "silk": return Response( - content=wav_data, - media_type="audio/wav", + content=payload, + media_type=media_type, + headers={"Content-Disposition": f"inline; filename=voice_{int(server_id)}.{ext}"}, ) # Fallback to raw SILK if conversion fails @@ -1821,11 +1821,16 @@ async def open_chat_media_folder( if not isinstance(data, (bytes, bytearray)): data = bytes(data) + payload, ext, _media_type = _convert_silk_to_browser_audio(data, preferred_format="mp3") + if not payload: + payload = data + ext = "silk" + export_dir = account_dir / "_exports" export_dir.mkdir(parents=True, exist_ok=True) - p = export_dir / f"voice_{int(server_id)}.silk" + p = export_dir / f"voice_{int(server_id)}.{ext}" try: - p.write_bytes(data) + p.write_bytes(payload) except Exception as e: raise HTTPException(status_code=500, detail=f"Failed to export voice: {e}") else: diff --git a/tests/test_chat_export_html_format.py b/tests/test_chat_export_html_format.py index 099c457..1f5047f 100644 --- a/tests/test_chat_export_html_format.py +++ b/tests/test_chat_export_html_format.py @@ -1,6 +1,8 @@ import os import json import hashlib +import logging +import re import sqlite3 import sys import unittest @@ -243,6 +245,22 @@ class TestChatExportHtmlFormat(unittest.TestCase): self._seed_media_files(account_dir) return account_dir + def _insert_missing_voice_message(self, account_dir: Path, *, username: str, server_id: int, duration_ms: int) -> None: + conn = sqlite3.connect(str(account_dir / "message_0.db")) + try: + table_name = f"msg_{hashlib.md5(username.encode('utf-8')).hexdigest()}" + row = conn.execute(f"SELECT COALESCE(MAX(local_id), 0), COALESCE(MAX(sort_seq), 0) FROM {table_name}").fetchone() + next_local_id = int((row[0] or 0)) + 1 + next_sort_seq = int((row[1] or 0)) + 1 + voice_xml = f'' + conn.execute( + f"INSERT INTO {table_name} (local_id, server_id, local_type, sort_seq, real_sender_id, create_time, message_content, compress_content) VALUES (?, ?, ?, ?, ?, ?, ?, ?)", + (next_local_id, int(server_id), 34, next_sort_seq, 2, 1735689700, voice_xml, None), + ) + conn.commit() + finally: + conn.close() + def _create_job(self, manager, *, account: str, username: str): job = manager.create_job( account=account, @@ -283,7 +301,14 @@ class TestChatExportHtmlFormat(unittest.TestCase): try: os.environ["WECHAT_TOOL_DATA_DIR"] = str(root) svc = self._reload_export_modules() - job = self._create_job(svc.CHAT_EXPORT_MANAGER, account=account, username=username) + original_converter = svc._convert_silk_to_browser_audio + svc._convert_silk_to_browser_audio = ( + lambda data, preferred_format="mp3": (bytes(data or b""), "silk", "audio/silk") + ) + try: + job = self._create_job(svc.CHAT_EXPORT_MANAGER, account=account, username=username) + finally: + svc._convert_silk_to_browser_audio = original_converter self.assertEqual(job.status, "done", msg=job.error) self.assertTrue(job.zip_path and job.zip_path.exists()) @@ -332,6 +357,8 @@ class TestChatExportHtmlFormat(unittest.TestCase): css_text = zf.read("assets/wechat-chat-export.css").decode("utf-8", errors="ignore") self.assertIn("wechat-transfer-card", css_text) + self.assertRegex(css_text, re.compile(r"\.wechat-voice-sent(?::|::)after")) + self.assertRegex(css_text, re.compile(r"\.wechat-voice-received(?::|::)before")) self.assertNotIn("wechat-transfer-card[data-v-", css_text) self.assertNotIn("bento-container", css_text) @@ -346,6 +373,87 @@ class TestChatExportHtmlFormat(unittest.TestCase): self.assertIn("wxemoji/Expression_1@2x.png", names) self.assertIn("../../wxemoji/Expression_1@2x.png", html_text) finally: + logging.shutdown() + if prev_data is None: + os.environ.pop("WECHAT_TOOL_DATA_DIR", None) + else: + os.environ["WECHAT_TOOL_DATA_DIR"] = prev_data + + def test_html_export_prefers_mp3_for_voice_assets(self): + with TemporaryDirectory() as td: + root = Path(td) + account = "wxid_test" + username = "wxid_friend" + self._prepare_account(root, account=account, username=username) + + prev_data = os.environ.get("WECHAT_TOOL_DATA_DIR") + try: + os.environ["WECHAT_TOOL_DATA_DIR"] = str(root) + svc = self._reload_export_modules() + + original_converter = svc._convert_silk_to_browser_audio + svc._convert_silk_to_browser_audio = ( + lambda data, preferred_format="mp3": (b"ID3FAKE_MP3_DATA", "mp3", "audio/mpeg") + ) + try: + job = self._create_job(svc.CHAT_EXPORT_MANAGER, account=account, username=username) + finally: + svc._convert_silk_to_browser_audio = original_converter + + self.assertEqual(job.status, "done", msg=job.error) + + self.assertTrue(job.zip_path and job.zip_path.exists()) + with zipfile.ZipFile(job.zip_path, "r") as zf: + names = set(zf.namelist()) + voice_path = f"media/voices/voice_{self._VOICE_SERVER_ID}.mp3" + self.assertIn(voice_path, names) + self.assertNotIn(f"media/voices/voice_{self._VOICE_SERVER_ID}.wav", names) + + html_path = next((n for n in names if n.endswith("/messages.html")), "") + self.assertTrue(html_path) + html_text = zf.read(html_path).decode("utf-8") + self.assertIn(f"../../{voice_path}", html_text) + finally: + logging.shutdown() + if prev_data is None: + os.environ.pop("WECHAT_TOOL_DATA_DIR", None) + else: + os.environ["WECHAT_TOOL_DATA_DIR"] = prev_data + + def test_html_export_keeps_voice_bubble_when_audio_file_missing(self): + with TemporaryDirectory() as td: + root = Path(td) + account = "wxid_test" + username = "wxid_friend" + account_dir = self._prepare_account(root, account=account, username=username) + self._insert_missing_voice_message(account_dir, username=username, server_id=999999, duration_ms=6543) + + prev_data = os.environ.get("WECHAT_TOOL_DATA_DIR") + try: + os.environ["WECHAT_TOOL_DATA_DIR"] = str(root) + svc = self._reload_export_modules() + original_converter = svc._convert_silk_to_browser_audio + svc._convert_silk_to_browser_audio = ( + lambda data, preferred_format="mp3": (bytes(data or b""), "silk", "audio/silk") + ) + try: + job = self._create_job(svc.CHAT_EXPORT_MANAGER, account=account, username=username) + finally: + svc._convert_silk_to_browser_audio = original_converter + self.assertEqual(job.status, "done", msg=job.error) + + self.assertTrue(job.zip_path and job.zip_path.exists()) + with zipfile.ZipFile(job.zip_path, "r") as zf: + names = set(zf.namelist()) + html_path = next((n for n in names if n.endswith("/messages.html")), "") + self.assertTrue(html_path) + html_text = zf.read(html_path).decode("utf-8") + self.assertIn("wechat-voice-wrapper", html_text) + self.assertIn('data-render-type="voice"', html_text) + self.assertIn('data-voice-id="message_0:msg_d5616d78f22fe35c632f66cabecfc82d:11"', html_text) + self.assertIn('class="wechat-voice-duration">7"', html_text) + finally: + logging.shutdown() if prev_data is None: os.environ.pop("WECHAT_TOOL_DATA_DIR", None) else: diff --git a/tests/test_chat_export_html_paging.py b/tests/test_chat_export_html_paging.py index 885fd81..ca71f61 100644 --- a/tests/test_chat_export_html_paging.py +++ b/tests/test_chat_export_html_paging.py @@ -1,6 +1,7 @@ import os import json import hashlib +import logging import sqlite3 import sys import unittest @@ -215,6 +216,7 @@ class TestChatExportHtmlPaging(unittest.TestCase): page1_text = zf.read(page1_js).decode("utf-8", errors="ignore") self.assertIn("MSG0001", page1_text) finally: + logging.shutdown() if prev_data is None: os.environ.pop("WECHAT_TOOL_DATA_DIR", None) else: