src/feedback_service.py - master - delbraun/AutoRAWCompressor

src/feedback_service.py
747 строк25 KB
Divan Games
fix: диалог Zona использует вшитый токен из сборки
06 июл 2026, 06:33
06 июл 2026, 06:33fea3c33
"""Отправка обратной связи из программы через Telegram (бот Zona)."""
from __future__ import annotations

import html
import io
import json
import mimetypes
import platform
import random
import struct
import subprocess
import sys
import threading
import urllib.request
from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import Enum
from pathlib import Path
from typing import Callable

from app_paths import resource_path, user_config_dir, ui_config_path
from version import APP_NAME, version_string

_IMAGE_EXTS = frozenset({".png", ".jpg", ".jpeg", ".webp", ".bmp", ".gif", ".tif", ".tiff"})
_WIN_CREATE_NO_WINDOW = 0x08000000


class FeedbackType(str, Enum):
    BUG = "bug"
    FEATURE = "feature"
    OTHER = "other"


FEEDBACK_TITLES: dict[FeedbackType, str] = {
    FeedbackType.BUG: "Сообщить об ошибке",
    FeedbackType.FEATURE: "Предложить улучшение",
    FeedbackType.OTHER: "Обратная связь",
}

FEEDBACK_LABELS: dict[FeedbackType, str] = {
    FeedbackType.BUG: "🐛 Ошибка",
    FeedbackType.FEATURE: "💡 Предложение",
    FeedbackType.OTHER: "💬 Другое",
}


@dataclass
class FeedbackContext:
    user_name: str = ""
    selected_folder: str | None = None
    current_frame: str | None = None
    export_mode: str | None = None
    load_mode: str | None = None
    system_specs: str = ""


@dataclass
class FeedbackPayload:
    feedback_type: FeedbackType
    text: str
    contact: str = ""
    include_diagnostics: bool = True
    attachments: list[Path] = field(default_factory=list)
    context: FeedbackContext = field(default_factory=FeedbackContext)


@dataclass
class SubmitResult:
    ok: bool
    ticket_id: str
    queued: bool = False
    error: str | None = None


def load_zona_data() -> dict[str, str]:
    result: dict[str, str] = {}

    # Читаем data.dat (URL, метаданные; используется при разработке и как запасной источник)
    candidates = [
        Path(__file__).resolve().parent / "zona" / "data.dat",
        resource_path("zona", "data.dat"),
    ]
    for path in candidates:
        if not path.is_file():
            continue
        try:
            for line in path.read_text(encoding="utf-8").splitlines():
                if ":" in line:
                    key, _, val = line.partition(":")
                    result[key.strip().upper()] = val.strip()
        except OSError:
            pass
        break

    # Токен и дефолтный Chat ID из builtin-модуля (вшит в бинарник при сборке).
    # Имеют приоритет над data.dat — токен не лежит как plaintext-файл в dist.
    try:
        from builtin_zona_token import BUILTIN_ZONA_TOKEN, BUILTIN_ZONA_CHAT_ID  # noqa: PLC0415
        if BUILTIN_ZONA_TOKEN:
            result["TOKEN"] = BUILTIN_ZONA_TOKEN
        if BUILTIN_ZONA_CHAT_ID:
            result.setdefault("ID", BUILTIN_ZONA_CHAT_ID)
            result.setdefault("FEEDBACK_ID", BUILTIN_ZONA_CHAT_ID)
    except ImportError:
        pass

    return result


def _feedback_chat_id(zona: dict[str, str]) -> str:
    return (zona.get("FEEDBACK_ID") or zona.get("ID") or "").strip()


def _queue_path() -> Path:
    return user_config_dir() / "feedback_queue.json"


def _attachments_dir() -> Path:
    path = user_config_dir() / "feedback_attachments"
    path.mkdir(parents=True, exist_ok=True)
    return path


def _state_path() -> Path:
    return user_config_dir() / "feedback_state.json"


