guanlan-src/app_common/app_common_guanlan/paths.py

# -*- coding: utf-8 -*-
r"""观澜 v2 路径中心 —— 跨平台部署的唯一路径真源 (2026-09-11 用户令)。

## 约定 (硬约束)

1. **代码与配置里只写相对路径**, 相对 **安装根 (ROOT)** —— 不写机器相关绝对路径
   (`/Users/…`、`/Volumes/…`、`C:\…`、`D:\…`、`F:\…`)。
2. 运行时**由本模块解析成绝对路径**, 解析基准是 ROOT —— **不是 cwd**。
   cwd 相对是隐形坑: 从别处调用同一个脚本 (`python <绝对路径>/scripts/x.py`) 或换工作目录,
   路径就会指到别处; 本仓 2026-09-11 实测有 20+ 处 `Path('outputs/rudong/…')` 属此类, 已全部归到本模块。
3. 需要写进产物/清单/页面的路径字符串, 用 **POSIX 相对形式** (`outputs/rudong/…`, 见 `rel()`),
   Windows 与 Linux 通用; 不要用 `os.sep` 拼库存字符串 (那只适合"给人看"的显示, 见 `disp()`)。
4. ROOT 的确定: 环境变量 `WINDSCADA_ROOT` (冻结打包时由启动器设) → 否则按本文件位置回溯
   (`src/paths.py` 的上一级) —— 因此**整个安装目录可以整个拷到别的电脑/别的盘**, 不需要改任何路径。

## 跨平台要点

- 分隔符: 一律 `pathlib`, 需要字符串时 `.as_posix()`;
- 大小写: Linux 区分大小写 —— 目录名只用本模块常量, 不靠大小写变体;
- 解释器: venv 按平台探测 (`.venv/Scripts/python.exe` / `.venv/bin/python`, 见 `venv_python()`);
- 中文目录名 (如 `data/raw/如东`、`故障报警`): UTF-8 存写, 两侧平台均可; 页面显示走 `disp()`。
"""
from __future__ import annotations

try:
    from ._root import install_root as _install_root
except ImportError:                     # 直接当脚本跑（python <本文件>）时没有包上下文
    from _root import install_root as _install_root

import os
import pathlib
import sys

# ---- 安装根: env → 本文件位置回溯 (src/paths.py → <安装目录>) ----------------------------
ROOT = pathlib.Path(os.environ.get('WINDSCADA_ROOT') or _install_root(__file__))

# ---- 输入侧 (离线数据): data/raw, 下一级目录 = 场站名 (见 docs/数据目录结构与落位约定) ----
RAW_ROOT = pathlib.Path(os.environ.get('WINDSCADA_RUDONG_SRC') or (ROOT / 'data' / 'raw'))

# ---- 非场站维度的固定位置 (全部为 ROOT 相对) -------------------------------------------
CONFIGS = ROOT / 'configs'
# P9（2026-09-28 用户令）：模块专属配置下放到各模块 configs/；安装根 configs/ 只留**跨模块共用单件**。
# 次序固定（先安装根，再按模块），同一逻辑配置只许出现在一处 —— 出现两处即报错（配置必须单源）。
MODULE_CONFIG_ROOTS = tuple(ROOT / m / 'configs' for m in
                            ('app_ETL', 'app_frontEnd', 'app_backEnd', 'app_qualityGate'))
REFERENCE = ROOT / 'reference'
RELEASE = ROOT / 'release'
PORTAL = RELEASE / 'portal.html'
VIEWER = RELEASE / 'viewer'
RESOURCES = ROOT / 'resources'
SIM_DIR = RESOURCES / 'oem_envision_sc1_rudong2014'
DOCS = ROOT / 'docs'
LOGS = ROOT / 'logs'
RUN = ROOT / 'run'
SCRIPTS = ROOT / 'scripts'
SRC = ROOT / 'src'
WHEELS = ROOT / 'wheels'
DATA = ROOT / 'data'
SERVE_JSON = CONFIGS / 'serve.json'
MODELS_JSON = CONFIGS / 'models.json'

# ---- 配置目录约定与唯一取用口 (2026-09-17 用户令 2: 统一配置目录及配置文件) ----------------
# 为什么要有这一节: 原来各模块自己拼 `ROOT / 'configs' / 'xxx.yaml'` (实测 9 个模块各拼各的),
# 于是"配置放哪"随时间漂移: 顶层散着 serve.json/models.json/portal_pages.yaml, 还混进过一个
# `serve.json.bak-bomfix`; `scripts/audit_chinese_terms.py` 甚至要**试三个位置**才找得到 terms 库。
# 现在: 目录分工写死在下面, 代码**只从 `config()` / `config_dir()` 取路径**, 不许再手拼字符串;
# `scripts/config_audit.py` 会按这套约定查实物与代码(见 docs §11)。
#
#   configs/<域>/<名字>.<yaml|json|csv>     域 = canonical | contracts | farms | terms | <新增>
#   configs/serve.json                      运行期单件配置 (端口/路径真源)
#   configs/models.json                     运行期单件配置 (本机模型档)
#   configs/portal_pages.yaml               运行期单件配置 (门户页面归口登记表, 见 §10)
TOP_LEVEL_CONFIGS = ('serve.json', 'models.json', 'portal_pages.yaml')
CONFIG_DOMAINS = ('canonical', 'contracts', 'farms', 'terms')
CONFIG_EXTS = ('.yaml', '.yml', '.json', '.csv')


