"""계획노선 추천 결과 영구 저장(PLAN 327 · 요구 224-3).

같은 입력이면 같은 결과 — 셈한 추천을 입력 지문 키로 프로젝트 폴더에 남기고, 같은 입력이 오면
셈하지 않고 꺼낸다. 바뀐 입력만 새로 셈해 더한다(덮어쓰기 없음 · 종류마다 최근 `KEEP_PER_KIND` 벌).

층 = 서버 계산 캐시(데이터 3층 밖 · 사용자 편집값 아님 — 자동저장 금지와 무관). 자리
`B05_Profile/route_recommend/<키>.json` — initial_snapshot 대상(`B05_Profile/route`) 밖이라
[초기화]에 안 휩쓸리고, 자료 교체(`purge_output_folders`) · 프로젝트 삭제 때 폴더째 지워진다.

키 = 후보 묶음 지문(`_pool_key` — 지형 · 예상노선 · 계류 · 상한 · 최소 R · 이격 · 고름 · 목표
기울기) + 갈래 · 비중 · 종단 반영 · 설계속도 · 표준 횡단면 · `ALGO_VERSION` · 엔진 소스 지문. 관
지점은 뺀다 — 덮개 수는 꺼낼 때 다시 센다. 추천 결과를 바꾸는 코드 고침은 `ALGO_VERSION` 을
올린다 — 빠뜨려도 엔진 소스 지문이 바뀌어 새로 셈. 다른 판 · 지문 파일은 안 읽고 남기며 30일 지난
것만 미리 셈 때 지운다(420-2 · storage 를 두 PC · 여러 워크트리가 같이 씀). 프로젝트 접근(화면
요청)은 같은 판 · 같은 입력이면 다른 소스 지문 저장본도 꺼냄(`read_stored_any` · 505-1 덧) — 지문
견줌 · 다시 셈은 전처리 체인에서만.
기본값 · 저장된 제약 결과(`default`)는 「최근 벌」을 따로 센다(351-1).
"""

from __future__ import annotations

import asyncio
import functools
import hashlib
import json
import time
from contextlib import asynccontextmanager
from datetime import datetime
from pathlib import Path
from typing import Any, AsyncIterator, Callable

from B05_Profile.B05_Profile_Engine_RouteInitial import (
    DEFAULT_WEIGHTS,
    Terrain,
    _pool_key,
    resolve_inputs,
    resolve_weights,
)
from common_util.common_util_json import atomic_write_json

# 3 = 곡선 MC 일반측점(351-2) · 4 = 다투는 곡선 끔(364-5) · 5 = 후보마다 맞춤(383-1) ·
# 6 = 같은 쪽 곡선 합치기 · 거리 한도 대신 직접 잼(406) · 7 = 측점 밖이 늘면 옛 차례 맞춤(406-4) ·
# 8 = 횡단 토공 · 표준 횡단면 = 저장 패널값 · 엔진 소스 지문(420)
ALGO_VERSION = 8
# 엔진 소스 지문에 드는 파일(저장소 뿌리 기준) — 추천 결과를 낳는 코드
SOURCE_GLOBS = (
    "B05_Profile/B05_Profile_Engine_RouteInitial*.py",
    "B05_Profile/B05_Profile_Engine_Sections_Core.py",
    "B06_Section/B06_Section_Engine_Design*.py",
    "common_util/common_util_route_polyline.py",
)
STORE_DIR = "B05_Profile/route_recommend"
# 종류(갈래 · 지형) 마다 남길 벌 수 — 최근에 적은 것이 남음(꺼내기는 읽기만 · 505-1)
KEEP_PER_KIND = 2
# 다른 판 · 소스 지문 저장본을 이 날수 지나면 미리 셈 때 지움(420-2)
STALE_DAYS = 30
# 업로드 체인 끝에 미리 셈할 기본 다섯 알약(화면 차례) · 추적은 묶음이 선 뒤라 `similar` 가 채워짐
DEFAULT_MODES = ("gradeline", "earthwork", "gentle", "weighted", "follow")

# 묶음 지문마다 잠금 — 진행 중 같은 묶음(같은 키 포함)은 기다렸다 꺼냄(두 번 셈 막음)
_LOCKS: dict[str, asyncio.Lock] = {}
# 묶음 지문마다 그 묶음을 짤 때 표준 횡단면(`forget_pool_on_standard`)
_POOL_STANDARDS: dict[str, str] = {}


def store_dir(project_root: Path) -> Path:
    return Path(project_root) / STORE_DIR