def _next_ticket_id() -> str:
    path = _state_path()
    counter = 0
    if path.is_file():
        try:
            counter = int(json.loads(path.read_text(encoding="utf-8")).get("counter", 0))
        except (OSError, ValueError, json.JSONDecodeError):
            counter = 0
    counter += 1
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps({"counter": counter}, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    stamp = datetime.now(timezone.utc).strftime("%Y%m%d")
    return f"FB-{stamp}-{counter:04d}"


def _save_pil_image(img) -> Path:
    """Конвертирует PIL-изображение в PNG и сохраняет во временную папку."""
    if img.mode == "P":
        img = img.convert("RGBA")
    elif img.mode not in ("RGB", "RGBA", "L"):
        img = img.convert("RGB")
    stamp = datetime.now().strftime("%Y%m%d_%H%M%S_%f")
    out = _attachments_dir() / f"clipboard_{stamp}.png"
    img.save(out, format="PNG")
    return out


def _is_image_path(path: Path) -> bool:
    return path.is_file() and path.suffix.lower() in _IMAGE_EXTS


def _read_clipboard_bytes(fmt: int) -> bytes | None:
    """Читает данные из буфера Windows для заданного формата. Не открывает буфер сам."""
    import ctypes
    kernel32 = ctypes.windll.kernel32
    handle = ctypes.windll.user32.GetClipboardData(fmt)
    if not handle:
        return None
    ptr = kernel32.GlobalLock(handle)
    if not ptr:
        return None
    try:
        size = kernel32.GlobalSize(handle)
        return ctypes.string_at(ptr, size)
    finally:
        kernel32.GlobalUnlock(handle)


def _dib_to_pil(data: bytes):
    """Собирает BMP-файл из CF_DIB/CF_DIBV5 и открывает через PIL."""
    from PIL import Image
    # BITMAPINFOHEADER: первые 4 байта — размер заголовка
    if len(data) < 4:
        return None
    header_size = struct.unpack_from("<I", data, 0)[0]
    # pixel data offset = 14 (file header) + header_size + размер таблицы цветов
    # Для простоты: пробуем через стандартный BMP offset = 14 + header_size
    pixel_offset = 14 + header_size
    # Добавляем цветовую таблицу если нужна (bit_count <= 8)
    if header_size >= 40:
        bit_count = struct.unpack_from("<H", data, 14)[0]
        if bit_count <= 8:
            colors = struct.unpack_from("<I", data, 32)[0]
            if colors == 0:
                colors = 1 << bit_count
            pixel_offset += colors * 4
    bmp_header = (
        b"BM"
        + struct.pack("<I", 14 + len(data))   # file size
        + b"\x00\x00\x00\x00"                  # reserved
        + struct.pack("<I", pixel_offset)       # pixel data offset
    )
    try:
        return Image.open(io.BytesIO(bmp_header + data))
    except Exception:
        return None


def _win_grab_via_powershell() -> Path | None:
    """Читает изображение из буфера через PowerShell/.NET в отдельном процессе.

    Это самый надёжный способ на Windows: не конфликтует с Tkinter,
    корректно обрабатывает формат PNG (Win+Shift+S) и CF_DIB (Print Screen).
    """
    stamp = datetime.now().strftime("%Y%m%d_%H%M%S_%f")
    out = _attachments_dir() / f"clipboard_{stamp}.png"

    # Путь через переменную окружения — без проблем с кавычками/кириллицей
    script = (
        "Add-Type -Assembly System.Windows.Forms; "
        "Add-Type -Assembly System.Drawing; "
        "$out = $env:_FB_OUT; "
        # Пробуем кастомный PNG-формат (Win+Shift+S, Snipping Tool)
        "$png = [System.Windows.Forms.Clipboard]::GetData('PNG'); "
        "if ($png) { "
        "  $ms = New-Object System.IO.MemoryStream; "
        "  $png.CopyTo($ms); "
        "  [System.IO.File]::WriteAllBytes($out, $ms.ToArray()); "
        "  exit 0 "
        "} "
        # Стандартный растровый буфер (Print Screen, другие приложения)
        "$img = [System.Windows.Forms.Clipboard]::GetImage(); "
        "if ($img) { "
        "  $img.Save($out, [System.Drawing.Imaging.ImageFormat]::Png); "
        "  exit 0 "
        "} "
        "exit 1"
    )

    import os
    env = os.environ.copy()
    env["_FB_OUT"] = str(out)

    try:
        r = subprocess.run(
            [
                "powershell",
                "-NoProfile", "-NonInteractive",
                "-WindowStyle", "Hidden",
                "-ExecutionPolicy", "Bypass",
                "-Command", script,
            ],
            capture_output=True,
            timeout=15,
            creationflags=_WIN_CREATE_NO_WINDOW,
            env=env,
        )
        if out.is_file() and out.stat().st_size > 0:
            return out
    except Exception:
        pass
    return None


def _win_grab_via_win32() -> Path | None:
    """Запасной путь: читаем буфер напрямую через ctypes Win32 API."""
    try:
        import ctypes
        from PIL import Image
    except ImportError:
        return None

    user32 = ctypes.windll.user32
    kernel32 = ctypes.windll.kernel32

    for _attempt in range(8):
        if user32.OpenClipboard(0):
            break
        import time
        time.sleep(0.05)
    else:
        return None

    try:
        # Кастомный PNG (Win+Shift+S)
        png_fmt = user32.RegisterClipboardFormatW("PNG")
        if png_fmt and user32.IsClipboardFormatAvailable(png_fmt):
            data = _read_clipboard_bytes(png_fmt)
            if data:
                try:
                    img = Image.open(io.BytesIO(data))
                    return _save_pil_image(img)
                except Exception:
                    pass

        # CF_DIBV5 / CF_DIB
        for cf_fmt in (17, 8):
            if user32.IsClipboardFormatAvailable(cf_fmt):
                data = _read_clipboard_bytes(cf_fmt)
                if data:
                    img = _dib_to_pil(data)
                    if img is not None:
                        try:
                            return _save_pil_image(img)
                        except Exception:
                            pass
                break

        # CF_HDROP — скопированные файлы
        CF_HDROP = 15
        if user32.IsClipboardFormatAvailable(CF_HDROP):
            shell32 = ctypes.windll.shell32
            handle = user32.GetClipboardData(CF_HDROP)
            if handle:
                count = shell32.DragQueryFileW(handle, 0xFFFFFFFF, None, 0)
                for i in range(count):
                    buf = ctypes.create_unicode_buffer(260)
                    shell32.DragQueryFileW(handle, i, buf, 260)
                    path = Path(buf.value)
                    if _is_image_path(path):
                        return path
    finally:
        user32.CloseClipboard()
    return None


def attachments_from_clipboard() -> list[Path]:
    """Все изображения из буфера: Win+Shift+S (PNG), DIB, скопированные файлы."""
    if sys.platform != "win32":
        return _pil_clipboard_fallback()

    # PowerShell — отдельный процесс, не конфликтует с Tkinter
    path = _win_grab_via_powershell()
    if path:
        return [path]

    # Win32 API — запасной путь (быстрее, но может конфликтовать с Tk)
    path = _win_grab_via_win32()
    if path:
        return [path]

    return []


def _pil_clipboard_fallback() -> list[Path]:
    """Запасной путь для не-Windows (macOS / Linux)."""
    try:
        from PIL import ImageGrab
        data = ImageGrab.grabclipboard()
    except Exception:
        return []
    if data is None:
        return []
    if isinstance(data, list):
        return [p for item in data if _is_image_path(p := Path(str(item).strip().strip('"')))]
    if hasattr(data, "save"):
        return [_save_pil_image(data)]
    return []


def attachment_from_clipboard() -> Path | None:
    items = attachments_from_clipboard()
    return items[0] if items else None


def os_info_string() -> str:
    return f"{platform.system()} {platform.release()} ({platform.machine()})"


def _run_wmic(args: list[str]) -> str:
    try:
        result = subprocess.run(
            args,
            capture_output=True,
            text=True,
            encoding="utf-8",
            errors="replace",
            timeout=4,
            creationflags=_WIN_CREATE_NO_WINDOW if sys.platform == "win32" else 0,
        )
        if result.returncode != 0:
            return ""
        return result.stdout
    except Exception:
        return ""


def _wmic_value(text: str, key: str) -> str:
    prefix = f"{key}="
    for line in text.splitlines():
        if line.startswith(prefix):
            return line[len(prefix):].strip()
    return ""


def collect_system_specs() -> str:
    """Характеристики ПК для диагностики ошибок."""
    lines: list[str] = []
    uname = platform.uname()
    lines.append(f"Имя ПК: {platform.node()}")
    lines.append(f"ОС: {uname.system} {uname.release} ({uname.version})")
    lines.append(f"Архитектура: {uname.machine}")

    cpu_name = platform.processor() or ""
    if cpu_name.strip():
        lines.append(f"Процессор: {cpu_name.strip()}")

    try:
        import psutil

        phys = psutil.cpu_count(logical=False) or 0
        logical = psutil.cpu_count(logical=True) or 0
        freq = psutil.cpu_freq()
        freq_note = f", max {freq.max:.0f} MHz" if freq and freq.max else ""
        lines.append(f"CPU: {phys} ядер / {logical} потоков{freq_note}")

        vm = psutil.virtual_memory()
        lines.append(f"RAM: {vm.total / (1024 ** 3):.1f} GB (занято {vm.percent:.0f}%)")

        root = "C:\\" if sys.platform == "win32" else "/"
        disk = psutil.disk_usage(root)
        lines.append(
            f"Диск {root}: {disk.total / (1024 ** 3):.0f} GB "
            f"(свободно {disk.free / (1024 ** 3):.0f} GB)"
        )
    except ImportError:
        pass

    if sys.platform == "win32":
        cpu_text = _run_wmic(["wmic", "cpu", "get", "Name", "/value"])
        wmic_cpu = _wmic_value(cpu_text, "Name")
        if wmic_cpu and not any(wmic_cpu in line for line in lines):
            lines.append(f"CPU модель: {wmic_cpu}")

        gpu_text = _run_wmic(["wmic", "path", "win32_VideoController", "get", "Name", "/value"])
        gpu_names = [
            _wmic_value(block, "Name")
            for block in gpu_text.split("\n\n")
            if _wmic_value(block, "Name")
        ]
        if gpu_names:
            lines.append("GPU: " + " · ".join(dict.fromkeys(gpu_names)))

        try:
            result = subprocess.run(
                [
                    "nvidia-smi",
                    "--query-gpu=name,memory.total,driver_version",
                    "--format=csv,noheader,nounits",
                ],
                capture_output=True,
                text=True,
                encoding="utf-8",
                errors="replace",
                timeout=4,
                creationflags=_WIN_CREATE_NO_WINDOW,
            )
            if result.returncode == 0:
                smi = result.stdout.strip().splitlines()
                if smi:
                    lines.append("NVIDIA: " + smi[0].strip())
        except Exception:
            pass

        screen_text = _run_wmic(["wmic", "path", "Win32_VideoController", "get", "CurrentHorizontalResolution,CurrentVerticalResolution", "/value"])
        w = _wmic_value(screen_text, "CurrentHorizontalResolution")
        h = _wmic_value(screen_text, "CurrentVerticalResolution")
        if w and h:
            lines.append(f"Экран: {w}x{h}")

    return "\n".join(lines)


def build_diagnostics(payload: FeedbackPayload) -> str:
    ctx = payload.context
    lines = [
        f"Версия: {version_string()}",
        f"ОС: {os_info_string()}",
        f"Python: {platform.python_version()}",
    ]
    specs = ctx.system_specs.strip() or collect_system_specs()
    if specs:
        lines.append("")
        lines.append("--- ПК ---")
        lines.extend(specs.splitlines())
    if ctx.user_name.strip():
        lines.append("")
        lines.append(f"Пользователь: {ctx.user_name.strip()}")
    if ctx.export_mode:
        lines.append(f"Режим экспорта: {ctx.export_mode}")
    if ctx.load_mode:
        lines.append(f"Режим загрузки: {ctx.load_mode}")
    if ctx.selected_folder:
        lines.append(f"Папка: {ctx.selected_folder}")
    if ctx.current_frame:
        lines.append(f"Кадр: {ctx.current_frame}")
    return "\n".join(lines)


def format_feedback_message(payload: FeedbackPayload, ticket_id: str) -> str:
    label = FEEDBACK_LABELS.get(payload.feedback_type, payload.feedback_type.value)
    body = html.escape(payload.text.strip())
    parts = [
        f"<b>{label}</b> · {APP_NAME}",
        f"<b>Заявка:</b> #{ticket_id}",
    ]
    user = payload.context.user_name.strip()
    if user:
        parts.append(f"<b>От:</b> {html.escape(user)}")
    contact = payload.contact.strip()
    if contact:
        parts.append(f"<b>Контакт:</b> {html.escape(contact)}")
    if payload.attachments:
        parts.append(f"<b>Файлов:</b> {len(payload.attachments)}")
    parts.append("")
    parts.append(body)
    if payload.include_diagnostics:
        diag = html.escape(build_diagnostics(payload))
        parts.append("")
        parts.append(f"<i>Диагностика:</i>\n<pre>{diag}</pre>")
    return "\n".join(parts)


def _load_queue() -> list[dict]:
    path = _queue_path()
    if not path.is_file():
        return []
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
        return data if isinstance(data, list) else []
    except (OSError, json.JSONDecodeError):
        return []


def _save_queue(items: list[dict]) -> None:
    path = _queue_path()
    path.parent.mkdir(parents=True, exist_ok=True)
    if items:
        path.write_text(json.dumps(items, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    elif path.is_file():
        path.unlink(missing_ok=True)


def _payload_to_record(payload: FeedbackPayload, ticket_id: str) -> dict:
    ctx = payload.context
    return {
        "ticket_id": ticket_id,
        "feedback_type": payload.feedback_type.value,
        "text": payload.text,
        "contact": payload.contact,
        "include_diagnostics": payload.include_diagnostics,
        "attachments": [str(p) for p in payload.attachments],
        "context": {
            "user_name": ctx.user_name,
            "selected_folder": ctx.selected_folder,
            "current_frame": ctx.current_frame,
            "export_mode": ctx.export_mode,
            "load_mode": ctx.load_mode,
            "system_specs": ctx.system_specs,
        },
        "created_at": datetime.now(timezone.utc).isoformat(),
    }


def _record_to_payload(record: dict) -> FeedbackPayload:
    ctx_raw = record.get("context") or {}
    attachments_raw = record.get("attachments")
    if not attachments_raw:
        legacy = str(record.get("attachment", "")).strip()
        attachments_raw = [legacy] if legacy else []
    attachments: list[Path] = []
    for item in attachments_raw:
        path = Path(str(item))
        if path.is_file():
            attachments.append(path)
    return FeedbackPayload(
        feedback_type=FeedbackType(str(record.get("feedback_type", FeedbackType.OTHER.value))),
        text=str(record.get("text", "")),
        contact=str(record.get("contact", "")),
        include_diagnostics=bool(record.get("include_diagnostics", True)),
        attachments=attachments,
        context=FeedbackContext(
            user_name=str(ctx_raw.get("user_name", "")),
            selected_folder=ctx_raw.get("selected_folder"),
            current_frame=ctx_raw.get("current_frame"),
            export_mode=ctx_raw.get("export_mode"),
            load_mode=ctx_raw.get("load_mode"),
            system_specs=str(ctx_raw.get("system_specs", "")),
        ),
    )


def _telegram_post_json(token: str, method: str, body: dict, *, timeout: float = 12.0) -> dict:
    url = f"https://api.telegram.org/bot{token}/{method}"
    data = json.dumps(body).encode("utf-8")
    req = urllib.request.Request(
        url,
        data=data,
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    with urllib.request.urlopen(req, timeout=timeout) as resp:
        payload = json.loads(resp.read())
    if not payload.get("ok"):
        raise RuntimeError(str(payload.get("description", "Telegram API error")))
    return payload


def _telegram_send_document(
    token: str,
    chat_id: str,
    file_path: Path,
    *,
    caption: str = "",
    timeout: float = 20.0,
) -> None:
    boundary = f"----AutoRAWFeedback{random.randint(10_000_000, 99_999_999)}"
    mime = mimetypes.guess_type(file_path.name)[0] or "application/octet-stream"
    file_bytes = file_path.read_bytes()

    parts: list[bytes] = []
    for name, value in (("chat_id", chat_id), ("parse_mode", "HTML")):
        parts.append(f"--{boundary}\r\n".encode())
        parts.append(f'Content-Disposition: form-data; name="{name}"\r\n\r\n'.encode())
        parts.append(f"{value}\r\n".encode())
    if caption.strip():
        parts.append(f"--{boundary}\r\n".encode())
        parts.append(b'Content-Disposition: form-data; name="caption"\r\n\r\n')
        parts.append(f"{caption}\r\n".encode())
    parts.append(f"--{boundary}\r\n".encode())
    parts.append(
        f'Content-Disposition: form-data; name="document"; filename="{file_path.name}"\r\n'.encode()
    )
    parts.append(f"Content-Type: {mime}\r\n\r\n".encode())
    parts.append(file_bytes)
    parts.append(b"\r\n")
    parts.append(f"--{boundary}--\r\n".encode())
    body = b"".join(parts)

    url = f"https://api.telegram.org/bot{token}/sendDocument"
    req = urllib.request.Request(
        url,
        data=body,
        headers={"Content-Type": f"multipart/form-data; boundary={boundary}"},
        method="POST",
    )
    with urllib.request.urlopen(req, timeout=timeout) as resp:
        payload = json.loads(resp.read())
    if not payload.get("ok"):
        raise RuntimeError(str(payload.get("description", "Telegram API error")))


def _send_payload_sync(payload: FeedbackPayload, ticket_id: str) -> None:
    zona = load_zona_data()
    token = zona.get("TOKEN", "").strip()
    chat_id = _feedback_chat_id(zona)
    if not token:
        raise RuntimeError("Не найден токен бота (zona/data.dat)")
    if not chat_id:
        raise RuntimeError("Не указан чат для обратной связи (FEEDBACK_ID в data.dat)")

    message = format_feedback_message(payload, ticket_id)
    _telegram_post_json(
        token,
        "sendMessage",
        {"chat_id": chat_id, "text": message, "parse_mode": "HTML"},
    )
    total = len(payload.attachments)
    for index, attachment in enumerate(payload.attachments, start=1):
        if not attachment.is_file():
            continue
        caption = f"📎 {index}/{total} к #{ticket_id}" if total > 1 else f"📎 Вложение к #{ticket_id}"
        _telegram_send_document(token, chat_id, attachment, caption=caption)


def _enqueue(record: dict) -> None:
    items = _load_queue()
    items.append(record)
    _save_queue(items)


def submit_feedback(
    payload: FeedbackPayload,
    *,
    ticket_id: str | None = None,
    on_done: Callable[[SubmitResult], None] | None = None,
) -> str:
    """Отправить обратную связь. Возвращает номер заявки сразу; отправка — в фоне."""
    tid = ticket_id or _next_ticket_id()
    record = _payload_to_record(payload, tid)

    def _worker() -> None:
        result: SubmitResult
        try:
            _send_payload_sync(payload, tid)
            result = SubmitResult(ok=True, ticket_id=tid)
        except Exception as exc:
            _enqueue(record)
            result = SubmitResult(ok=False, ticket_id=tid, queued=True, error=str(exc))
        if on_done is not None:
            on_done(result)

    threading.Thread(target=_worker, daemon=True).start()
    return tid


def flush_pending_queue(
    *,
    on_item: Callable[[SubmitResult], None] | None = None,
) -> int:
    """Попытаться отправить отложенные заявки. Возвращает число успешно отправленных."""
    items = _load_queue()
    if not items:
        return 0

    sent = 0
    remaining: list[dict] = []
    for record in items:
        ticket_id = str(record.get("ticket_id", _next_ticket_id()))
        payload = _record_to_payload(record)
        try:
            _send_payload_sync(payload, ticket_id)
            sent += 1
            if on_item is not None:
                on_item(SubmitResult(ok=True, ticket_id=ticket_id))
        except Exception as exc:
            remaining.append(record)
            if on_item is not None:
                on_item(SubmitResult(ok=False, ticket_id=ticket_id, queued=True, error=str(exc)))
    _save_queue(remaining)
    return sent


def load_feedback_contact() -> str:
    try:
        data = json.loads(ui_config_path().read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return ""
    return str(data.get("feedback_contact", "")).strip()


def save_feedback_contact(contact: str) -> None:
    path = ui_config_path()
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        data = {}
    data["feedback_contact"] = contact.strip()
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")