def _config_roots() -> tuple[pathlib.Path, ...]:
    """配置的候选根（安装根在前，模块在后）。"""
    return (CONFIGS,) + MODULE_CONFIG_ROOTS


def _config_hits(*parts: str, kind: str) -> list[pathlib.Path]:
    check = (lambda q: q.is_file()) if kind == 'file' else (lambda q: q.is_dir())
    return [r.joinpath(*parts) for r in _config_roots() if check(r.joinpath(*parts))]


def config(*parts: str, must_exist: bool = False) -> pathlib.Path:
    """配置文件的唯一取用口: `P.config('terms', 'display_map.yaml')` / `P.config('serve.json')`。

    只做路径解析, 不读文件 (读法由调用方决定: yaml/json/csv 各不相同);
    P9 起**多根查找**：安装根 `configs/` → 各模块 `configs/`（次序固定）。
    同一逻辑配置出现在两处 ⇒ 直接抛错（配置必须单源, 否则"改了没生效"是必然的）。
    `must_exist=True` 时不存在就抛 FileNotFoundError —— 配置缺失应该在启动时报出来, 别静默用默认值。
    """
    hits = _config_hits(*parts, kind='file')
    if len(hits) > 1:
        raise RuntimeError('同一配置出现在多处（配置必须单源）: '
                           + ' · '.join(rel(q) for q in hits))
    p = hits[0] if hits else CONFIGS.joinpath(*parts)
    if must_exist and not p.is_file():
        raise FileNotFoundError(f'缺配置文件 {p} (约定见 src/paths.py 配置一节 / docs §11)')
    return p


def config_dir(*parts: str) -> pathlib.Path:
    """配置目录 (域) 的取用口: `P.config_dir('farms')`（多根查找 + 单源校验）。"""
    hits = _config_hits(*parts, kind='dir')
    if len(hits) > 1:
        raise RuntimeError('同一配置域出现在多处（配置必须单源）: '
                           + ' · '.join(rel(q) for q in hits))
    return hits[0] if hits else CONFIGS.joinpath(*parts)


FARMS = config_dir('farms')                      # 场定义目录（P9 起在 app_ETL/configs/farms）


def farm_config(name: str | None = None) -> pathlib.Path | None:
    """场定义配置文件 —— **格式统一为 YAML**, 兼容历史 `.json` (有就优先用)。

    2026-09-17 实测的坑: `app_ETL/configs/farms/` 下有 8 个场定义是 `.yaml`, 而 `available()` 只认 `*.json`
    ⇒ 这些场**根本列不出来**(等于配置写了没人看见); 目录里那个模板还叫 `_模板.json.example`,
    与实物格式相反。现在两边都认, 且约定"新的场定义写 yaml"。
    """
    f = farm(name)
    for ext in ('.yaml', '.yml', '.json'):
        p = FARMS / f'{f}{ext}'
        if p.is_file():
            return p
    return None


def farm(name: str | None = None) -> str:
    """当前场名: 显式 → env WINDSCADA_FARM → 'rudong'。

    这里不 import src.windscada.config 以免循环 (config 反过来要用本模块)。
    需要 set_current() 那种运行期切换时, 调用方把场名显式传进来即可。"""
    return name or os.environ.get('WINDSCADA_FARM') or 'rudong'


# ---- 产物侧: outputs/<场名>/… (页面取数的仓) -------------------------------------------
def out_root(name: str | None = None) -> pathlib.Path:
    return ROOT / 'outputs' / farm(name)


def store(name: str | None = None) -> pathlib.Path:
    """L0 标准仓 (parquet) —— 页面主取数处。"""
    return out_root(name) / 'windscada'


def ont(name: str | None = None) -> pathlib.Path:
    """本体对象库目录 (objects.json / 检索索引 / turbine_params)。"""
    return out_root(name) / 'ontology'


def objects_json(name: str | None = None) -> pathlib.Path:
    return ont(name) / 'objects.json'


def cms(name: str | None = None) -> pathlib.Path:
    """CMS 振动诊断产物目录。"""
    return out_root(name) / 'windcms'


def m5(name: str | None = None) -> pathlib.Path:
    """振动线 handoff / TCM 兼容件目录。"""
    return out_root(name) / 'm5_cms_tcm'


def tcm_replay(name: str | None = None) -> pathlib.Path:
    return out_root(name) / 'tcm_compatible_replay'


def sop(name: str | None = None) -> pathlib.Path:
    """SOP 中间件与评审落盘目录。"""
    return out_root(name) / 'sop'


