# 西占 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 四项。地名只认单条地名,所以对「国家 省 城市」这类写法做候选重试:
``python 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 时区时按真实偏移(含夏令时),没有时区才回落按经度推算:
``python 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 逐个计算太阳、月亮、水星、金星、火星、木星、土星、天王星、海王星、冥王星,输出黄经、星座、所在宫位:
``python 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° 为第一宫起点):
```python 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°:
```python 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 宫时公式取反:
```python 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 本命盘组装
```python 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)。行运不改变上升点——上升、下降沿用本命盘,与旧工作流口径一致。
```python 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,处理顺序固定为「幂等 → 缓存 → 计算 → 扣费 → 落缓存」。
``javascript // 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_id24 小时内回放,不二次扣费; - 计算失败、参数错误、余额不足均不扣费;
- 返回体里的
billing明确标注本次是charged还是cached / replayed,以及余额。
扣费是单条原子语句,避免并发下透支:
``javascript 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 里保留全部中间基准,便于逐项核对。