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 · 여러 워크트리가 같이 씀).
기본값 · 저장된 제약 결과(`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

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
# 다른 판 · 소스 지문 저장본을 이 날수 지나면 미리 셈 때 지움(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,
) -> 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). 같은 판 ·
    소스 지문 안에서만 셈 — 다른 지문(다른 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