西占 Skill 后端实现公开（本命盘 / 行运盘）

# 西占 Skill 后端实现公开（本命盘 / 行运盘）

本篇公开「西占」技能从客户端到计算内核的完整实现方式，包含核心代码。所有代码均取自当前线上版本。

一、整体链路

客户端技能（xjx-astro-cn / xjx-astro-global，指导型）
        ↓  模型按 SKILL.md 组织参数
客户端核心内置工具 xjx_fetch_astro
        ↓  （登录凭证由客户端核心代管，技能本身不接触 token）
Node 路由  POST /api/astro/v1/natal  |  /api/astro/v1/transit
        ↓  本机调用，不经过公网
Python 计算服务  127.0.0.1:8098  /api/astro/natal  |  /api/astro/transit
        ↓
Swiss Ephemeris（pyswisseph）

各层职责是刻意分开的：

• 技能层只写「取什么数、怎么解读」，不含任何算法；
• 客户端核心代管凭证并封装网络请求，技能不拼 HTTP、不写域名；
• Node 层负责鉴权、扣费、缓存、幂等与错误归一；
• Python 层只做纯计算，无状态、不写库。

二、输入参数

本命盘：

year / month / day / h / m      出生钟表时间
birth_place                     出生地
country                         国家（可选，用于地名消歧）
time_mode                       true_solar（默认）| clock

行运盘在本命参数之外再加：

current_place / current_country 当前所在地
transit_at                      行运时刻（YYYY-MM-DD HH:MM，缺省为当下）

三、地点与时间基准

地点解析走与心鉴学同源的 GeoNames 库（小库优先、全量库兜底），拿到 lat / lon / tz / meridian 四项。地名只认单条地名，所以对「国家 省 城市」这类写法做候选重试：

def _resolve_place(place, country):
    info = geo_lookup.resolve(place, country)
    if info:
        return info
    try:
        for cand in true_solar._place_candidates(place, country):
            info = geo_lookup.resolve(cand, country)
            if info:
                return info
    except Exception:
        pass
    return None

时间基准默认是真太阳时：先由 true_solar.calc 把钟表时间换算成当地真太阳时，再转 UT 儒略日。有 IANA 时区时按真实偏移（含夏令时），没有时区才回落按经度推算：

def _jd_from_local(local_dt, tz_name, meridian):
    """把某地的当地钟表时间换算成 UT 儒略日；有 IANA 时区时按真实偏移（含 DST）。"""
    off = None
    if tz_name:
        try:
            off = ZoneInfo(tz_name).utcoffset(local_dt)
        except Exception:
            off = None
    if off is None:
        off = timedelta(minutes=(meridian or 120.0) * 4)
    utc = local_dt - off
    return swe.julday(
        utc.year, utc.month, utc.day,
        utc.hour + utc.minute / 60.0 + utc.second / 3600.0,
    )

time_mode=clock 只保留给旧工作流对照用，线上默认走真太阳时。

四、本命盘计算

4.1 十行星

swe.calc_ut 逐个计算太阳、月亮、水星、金星、火星、木星、土星、天王星、海王星、冥王星，输出黄经、星座、所在宫位：

def _bodies(jd, houses):
    out = {}
    for key, pid in (
        ("sun", swe.SUN), ("moon", swe.MOON), ("mercury", swe.MERCURY),
        ("venus", swe.VENUS), ("mars", swe.MARS), ("jupiter", swe.JUPITER),
        ("saturn", swe.SATURN), ("uranus", swe.URANUS),
        ("neptune", swe.NEPTUNE), ("pluto", swe.PLUTO),
    ):
        pos = swe.calc_ut(jd, pid)
        deg = _extract_degree(pos[0])
        out[key] = {
            "name": { ... }[key],
            "degree": round(deg, 2),
            "constellation": _constellation(deg),
            "house": _house(deg, houses),
        }
    return out

