Spaces:
Running
Running
| """Convert a FixSense answer into the recursive content tree the app team asked for. | |
| Every node has exactly these keys (empty values follow the app team's sketch): | |
| { | |
| "text": "...", # plain text, no markdown, no inline [citation ids] | |
| "image": "", # one image URL | |
| "images": [], # image URLs | |
| "video": "", # one video URL (may end in #t=start,end -> play only that part) | |
| "videos": [], # video URLs | |
| "obj": "", # one nested node (or "" when there is none) | |
| "objs": [] # nested nodes, in reading order | |
| } | |
| Nodes may carry a few extra keys (``type``, ``video_start``, ``video_end``, ``page``, ``source_url`` ...) | |
| that a client can use or ignore. The root node additionally has ``meta``. | |
| All media URLs are absolute and signed (they work without logging in, and expire); ``url_fn`` does that. | |
| """ | |
| from __future__ import annotations | |
| import re | |
| from typing import Callable | |
| _CITE_RE = re.compile(r"\s*\[[^\[\]]*::[^\[\]]*\]") | |
| _T = { | |
| "en": { | |
| "safety": "Safety", "parts": "Parts and materials", "tools": "Tools you need", "steps": "Steps", | |
| "escalate": "When to stop and go to a shop", "related": "You might also ask", "causes": "Most likely causes", | |
| "check": "How to check", "sources": "Sources", "manual": "Owner's manual", "page": "p.", "watch": "Watch", | |
| "likelihood": "likelihood", "tools_parts": "Tools and parts", "options": "Options", "details": "Details", | |
| }, | |
| "zh": { | |
| "safety": "安全提示", "parts": "零件与耗材", "tools": "所需工具", "steps": "操作步骤", | |
| "escalate": "什么时候该停手送修", "related": "你可能还想问", "causes": "最可能的原因", | |
| "check": "怎么检查", "sources": "来源", "manual": "车主手册", "page": "第", "watch": "看视频", | |
| "likelihood": "可能性", "tools_parts": "工具与零件", "options": "选项", "details": "补充说明", | |
| }, | |
| } | |
| def node(text: str = "", *, image: str = "", images=None, video: str = "", videos=None, obj="", objs=None, **extra) -> dict: | |
| n = {"text": text, "image": image, "images": list(images or []), "video": video, | |
| "videos": list(videos or []), "obj": obj, "objs": list(objs or [])} | |
| n.update({k: v for k, v in extra.items() if v not in (None, "")}) | |
| return n | |
| def _clean(text: str | None) -> str: | |
| return re.sub(r"[ \t]+", " ", _CITE_RE.sub("", text or "")).strip() | |
| def _secs(v) -> float | None: | |
| if v is None or v == "": | |
| return None | |
| try: | |
| return float(v) | |
| except (TypeError, ValueError): | |
| pass | |
| try: | |
| parts = [float(x) for x in str(v).split(":")] | |
| except ValueError: | |
| return None | |
| out = 0.0 | |
| for p in parts: | |
| out = out * 60 + p | |
| return out | |
| def _mmss(t: float) -> str: | |
| t = max(0, int(round(t))) | |
| return f"{t // 60:02d}:{t % 60:02d}" | |
| class _Ctx: | |
| def __init__(self, payload: dict, url_fn: Callable[[str], str], lang: str) -> None: | |
| self.U = lambda rel: url_fn(rel) if rel else "" | |
| self.t = _T.get(lang, _T["en"]) | |
| self.ev = {e["ref_id"]: e for e in (payload.get("evidence") or []) if e.get("ref_id")} | |
| def video(self, ev: dict | None, start=None, end=None) -> str: | |
| """Playable video URL for an evidence item with a media-fragment time range, or "".""" | |
| if not ev or not ev.get("video_url"): | |
| return "" | |
| s = start if start is not None else _secs(ev.get("video_start_seconds")) | |
| e = end if end is not None else _secs(ev.get("end_time")) | |
| frag = "" | |
| if s is not None and e is not None and e > s: | |
| frag = f"#t={s:.1f},{e:.1f}" | |
| elif s is not None: | |
| frag = f"#t={s:.1f}" | |
| return self.U(ev["video_url"]) + frag | |
| def figures(self, ev: dict | None) -> list[str]: | |
| """Pictures cut from the manual for this passage; the highlighted page image when there are none.""" | |
| if not ev: | |
| return [] | |
| figs = [self.U(u) for u in (ev.get("figure_urls") or [])] | |
| if not figs and ev.get("pdf_image_url"): | |
| figs = [self.U(ev["pdf_image_url"])] | |
| return figs | |
| def evidence_node(self, ev: dict) -> dict: | |
| kind = {"oem_manual": "manual", "action_chunk": "video", "manual_procedure": "procedure"}.get(ev.get("doc_type"), "source") | |
| title = ev.get("title") or "" | |
| passage = (ev.get("passage") or ev.get("snippet") or "").strip() | |
| if len(passage) > 320: | |
| passage = passage[:320].rstrip() + "..." | |
| text = f"{title}\n{passage}".strip() | |
| if kind == "manual": | |
| figs = self.figures(ev) | |
| return node(text, image=figs[0] if figs else "", images=figs, type="evidence", source=kind, | |
| page=ev.get("pdf_page") or ev.get("page_start"), section=ev.get("section")) | |
| s, e = _secs(ev.get("video_start_seconds")), _secs(ev.get("end_time")) | |
| return node(text, image=self.U(ev.get("image_url") or ""), video=self.video(ev), type="evidence", source=kind, | |
| video_start=s, video_end=e, source_url=ev.get("source_url")) | |
| def manual_node(self, m: dict) -> dict: | |
| """The verified manual quote of a guide step / quick answer, with its figures.""" | |
| t = self.t | |
| ev = self.ev.get(m.get("ref")) or {} | |
| figs = [self.U(u) for u in (m.get("figure_urls") or [])] or self.figures(ev) | |
| page = m.get("page") | |
| cite = f"{t['manual']} {t['page']}{page}" if page is not None else t["manual"] | |
| return node(f"“{m.get('quote', '').strip()}”\n— {cite}", image=figs[0] if figs else "", images=figs, | |
| type="manual_quote", page=page, section=m.get("section")) | |
| def list_block(self, title: str, items: list[dict], kind: str) -> dict | None: | |
| if not items: | |
| return None | |
| children = [] | |
| for it in items: | |
| line = it.get("name", "").strip() | |
| if it.get("spec"): | |
| line += f" — {it['spec']}" | |
| if it.get("note"): | |
| line += f" ({it['note']})" | |
| children.append(node(line, type=kind[:-1] if kind.endswith("s") else kind)) | |
| return node(title, objs=children, type=kind) | |
| def related(self, qs: list[str] | None) -> dict | None: | |
| if not qs: | |
| return None | |
| return node(self.t["related"], objs=[node(q, type="suggested_question") for q in qs], type="suggested_questions") | |
| # ------------------------------------------------------------------------------- guide (how-to) | |
| def _guide(g: dict, c: _Ctx) -> dict: | |
| t = c.t | |
| blocks: list[dict] = [] | |
| if g.get("safety_warnings"): | |
| blocks.append(node(t["safety"], objs=[node(s, type="warning") for s in g["safety_warnings"]], type="safety")) | |
| for title, key in ((t["parts"], "parts"), (t["tools"], "tools")): | |
| b = c.list_block(title, g.get(key) or [], key) | |
| if b: | |
| blocks.append(b) | |
| steps = [] | |
| for s in g.get("steps") or []: | |
| v = s.get("video") or {} | |
| ev = c.ev.get(v.get("ref")) if v else None | |
| text = f"{s.get('step')}. {s.get('title', '').strip()}\n{s.get('instruction', '').strip()}".strip() | |
| if s.get("safety_note"): | |
| text += f"\n! {s['safety_note']}" | |
| vs, ve = (v.get("start"), v.get("end")) if v else (None, None) | |
| video_url = c.video(ev, vs, ve) if v else "" | |
| frame = c.U(v.get("frame_url") or "") if v else "" | |
| objs = [] | |
| if v and vs is not None and ve is not None: | |
| note = f"{t['watch']} {_mmss(vs)}–{_mmss(ve)}" + (f": {v['watch']}" if v.get("watch") else "") | |
| objs.append(node(note, type="video_note", video_start=vs, video_end=ve)) | |
| steps.append(node( | |
| text, image=frame or (c.U((ev or {}).get("image_url") or "") if v else ""), video=video_url, | |
| obj=c.manual_node(s["manual"]) if s.get("manual") else "", objs=objs, type="step", step=s.get("step"), | |
| video_start=vs, video_end=ve, source_url=(ev or {}).get("source_url") if v and not video_url else None, | |
| )) | |
| blocks.append(node(t["steps"], objs=steps, type="steps")) | |
| if g.get("when_to_escalate"): | |
| blocks.append(node(f"{t['escalate']}: {g['when_to_escalate']}", type="escalate")) | |
| if g.get("confidence_note"): | |
| blocks.append(node(g["confidence_note"], type="confidence")) | |
| rel = c.related(g.get("related_questions")) | |
| if rel: | |
| blocks.append(rel) | |
| head = "\n".join(x for x in (g.get("title"), g.get("summary")) if x) | |
| return node(head, objs=blocks, type="guide", difficulty=g.get("difficulty"), grounding=g.get("grounding")) | |
| # ------------------------------------------------------------------------------- diagnosis (top-3 causes) | |
| def _diagnosis(d: dict, c: _Ctx) -> dict: | |
| t = c.t | |
| blocks: list[dict] = [] | |
| if d.get("immediate_safety_warnings"): | |
| blocks.append(node(t["safety"], objs=[node(_clean(s), type="warning") for s in d["immediate_safety_warnings"]], type="safety")) | |
| causes = [] | |
| for cause in d.get("top_causes") or []: | |
| refs = [r for r in (cause.get("source_refs") or []) if r in c.ev] | |
| evs = [c.ev[r] for r in refs] | |
| vid = next((e for e in evs if e.get("video_url")), None) or next((e for e in evs if e.get("doc_type") == "action_chunk"), None) | |
| figs: list[str] = [] | |
| for e in evs: | |
| if e.get("doc_type") == "oem_manual": | |
| figs += c.figures(e) | |
| steps = [node(_clean(f"{s.get('step')}. {s.get('instruction', '')}" + (f"\n! {s['safety_note']}" if s.get("safety_note") else "")), type="step", step=s.get("step")) | |
| for s in cause.get("repair_steps") or []] | |
| children = list(steps) | |
| if cause.get("verification_checks"): | |
| children.append(node(t["check"], objs=[node(_clean(x), type="check") for x in cause["verification_checks"]], type="checks")) | |
| if evs: | |
| children.append(node(t["sources"], objs=[c.evidence_node(e) for e in evs], type="sources")) | |
| text = f"#{cause.get('rank')} {cause.get('cause', '').strip()} ({t['likelihood']}: {cause.get('likelihood', '')})\n{_clean(cause.get('reasoning'))}" | |
| causes.append(node( | |
| text, image=c.U((vid or {}).get("image_url") or "") if vid else "", images=figs[:3], video=c.video(vid) if vid else "", | |
| objs=children, type="cause", rank=cause.get("rank"), likelihood=cause.get("likelihood"), | |
| video_start=_secs((vid or {}).get("video_start_seconds")) if vid else None, | |
| video_end=_secs((vid or {}).get("end_time")) if vid else None, | |
| )) | |
| blocks += causes | |
| if d.get("tools_and_parts"): | |
| blocks.append(node(t["tools_parts"], objs=[node(_clean(x), type="item") for x in d["tools_and_parts"]], type="tools_and_parts")) | |
| if d.get("when_to_escalate"): | |
| blocks.append(node(f"{t['escalate']}: {_clean(d['when_to_escalate'])}", type="escalate")) | |
| if d.get("confidence_note"): | |
| blocks.append(node(_clean(d["confidence_note"]), type="confidence")) | |
| return node(t["causes"], objs=blocks, type="diagnosis") | |
| # ------------------------------------------------------------------------------- quick answer | |
| def _lookup(a: dict, c: _Ctx) -> dict: | |
| t = c.t | |
| v = a.get("video") or {} | |
| ev = c.ev.get(v.get("ref")) if v else None | |
| objs = [] | |
| if a.get("details"): | |
| objs.append(node(t["details"], objs=[node(x, type="detail") for x in a["details"]], type="details")) | |
| rel = c.related(a.get("related_questions")) | |
| if rel: | |
| objs.append(rel) | |
| vs, ve = (v.get("start"), v.get("end")) if v else (None, None) | |
| return node( | |
| a.get("answer", ""), image=c.U(v.get("frame_url") or "") if v else "", video=c.video(ev, vs, ve) if v else "", | |
| obj=c.manual_node(a["manual"]) if a.get("manual") else "", objs=objs, type="answer", | |
| video_start=vs, video_end=ve, grounding=a.get("grounding"), | |
| ) | |
| # ------------------------------------------------------------------------------- everything else | |
| def _clarify(reply: str, clar: dict, c: _Ctx) -> dict: | |
| qs = [] | |
| for q in clar.get("questions") or []: | |
| qs.append(node(q.get("text", ""), objs=[node(o, type="option") for o in q.get("options") or []], type="question")) | |
| return node(reply, objs=qs, type="clarify") | |
| def _followup_answer(a: dict, c: _Ctx) -> dict: | |
| steps = [node(_clean(f"{s.get('step')}. {s.get('instruction', '')}" + (f"\n! {s['safety_note']}" if s.get("safety_note") else "")), type="step", step=s.get("step")) | |
| for s in a.get("steps") or []] | |
| objs = steps | |
| rel = c.related(a.get("suggested_questions")) | |
| if rel: | |
| objs = steps + [rel] | |
| return node(_clean(a.get("answer")), objs=objs, type="answer") | |
| def to_app_obj(payload: dict, url_fn: Callable[[str], str], lang: str = "en") -> dict: | |
| """The app-team JSON for one FixSense response (the dict /api/query or /api/followup returns).""" | |
| c = _Ctx(payload, url_fn, lang) | |
| intent = payload.get("intent") or ("diagnose" if payload.get("diagnosis") else "reply") | |
| reply = _clean(payload.get("reply") or "") | |
| if intent == "howto" and payload.get("guide"): | |
| root = _guide(payload["guide"], c) | |
| elif intent == "lookup" and payload.get("lookup"): | |
| root = _lookup(payload["lookup"], c) | |
| elif intent == "clarify": | |
| root = _clarify(reply or _clean((payload.get("answer") or {}).get("answer")), payload.get("clarify") or {}, c) | |
| elif intent in ("chitchat", "out_of_scope"): | |
| root = node(reply or _clean((payload.get("answer") or {}).get("answer")), type="reply") | |
| elif payload.get("diagnosis") and (payload["diagnosis"].get("top_causes") or not payload.get("answer")): | |
| root = _diagnosis(payload["diagnosis"], c) | |
| if not payload["diagnosis"].get("top_causes") and payload["diagnosis"].get("confidence_note"): | |
| root["text"] = _clean(payload["diagnosis"]["confidence_note"]) | |
| elif payload.get("answer"): | |
| root = _followup_answer(payload["answer"], c) | |
| else: | |
| root = node(reply, type="reply") | |
| root["meta"] = { | |
| "intent": intent, "request_id": payload.get("request_id"), "lang": lang, | |
| "refused": bool(payload.get("refused")), "error": payload.get("error"), | |
| "vehicle": payload.get("vehicle_summary"), | |
| } | |
| return root | |
| def validate(tree: dict, _path: str = "root") -> list[str]: | |
| """Structural check of the schema above; returns a list of problems (empty = valid).""" | |
| problems: list[str] = [] | |
| want = {"text": str, "image": str, "images": list, "video": str, "videos": list, "objs": list} | |
| for k, typ in want.items(): | |
| if k not in tree: | |
| problems.append(f"{_path}: missing '{k}'") | |
| elif not isinstance(tree[k], typ): | |
| problems.append(f"{_path}.{k}: expected {typ.__name__}, got {type(tree[k]).__name__}") | |
| if "obj" not in tree: | |
| problems.append(f"{_path}: missing 'obj'") | |
| for k in ("images", "videos"): | |
| for i, u in enumerate(tree.get(k) or []): | |
| if not isinstance(u, str) or not u.startswith(("http://", "https://")): | |
| problems.append(f"{_path}.{k}[{i}]: not an absolute URL: {u!r}") | |
| for k in ("image", "video"): | |
| u = tree.get(k) | |
| if isinstance(u, str) and u and not u.startswith(("http://", "https://")): | |
| problems.append(f"{_path}.{k}: not an absolute URL: {u!r}") | |
| if isinstance(tree.get("obj"), dict): | |
| problems += validate(tree["obj"], _path + ".obj") | |
| elif tree.get("obj") not in ("", None): | |
| problems.append(f"{_path}.obj: expected node or empty string") | |
| for i, ch in enumerate(tree.get("objs") or []): | |
| problems += validate(ch, f"{_path}.objs[{i}]") if isinstance(ch, dict) else [f"{_path}.objs[{i}]: not an object"] | |
| return problems | |