yhm-shop-data/store-ops/backend/codesync.py

# -*- coding: utf-8 -*-
"""
阈值「写回代码」

需求：在【指标配置】面板改数字，就是改规则本身 —— 不再有「代码默认值」和「库里自定义值」
两套数字打架。

做法：
  1. 面板保存 → 先写库（立即生效）
  2. 再把新值写回 rules.py 的 DEFAULT_THRESHOLDS（代码里的「标准」跟着变）
  3. 重新加载 rules 模块

出厂值单独留一份快照（data/factory_thresholds.json），
【恢复默认】= 回到出厂值，并且同样写回代码，保证任何时刻库和代码都一致。
"""
import importlib
import importlib.util
import json
import os
import re

BASE = os.path.dirname(os.path.abspath(__file__))
RULES_PY = os.path.join(BASE, "rules.py")
DATA_DIR = os.path.join(os.path.dirname(BASE), "data")
FACTORY_FILE = os.path.join(DATA_DIR, "factory_thresholds.json")


def _fmt(v):
    """把值格式化成 Python 源码里的写法"""
    if isinstance(v, (list, tuple)):
        return "[" + ", ".join(_fmt(x) for x in v) + "]"
    if isinstance(v, bool):
        return "True" if v else "False"
    if isinstance(v, float):
        if abs(v - round(v)) < 1e-12:
            return str(int(round(v)))
        return "%.10g" % v          # 去掉浮点噪音：0.0028000000000000004 → 0.0028
    if isinstance(v, int):
        return str(v)
    return repr(v)


def _value_re(v):
    """匹配某个值在源码里的形式（数字/列表），用于定位替换"""
    if isinstance(v, (list, tuple)):
        return r"\[[^\[\]]*\]"
    return r"[^,\]\}\n#]+"


def write_defaults(T):
    """
    把 T 写回 rules.py 的 DEFAULT_THRESHOLDS。
    只改已有键的「值」，保留键名、缩进、行尾注释和其它代码。
    返回 (ok, msg)。
    """
    try:
        with open(RULES_PY, encoding="utf-8") as f:
            src = f.read()
    except OSError as e:
        return False, "读不到 rules.py：%s" % e

    start = src.find("DEFAULT_THRESHOLDS = {")
    if start < 0:
        return False, "rules.py 里找不到 DEFAULT_THRESHOLDS"
    end = src.find("\n}", start)
    if end < 0:
        return False, "DEFAULT_THRESHOLDS 结构异常"
    head, block, tail = src[:start], src[start:end], src[end:]

    changed = 0
    for k, v in T.items():
        # 组1=键与冒号  组2=旧值  组3=逗号 + 行尾注释 + 换行
        pat = re.compile(
            r'("' + re.escape(k) + r'"\s*:\s*)(' + _value_re(v) + r')(\s*,[^\n]*\n)')
        m = pat.search(block)
        if not m:
            continue
        if m.group(2).strip() == _fmt(v):
            continue
        block = block[:m.start()] + m.group(1) + _fmt(v) + m.group(3) + block[m.end():]
        changed += 1

    if not changed:
        return True, "代码默认值已是最新，无需变更"

    try:
        with open(RULES_PY + ".bak", "w", encoding="utf-8") as f:
            f.write(src)
        tmp = RULES_PY + ".tmp"
        with open(tmp, "w", encoding="utf-8") as f:
            f.write(head + block + tail)
        os.replace(tmp, RULES_PY)
    except OSError as e:
        return False, "写 rules.py 失败：%s" % e

    # 关键：macOS 自带 Python 把 .pyc 放在 ~/Library/Caches/com.apple.python，
    # 不在这里清掉的话，同尺寸同秒的改动会被旧字节码顶掉（踩过）
    try:
        cache = importlib.util.cache_from_source(RULES_PY)
        if os.path.exists(cache):
            os.remove(cache)
    except Exception:
        pass

    try:
        import rules
        importlib.reload(rules)
    except Exception as e:
        return False, "已写入但重新加载失败：%s（重启服务后生效）" % e

    return True, "已同步写回代码（%d 项）" % changed


def get_factory():
    """出厂阈值：首次运行时从代码快照一份，之后固定不变"""
    os.makedirs(DATA_DIR, exist_ok=True)
    if os.path.isfile(FACTORY_FILE):
        try:
            with open(FACTORY_FILE, encoding="utf-8") as f:
                return json.load(f)
        except Exception:
            pass
    import rules
    snap = dict(rules.DEFAULT_THRESHOLDS)
    try:
        with open(FACTORY_FILE, "w", encoding="utf-8") as f:
            json.dump(snap, f, ensure_ascii=False, indent=1)
    except OSError:
        pass
    return snap