"""天问 (Tianwen) — gradio.Server 后端 + React (shadcn) 前端。 启动: python app.py -> http://127.0.0.1:7860/ 前端: frontend/ (Vite + React), 唯一前端; 构建产物 frontend/dist 由后端托管。 静态数据 (生肖图 / 城市库) 随前端置于 frontend/public/data, 构建后经 /data 提供。 架构 (参照 docs/Any Custom Frontend with Gradio's Backend.md): - @app.api() 端点走 Gradio 队列, 前端 fetch 两步调用: POST /gradio_api/call/ {"data": [...]} -> {event_id} GET /gradio_api/call// -> SSE "event: complete" - 流式解读走 FastAPI 原生 /stream/* (SSE), 见文末路由。 - 排盘 / 解读 / 安全护栏全部在本地 Python 完成, 无外部请求。 """ from __future__ import annotations import base64 import datetime as _dt import json import os import random import socket import sys import tempfile from pathlib import Path from typing import Any # 本机代理（如 Clash) 会拦截 Gradio 对 localhost 的启动自检，必须绕过 os.environ.setdefault("NO_PROXY", "127.0.0.1,localhost") os.environ.setdefault("no_proxy", "127.0.0.1,localhost") # HF Space: 强制 Gradio 绑 0.0.0.0:7860 并关闭 SSR 代理层（即使平台以 import 方式拉起也生效） if os.environ.get("SPACE_ID") or os.environ.get("SPACE_HOST"): os.environ.setdefault("GRADIO_SERVER_NAME", "0.0.0.0") os.environ.setdefault("GRADIO_SERVER_PORT", "7860") os.environ.setdefault("GRADIO_SSR_MODE", "False") # Windows shim: 阻止 spaces 拉起 GPU if sys.platform == "win32": import types _mock = types.ModuleType("spaces") _mock.GPU = lambda *a, **kw: (a[0] if (a and callable(a[0])) else (lambda f: f)) sys.modules["spaces"] = _mock # gradio 6 在 Windows 上 import 时会出现 gradio_client 子模块循环依赖， # 表现为 import gradio 静默挂起。预先 import gradio_client 解决。 try: from gradio_client import utils as _gc_utils # noqa: F401 from gradio_client.client import Client as _GCClient # noqa: F401 except Exception: pass from gradio import Server from gradio.data_classes import FileData from fastapi.responses import HTMLResponse, FileResponse, StreamingResponse from fastapi.staticfiles import StaticFiles from pydantic import BaseModel from lunar_python import Solar from bazi.engine import compute_five_dim from bazi.five_elements import SHENGXIAO_ORDER, describe_relation from bazi.shensha_cards import SHEN_SHA_CARDS, JI_SHEN as _JI_SHEN from divination.coin import Cast, Line, COIN_SUM_TO_LINE, LINE_VALUE_TO_NAME from divination.hexagram import hex_from_six_lines, get_hexagram from divination.interpret import build_reading from prompts.templates import ( daily_narrative, cast_narrative, micro_action, daily_narrative_stream, cast_narrative_stream, natal_narrative_stream, year_narrative_stream, month_narrative_stream, ) from prompts.llm_client import ( set_runtime_config as _llm_set_config, get_runtime_config as _llm_get_config, is_reachable as _llm_reachable, ) from lunar_python.util import LunarUtil from safety.intercept import check as safety_check, HELPLINES from storage import profiles as prof_store from storage import readings as read_store from storage import divination_log as div_store from storage import mood_log as mood_store from storage import backup as bk from storage.fs import EXPORTS_DIR from ui.summary_image import render_daily_image APP_TITLE = "天问" APP_VERSION = "V3.0" SAFETY_VERSION = "v3.0.0" LLM_VERSION = "MiniCPM5-1B (fine-tuned, llama.cpp) + 规则模板兜底" DIST_DIR = Path(__file__).resolve().parent / "frontend" / "dist" # React (shadcn) 构建产物 ONBOARDING_QUOTES = [ "天行健，君子以自强不息。", "地势坤，君子以厚德载物。", "穷则变，变则通，通则久。", "君子以反身修德。", "观乎天文，以察时变。", "云在青天，水在瓶。", ] # 大运十神关系 -> 积极心理学叙事 _DAYUN_NARRATIVE = { "同类": "同气相求的十年。自我力量得到同伴支援，适合结盟协作、巩固自己的立场，也请留意人际边界，把精力收回自己身上。", "生我": "滋养与吸收的十年。资源与贵人之气涌入，是学习、休整、向内积累的好时机，允许自己慢下来，被照顾也是一种能力。", "我生": "表达与创造的十年。才华向外流淌，适合输出作品、教学相长；留意精力管理，创造之余记得给自己蓄水。", "我克": "开拓与收获的十年。你对资源有更强的掌控力，适合主动布局事业与财务；收获时也练习分享，让流动持续。", "克我": "磨砺与塑形的十年。外部压力像刻刀，雕出更清晰的自我边界。压力大时，把它视作修炼心理韧性的道场，蓄力以待。", } # 干支 / 神煞 长按解释卡片（规则字典，防幻觉：不由生成式输出） _TERM_CARDS = { "甲": "甲木，参天大树之象。代表正直向上的生长力，适合开创。", "乙": "乙木，花草藤蔓之象。柔韧而善于借势，适合协作与渗透。", "丙": "丙火，太阳之火。热情坦荡，照亮他人，留意过度燃烧。", "丁": "丁火，灯烛之火。细腻持久的温暖，适合深耕与陪伴。", "戊": "戊土，高山厚土。稳重可靠，是他人依靠，也要给自己松土。", "己": "己土，田园之土。包容滋养，善于成全，记得也滋养自己。", "庚": "庚金，刀剑之金。果决锐利，长于裁断，用在创作上是利器。", "辛": "辛金，珠玉之金。精致敏锐，重视品质，留意完美主义。", "壬": "壬水，江河之水。奔流不息的行动力与适应力，宜疏不宜堵。", "癸": "癸水，雨露之水。安静的洞察与共情，是滋润万物的智慧。", "子": "子水，夜半之时。万物萌动之始，代表潜藏的智慧与直觉。", "丑": "丑土，凌晨之时。寒土孕金，代表沉默的积累与韧性。", "寅": "寅木，黎明之时。生机破土，代表勇气与新的开始。", "卯": "卯木，日出之时。草木舒展，代表生长、社交与柔韧。", "辰": "辰土，上午之时。水库之土，包容变化，代表蓄势与格局。", "巳": "巳火，近午之时。文明之火，代表表达、礼仪与洞见。", "午": "午火，正午之时。光明炽盛，代表行动力与影响力的顶点。", "未": "未土，午后之时。燥土养木，代表温和的坚持与照料。", "申": "申金，傍晚之时。代表变通、技艺与解决问题的巧思。", "酉": "酉金，日落之时。代表精纯、审美与对秩序的追求。", "戌": "戌土，黄昏之时。代表忠诚、守护与对信念的坚持。", "亥": "亥水，入夜之时。代表包容、想象力与回归本真。", "天乙贵人": "众神之首的吉星，象征贵人扶助。心理学视角：你善于建立支持系统，困境中总有人愿意伸手。", "文昌": "主聪明好学之星。代表表达与学习的天赋，适合以文字与思考立身。", "驿马": "主变动远行之星。心有四方，变化即机会，拥抱流动的人生。", "桃花": "主人缘魅力之星。亲和力是天赋，善用它建立滋养型关系。", "华盖": "主艺术孤高之星。独处时能量最盛，适合艺术、哲思与修行。", "羊刃": "锐利而专注的能量，用在创作与攻坚上是利器，留意刚极易折，配合柔性节奏。", "七杀": "外部压力较大的信号，适合借此修炼心理边界，把压力转译为目标。", "劫煞": "提醒守护自己资源与精力的信号，练习温和而坚定地说不。", "灾煞": "波动期的提示，放慢节奏、降低杠杆，把注意力放回可控之事。", "孤辰": "独处倾向的信号。孤独不是缺陷，是与自己深度相处的邀请。", "寡宿": "内省倾向的信号。安静的时光里，完成与自我的和解。", } # ---- 英文对照 (默认英语用户需要看懂; 五行 / 关系 / 大运 / 干支神煞) ---- _ELEMENT_EN = {"金": "Metal", "木": "Wood", "水": "Water", "火": "Fire", "土": "Earth"} _RELATION_EN = {"同类": "Companion", "生我": "Resource", "我生": "Output", "我克": "Wealth", "克我": "Influence"} _DAYUN_NARRATIVE_EN = { "同类": "A decade of kindred energy. Your sense of self is backed by peers — good for alliances and consolidating your ground; mind your boundaries and keep some energy for yourself.", "生我": "A decade of nourishment. Resources and mentors flow in — a fine time to learn, rest, and build inwardly. Let yourself slow down; being cared for is a skill too.", "我生": "A decade of expression. Talent flows outward — good for creating, teaching, and sharing. Watch your energy and refill the well between bursts.", "我克": "A decade of initiative. You hold firmer command of resources — good for shaping career and finances; practice sharing your gains so the flow continues.", "克我": "A decade of tempering. Outside pressure is like a chisel, carving a clearer self. When it weighs on you, treat it as a dojo for resilience and gather strength.", } _TERM_CARDS_EN = { "甲": "Jiǎ wood — the towering tree. Upright, upward-growing strength; suited to pioneering.", "乙": "Yǐ wood — vines and grasses. Supple, good at leveraging circumstance; suited to cooperation.", "丙": "Bǐng fire — the sun. Warm and open, lighting others; mind burning out.", "丁": "Dīng fire — lamp and candle. A gentle, lasting warmth; suited to deep, steady work and companionship.", "戊": "Wù earth — the high mountain. Steady and dependable, a rock for others; remember to loosen your own soil.", "己": "Jǐ earth — garden soil. Nurturing and accommodating, good at enabling others; nourish yourself too.", "庚": "Gēng metal — blade and sword. Decisive and sharp; a fine tool when turned to creation.", "辛": "Xīn metal — jewel and gem. Refined and keen, valuing quality; watch perfectionism.", "壬": "Rén water — the river. Ceaseless drive and adaptability; better guided than dammed.", "癸": "Guǐ water — dew and rain. Quiet insight and empathy; the wisdom that nourishes all things.", "子": "Zǐ (midnight). The stirring start of all things — latent wisdom and intuition.", "丑": "Chǒu (before dawn). Cold earth holding metal — silent accumulation and resilience.", "寅": "Yín (daybreak). Life breaking through soil — courage and new beginnings.", "卯": "Mǎo (sunrise). Plants unfurling — growth, sociability, suppleness.", "辰": "Chén (morning). Reservoir earth holding change — gathering momentum and scope.", "巳": "Sì (late morning). The fire of culture — expression, ritual, insight.", "午": "Wǔ (noon). Blazing light — the peak of drive and influence.", "未": "Wèi (afternoon). Dry earth feeding wood — gentle persistence and care.", "申": "Shēn (dusk). Adaptability, craft, and clever problem-solving.", "酉": "Yǒu (sunset). Refinement, aesthetics, and a love of order.", "戌": "Xū (twilight). Loyalty, guardianship, devotion to one's convictions.", "亥": "Hài (nightfall). Openness, imagination, and a return to one's true self.", "天乙贵人": "Noble Mentor star — a sign of helpful patrons. In plain terms: you build support systems well; in hard times, someone tends to lend a hand.", "文昌": "Scholar star — a gift for learning and expression; suited to a life built on words and thought.", "驿马": "Travelling Horse star — movement and journeys; change is opportunity, embrace a life in motion.", "桃花": "Peach Blossom star — charm and rapport; your warmth is a gift for building nourishing bonds.", "华盖": "Canopy star — artistry and solitude; your energy peaks alone, suited to art, philosophy, and practice.", "羊刃": "Sharp, focused energy — a fine tool for creating and breaking through; the rigid snaps, so pair it with a softer rhythm.", "七杀": "A sign of strong outside pressure — good for tempering your boundaries and turning pressure into goals.", "劫煞": "A reminder to guard your resources and energy — practice saying no, gently and firmly.", "灾煞": "A note of turbulence — slow down, reduce leverage, return attention to what you can control.", "孤辰": "A sign of solitude — being alone isn't a flaw; it's an invitation to deep time with yourself.", "寡宿": "A sign of inwardness — in the quiet, make peace with yourself.", } def _today() -> _dt.date: return _dt.date.today() def _today_day_gan() -> str: try: lunar = Solar.fromYmd(_today().year, _today().month, _today().day).getLunar() return lunar.getDayGan() except Exception: return "甲" _SHENGXIAO_BY_YEAR = list(SHENGXIAO_ORDER) def _profile_brief(p: dict[str, Any]) -> dict[str, Any]: zodiac = p.get("zodiac", "") if not zodiac: y = (p.get("birth") or {}).get("year") zodiac = _SHENGXIAO_BY_YEAR[(int(y) - 4) % 12] if y else "" # 四柱八字（名片展示用）: 创建时已存；旧档案懒算并回填缓存 bazi = p.get("bazi") if not bazi and (p.get("birth") or {}).get("year"): try: rd = _compute_reading(p) bazi = [pl["gan"] + pl["zhi"] for pl in rd["ming"]["pillars"]] p["bazi"] = bazi prof_store.save(p) except Exception: bazi = [] return {"id": p["id"], "name": p.get("name", "未命名"), "birth": p.get("birth", {}), "zodiac": zodiac, "bazi": bazi or []} def _list_profiles(device_id: str = "") -> list[dict[str, Any]]: """仅返回归属于该设备的档案（隐私：档案与创建它的设备/浏览器绑定）。""" did = (device_id or "").strip() return [ _profile_brief(p) for p in prof_store.list_profiles() if p.get("owner") == did and did ] def _load_owned(profile_id: str, device_id: str) -> dict[str, Any] | None: """加载档案并校验设备归属，不属于该设备视同不存在。""" prof = prof_store.load(profile_id) if not prof or prof.get("owner") != (device_id or "").strip(): return None return prof def _compute_reading(profile: dict[str, Any]) -> dict[str, Any]: b = profile["birth"] rd = compute_five_dim( b["year"], b["month"], b["day"], b["hour"], b.get("minute", 0), b.get("gender", 1), tz_offset=float(b.get("tz_offset", 8.0)), longitude=float(b.get("longitude", 120.0)), southern=bool(b.get("southern", False)), ).to_dict() rd["profile"] = {"name": profile.get("name", "您"), "id": profile["id"]} return rd def _dayun_brief_text(reading: dict[str, Any], index: int, lang: str = "zh") -> str: en = lang == "en" yun = reading.get("yun", {}) or {} stages = yun.get("dayun_list", []) or [] if not stages or index < 0 or index >= len(stages): return "This phase is still in the mist." if en else "此运尚在云雾之中。" st = stages[index] if st.get("ganzhi") == "童限": if en: return ( f"The Early Years (ages 0-{st['end_age']}) are the nurturing time before the luck phases begin. " f"Their tone is written by your year pillar and your parents' shelter — like a seed gathering its first breath " f"in the soil. Looking back at them is a clue to your original temperament." ) return ( f"童限 (0-{st['end_age']} 岁）, 是大运起步前的蒙养之年。这段时光的底色由年柱与父母的庇护写就，" f"如同种子在土壤中积蓄第一口气。回望它，是理解自己最初性情的线索。" ) ming = reading.get("ming", {}) pillars = ming.get("pillars") or [] day_el = pillars[2]["gan_element"] if len(pillars) > 2 else "土" rel = describe_relation(day_el, st.get("gan_element", day_el)) cur = yun.get("current_index", -1) if en: body = _DAYUN_NARRATIVE_EN.get(rel, "Ups and downs are the norm — cultivate in fair seasons, gather strength in lean ones.") when = "current" if index == cur else ("past" if index < cur else "coming") return ( f"The {st['ganzhi']} luck phase (ages {st['start_age']}-{st['end_age']}, {st['start_year']}-{st['end_year']}) " f"is your {when} stretch. Its stem {_ELEMENT_EN.get(st['gan_element'], st['gan_element'])} energy forms a " f"“{_RELATION_EN.get(rel, rel)}” dynamic with your day master's {_ELEMENT_EN.get(day_el, day_el)}. {body} " f"Ups and downs are natural — each phase carries its own lesson and gift." ) body = _DAYUN_NARRATIVE.get(rel, "起伏本是常态，顺境耕作，低谷蓄力。") when = "当前所处的" if index == cur else ("已经走过的" if index < cur else "未来将至的") return ( f"{st['ganzhi']} 大运 ({st['start_age']}-{st['end_age']} 岁，" f"{st['start_year']}-{st['end_year']} 年）, 是您{when}一程。" f"天干{st['gan_element']}气与您的日主{day_el}气构成「{rel}」之势。{body}" f" 起伏本是常态，每一运都有它的功课与礼物。" ) app = Server() # ============================ API: 启动与档案 ============================ @app.api() def bootstrap(device_id: str) -> dict: """前端启动握手：本设备的档案列表 + 今日信息 + 版本与热线。""" did = (device_id or "").strip() if did: # 一次性迁移：早期未标记归属的档案，归首个访问的设备 for p in prof_store.list_profiles(): if not p.get("owner"): p["owner"] = did prof_store.save(p) return { "profiles": _list_profiles(did), "quote": random.choice(ONBOARDING_QUOTES), "today": _today().isoformat(), "zodiacs": list(SHENGXIAO_ORDER), "version": {"app": APP_VERSION, "safety": SAFETY_VERSION, "llm": LLM_VERSION}, "helplines": [{"name": n, "tel": t} for n, t in HELPLINES], } @app.api() def create_profile( name: str, year: int, month: int, day: int, hour: int, minute: int, gender: int, tz_offset: float, longitude: float, southern: bool, device_id: str, tz_name: str, ) -> dict: did = (device_id or "").strip() if not (name or "").strip(): return {"ok": False, "error": "请填写姓名。", "profiles": _list_profiles(did)} if not did: return {"ok": False, "error": "设备标识缺失，请刷新页面。", "profiles": []} try: offset = float(tz_offset if tz_offset is not None else 8.0) tzn = (tz_name or "").strip() if tzn: # IANA 时区（来自 GeoNames 城市库）: 按出生日期算偏移，自动含夏令时 try: from zoneinfo import ZoneInfo birth_dt = _dt.datetime(int(year), int(month), int(day), int(hour), int(minute or 0), tzinfo=ZoneInfo(tzn)) offset = birth_dt.utcoffset().total_seconds() / 3600.0 except Exception: tzn = "" prof = prof_store.make_profile( name.strip(), int(year), int(month), int(day), int(hour), int(minute or 0), int(gender) ) prof["birth"].update({ "tz_offset": offset, "tz_name": tzn, "longitude": float(longitude if longitude is not None else 120.0), "southern": bool(southern), }) prof["owner"] = did reading = _compute_reading(prof) # 先验证生辰可排盘，再落盘保存 prof["zodiac"] = reading["ming"]["zodiac"] # 生肖以排盘（农历年界） 为准 prof["bazi"] = [pl["gan"] + pl["zhi"] for pl in reading["ming"]["pillars"]] prof_store.save(prof) read_store.save(_today(), prof["id"], reading) return { "ok": True, "profile_id": prof["id"], "profiles": _list_profiles(did), "zodiac": reading["ming"]["zodiac"], "pillars": reading["ming"]["pillars"], "solar_corrected": reading["birth"].get("solar_corrected", ""), } except Exception as exc: return {"ok": False, "error": f"暂不支持此时间范围：{exc}", "profiles": _list_profiles(did)} @app.api() def delete_profile(profile_id: str, device_id: str) -> dict: if _load_owned(profile_id, device_id): prof_store.delete(profile_id) return {"ok": True, "profiles": _list_profiles(device_id)} # ============================ API: 五维解读 ============================ def _daily_texts(profile_id: str, reading: dict[str, Any], lang: str = "zh") -> tuple[str, list[str]]: """微行动（规则引擎，快） + 缓存的日运叙事。 日运叙事不在此处阻塞生成：当日缓存命中则返回全文，否则返回空串， 由前端调流式端点 stream_daily 逐字生成 (PRD 10.2 流式输出）。 """ cached = read_store.load(_today(), profile_id) or {} narrative = cached.get("_narrative") or "" cache_key = "_actions" if lang == "zh" else f"_actions_{lang}" actions = cached.get(cache_key) or micro_action(reading, lang=lang) reading["_narrative"] = narrative reading["_actions"] = actions return narrative, actions @app.api() def get_reading(profile_id: str, device_id: str, lang: str = "zh") -> dict: """完整五维数据 + 日运叙事 + 微行动 + 今日心境。""" prof = _load_owned(profile_id, device_id) if not prof: return {"ok": False, "error": "档案不存在"} reading = _compute_reading(prof) narrative, actions = _daily_texts(profile_id, reading, lang) mood = mood_store.load(_today(), profile_id) return { "ok": True, "reading": reading, "narrative": narrative, "actions": actions, "mood": mood, "current_dayun_brief": _dayun_brief_text(reading, reading.get("yun", {}).get("current_index", 0), lang), } @app.api() def dayun_brief(profile_id: str, index: int, device_id: str, lang: str = "zh") -> dict: prof = _load_owned(profile_id, device_id) if not prof: return {"ok": False, "error": "档案不存在"} reading = _compute_reading(prof) return {"ok": True, "text": _dayun_brief_text(reading, int(index), lang)} @app.api() def explain_term(term: str, lang: str = "zh") -> dict: key = (term or "").strip() if lang == "en": text = _TERM_CARDS_EN.get(key) if not text and key in SHEN_SHA_CARDS: if key in _JI_SHEN: text = "An auspicious almanac spirit for this day — read it as a gentle tailwind: a good day to lean into what it favors, not a guarantee." else: text = "A cautionary almanac spirit — not an omen of doom, just a nudge to slow down, double-check, and protect your energy today." return {"ok": True, "term": term, "text": text or "A faint light in your chart. Every symbol here is a small window for looking at yourself — meet it with curiosity, not a rush to judge fortune."} text = _TERM_CARDS.get(key) or SHEN_SHA_CARDS.get(key) return {"ok": True, "term": term, "text": text or "这是命盘中的一处微光。每个符号都是观照自己的一扇小窗——带着好奇去看它，而非急着论断吉凶。"} # ── LLM API 接口配置 (可在设置中切换不同的 OpenAI-compatible 服务) ────────────── _LLM_CFG_FILE = Path(__file__).resolve().parent / "data" / "llm_config.json" def _load_llm_cfg() -> None: """启动时载入已保存的 LLM 配置 (endpoint/key/model)。""" try: if _LLM_CFG_FILE.exists(): cfg = json.loads(_LLM_CFG_FILE.read_text(encoding="utf-8-sig")) _llm_set_config(cfg.get("base"), cfg.get("key"), cfg.get("model")) except Exception: pass @app.api() def set_llm_config(endpoint: str = "", api_key: str = "", model: str = "") -> dict: """配置 LLM API: endpoint(base url) + secret(api key) + 可选 model。 空串表示清除该项, 回落到默认。配置持久化到 data/llm_config.json。""" _llm_set_config(endpoint, api_key, model) try: _LLM_CFG_FILE.parent.mkdir(parents=True, exist_ok=True) # key 也落盘 (本机单用户场景); 如需更安全可改为仅存内存 _LLM_CFG_FILE.write_text(json.dumps( {"base": endpoint.strip(), "key": api_key.strip(), "model": model.strip()}, ensure_ascii=False, ), encoding="utf-8") except Exception as e: return {"ok": False, "error": str(e)} reachable = _llm_reachable() cfg = _llm_get_config() return {"ok": True, "reachable": reachable, "base": cfg["base"], "model": cfg["model"]} @app.api() def get_llm_config() -> dict: """返回当前生效的 LLM 配置 (不回传 key 明文)，并探活。""" cfg = _llm_get_config() return {"ok": True, "base": cfg["base"], "model": cfg["model"], "has_key": cfg["has_key"], "reachable": _llm_reachable()} _load_llm_cfg() # ============================ API: 摇卦 ============================ @app.api() def cast_begin(question: str) -> dict: """书愿：危机词命中则熔断，直接返回安慰文案与热线。""" q = (question or "").strip() if not q: return {"ok": False, "error": "请先落笔心中所念之事。"} hit, payload = safety_check(q) if hit: return { "ok": False, "blocked": True, "message": payload, "helplines": [{"name": n, "tel": t} for n, t in HELPLINES], } return {"ok": True, "question": q} @app.api() def toss() -> dict: """掷三枚铜钱，成一爻。""" coins = [random.randint(0, 1) for _ in range(3)] # 1=正（阳面） 0=反 s = sum(3 if c == 0 else 2 for c in coins) value = COIN_SUM_TO_LINE[s] return { "coins": coins, "sum": s, "kind": LINE_VALUE_TO_NAME[value], "is_yang": value in (7, 9), "is_changing": value in (6, 9), } @app.api() def cast_complete(question: str, lines: list, lang: str = "zh") -> dict: """六爻已成：排出本卦/变卦，生成六亲六神与解卦叙事。""" if not lines or len(lines) != 6: return {"ok": False, "error": "六爻未成，请继续摇卦。"} line_objs = [ Line(index=i, coins=ld["coins"], sum=ld["sum"], kind=ld["kind"], is_yang=ld["is_yang"], is_changing=ld["is_changing"]) for i, ld in enumerate(lines, start=1) ] ben_bits = [1 if ld["is_yang"] else 0 for ld in lines] ben_code = hex_from_six_lines(ben_bits) bian_bits = [(1 - b) if ld["is_changing"] else b for b, ld in zip(ben_bits, lines)] bian_code = hex_from_six_lines(bian_bits) cast = Cast( lines=line_objs, ben_code=ben_code, bian_code=bian_code, ben_hex=get_hexagram(ben_code), bian_hex=get_hexagram(bian_code), changing_indices=[i for i, ld in enumerate(lines, start=1) if ld["is_changing"]], question=(question or "").strip(), created_at=_dt.datetime.now().isoformat(timespec="seconds"), ) cr = build_reading(cast, day_gan=_today_day_gan()).to_dict() cr["question"] = (question or "").strip() cr["created_at"] = cast.created_at # 解卦叙事由前端调 stream_cast 流式生成（不在此阻塞）; 微行动为规则引擎，即时 actions = micro_action(cr, lang=lang) return { "ok": True, "reading": cr, "ben": cr["ben"], "bian": cr["bian"], "ben_bits": ben_bits, "bian_bits": bian_bits, "changing": cast.changing_indices, "narrative": "", "actions": actions, } # ---- 流式解读： FastAPI 原生 SSE (gradio generator 不能序列化中间 yield, 故走原生路由） ---- # 每个事件为 data: {"text": <当前累积全文>}\n\n; 前端直接替换显示。 def _sse(text: str) -> str: return "data: " + json.dumps({"text": text}, ensure_ascii=False) + "\n\n" def _liunian_info(reading: dict[str, Any], birth: dict[str, Any]) -> dict[str, Any]: """今年流年：干支 / 五行 / 虚岁 / 与日主关系（流年按年中取，避开立春边界歧义）。""" year = _today().year gz = Solar.fromYmd(year, 6, 1).getLunar().getYearInGanZhi() gan_el = LunarUtil.WU_XING_GAN.get(gz[0], "—") if gz else "—" zhi_el = LunarUtil.WU_XING_ZHI.get(gz[1], "—") if len(gz) > 1 else "—" pillars = reading.get("ming", {}).get("pillars") or [] day_el = pillars[2]["gan_element"] if len(pillars) > 2 else "土" return { "year": year, "ganzhi": gz, "gan_element": gan_el, "zhi_element": zhi_el, "age": year - int(birth.get("year", year)) + 1, # 虚岁 "relation": describe_relation(day_el, gan_el), } _TIANGAN = "甲乙丙丁戊己庚辛壬癸" _MONTH_ZHI = ["寅", "卯", "辰", "巳", "午", "未", "申", "酉", "戌", "亥", "子", "丑"] def _liuyue_list(reading: dict[str, Any], ln: dict[str, Any]) -> list[dict[str, Any]]: """十二流月干支（五虎遁年起月） + 与日主关系。""" year_gan = (ln.get("ganzhi") or "甲")[0] yi = _TIANGAN.find(year_gan) base = ((yi % 5) * 2 + 2) % 10 if yi >= 0 else 2 # 正月天干索引（甲己丙作首…) pillars = reading.get("ming", {}).get("pillars") or [] day_el = pillars[2]["gan_element"] if len(pillars) > 2 else "土" out = [] for m in range(12): gan = _TIANGAN[(base + m) % 10] gan_el = LunarUtil.WU_XING_GAN.get(gan, "—") out.append({ "month": m + 1, "ganzhi": gan + _MONTH_ZHI[m], "gan_element": gan_el, "relation": describe_relation(day_el, gan_el), }) return out # 正念话固定话语集 (按语言 × 进度), 零延迟, 无需 LLM _MINDFUL = { "zh": { "none": "新的一年如十二段节气，尚未一一展开。不必心急，一月一月地看，自有分晓。", "all": "今年十二个月您都已照见。心里有数，便是最大的安稳——其余的，交给时间。", "some": "您已照见今年的一部分光景。看得见的，便不再只是焦虑；慢慢来，其余的会在合适的时候清晰起来。", }, "en": { "none": "The year is twelve seasons, not yet unfolded. No need to rush — take it one month at a time, and clarity will come.", "all": "You have seen all twelve months of this year. To know is the steadiest comfort; leave the rest to time.", "some": "You have glimpsed part of this year. What you can see is no longer mere worry — go slowly, the rest will clear in its own time.", }, } def _year_mindful(done_count: int, year_done: bool, lang: str = "zh") -> str: """据已推演进度生成一两句正念话 (固定话语集, 按语言切换, 零延迟)。""" table = _MINDFUL.get(lang, _MINDFUL["zh"]) if done_count <= 0 and not year_done: return table["none"] if done_count >= 12: return table["all"] return table["some"] def _year_cache(prof: dict[str, Any], year: int) -> dict[str, Any]: return (prof.get("year_cache") or {}).get(str(year), {}) @app.api() def year_overview(profile_id: str, device_id: str, lang: str = "zh") -> dict: """年运 Tab 的结构化数据：流年 + 十二流月（含已缓存的解读） + 正念话。""" prof = _load_owned(profile_id, device_id) if not prof: return {"ok": False, "error": "档案不存在"} reading = _compute_reading(prof) yun = reading.get("yun", {}) cur = (yun.get("dayun_list") or [{}])[yun.get("current_index", 0)] if yun.get("dayun_list") else {} ln = _liunian_info(reading, prof["birth"]) year = ln["year"] yc = _year_cache(prof, year) months = _liuyue_list(reading, ln) for m in months: m["text"] = yc.get(f"m{m['month']}", "") m["done"] = bool(m["text"]) done_count = sum(1 for m in months if m["done"]) year_text = yc.get("year", "") return { "ok": True, "liunian": ln, "current_dayun": cur, "months": months, "year_text": year_text, "year_done": bool(year_text), "mindful": _year_mindful(done_count, bool(year_text), lang), } class _StreamNatalReq(BaseModel): profile_id: str device_id: str lang: str = "zh" @app.post("/stream/natal") def stream_natal_route(req: _StreamNatalReq): def gen(): prof = _load_owned(req.profile_id, req.device_id) if not prof: yield _sse("档案不存在"); return # 仅缓存中文版；英文每次实时生成，避免旧的中文缓存被当英文返回 if req.lang == "zh": cached = (read_store.load(_today(), req.profile_id) or {}).get("_natal") if cached: yield _sse(cached); return reading = _compute_reading(prof) acc = "" for piece in natal_narrative_stream(reading, req.lang): acc += piece yield _sse(acc) hit, payload = safety_check(acc) if hit: acc = payload yield _sse(payload) if req.lang == "zh": rec = read_store.load(_today(), req.profile_id) or {} rec["_natal"] = acc try: read_store.save(_today(), req.profile_id, {**reading, **rec}) except Exception: pass return StreamingResponse(gen(), media_type="text/event-stream") def _save_year_cache(prof: dict[str, Any], year: int, key: str, text: str) -> None: """把本年/某月解读写入档案的 year_cache （中文版，推完即缓存，再进直接展示）。""" fresh = prof_store.load(prof["id"]) or prof yc = fresh.setdefault("year_cache", {}) yc.setdefault(str(year), {})[key] = text try: prof_store.save(fresh) except Exception: pass @app.post("/stream/year") def stream_year_route(req: _StreamNatalReq): def gen(): prof = _load_owned(req.profile_id, req.device_id) if not prof: yield _sse("档案不存在"); return reading = _compute_reading(prof) ln = _liunian_info(reading, prof["birth"]) reading["liunian"] = ln if req.lang == "zh": cached = _year_cache(prof, ln["year"]).get("year") if cached: yield _sse(cached); return acc = "" for piece in year_narrative_stream(reading, req.lang): acc += piece yield _sse(acc) hit, payload = safety_check(acc) if hit: acc = payload yield _sse(payload) if req.lang == "zh": _save_year_cache(prof, ln["year"], "year", acc) return StreamingResponse(gen(), media_type="text/event-stream") class _StreamMonthReq(BaseModel): profile_id: str device_id: str month: int lang: str = "zh" @app.post("/stream/month") def stream_month_route(req: _StreamMonthReq): def gen(): prof = _load_owned(req.profile_id, req.device_id) if not prof: yield _sse("档案不存在"); return m = int(req.month) if not 1 <= m <= 12: yield _sse("月份无效"); return reading = _compute_reading(prof) ln = _liunian_info(reading, prof["birth"]) if req.lang == "zh": cached = _year_cache(prof, ln["year"]).get(f"m{m}") if cached: yield _sse(cached); return reading["liunian"] = ln reading["liuyue"] = _liuyue_list(reading, ln)[m - 1] acc = "" for piece in month_narrative_stream(reading, req.lang): acc += piece yield _sse(acc) hit, payload = safety_check(acc) if hit: acc = payload yield _sse(payload) if req.lang == "zh": _save_year_cache(prof, ln["year"], f"m{m}", acc) return StreamingResponse(gen(), media_type="text/event-stream") class _StreamDailyReq(BaseModel): profile_id: str device_id: str lang: str = "zh" class _StreamCastReq(BaseModel): cast_reading: dict device_id: str lang: str = "zh" @app.post("/stream/daily") def stream_daily_route(req: _StreamDailyReq): def gen(): prof = _load_owned(req.profile_id, req.device_id) if not prof: yield _sse("档案不存在"); return cached = (read_store.load(_today(), req.profile_id) or {}) if req.lang == "zh" else {} if cached.get("_narrative"): yield _sse(cached["_narrative"]); return reading = _compute_reading(prof) acc = "" for piece in daily_narrative_stream(reading, req.lang): acc += piece yield _sse(acc) hit, payload = safety_check(acc) if hit: acc = payload yield _sse(payload) if req.lang == "zh": # 仅缓存中文版，英文每次实时生成 reading["_narrative"] = acc reading["_actions"] = micro_action(reading) try: read_store.save(_today(), req.profile_id, reading) except Exception: pass return StreamingResponse(gen(), media_type="text/event-stream") @app.post("/stream/cast") def stream_cast_route(req: _StreamCastReq): def gen(): cr = req.cast_reading if not cr or not cr.get("ben"): yield _sse("卦象待生成，让我们一起静心。"); return acc = "" for piece in cast_narrative_stream(cr, req.lang): acc += piece yield _sse(acc) hit, payload = safety_check(acc) if hit: yield _sse(payload) return StreamingResponse(gen(), media_type="text/event-stream") @app.api() def save_cast(cast_reading: dict, device_id: str) -> dict: if not cast_reading or not cast_reading.get("ben"): return {"ok": False, "error": "无可保存的卦例。"} rec = dict(cast_reading) rec["owner"] = (device_id or "").strip() rec_id = div_store.save(rec) return {"ok": True, "id": rec_id, "casts": list_casts(device_id)["casts"]} @app.api() def list_casts(device_id: str) -> dict: did = (device_id or "").strip() out = [] records = [d for d in div_store.list_all() if d.get("owner") == did and did] for d in records[:50]: ben = d.get("ben") if isinstance(d.get("ben"), dict) else {} bian = d.get("bian") if isinstance(d.get("bian"), dict) else {} out.append({ "id": d.get("id", ""), "question": d.get("question", "") or "心中挂碍", "ben_name": ben.get("name", "—"), "ben_nature": ben.get("nature", ""), "bian_name": bian.get("name", ""), "ben_code": ben.get("code", 0), "bian_code": bian.get("code", 0), "changing": d.get("changing_indices", []), "created_at": d.get("created_at", ""), "outcome": d.get("outcome", ""), "summary": d.get("summary", ""), }) return {"ok": True, "casts": out} @app.api() def set_outcome(div_id: str, outcome: str, device_id: str) -> dict: rec = div_store.load(div_id) if not rec or rec.get("owner") != (device_id or "").strip(): return {"ok": False} rec = div_store.update_outcome(div_id, (outcome or "").strip()) return {"ok": bool(rec)} # ============================ API: 心境 ============================ @app.api() def save_mood(profile_id: str, mood: str, note: str, device_id: str) -> dict: if not _load_owned(profile_id, device_id): return {"ok": False, "error": "请先选择档案。"} hit, payload = safety_check(note or "") if hit: return {"ok": False, "blocked": True, "message": payload} rec = mood_store.save(_today(), profile_id, mood or "cloud", (note or "").strip()) return {"ok": True, "mood": rec} @app.api() def mood_series(profile_id: str, days: int, device_id: str) -> dict: if not _load_owned(profile_id, device_id): return {"ok": True, "series": []} return {"ok": True, "series": mood_store.series_for_profile(profile_id, int(days or 30))} # ============================ API: 日签 ============================ @app.api() def summary_image(profile_id: str, style: str, device_id: str, lang: str = "zh") -> FileData: """生成今日日签 PNG, 返回可下载的文件。lang 控制文字语言。""" prof = _load_owned(profile_id, device_id) if not prof: raise ValueError("档案不存在") reading = _compute_reading(prof) _daily_texts(profile_id, reading, lang) mood_rec = mood_store.load(_today(), profile_id) mood = (mood_rec or {}).get("mood", "") el_map = {"rain": "水", "sun": "火", "wood": "木", "metal": "金", "ink": "墨"} if style in el_map: reading["riyun"] = {**reading["riyun"], "day_element": el_map[style]} divs = div_store.list_all() img = render_daily_image(reading, mood=mood, divination=divs[0] if divs else None, lang=lang) fname = f"tianwen-daily-{_today().isoformat()}" if lang == "en" else f"天问日签_{_today().isoformat()}" out = Path(tempfile.gettempdir()) / f"celestial_{_today().isoformat()}_{style}_{lang}.png" img.save(out) return FileData(path=str(out), orig_name=f"{fname}.png") # ============================ API: 备份与清除 ============================ @app.api() def export_backup(password: str) -> FileData: if not password or len(password) < 4: raise ValueError("密码至少 4 个字符") out = bk.export(password) return FileData(path=str(out), orig_name=out.name) @app.api() def restore_backup(content_b64: str, password: str, device_id: str) -> dict: if not content_b64 or not password: return {"ok": False, "error": "请选择备份文件并输入密码。"} try: raw = base64.b64decode(content_b64) EXPORTS_DIR.mkdir(parents=True, exist_ok=True) tmp = EXPORTS_DIR / "_restore_upload.celestial" tmp.write_bytes(raw) n = bk.restore(tmp, password) tmp.unlink(missing_ok=True) # 跨设备迁移：还原的档案与卦例归属导入它们的设备 did = (device_id or "").strip() if did: for p in prof_store.list_profiles(): p["owner"] = did prof_store.save(p) for d in div_store.list_all(): d["owner"] = did div_store.save(d) return {"ok": True, "restored": n, "profiles": _list_profiles(did)} except Exception as exc: return {"ok": False, "error": f"还原失败（密码错误或文件损坏）: {exc}"} @app.api() def clear_all(confirm: str) -> dict: if (confirm or "").strip() != "DELETE": return {"ok": False, "error": "请输入 DELETE 以确认。"} import shutil data_dir = Path(__file__).resolve().parent / "data" if data_dir.exists(): shutil.rmtree(data_dir) return {"ok": True, "profiles": []} # ============================ 静态前端 (React, frontend/dist) ============================ # 单一前端: frontend/ (Vite + React)。静态数据 (生肖图 / 城市库) 随前端置于 public/data, # 构建后落在 dist/data, 经 /data 提供。后端只用 gradio.Server 托管这份构建产物。 USE_DIST = DIST_DIR.exists() and (DIST_DIR / "index.html").exists() if not USE_DIST: print("\n[警告] 未找到 frontend/dist —— 请先构建前端: cd frontend && pnpm install && pnpm build\n") @app.get("/", response_class=HTMLResponse) async def homepage(): if not USE_DIST: return HTMLResponse("

