243 lines
13 KiB
Python
243 lines
13 KiB
Python
"""Questions and dated evidence for the multi-round chart-choice flow."""
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
from datetime import datetime, timedelta, timezone
|
||
from typing import Any
|
||
|
||
from scripts.chart_choice import astrologer
|
||
from scripts.chart_choice.boundaries import ChartChoiceInputError, ChartRequest
|
||
from chart_dasha.birth_moment import YEAR_DAYS, split_period
|
||
|
||
TOPICS = {
|
||
"D9": ("感情", "开始恋爱、结婚或分开", ("relationship", "marriage", "divorce")),
|
||
"D10": ("工作", "入职、升职、离职、换工作或创业", ("career", "job_change", "business")),
|
||
"D2": ("收入和支出", "收入明显增加、意外支出或较大的损失", ("finance_pos", "finance_neg")),
|
||
}
|
||
CHOICES = ("before", "after", "both", "neither", "unsure")
|
||
|
||
|
||
def _today(req: ChartRequest) -> datetime:
|
||
return (datetime.now(timezone.utc) + timedelta(hours=req.tz)).replace(
|
||
tzinfo=None, hour=0, minute=0, second=0, microsecond=0,
|
||
)
|
||
|
||
|
||
def _boundaries(row: dict, start: datetime, end: datetime) -> set[datetime]:
|
||
points = set()
|
||
for major in row["timeline"]:
|
||
if major["end"] <= start or major["start"] >= end:
|
||
continue
|
||
for minor in split_period(major):
|
||
if minor["end"] > start and minor["start"] < end:
|
||
points.update(p["start"] for p in split_period(minor))
|
||
majors, minors = row["narayana"]
|
||
periods = list(majors or [])
|
||
if minors:
|
||
for children in minors.values():
|
||
periods.extend(children or [])
|
||
for period in periods:
|
||
points.add(row["at"] + timedelta(days=float(period["start_age"]) * YEAR_DAYS))
|
||
points.add(row["at"] + timedelta(days=float(period["end_age"]) * YEAR_DAYS))
|
||
return {at for at in points if start < at < end}
|
||
|
||
|
||
def _age(at: datetime, birth: datetime) -> int:
|
||
return at.year - birth.year - ((at.month, at.day) < (birth.month, birth.day))
|
||
|
||
|
||
def _age_label(ages: tuple[int, int]) -> str:
|
||
return f"{ages[0]} 岁" if ages[0] == ages[1] else f"{ages[0]}–{ages[1]} 岁"
|
||
|
||
|
||
def _date_span(at: datetime, precision: str) -> tuple[datetime, datetime]:
|
||
if precision == "year":
|
||
return at.replace(month=1, day=1), at.replace(month=12, day=31)
|
||
start = at.replace(day=1)
|
||
next_month = (start.replace(day=28) + timedelta(days=4)).replace(day=1)
|
||
return start, next_month - timedelta(days=1)
|
||
|
||
|
||
def _follow_up_options(chart: str, categories: tuple[str, ...]) -> list[dict]:
|
||
if chart == "D2":
|
||
return [
|
||
{"id": "income_increase", "value": "finance_pos", "text": "收入明显增加"},
|
||
{"id": "unexpected_expense", "value": "finance_neg", "text": "一笔较大的意外支出"},
|
||
{"id": "financial_loss", "value": "finance_neg", "text": "一笔较大的损失"},
|
||
]
|
||
return [{"id": key, "value": key, "text": astrologer.EVENT_CATEGORIES[key]["label"]} for key in categories]
|
||
|
||
|
||
def _event_question(chart: str, rows: tuple[dict, dict], points: list[datetime], birth: datetime) -> dict | None:
|
||
label, examples, categories = TOPICS[chart]
|
||
windows: list[dict[tuple[int, int], tuple[float, float]]] = [{}, {}]
|
||
for category in categories:
|
||
event = {"category": category}
|
||
varga = [astrologer._varga_score(event, row) for row in rows]
|
||
for lo, hi in zip(points, points[1:]):
|
||
event["at"] = lo + (hi - lo) / 2
|
||
(before, left), (after, right) = (astrologer._dasha_score(event, row) for row in rows)
|
||
if left.get("unavailable") != right.get("unavailable"):
|
||
continue
|
||
scores = (before, after)
|
||
ages = (_age(lo, birth), _age(hi - timedelta(microseconds=1), birth))
|
||
strong = [score >= 4 or (score > 0 and v > 0) for score, v in zip(scores, varga)]
|
||
for i, score in enumerate(scores):
|
||
difference = score - scores[1 - i] + varga[i] - varga[1 - i]
|
||
if strong[i] and scores[1 - i] == 0 and difference > 0:
|
||
rank = (difference, (hi - lo).total_seconds())
|
||
windows[i][ages] = max(windows[i].get(ages, rank), rank)
|
||
pairs = (
|
||
(left, right) for left in windows[0] for right in windows[1]
|
||
if left[1] < right[0] or right[1] < left[0]
|
||
)
|
||
pair = max(pairs, key=lambda p: (windows[0][p[0]][0] + windows[1][p[1]][0],
|
||
windows[0][p[0]][1] + windows[1][p[1]][1]), default=None)
|
||
if pair is None:
|
||
return None
|
||
before, after = map(_age_label, pair)
|
||
return {
|
||
"id": chart, "chart": chart, "before_ages": pair[0], "after_ages": pair[1],
|
||
"prompt": f"{label}上的变化，线前候选对应 {before}，线后候选对应 {after}。你在哪一段有过这些变化？比如{examples}。",
|
||
"options": [{"value": key, "text": text} for key, text in zip(
|
||
CHOICES, (f"{before}有", f"{after}有", "两段都有", "两段都没有", "记不清"),
|
||
)],
|
||
"follow_up": {
|
||
"prompt": "大概是哪一年？月份记得的话，也可以填。那次发生了什么？",
|
||
"categories": _follow_up_options(chart, categories),
|
||
},
|
||
}
|
||
|
||
|
||
def questions(req: ChartRequest, charts: list[str], before: datetime, after: datetime) -> dict:
|
||
eligible = [key for key in TOPICS if key in charts]
|
||
if not eligible and "D1" not in charts:
|
||
return {"event_questions": []}
|
||
rows = (astrologer._minute_row(req, before), astrologer._minute_row(req, after))
|
||
birth = min(before, after)
|
||
result: dict[str, Any] = {"event_questions": []}
|
||
if eligible:
|
||
start = birth.replace(year=birth.year + 18, day=28 if (birth.month, birth.day) == (2, 29) else birth.day)
|
||
end = min(_today(req) + timedelta(days=1), *(row["timeline"][-1]["end"] for row in rows))
|
||
if start < end:
|
||
points = sorted({start, end} | _boundaries(rows[0], start, end) | _boundaries(rows[1], start, end))
|
||
for chart in eligible:
|
||
if question := _event_question(chart, rows, points, birth):
|
||
result["event_questions"].append(question)
|
||
if len(result["event_questions"]) == 2:
|
||
break
|
||
if "D1" in charts:
|
||
options = astrologer.appearance_options(list(rows))
|
||
if len(options) == 2:
|
||
result["appearance_question"] = {
|
||
"prompt": "哪一组外表描述更接近你？",
|
||
"options": [
|
||
{"value": side, "text": item["features"]} for side, item in zip(("before", "after"), options)
|
||
] + [{"value": "unsure", "text": "说不清"}],
|
||
}
|
||
result["birthmark_question"] = {
|
||
"prompt": "你有胎记吗？有的话，可以选它所在的位置。",
|
||
"multiple": True,
|
||
"options": [{"value": part, "text": spec["label"]} for part, spec in astrologer.BIRTHMARK_PARTS.items()]
|
||
+ [{"value": "none", "text": "没有"}],
|
||
}
|
||
return result
|
||
|
||
|
||
def _event_answers(packet: dict, raw: Any, birth: datetime, today: datetime) -> tuple[dict, list[dict], list[dict]]:
|
||
if not isinstance(raw, dict):
|
||
raise ChartChoiceInputError("event_answers must be an object")
|
||
normalized, events, sources = {}, [], []
|
||
asked = {q["id"]: q for q in packet["event_questions"]}
|
||
for question_id, response in raw.items():
|
||
if question_id not in asked or not isinstance(response, dict) or response.get("choice") not in CHOICES:
|
||
raise ChartChoiceInputError("event answer must match an offered question and option")
|
||
choice = response["choice"]
|
||
sides = ("before", "after") if choice == "both" else (choice,) if choice in CHOICES[:2] else ()
|
||
dates = response.get("dates", {})
|
||
if not isinstance(dates, dict) or dates.keys() != set(sides):
|
||
raise ChartChoiceInputError("each selected event window needs a date")
|
||
categories = TOPICS[question_id][2]
|
||
recorded = {}
|
||
for side, event in dates.items():
|
||
if (not isinstance(event, dict) or event.get("category") not in categories
|
||
or not isinstance(event.get("date"), str) or not re.fullmatch(r"\d{4}(?:-\d{2})?", event["date"])):
|
||
raise ChartChoiceInputError("event follow-up needs an offered category and a year or month")
|
||
item = {key: event[key] for key in ("date", "category")}
|
||
parsed = astrologer.parse_events([item], birth)[0]
|
||
if parsed["at"].year > today.year:
|
||
raise ChartChoiceInputError("event date is outside the selected age window")
|
||
lower, upper = _date_span(parsed["at"], parsed["precision"])
|
||
upper = min(today, upper)
|
||
ages = asked[question_id][f"{side}_ages"]
|
||
if lower > upper or _age(upper, birth) < ages[0] or _age(lower, birth) > ages[1]:
|
||
raise ChartChoiceInputError("event date is outside the selected age window")
|
||
recorded[side] = item
|
||
events.append(item)
|
||
normalized[question_id] = {"choice": choice, "dates": recorded}
|
||
sources.append({"kind": "event", "id": question_id, "choice": choice,
|
||
**{side: (1 if choice == "both" else 2) if side in sides else 0 for side in CHOICES[:2]}})
|
||
return normalized, events, sources
|
||
|
||
|
||
def evidence(req: ChartRequest, packet: dict, body: dict, birth: datetime) -> tuple[dict, list[dict], list[dict]]:
|
||
result: dict[str, Any] = {}
|
||
answers, events, sources = _event_answers(packet, body.get("event_answers", {}), birth, _today(req))
|
||
if "event_answers" in body:
|
||
result["event_answers"] = answers
|
||
if "appearance" in body:
|
||
choice = body["appearance"]
|
||
if "appearance_question" not in packet or choice not in (*CHOICES[:2], "unsure"):
|
||
raise ChartChoiceInputError("appearance answer requires a D1 question")
|
||
result["appearance"] = choice
|
||
sources.append({"kind": "appearance", "choice": choice, "before": int(choice == "before"), "after": int(choice == "after")})
|
||
if "birthmarks" in body:
|
||
marks = body["birthmarks"]
|
||
if ("birthmark_question" not in packet or not isinstance(marks, list)
|
||
or any(not isinstance(part, str) or part not in (*astrologer.BIRTHMARK_PARTS, "none") for part in marks)
|
||
or ("none" in marks and marks != ["none"])):
|
||
raise ChartChoiceInputError("birthmark answers require D1 and valid positions")
|
||
result["birthmarks"] = list(dict.fromkeys(marks))
|
||
if marks and marks != ["none"]:
|
||
points = {}
|
||
for side in CHOICES[:2]:
|
||
row = astrologer._minute_row(req, datetime.fromisoformat(packet["sides"][side]["saved_at"]))
|
||
points[side] = int(any(astrologer._birthmark_supported(part, row) for part in marks))
|
||
sources.append({"kind": "birthmark", **points})
|
||
elif marks:
|
||
sources.append({"kind": "birthmark", "choice": "none", "before": 0, "after": 0})
|
||
return result, events, sources
|
||
|
||
|
||
def summarize(
|
||
req: ChartRequest, segment: tuple[datetime, datetime], reported: datetime, raw: list[dict], *,
|
||
representative: datetime | None = None,
|
||
) -> dict:
|
||
events, _ = astrologer._independent_events(astrologer.parse_events(raw, reported))
|
||
cov = astrologer.coverage(events)
|
||
base = astrologer._minute_row(req, reported) if events else None
|
||
start, end = segment
|
||
first = start.replace(second=0, microsecond=0)
|
||
if first < start:
|
||
first += timedelta(minutes=1)
|
||
rows = astrologer.scan(req, first, end) if base and first <= end else []
|
||
if base and not rows:
|
||
rows = [astrologer._minute_row(req, start + (end - start) // 2)]
|
||
scored = {row["at"]: astrologer.score_minute(events, row, base) for row in rows}
|
||
ranked = sorted(scored.values(), key=lambda row: -row["total"])
|
||
exact_tie = len(ranked) > 1 and ranked[0]["total"] == ranked[1]["total"]
|
||
confidence = ("tendency_only" if len(events) < 3
|
||
else astrologer.engine_confidence(ranked, cov["level"] == "thin", exact_tie)["key"])
|
||
result = {"event_count": len(events), "matched": ranked[0]["matched"] if ranked else 0,
|
||
"confidence": confidence, "coverage": cov, "caveat": astrologer.CAVEAT}
|
||
if representative is not None:
|
||
selected = scored.get(representative)
|
||
if selected is None and base is not None:
|
||
selected = astrologer.score_minute(events, astrologer._minute_row(req, representative), base)
|
||
result["matched"] = selected["matched"] if selected else 0
|
||
result["events_detail"] = [
|
||
{"date": event["date"], "category": event["category"], "matched": score["matched"]}
|
||
for event, score in zip(events, selected["events"], strict=True)
|
||
] if selected else []
|
||
return result
|