4.2 上升 / 下降与宫位制

上升点由 swe.houses(jd, lat, lon) 给出，下降点取对冲 180°；宫位采用整宫制（以上升点所在星座 0° 为第一宫起点）：

def _asc_desc(jd, lat, lon):
    res = swe.houses(jd, lat, lon)
    asc_deg = _extract_degree(res[1])
    asc = {"name": "上升点", "degree": round(asc_deg, 2),
           "constellation": _constellation(asc_deg), "house": 1}
    desc_deg = (asc_deg + 180) % 360
    desc = {"name": "下降点", "degree": round(desc_deg, 2),
            "constellation": _constellation(desc_deg), "house": 7}
    return asc, desc, asc_deg


def _whole_sign_houses(asc_deg):
    first = math.floor(asc_deg / 30) * 30
    return [((first + i * 30) % 360, (first + (i + 1) * 30) % 360) for i in range(12)]

4.3 相位

十行星 + 上升 + 下降 + 四个特殊点两两配对，容许度 ±3°，支持合相 0°、六分相 60°、四分相 90°、三分相 120°、对相 180°：

ASPECTS = {0: "合相", 60: "六分相", 90: "四分相", 120: "三分相", 180: "对相"}

def _aspects(bodies):
    out = []
    for i in range(len(bodies)):
        for j in range(i + 1, len(bodies)):
            b1, b2 = bodies[i], bodies[j]
            diff = abs(b1["degree"] - b2["degree"])
            diff = min(diff, 360 - diff)
            for deg, name in ASPECTS.items():
                if abs(diff - deg) <= 3:
                    out.append({
                        "body1": b1["name"], "body2": b2["name"],
                        "aspect": name, "angle": round(diff, 2),
                        "deviation": round(abs(diff - deg), 2),
                    })
    return out

4.4 特殊点

福点、北交点、南交点、莉莉丝。交点与莉莉丝用 T 表达式推算，福点按传统公式，太阳落 7–12 宫时公式取反：

def _special_points(jd, asc, sun, moon, houses):
    T = (jd - 2451545.0) / 36525.0

    fu_longitude = (
        asc["degree"] + moon["degree"] - sun["degree"]
        if sun["house"] >= 7
        else asc["degree"] + sun["degree"] - moon["degree"]
    ) % 360

    north = (125.044555 - 1934.1361849 * T + 0.0020762 * T ** 2
             + T ** 3 / 467410 - T ** 4 / 60616000) % 360
    south = (north + 180) % 360
    lilith = (83.353243 + 4069.013711 * T - 0.010324 * T ** 2
              - T ** 3 / 80053 + T ** 4 / 18999000 + 180) % 360
    ...

4.5 本命盘组装

def _natal_core(body):
    clock, place, country, place_info = _birth_inputs(body)
    mode = _resolve_mode(body)

    ts = true_solar.calc(clock.strftime("%Y-%m-%d %H:%M"), place, country)
    true_solar_str = ts.get("true_solar")
    chart_local = (datetime.strptime(true_solar_str, "%Y-%m-%d %H:%M")
                   if mode == "true_solar" else clock)

    lat, lon = float(place_info["lat"]), float(place_info["lon"])
    meridian = float(place_info.get("meridian") or 120.0)
    tz = place_info.get("tz")

    jd = _jd_from_local(chart_local, tz, meridian)
    asc, desc, asc_deg = _asc_desc(jd, lat, lon)
    houses = _whole_sign_houses(asc_deg)
    basic = _bodies(jd, houses)
    special = _special_points(jd, asc, basic["sun"], basic["moon"], houses)
    ...

返回值只暴露 report 与 meta：meta 里带 time_mode、birth_clock、true_solar、chart_time 与地点四项（含经纬度、时区、匹配方式），便于排查「为什么这张盘是这样」。

五、行运盘

