guanlan-src/app_ontology/app_ontology_guanlan/sop/farm_paths.py

# -*- coding: utf-8 -*-
"""farm_paths.py — Windows→Mac/Linux 路径规范化助手 (零改迁移用).

老的 per-farm 脚本写死了 Windows 盘符路径 (F:\\datang\\..., D:\\wind-analytics\\...,
D:\\data\\...). 本机无盘符。本模块把这些 win 路径按"有序前缀映射表"翻译成本机 Path,
让脚本只需把 `r"F:\\..."` 包一层 `winpath(...)` 即可,不必逐个硬改、不必批量 sed。

为什么不能盲目把 `X:\\` → `/Volumes/T9/`:
  - `D:\\wind-analytics\\...` 是**仓库**本身 → ~/wind-analytics (不在 T9)
  - `D:\\data\\...` 是旧 WIND_DATA_ROOT → /Volumes/T9/data
  - `F:\\...` 才是 T9 盘根 → /Volumes/T9/...
  - `D:\\BaiduNetdiskDownload`、`C:\\Users` 等本机无对应物 → 必须报错,不能静默乱指。

映射目标可被环境变量覆盖 (.env 即可):
  WIND_T9_ROOT     默认 /Volumes/T9        (F:\\ 盘根)
  WIND_DATA_ROOT   默认 /Volumes/T9/data   (D:\\data 替身)
  WIND_REPO_ROOT   默认 = 本仓库根(由本文件位置推出)
  WIND_PATH_STRICT 默认 1 = 遇未知盘符前缀抛错;设 0 = 仅告警并尽力按 T9 兜底。

Windows 上 (os.name=='nt') 本模块为 no-op,原样返回 → 旧机继续可跑,跨平台安全。

用法:
    from src.sop.farm_paths import winpath
    SEC = winpath(r"F:\\datang\\景泰兴泉风电场\\数据转换及清洗\\sec2")  # -> Path(/Volumes/T9/datang/...)
    files = sorted((SEC).glob("DJ*.csv"))
自检: python src/sop/farm_paths.py
"""
from __future__ import annotations

try:
    from app_common.app_common_guanlan.api import install_root as _install_root
except ImportError:                     # 理论不可达；包结构异常时回退到按位置上跳
    from pathlib import Path as _P
    def _install_root(_f): return _P(_f).resolve().parents[4]
import os
import re
import warnings
from pathlib import Path

# 仓库根 = 安装根（P6 起按标记查找；原按 src/sop/farm_paths.py 的 parents[2] 推算）
_REPO_ROOT_DEFAULT = _install_root(__file__)

T9_ROOT   = os.environ.get("WIND_T9_ROOT", "/Volumes/T9")
DATA_ROOT = os.environ.get("WIND_DATA_ROOT", str(Path(T9_ROOT) / "data"))
REPO_ROOT = os.environ.get("WIND_REPO_ROOT", str(_REPO_ROOT_DEFAULT))
STRICT    = os.environ.get("WIND_PATH_STRICT", "1") not in ("0", "false", "False", "")

# 有序前缀映射 (win 前缀用小写正斜杠形式; 最具体的在前 —— D:/wind-analytics 必须先于 D:/ 命中)
PREFIX_MAP: list[tuple[str, str]] = [
    ("d:/wind-analytics", REPO_ROOT),   # 仓库自身 -> ~/wind-analytics
    ("d:/data",           DATA_ROOT),   # 旧 WIND_DATA_ROOT -> /Volumes/T9/data
    ("f:/data",           DATA_ROOT),   # T9 上的 data (与 D:\data 同一替身)
    ("f:/",               T9_ROOT),     # T9 盘根 -> /Volumes/T9
]

_DRIVE_RE = re.compile(r"^([A-Za-z]):[\\/]?")