@functools.cache
def source_fingerprint() -> str:
    """엔진 소스 지문 — `SOURCE_GLOBS` 파일 내용(줄끝 LF 로 맞춤 · 두 PC 같은 코드 = 같은 지문)."""
    root = Path(__file__).resolve().parents[1]
    digest = hashlib.blake2b(digest_size=8)
    for path in sorted({p for pattern in SOURCE_GLOBS for p in root.glob(pattern)}):
        digest.update(path.name.encode())
        digest.update(path.read_bytes().replace(b"\r\n", b"\n"))
    return digest.hexdigest()


def request_fingerprint(
    expected: list[tuple[float, float]],
    terrain: Terrain,
    streams: list[list[tuple[float, float]]],
    *,
    mode: str,
    weights: dict[str, float] | None,
    max_grade: float,
    min_radius_m: float,
    stream_offset_m: float | None,
    grade_smooth_m: float | None,
    target_grade: float | None,
    with_profile: bool,
    design_speed: int,
    standard: dict[str, Any] | None = None,
    source: str | None = None,
) -> tuple[str, str]:
    """(저장 키, 묶음 지문) — `generate_initial_route` 와 같은 기본값 풀이로 같은 입력 = 같은 키.

    `source` = 그 엔진 소스 지문의 키(없으면 지금 코드 · `read_stored_any`)."""
    pool = _pool_key(
        expected,
        terrain,
        streams,
        max_grade,
        min_radius_m,
        *resolve_inputs(max_grade, stream_offset_m, grade_smooth_m, target_grade),
    ).hex()
    used = resolve_weights(mode, weights)
    tail = (
        mode,
        sorted(used.items()),
        bool(with_profile),
        int(design_speed),
        json.dumps(standard, sort_keys=True, default=str) if standard else None,
        ALGO_VERSION,
        source or source_fingerprint(),
    )
    key = hashlib.blake2b((pool + repr(tail)).encode(), digest_size=16).hexdigest()
    return key, pool


def is_default(
    mode: str,
    weights: dict[str, float] | None,
    criteria: dict[str, Any] | None,
    terrain: Any,
    saved: dict[str, Any] | None = None,
) -> bool:
    """지킬 결과 — 기본값(빈 제약 · 50/25/25 · 설계 지형) 또는 저장된 제약(`initial_choice`) 그대로.

    모달이 여는 값이라 「최근 2벌」 정리에서 따로 센다(351-1 · 사용자 시도에 안 밀림).
    """
    if terrain is not None:
        return False
    filled = {key: value for key, value in (criteria or {}).items() if value is not None}
    used = resolve_weights(mode, weights)
    if not filled and (mode != "weighted" or used == DEFAULT_WEIGHTS):
        return True
    saved = saved or {}
    return filled == (saved.get("criteria") or {}) and (
        mode != "weighted" or used == saved.get("weights")
    )


def _current(data: Any) -> bool:
    return (
        isinstance(data, dict)
        and data.get("version") == ALGO_VERSION
        and data.get("source") == source_fingerprint()
    )


def read_stored(project_root: Path, key: str) -> dict[str, Any] | None:
    """저장본 한 벌 — 없거나 못 읽거나 판 · 소스 지문이 다르면 None · 파일을 안 건드림(505-1)."""
    path = store_dir(project_root) / f"{key}.json"
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, ValueError):
        return None
    return data if _current(data) else None


def read_stored_any(
    project_root: Path, pool: str, key_of: Callable[[str], str]
) -> tuple[str, dict[str, Any]] | None:
    """(키, 저장본) — 같은 판 · 묶음이고 그 파일의 소스 지문으로 다시 낸 키(`key_of`)가 파일 이름과
    같은 것 중 최근 것 · 없으면 None. 프로젝트 접근 전용(505-1 덧 · 엔진 고쳐도 안내 안 쏟아짐)."""
    found: list[tuple[float, str, dict[str, Any]]] = []
    for path in store_dir(project_root).glob("*.json"):
        try:
            data = json.loads(path.read_text(encoding="utf-8"))
            mtime = path.stat().st_mtime
        except (OSError, ValueError):
            continue
        if not isinstance(data, dict) or data.get("version") != ALGO_VERSION:
            continue
        if data.get("pool") == pool and key_of(str(data.get("source"))) == path.stem:
            found.append((mtime, path.stem, data))
    if not found:
        return None
    _, key, data = max(found, key=lambda item: item[0])
    return key, data


