# FILE: _St/Auto8700/st_update8700.py | ROLE: Auto8700 시작 전 _St build 자동 업데이트 확인·다운로드
from __future__ import annotations

import hashlib
import html
import os
import re
import shutil
import subprocess
import sys
import tempfile
from datetime import datetime, timedelta
import urllib.parse
import urllib.request
import zipfile
from dataclasses import dataclass
from pathlib import Path, PurePosixPath
from typing import Callable, Mapping

DEFAULT_VERSION_URL = "http://aiwk.yjm.kr/down/_St_ver.html"
LOCAL_VERSION_NAME = "_St_ver.html"
PUBLISHER_MARKER = Path("AutoDistribute") / "AutoDistribute.py"
META_RE = re.compile(
    r'<meta\s+name=["\'](?P<name>st-[^"\']+)["\']\s+content=["\'](?P<value>.*?)["\']\s*/?>',
    re.IGNORECASE,
)


@dataclass(frozen=True)
class VersionInfo:
    version: str
    build: int
    file: str
    sha256: str = ""


@dataclass(frozen=True)
class UpdateDecision:
    action: str
    local: VersionInfo
    remote: VersionInfo
    publisher: bool
    reason: str


class UpdateError(RuntimeError):
    pass


def empty_version() -> VersionInfo:
    return VersionInfo(version="0", build=0, file="_St.zip", sha256="")


def parse_version_html(text: str) -> VersionInfo:
    values: dict[str, str] = {}
    for match in META_RE.finditer(text or ""):
        values[match.group("name").casefold()] = html.unescape(match.group("value").strip())
    try:
        build = int(values.get("st-build", "0"))
    except ValueError as exc:
        raise UpdateError("_St_ver.html st-build가 정수가 아닙니다.") from exc
    if build < 0:
        raise UpdateError("_St_ver.html st-build는 0 이상이어야 합니다.")
    return VersionInfo(
        version=values.get("st-version", "0") or "0",
        build=build,
        file=values.get("st-file", "_St.zip") or "_St.zip",
        sha256=values.get("st-sha256", "").strip().casefold(),
    )


def read_local_version(st_root: Path) -> VersionInfo:
    path = st_root / LOCAL_VERSION_NAME
    if not path.is_file():
        return empty_version()
    try:
        return parse_version_html(path.read_text(encoding="utf-8-sig"))
    except Exception as exc:
        raise UpdateError(f"로컬 {LOCAL_VERSION_NAME} 읽기 실패: {exc}") from exc


def fetch_text(url: str, timeout: float = 3.0) -> str:
    request = urllib.request.Request(url, headers={"Cache-Control": "no-cache", "User-Agent": "Auto8700-Updater/1"})
    with urllib.request.urlopen(request, timeout=max(0.5, float(timeout))) as response:
        return response.read().decode("utf-8-sig", errors="strict")


def fetch_remote_version(url: str, timeout: float = 3.0, getter: Callable[[str, float], str] = fetch_text) -> VersionInfo:
    try:
        return parse_version_html(getter(url, timeout))
    except Exception as exc:
        if isinstance(exc, UpdateError):
            raise
        raise UpdateError(f"서버 버전 확인 실패: {exc}") from exc


def is_publisher(st_root: Path, marker: str | Path = PUBLISHER_MARKER) -> bool:
    return (st_root / Path(marker)).is_file()


def decide_update(local: VersionInfo, remote: VersionInfo, *, publisher: bool) -> UpdateDecision:
    if publisher:
        return UpdateDecision("PUBLISHER_SKIP", local, remote, True, "배포자 AutoDistribute marker 감지")
    if remote.build > local.build:
        return UpdateDecision("UPDATE", local, remote, False, "서버 build가 로컬보다 큼")
    if remote.build == local.build:
        return UpdateDecision("SAME", local, remote, False, "build 동일")
    return UpdateDecision("LOCAL_NEWER", local, remote, False, "로컬 build가 서버보다 큼")


def remote_zip_url(version_url: str, remote: VersionInfo) -> str:
    return urllib.parse.urljoin(version_url, remote.file or "_St.zip")


