Aislo/B05_Profile/B05_Profile_Engine_RouteInitial_Store.py

"""계획노선 추천 결과 영구 저장(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` 을 올린다(옛 파일은 다른 키라 안 쓰임).
"""

from __future__ import annotations

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

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

ALGO_VERSION = 2
STORE_DIR = "B05_Profile/route_recommend"
# 종류(갈래 · 지형) 마다 남길 벌 수 — 꺼낼 때 수정시각을 새로 해 최근에 쓴 것이 남음
KEEP_PER_KIND = 2
# 업로드 체인 끝에 미리 셈할 기본 다섯 알약(화면 차례) · 추적은 묶음이 선 뒤라 `similar` 가 채워짐
DEFAULT_MODES = ("gradeline", "earthwork", "gentle", "weighted", "follow")

# 묶음 지문마다 잠금 — 진행 중 같은 묶음(같은 키 포함)은 기다렸다 꺼냄(두 번 셈 막음)
_LOCKS: dict[str, asyncio.Lock] = {}


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


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,
) -> tuple[str, str]:
    """(저장 키, 묶음 지문) — `generate_initial_route` 와 같은 기본값 풀이로 같은 입력 = 같은 키."""
    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), ALGO_VERSION)
    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
) -> bool:
    """기본값 결과 — 빈 제약 · 50/25/25 · 설계 지형. 표시일 뿐 지우는 데 쓰지 않음."""
    empty = not any(value is not None for value in (criteria or {}).values())
    plain = mode != "weighted" or resolve_weights(mode, weights) == DEFAULT_WEIGHTS
    return empty and plain and terrain is None


def read_stored(project_root: Path, key: str) -> dict[str, Any] | None:
    """저장본 한 벌 — 없거나 못 읽거나 판이 다르면 None."""
    path = store_dir(project_root) / f"{key}.json"
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, ValueError):
        return None
    if not isinstance(data, dict) or data.get("version") != ALGO_VERSION:
        return None
    try:
        path.touch()  # 최근에 쓴 것 = 정리 때 남김
    except OSError:
        pass
    return 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,
            "key": key,
            "pool": pool,
            "default": default,
            "created_at": datetime.now().isoformat(timespec="seconds"),
            "request": request,
            "result": result,
        },
    )
    prune_stored(project_root, request)
    return True


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


def prune_stored(project_root: Path, request: dict[str, Any]) -> int:
    """같은 종류(갈래 · 지형) 저장본을 최근 `KEEP_PER_KIND` 벌만 남기고 지운다. 지운 수."""
    kind = _kind(request)
    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 _kind(data.get("request")) == 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)


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