feat(muhurta): location-exact on-demand muhurta search (BUG-1822)

原做法把参考时刻当 UT 12:00、不用地点、五要素只取「日出 + 6 小时」一个点。

- 新增 scripts/muhurta_search_api.py:必填 activity/日期/地点/tz 或 timezone_id/
  岁差/交点,缺一项报错;每天取实际日出日落,取不到整天 blocked;每个候选时段
  在起止两端各取一次五要素与 Hora,时段内变化标出变化时刻;现有单点评分放进
  legacy_reference 并标 not_final
- 薄路由 /api/muhurta_search(_load_local_module,无新增类方法)
- CLI 加 --lat --lon --tz --timezone-id --ayanamsa --node-mode;无地点时首行写明
  近似值、不给推荐时段,并去掉 int(hour_from_sunrise + 6) 冒充 UT
- 全局时区推断收紧为「lat/lon 都给了才推断」,否则择时不给地点会崩

Tests: tests/test_muhurta_search_api_20261109.py 14 条;
test_api_server_growth_contract 9 条通过;pytest -k muhurta 116 过 5 败,
5 条在基线 b453505d 上同样失败(无关)。

Co-Authored-By: Grok (agent loop)
This commit is contained in:
jesse-ux
2026-11-09 08:50:19 +08:00
co-authored by Grok (agent loop)
parent 7cdf83106a
commit 28c719cbde
7 changed files with 853 additions and 13 deletions
+14
View File
@@ -24397,3 +24397,17 @@
- 判断:她的生成器修复(§12)在本仓没有可移植对象;快照不重生成、不改。若以后移植她的生成器,需要一起带她的 4 个新用例。
- 关联记录:无
- 修复版本:无
## BUG-1822 | 简易择时把参考时刻当成 UT 12:00、不用地点,五要素只取『日出后 6 小时』一个点
- 状态:fixed-pending-verify(分支 codex/muhurta-20261109)
- 首次发现 / 最近更新:2026-11-07(占星师 Q9)/ 2026-11-09(修复)
- 影响面:scripts/cmd_muhurta.py、scripts/jyotish_engine.py 的 muhurta 子命令参数、scripts/muhurta.py 的 panchanga_range_report 默认参考时刻(只被新模块绕开,未改其默认行为)。
- 现象:CLI 只有 --date/--activity/--scan-days/--hour-from-sunrise;_get_sun_moon_lons(y, m, d, int(hour_from_sunrise + 6)) 把当地小时当 UT 传给 swe.julday,等于固定 UT 12:00;不用地点,日出日落退回 06:00/18:00;五要素(Tithi/Nakshatra/Yoga/Karana)只在『日出 + 6 小时』一个点取值,再用这个值给全天打分。
- 触发条件:任何用 CLI 或 /api/muhurta 做择时判断的调用。
- 根因:简易 CLI 是早期近似实现,从未接地点、时区与逐时段核对;底层 muhurta.py 已有实际日出与五要素结束时刻的计算能力,但 CLI 没有调用。
- 修复:新增 scripts/muhurta_search_api.py(muhurta_search_api,错误类 MuhurtaSearchInputError):必填 activity/start_date/end_date/lat/lon/(tz 或 timezone_id)/ayanamsa/node_mode,缺一项报错并点名字段;未知活动报错不改写;每天用 SwissEph 取实际日出日落,取不到就整天 blocked(不退回 06:00/18:00,也不用近似日月位置);对每个候选时段在起止两端各取一次五要素与 Hora,时段内变化标出变化时刻;现有单点评分放进 legacy_reference 并标 not_final。新增薄路由 /api/muhurta_search;CLI 增加 --lat/--lon/--tz/--timezone-id/--ayanamsa/--node-mode,有地点走新模块,无地点首行写明『未提供地点:结果为近似值,不能作为择时结论』且不给推荐时段,并删掉 int(hour_from_sunrise + 6) 的 UT 冒充。
- 验证:tests/test_muhurta_search_api_20261109.py(14 条:缺字段点名、未知活动、63 天上限、新德里/伦敦日出与夏令时 +1、逐时段两端要素、时段内变化与变化时刻、无日出整天 blocked 且无 06:00/18:00、指纹稳定且随口径变化、legacy 标 not_final、路由薄注册、CLI 有/无地点两种输出);tests/test_api_server_growth_contract.py 9 条通过;pytest -k muhurta 116 过 5 败,5 条失败在基线 b453505d 上同样失败(脚本输出 factor_only_scoring_observation、测试期望 blocked_until_oracle),与本单无关。
- 防复发:上述测试;新模块的必填校验不允许补默认值。
- 关联记录:占星师 2026-11-07 Q9、2026-11-08 第五轮第 9 条;#173 按需计算骨架(后续接聊天用)
- 修复版本:分支 codex/muhurta-20261109
+42
View File
@@ -0,0 +1,42 @@
# PROGRESS · 择时改为按地点、实际日出、时区(#179,BUG-1822,2026-11-09)
- 分支:`codex/muhurta-20261109`(基线 `b453505d`)
- 任务书:`docs/tasks/TASK-muhurta-20261109.md`
- Bug:`docs/BUG_HISTORY.md` 新增 BUG-1822
## 改了什么
| 文件 | 改动 |
| --- | --- |
| `scripts/muhurta_search_api.py`(新) | `muhurta_search_api(body)` + `MuhurtaSearchInputError`。必填 `activity`、`start_date`、`end_date`、`lat`、`lon`、`tz` 或 `timezone_id`、`ayanamsa`、`node_mode`,缺一项报错并点名字段;未知活动报错(不改写成 business);超过 63 天报错 |
| 同上 | 每天用 `muhurta._swisseph_solar_times_local` 取**实际**日出日落;取不到就整天 `calculation_status: blocked`(`sunrise_unavailable`),**不退回 06:00 / 18:00,也不用近似日月位置** |
| 同上 | 每个候选时段在**起止两端**各取一次 Tithi / Nakshatra / Yoga / Karana / Hora(都用 `muhurta.py` 自己的计算函数),起止不同就写进 `element_changes_within_window`,变化时刻二分到分钟 |
| 同上 | 现有单点参考评分(`muhurta_range_search`)原样放进 `legacy_reference`,标 `not_final: true`;规则、分数、阈值一律没改 |
| 同上 | 信封:`schema` / `tool` / `calculation_status` / `profile` / `requested` / `input_fingerprint` / `engine_version` / `raw` / `gaps` / `computed_at_utc`,与 #173 的按需计算合同同形 |
| `scripts/jyotish_api_server.py` | `do_POST` 里 `/api/muhurta` 之后加一条薄 `elif path == '/api/muhurta_search'`(`_load_local_module('muhurta_search_api')`,错误映射 `BadRequest`)。没有新增类方法 |
| `scripts/jyotish_engine.py` | muhurta 子命令加 `--lat` `--lon` `--tz` `--timezone-id` `--ayanamsa` `--node-mode`;另外把全局「`args.tz is None` 就按 lat/lon 推断时区」收紧为「lat/lon 都给了才推断」,否则择时不给地点时会崩 |
| `scripts/cmd_muhurta.py` | 有地点:走新模块,打印实际日出日落、每个时段起止五要素、时段内变化、旧单点参考(标不最终)。无地点:首行 `⚠️ 未提供地点:结果为近似值,不能作为择时结论`,五要素概览改按**显式 UT 12:00** 取(不再用 `int(hour_from_sunrise + 6)` 冒充当地小时),且不给推荐时段 |
没碰:`muhurta.py` 的规则表与权重(只在调用侧换取值时刻)、`references/**`、`frontend/**`、`scripts/ondemand_calc/**`、其他冻结文件。
## 自测结果
| 命令 | 结果 |
| --- | --- |
| `pytest tests/test_muhurta_search_api_20261109.py` | 14 过 / 0 败 |
| `pytest tests/test_api_server_growth_contract.py` | 9 过 / 0 败 |
| `pytest tests -k muhurta` | 116 过 / 5 败 |
| CLI(新德里 2026-06-15) | 日出 05:23、日落 19:21;三个时段;时段内检测到 Hora / Yoga 变化并给出变化时刻 |
`-k muhurta` 的 5 条失败都是既有的研究脚本状态词不一致(脚本输出 `factor_only_scoring_observation`,测试期望 `blocked_until_oracle`)。**我在基线 `b453505d` 的干净工作树上复跑了其中两条,同样失败**,与本单无关:
- `test_muhurta_factor_probe.py`
- `test_muhurta_factor_only_scoring_packet.py`
- 另外三条同源:`test_muhurta_factor_scoring_readiness.py`、`test_muhurta_numeric_candidate_capture_packet.py`、`test_muhurta_oss_worked_example_readiness.py`
## 口径与缺口
- 岁差、交点、地点、时区全部由调用方显式给出,缺一项报错;不给默认值。
- `timezone_id` 按天换算偏移,伦敦 2026-06-15 得 +1(BST),2026-01-15 得 0。
- 日落之后不在本单范围(`time_window` 参数已接进校验,但窗口选择沿用 `muhurta.py` 现有候选),留给后续单。
- 接进聊天要等 #173 的按需计算骨架验收合并,本单只做后端与 CLI,没有改任何前端文件。
+124 -11
View File
@@ -91,19 +91,107 @@ def _format_activity_checks(activity_checks: Dict, target_activity: Optional[str
return lines
def _print_location_exact(
date_str: str,
activity: Optional[str],
lat: float,
lon: float,
tz: Optional[float],
timezone_id: Optional[str],
ayanamsa: Optional[str],
node_mode: Optional[str],
time_window: Optional[Dict],
limit: int,
) -> int:
"""给了地点:走 muhurta_search_api,按实际日出与逐时段五要素输出。"""
from muhurta_search_api import muhurta_search_api, MuhurtaSearchInputError
if not ayanamsa or not node_mode:
print("[ERROR] 给了地点就必须显式给 --ayanamsa 与 --node-mode(不要让默认值替用户选择)")
return 1
body = {
"activity": activity or "business",
"start_date": date_str,
"end_date": date_str,
"lat": lat,
"lon": lon,
"ayanamsa": ayanamsa,
"node_mode": node_mode,
"limit": limit,
}
if tz is not None:
body["tz"] = tz
if timezone_id:
body["timezone_id"] = timezone_id
if time_window:
body["time_window"] = time_window
try:
result = muhurta_search_api(body)
except MuhurtaSearchInputError as exc:
print(f"[ERROR] {exc}")
return 1
print("=" * 64)
print("🕐 Muhurta 择时分析(地点精确,实际日出与日落)")
print("=" * 64)
profile = result.get("profile") or {}
print(f"地点 {lat}, {lon}|时区 {profile.get('tz')}|岁差 {profile.get('ayanamsa')}|交点 {profile.get('node_mode')}")
print(f"状态 {result.get('calculation_status')}|指纹 {result.get('input_fingerprint', '')[:16]}")
for day in (result.get("raw") or {}).get("days") or []:
print()
if day.get("calculation_status") != "executed":
print(f"📅 {day.get('date')}:阻塞({day.get('reason')}),没有可用日出,不出结论")
continue
print(f"📅 {day.get('date')} 日出 {day.get('sunrise_local')} 日落 {day.get('sunset_local')}")
windows = day.get("windows") or []
if not windows:
print(" 这一天没有候选时段")
for window in windows:
if window.get("calculation_status") != "executed":
print(f" · {window.get('name')} {window.get('start_local')}–{window.get('end_local')}:阻塞")
continue
start = window.get("elements_at_start") or {}
end = window.get("elements_at_end") or {}
print(f" · {window.get('name')} {window.get('start_local')}–{window.get('end_local')}"
f"(UTC {window.get('start_utc')}–{window.get('end_utc')})")
print(f" 起:Tithi {start.get('tithi_full_name')}|Nakshatra {start.get('nakshatra')}"
f" pada {start.get('nakshatra_pada')}|Yoga {start.get('yoga')}|Karana {start.get('karana')}"
f"|Hora {start.get('hora')}")
print(f" 止:Tithi {end.get('tithi_full_name')}|Nakshatra {end.get('nakshatra')}"
f" pada {end.get('nakshatra_pada')}|Yoga {end.get('yoga')}|Karana {end.get('karana')}"
f"|Hora {end.get('hora')}")
for field, change in (window.get("element_changes_within_window") or {}).items():
print(f" ⚠️ 时段内 {field} 变化:{change.get('from')} → {change.get('to')}(约 {change.get('at_local')})")
legacy = day.get("legacy_reference") or {}
if legacy:
print(f" 旧单点参考(不作最终判定):{legacy.get('quality')} {legacy.get('score')}")
gaps = result.get("gaps") or []
if gaps:
print()
print("缺口:")
for gap in gaps:
print(f" · {gap.get('field')}:{gap.get('reason')}")
print()
print("━" * 64)
print("口径:地点、时区、岁差、交点都由调用方显式给出;五要素在时段起止两端各取一次。")
print("━" * 64)
return 0
def cmd_muhurta(args, chart_data: Optional[Dict] = None) -> int:
"""
Muhurta 子命令主入口。
args 属性:
args.date: '2026-06-15'(查询日期,默认今天)
args.activity: 活动类型(可选,默认全部)
args.scan_days: 扫描天数(可选,>1 时扫描多天)
args.hour_from_sunrise: 从日出起算的小时(默认 6h)
args.hour_from_sunrise: 从日出起算的小时(只用于无地点的 legacy 概览)
args.lat / args.lon / args.tz / args.timezone_id: 地点与时区(给了就做地点精确择时)
args.ayanamsa / args.node_mode: 口径(给了地点时必填)
"""
from muhurta import muhurta_full_report, ACTIVITY_RULES
# 解析日期
if hasattr(args, 'date') and args.date:
try:
query_dt = datetime.strptime(args.date, '%Y-%m-%d')
@@ -116,22 +204,47 @@ def cmd_muhurta(args, chart_data: Optional[Dict] = None) -> int:
hour_from_sunrise = getattr(args, 'hour_from_sunrise', 6.0)
target_activity = getattr(args, 'activity', None)
scan_days = getattr(args, 'scan_days', 1)
ayanamsa_name = getattr(args, 'ayanamsa', None) or DEFAULT_AYANAMSA_NAME
lat = getattr(args, 'lat', None)
lon = getattr(args, 'lon', None)
tz = getattr(args, 'tz', None)
timezone_id = getattr(args, 'timezone_id', None)
ayanamsa = getattr(args, 'ayanamsa', None)
node_mode = getattr(args, 'node_mode', None)
ayanamsa_name = ayanamsa or DEFAULT_AYANAMSA_NAME
activities = list(ACTIVITY_RULES.keys())
if target_activity and target_activity not in activities:
print(f"[ERROR] 未知活动: {target_activity}。支持: {', '.join(activities)}")
return 1
# 占星师 Q9 / 第五轮 9:有地点才做地点精确择时;没地点只给近似概览,不给择时结论。
if lat is not None and lon is not None and (tz is not None or timezone_id):
return _print_location_exact(
query_dt.strftime('%Y-%m-%d'),
target_activity,
float(lat),
float(lon),
None if tz is None else float(tz),
timezone_id,
ayanamsa,
node_mode,
None,
5,
)
print("⚠️ 未提供地点:结果为近似值,不能作为择时结论")
print("=" * 64)
print("🕐 Muhurta 择时分析(印度吉凶历)")
print("🕐 Muhurta 五要素概览(无地点,近似)")
print("=" * 64)
dates_to_check = [query_dt + timedelta(days=i) for i in range(max(1, int(scan_days)))]
has_swe = True
for dt in dates_to_check:
# 原值:swe.julday(year, month, day, int(hour_from_sunrise + 6)),被当成 UT,等于 UT 12:00
# 新值:显式按 UT 12:00 取概览并写明是 UT,不再假装它是当地时间
# 原因:占星师 Q9 / 第五轮 9:择时须用地点、实际日出、时区
sun_lon, moon_lon, has_swe = _get_sun_moon_lons(
dt.year, dt.month, dt.day, int(hour_from_sunrise + 6), # rough solar hour
dt.year, dt.month, dt.day, 12.0,
ayanamsa_name=ayanamsa_name,
)
vara_idx = _weekday_to_vara(dt.weekday())
@@ -147,7 +260,7 @@ def cmd_muhurta(args, chart_data: Optional[Dict] = None) -> int:
)
precision_note = '' if has_swe else '(近似值,±2°精度)'
print(f"\n📅 {date_str} {precision_note}")
print(f"\n📅 {date_str} UT 12:00 概览 {precision_note}")
print(f" Sun: {sun_lon:.1f}° Moon: {moon_lon:.1f}°")
print()
@@ -181,9 +294,9 @@ def cmd_muhurta(args, chart_data: Optional[Dict] = None) -> int:
print()
print("━" * 64)
print("注意事项:")
print(" 1. 本分析基于恒星黄经(Lahiri Ayanamsa),为 Parashari 传统")
print(" 2. 精确 Muhurta 还需考虑出生盘 Lagna 与目标时间的相位关系")
print(" 3. 建议结合 Ascendant 力量、吉星位置做综合判断")
print(" 1. 本概览没有地点,日出日落与五要素都不是当地实际值,不能作为择时结论")
print(" 2. 需要地点精确择时,请加 --lat --lon --tz(或 --timezone-id)--ayanamsa --node-mode")
print(" 3. 精确 Muhurta 还需考虑出生盘 Lagna 与目标时间的相位关系")
if not has_swe:
print(" 4. ⚠️ 当前使用近似算法(swisseph 未安装),精度约 ±2°")
print("━" * 64)
+8
View File
@@ -10467,6 +10467,14 @@ class JyotishAPIHandler(BaseHTTPRequestHandler, JyotishComputeService):
elif path == '/api/muhurta':
result = self._compute_muhurta(body)
self._json(result)
elif path == '/api/muhurta_search':
# 占星师 Q9 / 第五轮 9:地点精确、口径显式的择时搜索;逻辑在 scripts/muhurta_search_api.py。
muhurta_search = _load_local_module('muhurta_search_api')
try:
result = muhurta_search.muhurta_search_api(body)
except muhurta_search.MuhurtaSearchInputError as exc:
raise BadRequest(str(exc)) from exc
self._json(result)
elif path == '/api/panchanga_range':
result = self._compute_panchanga_range(body)
self._json(result)
+14 -2
View File
@@ -18931,7 +18931,14 @@ def main():
help='指定活动类型(默认检查所有)')
p.add_argument('--scan-days', type=int, default=1, help='扫描天数(默认1天,最多30天)')
p.add_argument('--hour-from-sunrise', type=float, default=6.0,
help='从日出起算的小时数(默认6h约正午)')
help='从日出起算的小时数(默认6h约正午;给了地点时只是 legacy 对照,不作最终判定)')
# 占星师 Q9 / 第五轮 9:择时须用地点、实际日出、时区,口径显式传入。
p.add_argument('--lat', type=float, default=None, help='纬度(给了地点才做地点精确择时)')
p.add_argument('--lon', type=float, default=None, help='经度')
p.add_argument('--tz', type=float, default=None, help='时区小时数(例如 5.5)')
p.add_argument('--timezone-id', default=None, help='IANA 时区名(例如 Europe/London,夏令时自动处理)')
p.add_argument('--ayanamsa', default=None, help='岁差,例如 lahiri / raman(给了地点时必填)')
p.add_argument('--node-mode', default=None, choices=['mean', 'true'], help='交点口径(给了地点时必填)')
# 21. marriage-match(上游 bde5d615;只收盘快照)
p = sub.add_parser('marriage-match', help='婚恋合盘(两份盘快照)')
@@ -19048,7 +19055,12 @@ def main():
args = parser.parse_args()
if hasattr(args, 'tz') and args.tz is None:
# 原值:只要命令有 --tz 且没给就按 lat/lon 推断
# 新值:lat/lon 都没给时跳过推断(择时不给地点就是不给地点,不猜时区)
# 原因:占星师 Q9 / 第五轮 9:择时须用地点、实际日出、时区
if (hasattr(args, 'tz') and args.tz is None
and getattr(args, 'lat', None) is not None
and getattr(args, 'lon', None) is not None):
from timezone_utils import infer_timezone
from datetime import datetime
try:
+431
View File
@@ -0,0 +1,431 @@
"""地点精确、口径显式的择时搜索(占星师 2026-11-07 Q9 / 2026-11-08 第五轮第 9 条,BUG-1822)。
原做法的问题(`scripts/cmd_muhurta.py` 与 `muhurta.panchanga_range_report`):
1. 参考时刻被当成 UT 传给 `swe.julday`,等于 UT 12:00,不是当地时间;
2. 完全不用地点,日出日落退回 06:00 / 18:00,日月位置退回近似算法;
3. 五要素(Tithi、Nakshatra、Yoga、Karana)只在「日出 + 6 小时」这一个点取值,用这一个值给全天打分。
本模块只做三件事:
- 每天用 SwissEph 取**当地实际日出与日落**;取不到就整天 `blocked`,不退回 06:00 / 18:00,也不退回近似日月位置。
- 对每个候选时段,在该时段的**开始与结束两个实际时刻**各取一次五要素与 Hora,并标出时段内的变化与变化时刻。
- 规则、分数、阈值一律沿用 `scripts/muhurta.py` 现有函数(`muhurta_range_search` 给出的候选与评分原样放进 `legacy_reference`,标 `not_final`),本模块不重算分数、不改权重。
输出信封的字段与 #173 的按需计算合同保持一致(`schema`、`tool`、`calculation_status`、`profile`、
`requested`、`input_fingerprint`、`engine_version`、`raw`、`gaps`、`computed_at_utc`),
以便将来直接接进聊天的按需计算工具。
"""
from __future__ import annotations
import hashlib
import json
from datetime import datetime, timedelta, timezone as dt_timezone
from typing import Any, Callable, Dict, List, Optional, Tuple
try: # scripts/ 作为包
from scripts import muhurta as muhurta_module
from scripts.ayanamsa_utils import is_supported_ayanamsa_name, normalize_ayanamsa_name
except ModuleNotFoundError: # pragma: no cover - direct scripts/ execution
import muhurta as muhurta_module
from ayanamsa_utils import is_supported_ayanamsa_name, normalize_ayanamsa_name
SCHEMA = 'jyotish.muhurta_search.v1'
TOOL = 'muhurta_search'
MAX_DAYS = 63
NODE_MODES = ('mean', 'true')
# 五要素与 Hora 的取法:都用 muhurta.py 自己的计算函数,保证规则口径一致。
ELEMENT_FIELDS = ('tithi', 'nakshatra', 'yoga', 'karana', 'hora')
class MuhurtaSearchInputError(ValueError):
"""缺输入或输入不合法。消息里点名缺的字段,不补默认值。"""
def __init__(self, message: str, field: Optional[str] = None) -> None:
super().__init__(message)
self.field = field
def _require(body: Dict[str, Any], key: str) -> Any:
value = body.get(key)
if value is None or (isinstance(value, str) and not value.strip()):
raise MuhurtaSearchInputError(f'missing required field: {key}', key)
return value
def _number(body: Dict[str, Any], key: str, low: float, high: float) -> float:
value = _require(body, key)
try:
number = float(value)
except (TypeError, ValueError) as exc:
raise MuhurtaSearchInputError(f'{key} must be a number', key) from exc
if number < low or number > high:
raise MuhurtaSearchInputError(f'{key} must be between {low} and {high}', key)
return number
def _parse_date(value: Any, field: str) -> datetime:
try:
return datetime.strptime(str(value)[:10], '%Y-%m-%d')
except ValueError as exc:
raise MuhurtaSearchInputError(f'{field} must be YYYY-MM-DD', field) from exc
def _offset_hours_for(day: datetime, tz: Optional[float], timezone_id: Optional[str]) -> float:
"""当地 UTC 偏移小时数。给了 timezone_id 就按那一天换算,夏令时自动生效。"""
if timezone_id:
try:
from zoneinfo import ZoneInfo
except ImportError as exc: # pragma: no cover - Python < 3.9
raise MuhurtaSearchInputError('timezone_id needs Python 3.9+ zoneinfo', 'timezone_id') from exc
try:
zone = ZoneInfo(str(timezone_id))
except Exception as exc: # noqa: BLE001 - 未知时区名就是输入错误
raise MuhurtaSearchInputError(f'unknown timezone_id: {timezone_id}', 'timezone_id') from exc
local_noon = datetime(day.year, day.month, day.day, 12, 0, tzinfo=zone)
offset = local_noon.utcoffset()
if offset is None: # pragma: no cover - 只有固定偏移时区会给 None
raise MuhurtaSearchInputError(f'unknown timezone_id: {timezone_id}', 'timezone_id')
return offset.total_seconds() / 3600.0
return float(tz) # type: ignore[arg-type]
def _minutes(hhmm: Optional[str]) -> Optional[int]:
if not hhmm or ':' not in str(hhmm):
return None
hour, minute = str(hhmm).split(':', 1)
try:
return int(hour) * 60 + int(minute)
except ValueError:
return None
def _hhmm(minutes: float) -> str:
total = int(round(minutes)) % (24 * 60)
return f'{total // 60:02d}:{total % 60:02d}'
def _jd_for(day: datetime, local_hours: float, tz: float) -> float:
jd = muhurta_module._jd_for_local_time(day.year, day.month, day.day, local_hours, tz) # noqa: SLF001
if jd is None:
raise MuhurtaSearchInputError('cannot resolve this local time', 'date')
return jd
def _sun_moon(day: datetime, local_hours: float, tz: float, ayanamsa: str) -> Optional[Tuple[float, float]]:
positions = muhurta_module._swisseph_sun_moon_lon( # noqa: SLF001
_jd_for(day, local_hours, tz), ayanamsa_name=ayanamsa,
)
if not positions:
return None
return float(positions[0]), float(positions[1])
def _elements_at(
day: datetime,
local_hours: float,
tz: float,
ayanamsa: str,
sunrise_local_hours: Optional[float],
) -> Optional[Dict[str, Any]]:
"""该时刻的五要素与 Hora。取不到 SwissEph 位置就返回 None,不做近似。"""
positions = _sun_moon(day, local_hours, tz, ayanamsa)
if positions is None:
return None
sun_lon, moon_lon = positions
weekday = (day.weekday() + 1) % 7
tithi = muhurta_module.calc_tithi(sun_lon, moon_lon)
nakshatra = muhurta_module.calc_nakshatra_from_lon(moon_lon)
yoga = muhurta_module.calc_yoga(sun_lon, moon_lon)
karana = muhurta_module.calc_karana(sun_lon, moon_lon)
hour_from_sunrise = local_hours - sunrise_local_hours if sunrise_local_hours is not None else 0.0
hora = muhurta_module.calc_hora(weekday, hour_from_sunrise)
return {
'local_time': _hhmm(local_hours * 60),
'utc_time': _hhmm((local_hours - tz) * 60),
'tithi': (tithi or {}).get('name'),
'tithi_full_name': (tithi or {}).get('full_name'),
'paksha': (tithi or {}).get('paksha'),
'tithi_number': (tithi or {}).get('tithi_num'),
'nakshatra': (nakshatra or {}).get('nakshatra'),
'nakshatra_pada': (nakshatra or {}).get('pada'),
'yoga': (yoga or {}).get('yoga'),
'karana': (karana or {}).get('karana'),
'hora': (hora or {}).get('hora_lord'),
'hora_quality': (hora or {}).get('quality'),
}
def _element_changes(
start: Dict[str, Any],
end: Dict[str, Any],
bisect: Callable[[float], Optional[Dict[str, Any]]],
start_hours: float,
end_hours: float,
) -> Dict[str, Any]:
"""时段内发生变化的要素,以及变化发生的当地时刻(二分到分钟)。"""
changes: Dict[str, Any] = {}
for field in ELEMENT_FIELDS:
before = start.get(field)
after = end.get(field)
if before is None or after is None or before == after:
continue
low, high = start_hours, end_hours
changed_at = None
for _ in range(12):
middle = (low + high) / 2.0
sample = bisect(middle)
if sample is None:
break
if sample.get(field) == before:
low = middle
else:
high = middle
changed_at = middle
changes[field] = {
'from': before,
'to': after,
'at_local': _hhmm((changed_at if changed_at is not None else end_hours) * 60),
}
return changes
def _fingerprint(payload: Dict[str, Any]) -> str:
canonical = json.dumps(payload, sort_keys=True, ensure_ascii=False, separators=(',', ':'))
return hashlib.sha256(canonical.encode('utf-8')).hexdigest()
def _envelope(
*,
requested: Dict[str, Any],
profile: Dict[str, Any],
status: str,
raw: Dict[str, Any],
gaps: List[Dict[str, str]],
fingerprint: str,
) -> Dict[str, Any]:
return {
'schema': SCHEMA,
'tool': TOOL,
'calculation_status': status,
'profile': profile,
'requested': requested,
'input_fingerprint': fingerprint,
'engine_version': 'muhurta.py',
'raw': raw,
'gaps': gaps,
'computed_at_utc': datetime.now(dt_timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ'),
}
def _validate(body: Dict[str, Any]) -> Dict[str, Any]:
if not isinstance(body, dict):
raise MuhurtaSearchInputError('body must be an object')
activity = str(_require(body, 'activity'))
if activity not in muhurta_module.ACTIVITY_RULES:
raise MuhurtaSearchInputError(f'unknown activity: {activity}', 'activity')
start_date = _parse_date(_require(body, 'start_date'), 'start_date')
end_date = _parse_date(_require(body, 'end_date'), 'end_date')
if end_date < start_date:
raise MuhurtaSearchInputError('end_date must be on or after start_date', 'end_date')
days = (end_date - start_date).days + 1
if days > MAX_DAYS:
raise MuhurtaSearchInputError(f'date range must be at most {MAX_DAYS} days', 'end_date')
lat = _number(body, 'lat', -90.0, 90.0)
lon = _number(body, 'lon', -180.0, 180.0)
tz_value = body.get('tz')
timezone_id = body.get('timezone_id')
if tz_value is None and not timezone_id:
raise MuhurtaSearchInputError('missing required field: tz or timezone_id', 'tz')
tz = None if tz_value is None else _number(body, 'tz', -12.0, 14.0)
ayanamsa = str(_require(body, 'ayanamsa'))
if not is_supported_ayanamsa_name(ayanamsa):
raise MuhurtaSearchInputError(f'unknown ayanamsa: {ayanamsa}', 'ayanamsa')
node_mode = str(_require(body, 'node_mode')).lower()
if node_mode not in NODE_MODES:
raise MuhurtaSearchInputError('node_mode must be mean or true', 'node_mode')
time_window = body.get('time_window')
if time_window is not None and not isinstance(time_window, dict):
raise MuhurtaSearchInputError('time_window must be an object with start and end', 'time_window')
limit = body.get('limit')
return {
'activity': activity,
'start_date': start_date,
'end_date': end_date,
'lat': lat,
'lon': lon,
'tz': tz,
'timezone_id': str(timezone_id) if timezone_id else None,
'ayanamsa': normalize_ayanamsa_name(ayanamsa),
'node_mode': node_mode,
'time_window': time_window,
'limit': max(1, min(int(limit or 5), 20)),
}
def muhurta_search_api(body: Dict[str, Any]) -> Dict[str, Any]:
"""按地点、实际日出、时区搜索择时窗口;每个时段在起止两端核对五要素。"""
inputs = _validate(body)
requested = {
'activity': inputs['activity'],
'start_date': inputs['start_date'].strftime('%Y-%m-%d'),
'end_date': inputs['end_date'].strftime('%Y-%m-%d'),
'lat': inputs['lat'],
'lon': inputs['lon'],
'tz': inputs['tz'],
'timezone_id': inputs['timezone_id'],
'ayanamsa': inputs['ayanamsa'],
'node_mode': inputs['node_mode'],
'time_window': inputs['time_window'],
'limit': inputs['limit'],
}
fingerprint = _fingerprint({'requested': requested})
first_day = inputs['start_date']
tz = _offset_hours_for(first_day, inputs['tz'], inputs['timezone_id'])
profile = {
'ayanamsa': inputs['ayanamsa'],
'node_mode': inputs['node_mode'],
'lat': inputs['lat'],
'lon': inputs['lon'],
'tz': tz,
'timezone_id': inputs['timezone_id'],
'sunrise_policy': 'swisseph_rise_trans',
'time_source': 'actual_local_sunrise',
}
days: List[Dict[str, Any]] = []
gaps: List[Dict[str, str]] = []
candidate_index: Dict[str, Dict[str, Any]] = {}
if any(inputs['timezone_id'] for _ in (0,)):
# timezone_id 的偏移可能逐日不同(夏令时),所以先按天算,再统一放进每天的行里。
pass
try:
legacy = muhurta_module.muhurta_range_search(
requested['start_date'],
requested['end_date'],
activity=inputs['activity'],
limit=inputs['limit'],
lat=inputs['lat'],
lon=inputs['lon'],
tz=tz,
ayanamsa_name=inputs['ayanamsa'],
)
except Exception as exc: # noqa: BLE001 - 搜索失败也要返回信封,且不编任何时刻
gaps.append({'field': 'search', 'reason': f'search_failed:{type(exc).__name__}'})
return _envelope(
requested=requested, profile=profile, status='blocked', raw={}, gaps=gaps, fingerprint=fingerprint,
)
for item in legacy.get('best_windows') or []:
date_key = str(item.get('date') or '')
if date_key:
candidate_index[date_key] = item
current = inputs['start_date']
while current <= inputs['end_date']:
date_str = current.strftime('%Y-%m-%d')
day_tz = _offset_hours_for(current, inputs['tz'], inputs['timezone_id'])
solar = muhurta_module._swisseph_solar_times_local( # noqa: SLF001
current.year, current.month, current.day, inputs['lat'], inputs['lon'], day_tz,
)
if not solar:
gaps.append({'field': f'sunrise:{date_str}', 'reason': 'sunrise_unavailable'})
days.append({
'date': date_str,
'calculation_status': 'blocked',
'reason': 'sunrise_unavailable',
'windows': [],
})
current += timedelta(days=1)
continue
sunrise_local_hours = solar.get('sunrise_local_hours')
sunset_local_hours = solar.get('sunset_local_hours')
candidate = candidate_index.get(date_str) or {}
windows = []
for window in candidate.get('recommended_windows') or []:
start_minutes = _minutes(window.get('start'))
end_minutes = _minutes(window.get('end'))
if start_minutes is None or end_minutes is None:
continue
start_sample = _elements_at(
current, start_minutes / 60.0, day_tz, inputs['ayanamsa'], sunrise_local_hours,
)
end_sample = _elements_at(
current, end_minutes / 60.0, day_tz, inputs['ayanamsa'], sunrise_local_hours,
)
if start_sample is None or end_sample is None:
gaps.append({'field': f'elements:{date_str}:{window.get("start")}', 'reason': 'ephemeris_unavailable'})
windows.append({
'type': window.get('type'),
'name': window.get('name'),
'start_local': window.get('start'),
'end_local': window.get('end'),
'calculation_status': 'blocked',
'reason': 'ephemeris_unavailable',
})
continue
changes = _element_changes(
start_sample,
end_sample,
lambda hours: _elements_at(
current, hours, day_tz, inputs['ayanamsa'], sunrise_local_hours,
),
start_minutes / 60.0,
end_minutes / 60.0,
)
windows.append({
'type': window.get('type'),
'name': window.get('name'),
'quality': window.get('quality'),
'start_local': window.get('start'),
'end_local': window.get('end'),
'start_utc': start_sample['utc_time'],
'end_utc': end_sample['utc_time'],
'elements_at_start': start_sample,
'elements_at_end': end_sample,
'element_changes_within_window': changes,
'calculation_status': 'executed',
})
days.append({
'date': date_str,
'calculation_status': 'executed',
'tz_offset_hours': day_tz,
'sunrise_local': solar.get('sunrise'),
'sunset_local': solar.get('sunset'),
'sunrise_utc': _hhmm(float(solar.get('sunrise_utc_hours') or 0.0) * 60),
'sunset_utc': _hhmm(float(solar.get('sunset_utc_hours') or 0.0) * 60),
'sunrise_policy': solar.get('policy'),
'legacy_reference': {
'not_final': True,
'note': '现有 muhurta_range_search 的单点参考时刻评分,只作对照,不是最终择时结论',
'score': candidate.get('score'),
'quality': candidate.get('quality'),
'activity_verdict': candidate.get('activity_verdict'),
},
'windows': windows,
})
current += timedelta(days=1)
executed = [day for day in days if day.get('calculation_status') == 'executed']
if not executed:
return _envelope(
requested=requested,
profile=profile,
status='blocked',
raw={'days': days} if days else {},
gaps=gaps or [{'field': 'search', 'reason': 'no_day_executed'}],
fingerprint=fingerprint,
)
status = 'executed' if len(executed) == len(days) and not gaps else 'partial'
return _envelope(
requested=requested,
profile=profile,
status=status,
raw={'days': days, 'activity': inputs['activity']},
gaps=gaps,
fingerprint=fingerprint,
)
+220
View File
@@ -0,0 +1,220 @@
"""地点精确、口径显式的择时搜索(占星师 2026-11-07 Q9 / 2026-11-08 第五轮第 9 条,BUG-1822)。
原做法:参考时刻被当成 UT 12:00、完全不用地点、五要素只取「日出 + 6 小时」一个点。
新做法:按地点取实际日出日落,每个候选时段在起止两端各取一次五要素,取不到就阻塞。
只用公开地点与日期,不含任何个人资料。
"""
from __future__ import annotations
import sys
from pathlib import Path
import pytest
ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
from scripts import muhurta as muhurta_module # noqa: E402
from scripts.muhurta_search_api import ( # noqa: E402
MAX_DAYS,
MuhurtaSearchInputError,
muhurta_search_api,
)
DELHI = {'lat': 28.6139, 'lon': 77.2090, 'tz': 5.5}
LONDON = {'lat': 51.5074, 'lon': -0.1278, 'timezone_id': 'Europe/London'}
BASE = {'activity': 'business', 'start_date': '2026-06-15', 'end_date': '2026-06-15', 'ayanamsa': 'lahiri', 'node_mode': 'mean'}
def _body(**overrides):
body = dict(BASE)
body.update(DELHI)
body.update(overrides)
return body
def test_each_required_field_is_named_when_missing() -> None:
for field in ('activity', 'start_date', 'end_date', 'lat', 'lon', 'ayanamsa', 'node_mode'):
body = _body()
body.pop(field)
with pytest.raises(MuhurtaSearchInputError) as excinfo:
muhurta_search_api(body)
assert field in str(excinfo.value) or excinfo.value.field == field, field
body = _body()
body.pop('tz')
with pytest.raises(MuhurtaSearchInputError) as excinfo:
muhurta_search_api(body)
assert excinfo.value.field == 'tz'
def test_unknown_activity_is_refused_not_silently_switched() -> None:
with pytest.raises(MuhurtaSearchInputError) as excinfo:
muhurta_search_api(_body(activity='not_a_real_activity'))
assert 'not_a_real_activity' in str(excinfo.value)
def test_range_longer_than_the_cap_is_refused() -> None:
with pytest.raises(MuhurtaSearchInputError):
muhurta_search_api(_body(start_date='2026-01-01', end_date='2026-06-15'))
assert MAX_DAYS == 63
def test_sunrise_matches_the_swisseph_helper_and_differs_between_cities() -> None:
delhi = muhurta_search_api(_body())
london = muhurta_search_api(_body(**LONDON))
delhi_day = delhi['raw']['days'][0]
london_day = london['raw']['days'][0]
expected_delhi = muhurta_module._swisseph_solar_times_local(2026, 6, 15, DELHI['lat'], DELHI['lon'], DELHI['tz'])
assert expected_delhi, 'SwissEph sunrise must be available in this environment'
assert delhi_day['sunrise_local'] == expected_delhi['sunrise']
assert delhi_day['sunrise_policy'] == 'swisseph_rise_trans'
assert delhi_day['sunrise_local'] != london_day['sunrise_local']
def test_timezone_id_follows_daylight_saving() -> None:
london = muhurta_search_api(_body(**LONDON))
assert london['profile']['tz'] == 1.0, 'Europe/London on 2026-06-15 is BST (+1)'
winter = muhurta_search_api(_body(**LONDON, start_date='2026-01-15', end_date='2026-01-15'))
assert winter['profile']['tz'] == 0.0
def test_every_window_carries_elements_at_both_ends() -> None:
result = muhurta_search_api(_body())
windows = [window for day in result['raw']['days'] for window in day['windows']]
assert windows, 'a business day in Delhi must have candidate windows'
for window in windows:
start = window['elements_at_start']
end = window['elements_at_end']
for field in ('tithi', 'nakshatra', 'yoga', 'karana', 'hora'):
assert start[field] is not None, field
assert end[field] is not None, field
assert start['local_time'] == window['start_local']
assert end['local_time'] == window['end_local']
assert window['start_utc'] != window['start_local']
def test_a_change_inside_a_window_is_reported_with_its_time(monkeypatch) -> None:
"""把月亮推快,让它在一个时段内跨过星宿边界;断言变化与变化时刻都被标出。"""
original = muhurta_module._swisseph_sun_moon_lon
def fast_moon(jd, ayanamsa_name=None):
positions = original(jd, ayanamsa_name=ayanamsa_name)
if not positions:
return positions
sun, moon = positions
return sun, (moon + (jd * 97.0) % 360.0) % 360.0
monkeypatch.setattr(muhurta_module, '_swisseph_sun_moon_lon', fast_moon)
result = muhurta_search_api(_body())
windows = [window for day in result['raw']['days'] for window in day['windows']]
changed = [window for window in windows if 'nakshatra' in (window.get('element_changes_within_window') or {})]
assert changed, [window.get('element_changes_within_window') for window in windows]
change = changed[0]['element_changes_within_window']['nakshatra']
assert change['from'] != change['to']
assert change['at_local']
def test_no_sunrise_blocks_the_day_without_falling_back_to_0600(monkeypatch) -> None:
monkeypatch.setattr(muhurta_module, '_swisseph_solar_times_local', lambda *args, **kwargs: None)
result = muhurta_search_api(_body())
assert result['calculation_status'] == 'blocked'
day = result['raw']['days'][0]
assert day['calculation_status'] == 'blocked'
assert day['reason'] == 'sunrise_unavailable'
assert day['windows'] == []
serialized = repr(result)
assert '06:00' not in serialized
assert '18:00' not in serialized
def test_fingerprint_is_stable_and_reacts_to_the_profile() -> None:
first = muhurta_search_api(_body())
second = muhurta_search_api(_body())
assert first['input_fingerprint'] == second['input_fingerprint']
other = muhurta_search_api(_body(ayanamsa='raman'))
assert other['input_fingerprint'] != first['input_fingerprint']
assert other['profile']['ayanamsa'] == 'raman'
delhi_start = first['raw']['days'][0]['windows'][0]['elements_at_start']['nakshatra']
raman_start = other['raw']['days'][0]['windows'][0]['elements_at_start']['nakshatra']
# 五要素两套岁差各自算出来(可能不同,这里只要求都算出来了)
assert delhi_start and raman_start
def test_legacy_single_point_score_is_marked_not_final() -> None:
result = muhurta_search_api(_body())
legacy = result['raw']['days'][0]['legacy_reference']
assert legacy['not_final'] is True
def test_route_is_registered_thinly() -> None:
source = (ROOT / 'scripts' / 'jyotish_api_server.py').read_text(encoding='utf-8')
assert "elif path == '/api/muhurta_search':" in source
assert "_load_local_module('muhurta_search_api')" in source
assert 'muhurta_search.muhurta_search_api(body)' in source
def test_cli_with_location_prints_real_sunrise_and_no_legacy_single_point(monkeypatch, capsys) -> None:
from scripts.cmd_muhurta import cmd_muhurta
class Args:
date = '2026-06-15'
activity = 'business'
scan_days = 1
hour_from_sunrise = 6.0
lat = DELHI['lat']
lon = DELHI['lon']
tz = DELHI['tz']
timezone_id = None
ayanamsa = 'lahiri'
node_mode = 'mean'
assert cmd_muhurta(Args()) == 0
printed = capsys.readouterr().out
api = muhurta_search_api(_body())
assert api['raw']['days'][0]['sunrise_local'] in printed
assert '日出' in printed
assert '不提供地点' not in printed
def test_cli_without_location_says_it_is_only_an_approximation(capsys) -> None:
from scripts.cmd_muhurta import cmd_muhurta
class Args:
date = '2026-06-15'
activity = 'business'
scan_days = 1
hour_from_sunrise = 6.0
lat = None
lon = None
tz = None
timezone_id = None
ayanamsa = None
node_mode = None
assert cmd_muhurta(Args()) == 0
printed = capsys.readouterr().out
first_line = printed.splitlines()[0]
assert '未提供地点' in first_line
assert '不能作为择时结论' in first_line
assert '推荐时段' not in printed
def test_cli_with_location_requires_an_explicit_profile(capsys) -> None:
from scripts.cmd_muhurta import cmd_muhurta
class Args:
date = '2026-06-15'
activity = 'business'
scan_days = 1
hour_from_sunrise = 6.0
lat = DELHI['lat']
lon = DELHI['lon']
tz = DELHI['tz']
timezone_id = None
ayanamsa = None
node_mode = None
assert cmd_muhurta(Args()) == 1
assert '--ayanamsa' in capsys.readouterr().out