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` 을
올린다 — 빠뜨려도 엔진 소스 지문이 바뀌어 새로 셈. 옛 판 · 옛 지문 파일은 미리 셈 때 지운다(420).
기본값 · 저장된 제약 결과(`default`)는 「최근 벌」을 따로 센다(351-1).
"""

from __future__ import annotations

import asyncio
import functools
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

# 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"
# 종류(갈래 · 지형) 마다 남길 벌 수 — 꺼낼 때 수정시각을 새로 해 최근에 쓴 것이 남음
KEEP_PER_KIND = 2
# 업로드 체인 끝에 미리 셈할 기본 다섯 알약(화면 차례) · 추적은 묶음이 선 뒤라 `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,
) -> 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),
        json.dumps(standard, sort_keys=True, default=str) if standard else None,
        ALGO_VERSION,
        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."""
    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 _current(data):
        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,
            "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). 지운 수.
    """
    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 _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)` 가 참인 파일을 지운다(못 읽는 파일도 지움) — 지운 수."""
    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 data is None or drop(data):
            path.unlink(missing_ok=True)
            removed += 1
    return removed


def purge_outdated(project_root: Path) -> int:
    """옛 판 · 옛 소스 지문 저장본을 지운다(미리 셈 첫머리 · 420) — 지운 수."""
    return _remove_where(project_root, lambda data: not _current(data))


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

    `pool_lock(pool)` 안에서 부름(미리 셈이 다시 써 버리지 않게). 지운 수."""
    from B05_Profile.B05_Profile_Engine_RouteInitial import _POOLS

    _POOLS.pop(bytes.fromhex(pool), None)
    return _remove_where(project_root, lambda data: 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