def winpath(p) -> Path:
    """把可能含 Windows 盘符的路径翻译成本机 Path。无盘符/已是 POSIX 路径则原样返回。"""
    s = str(p)
    # Windows 原生运行: 不翻译,原样 (旧机兼容)
    if os.name == "nt":
        return Path(s)
    # 无盘符前缀 (相对路径 / 已是 /Volumes/... 的 POSIX 路径) -> 原样
    if not _DRIVE_RE.match(s):
        return Path(s)

    norm = s.replace("\\", "/")            # 反斜杠 -> 正斜杠
    low = norm.lower()
    for prefix, target in PREFIX_MAP:
        # 前缀以 '/' 结尾 (如 'f:/') 用 startswith; 否则需整段相等或后随 '/'
        if prefix.endswith("/"):
            hit = low.startswith(prefix)
            cut = len(prefix)
        else:
            hit = low == prefix or low.startswith(prefix + "/")
            cut = len(prefix)
        if hit:
            tail = norm[cut:].lstrip("/")
            return Path(target) / tail if tail else Path(target)

    # 未知盘符前缀: 本机无对应物 (D:\BaiduNetdiskDownload, C:\Users, 裸 D:\ ...)
    drive = _DRIVE_RE.match(s).group(1).upper()
    known = ", ".join(pfx for pfx, _ in PREFIX_MAP)
    msg = (f"farm_paths: 无法映射的 Windows 路径根 {drive!r}: {s!r}\n"
           f"  已知映射: {known}\n"
           f"  这是旧机本地路径,T9 上无对应物。请确认数据是否已搬到 T9,"
           f"或扩充 PREFIX_MAP / 设 WIND_PATH_STRICT=0 兜底。")
    if STRICT:
        raise ValueError(msg)
    warnings.warn(msg, stacklevel=2)
    # 非严格兜底: 把任意盘符当 T9 根 (大概率仍不对,仅为不中断流程)
    tail = norm[_DRIVE_RE.match(norm).end():].lstrip("/")
    return Path(T9_ROOT) / tail


def winstr(p) -> str:
    """winpath 的字符串版 (给 glob.glob / os.path.join 等吃 str 的旧代码)。"""
    return str(winpath(p))


if __name__ == "__main__":
    # 控制台可能是 GBK(中文 Windows 代码页 936): 正文里的 ✔ ✗ ✅ ⚠ 这类字符编不出来会抛
    # UnicodeEncodeError, 脚本干成了事却以退出码 1 结束(同类坑见 src/console.py)。降级为 '?' 而不是崩;
    # 不用 import 是为了兼顾 python -m 与直接当脚本跑两种启动方式。
    import sys as _sys
    for _s in (_sys.stdout, _sys.stderr):
        try: _s.reconfigure(errors='replace')
        except Exception: pass
    # 零依赖自检
    ok = True
    cases = {
        r"F:\datang\景泰兴泉风电场\sec2": f"{T9_ROOT}/datang/景泰兴泉风电场/sec2",
        r"F:\data\foo.csv":              f"{DATA_ROOT}/foo.csv",
        r"D:\data\bar":                  f"{DATA_ROOT}/bar",
        r"D:\wind-analytics\outputs\x":  f"{REPO_ROOT}/outputs/x",
        r"F:\\zhangyaoxian\\a":          f"{T9_ROOT}/zhangyaoxian/a",   # 双反斜杠
        "F:/datang/b":                   f"{T9_ROOT}/datang/b",          # 已是正斜杠
        "outputs/rel/path":              "outputs/rel/path",             # 相对路径原样
        "/Volumes/T9/already/posix":     "/Volumes/T9/already/posix",    # POSIX 原样
    }
    for win, exp in cases.items():
        got = str(winpath(win))
        flag = "PASS" if got == exp else "FAIL"
        if got != exp:
            ok = False
        print(f"  {flag}  {win!r:48} -> {got}")
    # 未知根应报错 (STRICT)
    try:
        winpath(r"D:\BaiduNetdiskDownload\x")
        print("  FAIL  未知盘符未报错"); ok = False
    except ValueError:
        print("  PASS  未知盘符 D:\\BaiduNetdiskDownload 正确报错")
    # 映射根存在性提示 (不计入 PASS/FAIL)
    print(f"\n  T9_ROOT={T9_ROOT} exists={Path(T9_ROOT).exists()}")
    print(f"  DATA_ROOT={DATA_ROOT} exists={Path(DATA_ROOT).exists()}")
    print(f"  REPO_ROOT={REPO_ROOT} exists={Path(REPO_ROOT).exists()}")
    print("\n" + ("✅ farm_paths 自检全过" if ok else "❌ farm_paths 自检有 FAIL"))
    raise SystemExit(0 if ok else 1)