def sha256_file(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest().casefold()


def validate_update_zip(path: Path) -> int:
    count = 0
    with zipfile.ZipFile(path, "r") as zf:
        bad = zf.testzip()
        if bad:
            raise UpdateError(f"업데이트 ZIP CRC 오류: {bad}")
        for info in zf.infolist():
            normalized = info.filename.replace("\\", "/")
            p = PurePosixPath(normalized)
            if p.is_absolute() or ".." in p.parts:
                raise UpdateError(f"업데이트 ZIP 위험 경로: {info.filename}")
            if not p.parts or p.parts[0].casefold() != "_st":
                raise UpdateError(f"업데이트 ZIP base_root 위반: {info.filename}")
            if not info.is_dir():
                count += 1
    if count == 0:
        raise UpdateError("업데이트 ZIP이 비어 있습니다.")
    return count



def read_zip_version(path: Path) -> VersionInfo:
    with zipfile.ZipFile(path, "r") as zf:
        try:
            raw = zf.read("_St/_St_ver.html")
        except KeyError as exc:
            raise UpdateError("업데이트 ZIP 내부 _St/_St_ver.html이 없습니다.") from exc
    try:
        return parse_version_html(raw.decode("utf-8-sig", errors="strict"))
    except Exception as exc:
        if isinstance(exc, UpdateError):
            raise
        raise UpdateError(f"ZIP 내부 버전 파일 읽기 실패: {exc}") from exc

def download_zip(url: str, target: Path, timeout: float = 60.0) -> Path:
    target.parent.mkdir(parents=True, exist_ok=True)
    temp = target.with_suffix(target.suffix + ".part")
    temp.unlink(missing_ok=True)
    request = urllib.request.Request(url, headers={"Cache-Control": "no-cache", "User-Agent": "Auto8700-Updater/1"})
    try:
        with urllib.request.urlopen(request, timeout=max(3.0, float(timeout))) as response, temp.open("wb") as out:
            shutil.copyfileobj(response, out, length=1024 * 1024)
        os.replace(temp, target)
    except Exception:
        temp.unlink(missing_ok=True)
        raise
    return target


def _spawn_detached(command: list[str], cwd: Path) -> subprocess.Popen:
    kwargs: dict = {"cwd": str(cwd), "stdin": subprocess.DEVNULL, "stdout": subprocess.DEVNULL, "stderr": subprocess.DEVNULL}
    if sys.platform.startswith("win"):
        flags = 0
        flags |= int(getattr(subprocess, "DETACHED_PROCESS", 0))
        flags |= int(getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0))
        kwargs["creationflags"] = flags
        kwargs["close_fds"] = True
    return subprocess.Popen(command, **kwargs)


def prepare_update(
    *,
    st_root: Path,
    remote: VersionInfo,
    version_url: str,
    timeout: float,
    logger: Callable[[str], None],
    downloader: Callable[[str, Path, float], Path] = download_zip,
    spawner: Callable[[list[str], Path], object] = _spawn_detached,
) -> Path:
    if not remote.sha256 or not re.fullmatch(r"[0-9a-f]{64}", remote.sha256):
        raise UpdateError("서버 _St_ver.html의 SHA256이 없거나 형식이 잘못되었습니다.")

    runtime = Path(tempfile.gettempdir()) / "st_update8700" / f"build_{remote.build}"
    runtime.mkdir(parents=True, exist_ok=True)
    zip_path = runtime / "_St.zip"
    url = remote_zip_url(version_url, remote)
    logger(f"[자동업데이트][UP-03][체크시작] ZIP 다운로드 / build={remote.build}")
    downloader(url, zip_path, timeout)
    logger(f"[자동업데이트][UP-03][체크완료] ZIP 다운로드 완료 / {zip_path}")

    logger("[자동업데이트][UP-04][체크시작] SHA256·ZIP 구조 검증")
    actual_hash = sha256_file(zip_path)
    if actual_hash != remote.sha256:
        zip_path.unlink(missing_ok=True)
        raise UpdateError(f"SHA256 불일치 / server={remote.sha256} local={actual_hash}")
    validate_update_zip(zip_path)
    embedded = read_zip_version(zip_path)
    if embedded.build != remote.build:
        zip_path.unlink(missing_ok=True)
        raise UpdateError(f"서버/ZIP build 불일치 / server={remote.build} zip={embedded.build}")
    logger(f"[자동업데이트][UP-04][체크완료] SHA256·ZIP·build 검증 OK / {actual_hash} / build={embedded.build}")

    source_apply = Path(__file__).resolve().parent / "st_update_apply.py"
    if not source_apply.is_file():
        raise UpdateError(f"updater 적용 파일 없음: {source_apply}")
    apply_copy = runtime / "st_update_apply.py"
    shutil.copy2(source_apply, apply_copy)
    manager = st_root / "Auto8700" / "auto_service8700_manager.py"
    command = [
        sys.executable,
        str(apply_copy),
        "--parent-pid",
        str(os.getpid()),
        "--zip",
        str(zip_path),
        "--st-root",
        str(st_root),
        "--manager",
        str(manager),
        "--expected-build",
        str(remote.build),
    ]
    logger("[자동업데이트][UP-05][체크시작] TEMP updater 실행 예약")
    spawner(command, runtime)
    logger(f"[자동업데이트][UP-05][체크완료] updater 실행 예약 / build={remote.build}")
    return zip_path