def guanlan(name: str | None = None) -> pathlib.Path:
    """事实契约与对外派生 (可上云面孔)。"""
    return out_root(name) / 'guanlan'


def pitch(name: str | None = None) -> pathlib.Path:
    return out_root(name) / 'pitch'


def paradigm(name: str | None = None) -> pathlib.Path:
    """范式实验件 (E3/E5/E8 底稿) —— 事实契约的输入之一 (2026-09-16 补: 原先直接用
    `ROOT/'outputs'/'rudong'/'paradigm_r1'` 拼, 既写死场名又绕过了本模块)。"""
    return out_root(name) / 'paradigm_r1'


def report_dir(name: str | None = None) -> pathlib.Path:
    """报告交付件目录 (`交接_振动→状态评估报告_*.md` / `现场单_*.md`) —— 由振动线出件,
    并被 `src/windcms/config.py` 的 knowledge_docs 引用 (2026-09-16 补: 该目录在 v0.2.0 里
    没有明确归属, 一直以 `out_root()/'report'` 的裸拼形式出现)。"""
    return out_root(name) / 'report'


def cloud(name: str | None = None) -> pathlib.Path:
    """可上云面孔 (脱敏后的契约/派生件/页面) —— `scripts/guanlan_cloud_*.py` 的落点。"""
    return guanlan(name) / 'cloud'


def contract(name: str | None = None) -> pathlib.Path:
    """场契约 (机型判据参数), 属 reference 侧, 不在 outputs。"""
    return REFERENCE / farm(name) / 'windscada_contract.yaml'


def station_dir(name: str | None = None) -> pathlib.Path:
    """本场原始件目录 data/raw/<场站名称> —— 具体由场配置扫描结果决定,
    这里只给"约定位置"兜底 (场配置里 windscada.config.farm()['raw_station_dir'] 才是权威)。"""
    return RAW_ROOT / farm(name)


# ---- 解释器与字符串形式 ----------------------------------------------------------------
def venv_python() -> pathlib.Path | None:
    """本安装目录下的 venv 解释器 (跨平台); 不存在返回 None。"""
    for rel in (('Scripts', 'python.exe'), ('bin', 'python'), ('bin', 'python3')):
        p = ROOT / '.venv' / pathlib.Path(*rel)
        if p.exists():
            return p
    return None


def resolve(p) -> pathlib.Path:
    """把"可能是相对路径"的值解析成绝对路径: 相对基准是 ROOT (**不是 cwd**)。"""
    q = pathlib.Path(p)
    return q if q.is_absolute() else (ROOT / q)


def rel(p) -> str:
    """给人/给清单的**相对**路径字符串 (POSIX 形式); 不在 ROOT 内则给绝对 POSIX。"""
    q = pathlib.Path(p)
    try:
        return q.resolve().relative_to(ROOT.resolve()).as_posix()
    except ValueError:
        return q.as_posix()


def disp(p) -> str:
    """给人看的显示路径: 安装目录内的写成 `<安装目录>/…`, 分隔符随本机; 其余原样绝对路径。
    (只用于显示 —— 不要把它写进产物或清单。)"""
    q = pathlib.Path(p)
    try:
        return str(pathlib.Path('<安装目录>') / q.resolve().relative_to(ROOT.resolve()))
    except ValueError:
        return str(q)


# 显示用分隔符: 只服务于给人看的文本 (页面「位置」列)。写进产物/清单的路径请用 rel() 的 POSIX 形式。
SEP = os.sep


def disp_dir(p) -> str:
    """目录的**显示形**: 末尾带本机分隔符 (页面「位置」列用).
    路径本身一律走 `rel()` 的 POSIX 形式; 这里的 os.sep 只服务于"给人看"。"""
    return disp(p) + os.sep


def python_exe() -> str:
    """跑子进程/脚本用的解释器: venv → 否则当前解释器。"""
    v = venv_python()
    return str(v) if v else sys.executable


def platform_tag() -> str:
    """平台标识 (探测脚本/日志用): windows / linux / darwin。"""
    return {'nt': 'windows', 'posix': 'linux'}.get(os.name, os.name) if sys.platform != 'darwin' else 'darwin'


if __name__ == '__main__':       # 自检: python -m src.paths  或  python src/paths.py
    print(f'ROOT       : {ROOT}   (存在: {ROOT.is_dir()})')
    print(f'平台       : {platform_tag()}   cwd: {pathlib.Path.cwd()}')
    print(f'原始件根   : {rel(RAW_ROOT)}   (存在: {RAW_ROOT.is_dir()})')
    print(f'venv 解释器: {venv_python() or "(无, 用 " + sys.executable + ")"}')
    for label, p in (('store', store()), ('ontology', ont()), ('windcms', cms()),
                     ('m5', m5()), ('sop', sop()), ('guanlan', guanlan()),
                     ('release', RELEASE), ('sim_dir', SIM_DIR)):
        print(f'  {label:9s} {rel(p):42s} 存在={p.exists()}')