行运时刻按当前所在地的当地时间确定：传了 transit_at 就按当地时间解析，没传就用当下的 UTC 转当地时区（含 DST）。行运不改变上升点——上升、下降沿用本命盘，与旧工作流口径一致。

def compute_transit(body):
    core = _natal_core(body)
    transit_jd, transit_local, current_info = _transit_moment(body)

    houses = core["houses"]
    natal_all = core["all_bodies"]

    transit_basic = _bodies(transit_jd, houses)
    transit_all = {}
    transit_all.update(transit_basic)
    transit_all["asc"] = natal_all["asc"]
    transit_all["desc"] = natal_all["desc"]
    transit_special = _special_points(
        transit_jd, natal_all["asc"], transit_basic["sun"], transit_basic["moon"], houses)
    transit_all.update(transit_special)

    constellations = _constellation_output(natal_all, transit_all, houses)
    transit_natal_aspects = _transit_aspects(
        list(transit_all.values()), list(natal_all.values()))
    ...

输出三块：

1. constellations：十二宫各自的本命星体 / 行运星体； 2. natal_aspects：本命盘相位； 3. transit_natal_aspects：行运星体对本命星体的相位（含实际角度与偏差）。

六、API 层：鉴权、扣费、缓存

入口是 Node 的 /api/astro/v1/natal 与 /api/astro/v1/transit，处理顺序固定为「幂等 → 缓存 → 计算 → 扣费 → 落缓存」。

// 1) 幂等：同一 request_id 24 小时内直接回放，不重新计算、不二次扣费
if (requestId) {
  const rows = await q(
    `SELECT response_json FROM astro_request_logs
     WHERE user_id = ? AND request_id = ? AND expires_at > NOW() LIMIT 1`,
    [user.id, requestId]);
  if (rows.length) { /* 回放，billing.replayed = true */ }
}

// 2) 同参短期缓存：input_fp + 2 小时窗口命中即免费
const hit = await q(
  `SELECT report_body FROM astro_report_cache
   WHERE user_id = ? AND kind = ? AND input_fp = ? AND window_key = ? AND expires_at > NOW()
   ORDER BY id DESC LIMIT 1`, [user.id, kind, fp, wk]);

// 3) 计算（失败不扣费）
const py = await callPy(kind === 'natal' ? '/api/astro/natal' : '/api/astro/transit', body);

// 4) 先算成功再扣费，避免算失败还扣钱
balance = await chargeToken(user.id, BUY_FEE, title);

计费口径：

• 每成功生成一份报告 6 token；
• 同参数 120 分钟内重复请求走缓存，不再扣费；
• 同一 request_id 24 小时内回放，不二次扣费；
• 计算失败、参数错误、余额不足均不扣费；
• 返回体里的 billing 明确标注本次是 charged 还是 cached / replayed，以及余额。

扣费是单条原子语句，避免并发下透支：

const r = await q('UPDATE users SET token_balance = token_balance - ? WHERE id = ? AND token_balance >= ?',
  [amount, uid, amount]);
if (!r || !r.affectedRows) throw new Error('Token 不足或扣费冲突，请重试');

七、部署与依赖

• Node：api-server（PM2 托管），挂载 astro-api.js；
• Python：xinjianxue-py 的 server.py，监听 8098，只接受本机调用；
• 计算依赖 pyswisseph；地点库为 GeoNames 构建的 SQLite（小库 + 全量库）；
• 计算服务无状态，重启不影响数据；报告缓存与幂等记录都在 MySQL，按过期时间自动清理。

八、这套实现的特点

1. 技能与算法彻底分离：换模型、换提示词不影响计算口径。 2. 只认真太阳时：钟表时间与真太阳时的换算与心鉴学其他模块共用同一套代码（含夏令时）。 3. 先算后扣、失败不计费：扣费永远发生在计算成功之后。 4. 可复现：同样的输入 + 同样的窗口，返回同一份结果；meta 里保留全部中间基准，便于逐项核对。