def update_check_once(
    cfg: Mapping[str, object],
    *,
    st_root: Path,
    source: str,
    logger: Callable[[str], None] = print,
    remote_getter: Callable[[str, float], str] = fetch_text,
    downloader: Callable[[str, Path, float], Path] = download_zip,
    spawner: Callable[[list[str], Path], object] = _spawn_detached,
) -> bool:
    """한 번 build를 확인한다. True면 updater를 예약했으므로 현재 Auto8700를 종료해야 한다."""
    update_cfg = dict(cfg.get("auto_update") or {}) if isinstance(cfg.get("auto_update"), Mapping) else {}
    if not bool(update_cfg.get("enabled", True)):
        logger(f"[자동업데이트][UP-00][체크완료] 자동업데이트 비활성 / source={source}")
        return False

    version_url = str(update_cfg.get("version_url") or DEFAULT_VERSION_URL).strip() or DEFAULT_VERSION_URL
    timeout = float(update_cfg.get("timeout_sec") or 3.0)
    marker = str(update_cfg.get("publisher_marker") or str(PUBLISHER_MARKER)).strip()

    logger(f"[자동업데이트][UP-01][체크시작] 로컬/서버 build 확인 / source={source}")
    local = read_local_version(st_root)
    publisher = is_publisher(st_root, marker)
    try:
        remote = fetch_remote_version(version_url, timeout, getter=remote_getter)
    except Exception as exc:
        logger(
            f"[자동업데이트][UP-01][체크실패] 서버 버전 확인 실패 / source={source} / "
            f"{exc} / 현재 버전 계속 실행"
        )
        return False
    logger(
        f"[자동업데이트][UP-01][체크완료] local_build={local.build} server_build={remote.build} "
        f"publisher={int(publisher)} source={source}"
    )

    decision = decide_update(local, remote, publisher=publisher)
    logger(f"[자동업데이트][UP-02][체크완료] 판단={decision.action} / {decision.reason} / source={source}")
    if decision.action != "UPDATE":
        return False

    try:
        prepare_update(
            st_root=st_root,
            remote=remote,
            version_url=version_url,
            timeout=float(update_cfg.get("download_timeout_sec") or 120.0),
            logger=logger,
            downloader=downloader,
            spawner=spawner,
        )
    except Exception as exc:
        logger(
            f"[자동업데이트][UP-05][체크실패] 업데이트 준비 실패 / source={source} / "
            f"{exc} / 현재 버전 계속 실행"
        )
        return False
    return True


def startup_update_check(
    cfg: Mapping[str, object],
    *,
    st_root: Path,
    logger: Callable[[str], None] = print,
    remote_getter: Callable[[str, float], str] = fetch_text,
    downloader: Callable[[str, Path, float], Path] = download_zip,
    spawner: Callable[[list[str], Path], object] = _spawn_detached,
) -> bool:
    """시작 시 1회 확인. True면 updater를 실행했으므로 현재 Auto8700는 즉시 종료해야 한다."""
    update_cfg = dict(cfg.get("auto_update") or {}) if isinstance(cfg.get("auto_update"), Mapping) else {}
    if not bool(update_cfg.get("enabled", True)) or not bool(update_cfg.get("check_on_start", True)):
        logger("[자동업데이트][UP-00][체크완료] 시작 버전 체크 비활성 / source=startup")
        return False
    return update_check_once(
        cfg,
        st_root=st_root,
        source="startup",
        logger=logger,
        remote_getter=remote_getter,
        downloader=downloader,
        spawner=spawner,
    )


def hourly_update_check(
    cfg: Mapping[str, object],
    *,
    st_root: Path,
    logger: Callable[[str], None] = print,
    remote_getter: Callable[[str, float], str] = fetch_text,
    downloader: Callable[[str, Path, float], Path] = download_zip,
    spawner: Callable[[list[str], Path], object] = _spawn_detached,
) -> bool:
    """실행 중 정시 확인. True면 updater 예약 완료 상태다."""
    update_cfg = dict(cfg.get("auto_update") or {}) if isinstance(cfg.get("auto_update"), Mapping) else {}
    if not bool(update_cfg.get("enabled", True)) or not bool(update_cfg.get("check_hourly", True)):
        return False
    return update_check_once(
        cfg,
        st_root=st_root,
        source="hourly",
        logger=logger,
        remote_getter=remote_getter,
        downloader=downloader,
        spawner=spawner,
    )


def seconds_until_next_hour(now: datetime | None = None) -> float:
    """로컬 시스템 시각 기준 다음 HH:00:00까지 남은 초. 항상 다음 정시를 반환한다."""
    current = now or datetime.now()
    next_hour = current.replace(minute=0, second=0, microsecond=0) + timedelta(hours=1)
    return max(0.001, (next_hour - current).total_seconds())