def write_stored(
    project_root: Path,
    key: str,
    pool: str,
    request: dict[str, Any],
    result: dict[str, Any],
    default: bool,
) -> bool:
    """새 키만 적는다 — 이미 있으면 그대로 두고 False(덮어쓰기 없음)."""
    path = store_dir(project_root) / f"{key}.json"
    if path.exists():
        return False
    atomic_write_json(
        path,
        {
            "version": ALGO_VERSION,
            "source": source_fingerprint(),
            "key": key,
            "pool": pool,
            "default": default,
            "created_at": datetime.now().isoformat(timespec="seconds"),
            "request": request,
            "result": result,
        },
    )
    prune_stored(project_root, request, default)
    return True


def _kind(request: Any, default: Any) -> tuple[Any, Any, bool]:
    request = request if isinstance(request, dict) else {}
    return request.get("mode"), request.get("terrain"), bool(default)


def prune_stored(project_root: Path, request: dict[str, Any], default: bool = False) -> int:
    """같은 종류(갈래 · 지형 · 기본값 여부) 저장본을 최근 `KEEP_PER_KIND` 벌만 남기고 지운다.

    기본값 · 저장된 제약 결과(`default`)는 따로 셈 — 사용자 시도가 밀어내지 않음(351-1). 같은 판 ·
    소스 지문 안에서만 셈 — 다른 지문(다른 PC · 워크트리 코드) 파일은 안 건드림(420-2). 지운 수.
    """
    kind = _kind(request, default)
    same: list[tuple[int, Path]] = []
    for path in store_dir(project_root).glob("*.json"):
        try:
            data = json.loads(path.read_text(encoding="utf-8"))
            if _current(data) and _kind(data.get("request"), data.get("default")) == kind:
                same.append((path.stat().st_mtime_ns, path))
        except (OSError, ValueError, AttributeError):
            continue
    same.sort(reverse=True)
    for _, path in same[KEEP_PER_KIND:]:
        path.unlink(missing_ok=True)
    return max(0, len(same) - KEEP_PER_KIND)


def _remove_where(project_root: Path, drop: Any) -> int:
    """저장본 중 `drop(data, path)` 가 참인 파일을 지운다(못 읽는 파일 = data None) — 지운 수."""
    removed = 0
    for path in store_dir(project_root).glob("*.json"):
        try:
            data = json.loads(path.read_text(encoding="utf-8"))
        except (OSError, ValueError):
            data = None
        if drop(data, path):
            path.unlink(missing_ok=True)
            removed += 1
    return removed


def purge_outdated(project_root: Path, now: float | None = None) -> int:
    """다른 판 · 소스 지문 저장본 중 `STALE_DAYS` 지난 것만 지운다(미리 셈 첫머리 · 420-2).

    storage 를 두 PC · 여러 워크트리가 같이 써 코드가 다르면 지문이 다름 — 새것은 안 읽고 둠.
    지운 수."""
    limit = (time.time() if now is None else now) - STALE_DAYS * 86400.0

    def stale(data: Any, path: Path) -> bool:
        try:
            return not _current(data) and path.stat().st_mtime < limit
        except OSError:
            return False

    return _remove_where(project_root, stale)


def drop_pool_results(project_root: Path, pool: str) -> int:
    """[재계산] — 이 후보 묶음(다섯 갈래 · 종단 켬끔)의 저장본을 지우고 메모리 묶음도 버린다.

    `pool_lock(pool)` 안에서 부름(미리 셈이 다시 써 버리지 않게) · 지금 판 · 소스 지문 것만
    (다른 지문 = 다른 PC · 워크트리 것 · 420-2). 지운 수."""
    from B05_Profile.B05_Profile_Engine_RouteInitial import _POOLS

    _POOLS.pop(bytes.fromhex(pool), None)
    return _remove_where(
        project_root, lambda data, _path: _current(data) and data.get("pool") == pool
    )


def forget_pool_on_standard(pool: str, standard: dict[str, Any] | None) -> None:
    """메모리 묶음 후보에 붙은 균형 종단은 그때 표준 횡단면으로 셈 — 바뀌면 묶음을 버린다."""
    from B05_Profile.B05_Profile_Engine_RouteInitial import _POOLS

    mark = json.dumps(standard, sort_keys=True, default=str)
    if _POOL_STANDARDS.get(pool, mark) != mark:
        _POOLS.pop(bytes.fromhex(pool), None)
    _POOL_STANDARDS[pool] = mark


@asynccontextmanager
async def pool_lock(pool: str) -> AsyncIterator[None]:
    # ponytail: 잠금은 지우지 않음 — 입력 묶음 수만큼(작음) · 다중 프로세스면 파일 잠금으로
    async with _LOCKS.setdefault(pool, asyncio.Lock()):
        yield