天问

前端尚未构建。请运行: cd frontend && pnpm install && pnpm build

", status_code=503) return (DIST_DIR / "index.html").read_text(encoding="utf-8") if USE_DIST: # Vite 构建: JS/CSS/字体 -> dist/assets; 静态数据 (生肖图/城市库) -> dist/data app.mount("/assets", StaticFiles(directory=str(DIST_DIR / "assets")), name="assets") if (DIST_DIR / "data").exists(): app.mount("/data", StaticFiles(directory=str(DIST_DIR / "data")), name="data") # PWA 与根静态文件 (sw.js / manifest.webmanifest / workbox-*.js / icon-*.png) def _dist_file_route(path: Path): async def _route(): return FileResponse(str(path)) return _route for _f in DIST_DIR.iterdir(): if _f.is_file() and _f.name != "index.html": app.add_api_route("/" + _f.name, _dist_file_route(_f), methods=["GET"], include_in_schema=False) # Service Worker 预缓存清单含 /index.html, 单独补一条（上面循环刻意跳过它） @app.get("/index.html", response_class=HTMLResponse, include_in_schema=False) async def _index_html(): return (DIST_DIR / "index.html").read_text(encoding="utf-8") def _find_port(start: int = 7860, max_tries: int = 10) -> int: for p in range(start, start + max_tries): with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: try: s.bind(("127.0.0.1", p)) return p except OSError: continue return start def _maybe_start_local_llm() -> None: """进程内起一个 CPU llama.cpp OpenAI 兼容服务于 127.0.0.1:8888， 供 llm_client 调用（用微调后的 Q4 GGUF）。仅当 TIANWEN_SELF_LLM=1 时启用， 后台线程下载模型并拉起，不阻塞 app 启动；模型就绪前解读自动走规则兜底。""" if os.environ.get("TIANWEN_SELF_LLM") != "1": return import threading def _run(): try: import subprocess from huggingface_hub import hf_hub_download repo = os.environ.get("MODEL_REPO", "build-small-hackathon/tianwen-minicpm5-1b") fname = os.environ.get("MODEL_FILE", "tianwen-minicpm1b-q4_k_m.gguf") gguf = hf_hub_download(repo_id=repo, filename=fname) print(f">>> 进程内 llama.cpp (CPU) 加载: {gguf}", flush=True) subprocess.Popen([ sys.executable, "-m", "llama_cpp.server", "--model", gguf, "--host", "127.0.0.1", "--port", "8888", "--n_ctx", "4096", ]) except Exception as e: print(f"自托管 LLM 启动失败（继续用规则兜底）: {e}", flush=True) threading.Thread(target=_run, daemon=True).start() if __name__ == "__main__": _maybe_start_local_llm() # HF Space (有 SPACE_ID) 必须绑 0.0.0.0:7860, 否则平台反代无法探活, 一直 "starting"。 # 本地默认 127.0.0.1 + 自动找空闲端口。 on_space = bool(os.environ.get("SPACE_ID") or os.environ.get("SPACE_HOST")) host = "0.0.0.0" if on_space else "127.0.0.1" port = 7860 if on_space else _find_port(7860, 10) print(f"\n>>> {APP_TITLE} {APP_VERSION} 已启动 -> http://{host}:{port}/\n") # ssr_mode=False: 关掉 Gradio 6 的 Node SSR 代理层 —— 我们自带 React 前端, 不需要它, # 且该代理层在 Space 上容易绑错地址导致探活失败。 app.launch(server_name=host, server_port=port, show_error=True, ssr_mode=False)