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` 을
올린다 — 빠뜨려도 엔진 소스 지문이 바뀌어 새로 셈. 다른 판 · 지문 파일은 안 읽고 남기며 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 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
from common_util.common_util_time import utc_now_text

# 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`) 그대로 · 지형은
    안 따짐(지형마다 미리 셈 · 509-5 · `terrain` 은 부르는 쪽 꼴 맞춤).

    모달이 여는 값이라 「최근 2벌」 정리에서 따로 센다(351-1 · 사용자 시도에 안 밀림).
    """
    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": utc_now_text(),
            "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