app.py

import gradio as gr
import os

# ===== 你的自定义模块 =====
from config import APP_TITLE, CHATBOT_HEIGHT, MIN_WIDTH_LEFT, MIN_WIDTH_RIGHT
from styles import MAHJONG_THEME_CSS

# Extra CSS to force dark & counter browser auto-invert
EXTRA_FIX_CSS = '''/* === Force dark site and neutralize browser-side auto-invert === */
html, body{ background-color:#0b1220 !important; color-scheme: dark !important; }
/* When browser/extension applies global invert, add a counter-invert on our app root */
html.force-dark-fix #app-root{ filter: invert(1) hue-rotate(180deg) !important; }'''
# Redirect to ?__theme=dark
FORCE_DARK_REDIRECT = '''<script>
(function(){
  try{
    const url=new URL(window.location.href);
    if(url.searchParams.get("__theme")!=="dark"){
      url.searchParams.set("__theme","dark");
      window.location.replace(url.toString());
    }
  }catch(e){}
})();
</script>'''
# Detect browser force-dark and set html class
DETECT_FORCE_DARK = '''<script>
(function(){
  try{
    const html=document.documentElement;
    const cs=getComputedStyle(html);
    const hasGlobalFilter=(cs.filter && cs.filter!=="none");
    const darkReaderOn=!!document.querySelector('style#dark-reader-style')||!!document.querySelector('meta[name="darkreader"]');
    if(hasGlobalFilter||darkReaderOn){ html.classList.add('force-dark-fix'); }
  }catch(e){}
})();
</script>'''
from ai_service import design_mahjong_game
# 优先尝试原生流式；没有则自动回退为伪流式
try:
    from ai_service import design_mahjong_game_stream
except Exception:
    design_mahjong_game_stream = None


# ====== 🔧 Hotfix: 兼容 gradio_client 对 additionalProperties(bool) 的解析 ======
try:
    import gradio_client.utils as _gc_utils

    _orig_get_type = _gc_utils.get_type
    def _safe_get_type(schema):
        if isinstance(schema, bool):
            return "any" if schema else "never"
        return _orig_get_type(schema)
    _gc_utils.get_type = _safe_get_type

    _orig_json2py = _gc_utils._json_schema_to_python_type
    def _safe_json2py(schema, defs):
        if isinstance(schema, bool):
            return "any" if schema else "never"
        if isinstance(schema, dict) and "additionalProperties" in schema:
            ap = schema["additionalProperties"]
            if isinstance(ap, bool):
                inner = "any" if ap else "never"
                return f"dict[str, {inner}]"
        return _orig_json2py(schema, defs)
    _gc_utils._json_schema_to_python_type = _safe_json2py

    _orig_json_schema_to_python_type = _gc_utils.json_schema_to_python_type
    def _safe_json_schema_to_python_type(schema):
        if isinstance(schema, bool):
            return "any" if schema else "never"
        return _orig_json_schema_to_python_type(schema)
    _gc_utils.json_schema_to_python_type = _safe_json_schema_to_python_type
except Exception:
    pass
# ====== 🔧 Hotfix 结束 ======


# ==================== 工具函数 ====================
def _messages_to_tuples(history):
    """
    将 Chatbot 的 messages 格式（[{role, content}, ...]）转换为 [(user, bot), ...]。
    若已是 tuples，则原样返回。
    """
    if not history:
        return []

    if isinstance(history, list) and history and isinstance(history[0], dict):
        pairs = []
        last_user = None
        for msg in history:
            role = msg.get("role")
            content = msg.get("content", "")
            if role == "user":
                last_user = content
            elif role == "assistant":
                pairs.append((last_user or "", content))
                last_user = None
        return pairs

    return history


def _chunk_fake_stream(text, step=40):
    """把整段文本切成小块，伪流式输出。"""
    s = str(text or "")
    for i in range(0, len(s), step):
        yield s[i:i + step]


def _load_base64_image(path):
    """读取本地图片并返回 data URI"""
    import base64
    import pathlib

    p = pathlib.Path(path)
    if not p.exists():
        return ""
    try:
        data = base64.b64encode(p.read_bytes()).decode("ascii")
        suffix = p.suffix.lower()
        mime = "image/png" if suffix == ".png" else "image/jpeg"
        return f"data:{mime};base64,{data}"
    except Exception:
        return ""


HERO_TILE_PATHS = ["UI00001.jpg", "UI00002.jpg", "UI00003.jpg", "UI00004.jpg"]
HERO_TILE_IMAGES = [_load_base64_image(p) for p in HERO_TILE_PATHS]


def _render_hero_tiles():
    tiles = [
        f'<span class="hero-photo" style="background-image:url(\'{src}\');" title="参考图 {idx + 1}"></span>'
        for idx, src in enumerate(HERO_TILE_IMAGES)
        if src
    ]
    return "".join(tiles)


def clear_cache():
    """清空缓存"""
    from cache_manager import file_cache, request_cache
    file_cache.clear()
    request_cache.clear()
    return "✅ 缓存已清空"


def clear_files():
    """清空文件上传"""
    return None


def update_file_status(files):
    """更新文件状态显示"""
    if not files:
        return "📁 文件状态：未上传"
    file_list = files if isinstance(files, (list, tuple)) else [files]
    names = []
    for f in file_list:
        if isinstance(f, str):
            names.append(os.path.basename(f))
        else:
            names.append(os.path.basename(getattr(f, "name", str(f))))
    head = "\n".join(f"  • {n}" for n in names[:3])
    tail = "\n  ..." if len(names) > 3 else ""
    return f"📁 文件状态：已上传 {len(names)} 个文件\n{head}{tail}"


def export_history_to_markdown(history):
    """将聊天历史导出为 Markdown 文件，并返回文件路径供下载"""
    import time
    import pathlib
    from datetime import datetime

    pairs = _messages_to_tuples(history)

    lines = ["# 对话记录", ""]
    lines.append(f"- 导出时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    lines.append("")
    for idx, (user_msg, bot_msg) in enumerate(pairs, start=1):
        lines.append(f"## 轮次 {idx}")
        if user_msg:
            lines.append("**用户**:\n")
            lines.append(user_msg)
            lines.append("")
        if bot_msg:
            lines.append("**助手**:\n")
            lines.append(bot_msg)
            lines.append("")

    content = "\n".join(lines).strip() + "\n"

    export_dir = pathlib.Path("exports")
    export_dir.mkdir(parents=True, exist_ok=True)
    filename = export_dir / f"chat_history_{int(time.time())}.md"
    with open(filename, "w", encoding="utf-8") as f:
        f.write(content)
    return str(filename)

# ==================== 新增：提取 GDL 和自然语言描述 ====================
def extract_gdl_and_narrative(content):
    """提取 GDL 和自然语言部分（支持多种格式变体）"""
    import re
    
    # 定义多种可能的标记格式（按优先级排序，支持更多变体）
    gdl_patterns = [
        r"##\s*m?GDL\s*描述",          # ## GDL描述 / ## mGDL 描述
        r"m?GDL\s*描述",               # GDL描述 / mGDL 描述
        r"##\s*GDL\s*Description",    # ## GDL Description（英文）
        r"##\s*m?GDL",                 # ## GDL / ## mGDL
        r"m?GDL\s*规则代码",            # mGDL 规则代码
        r"m?GDL\s*规则",               # mGDL 规则
    ]
    
    narrative_patterns = [
        r"##\s*自然语言规则说明",      # ## 自然语言规则说明
        r"##\s*自然语言规则",          # ## 自然语言规则
        r"自然语言规则说明",            # 自然语言规则说明
        r"自然语言规则",                # 自然语言规则
        r"##\s*规则说明",              # ## 规则说明
        r"规则说明",                    # 规则说明
    ]
    
    # 尝试查找 GDL 部分（支持大小写不敏感）
    gdl_start = -1
    for pattern in gdl_patterns:
        match = re.search(pattern, content, re.IGNORECASE)
        if match:
            gdl_start = match.start()
            break
    
    # 尝试查找自然语言部分（支持大小写不敏感）
    narrative_start = -1
    for pattern in narrative_patterns:
        match = re.search(pattern, content, re.IGNORECASE)
        if match:
            narrative_start = match.start()
            break
    
    if gdl_start != -1 and narrative_start != -1:
        # 确保顺序正确（GDL应该在自然语言之前）
        if gdl_start >= narrative_start:
            gdl_start, narrative_start = narrative_start, gdl_start
        
        # 获取 GDL 部分（从标记开始到自然语言标记之前）
        gdl_content = content[gdl_start:narrative_start].strip()
        
        # 获取自然语言部分（从标记开始到结尾）
        narrative_content = content[narrative_start:].strip()
        
        return gdl_content, narrative_content
    
    elif gdl_start != -1:
        # 只找到GDL，将后面全部作为GDL
        print(f"⚠️ 仅找到GDL标记，将其后内容作为GDL")
        return content[gdl_start:].strip(), ""
    
    elif narrative_start != -1:
        # 只找到自然语言，将后面全部作为自然语言
        print(f"⚠️ 仅找到自然语言标记，将其后内容作为自然语言")
        return "", content[narrative_start:].strip()
    
    else:
        # 都没找到，返回空
        print(f"⚠️ 提取警告: 未找到GDL和自然语言标记")
        return "", ""

def save_gdl_and_narrative(gdl_content, narrative_content):
    """保存 GDL 和自然语言内容到文件"""
    # 定义文件存储路径
    export_dir = os.path.join("exports")
    os.makedirs(export_dir, exist_ok=True)
    
    gdl_file_path = os.path.join(export_dir, "gdl_output.txt")
    narrative_file_path = os.path.join(export_dir, "narrative_output.txt")

    # 保存 GDL
    with open(gdl_file_path, "w", encoding="utf-8") as f:
        f.write(gdl_content)

    # 保存自然语言
    with open(narrative_file_path, "w", encoding="utf-8") as f:
        f.write(narrative_content)

    return gdl_file_path, narrative_file_path


# ==================== Gradio 界面（Mahjong Skin） ====================
with gr.Blocks(
    theme=gr.themes.Soft(primary_hue='green', neutral_hue='slate'),
    title=APP_TITLE,
    css=MAHJONG_THEME_CSS + EXTRA_FIX_CSS,
    elem_id='app-root'
) as demo:
    gr.HTML(FORCE_DARK_REDIRECT)
    gr.HTML(DETECT_FORCE_DARK)
    # 顶部 Hero
    hero_tiles_html = _render_hero_tiles()
    gr.HTML(f"""
    <div class="hero">
      <div style="display:flex;align-items:center;gap:12px;">
        <div style="font-size:30px;">🀄️ 麻将玩法灵感工坊 · KOI Lab</div>
      </div>
      <div style="margin-top:6px;opacity:.92;">与 AI 麻将策划智囊对话，快速生成番型体系与玩法流程</div>
      <div class="hero-icons" aria-hidden="true">
        {hero_tiles_html or '<span class="hero-photo"></span>'}
      </div>
    </div>
    """)

    with gr.Row(equal_height=True):
        # 左侧：设置区
        with gr.Column(scale=1, min_width=MIN_WIDTH_LEFT):
            with gr.Group(elem_classes="side-card"):
                gr.Markdown("### 输入与约束")
                gr.Markdown("✅ **系统已预加载 Mahjong GDL 规范与提示词**，可直接开局。", elem_classes="hint")
                file_uploader = gr.File(
                    label="上传自定义 Mahjong GDL/番型示例（.txt，可多选）",
                    file_types=[".txt"],
                    file_count="multiple",
                    type="filepath",
                    interactive=True,
                    show_label=True,
                    container=True,
                    scale=1
                )
                custom_prompt_box = gr.Textbox(
                    label="自定义 System Prompt（可选）",
                    placeholder="系统已经载入麻将玩法专家提示词。如需特殊规则，可在此粘贴。",
                    lines=10,
                    max_lines=20,
                    show_copy_button=True
                )
                prompt_mode = gr.Radio(
                    choices=["覆盖默认SYSTEM_PROMPT", "合并（默认在前）"],
                    value="覆盖默认SYSTEM_PROMPT",
                    label="自定义 Prompt 的使用方式"
                )
                gr.Markdown("💡 提示：系统默认包含麻将通用语法、番型模板与 Prompt；仅当你要替换地方规则时再上传或输入。", elem_classes="hint")

            with gr.Group(elem_classes="side-card"):
                gr.Markdown("### 使用说明")
                gr.Markdown(
                    "- ✅ **快速开始**：直接描述你想要的麻将玩法主题或目标。\n"
                    "- 📁 **可选上传**：导入地方番型表或自定义 GDL 模板以便引用。\n"
                    "- ✏️ **可选自定义**：针对特殊风格重写 System Prompt。\n"
                    "- 🔄 **清空缓存**：切换规则后建议刷新缓存，避免残留设定。"
                )

            with gr.Group(elem_classes="side-card"):
                gr.Markdown("### 快捷操作")
                with gr.Row():
                    clear_cache_btn = gr.Button("清空缓存", variant="secondary")
                    clear_files_btn = gr.Button("清空文件", variant="secondary")
                cache_status = gr.Markdown("💾 缓存状态：正常", elem_classes="hint")
                file_status = gr.Markdown("📁 文件状态：使用默认麻将配置（GDL + Prompt 已预加载）", elem_classes="hint")

        # 右侧：聊天区（手动事件绑定 + 流式输出）
        with gr.Column(scale=2, min_width=MIN_WIDTH_RIGHT):
            with gr.Group(elem_classes="table"):
                chatbot = gr.Chatbot(
                    height=CHATBOT_HEIGHT,
                    type="messages",
                    elem_classes="custom-chatbot",
                    avatar_images=("landlord.png", "bot.png"),
                )
                # 用 State 保存 messages 历史
                chat_state = gr.State([])  # list[dict]: [{"role":"user","content":...}, {"role":"assistant","content":...}]
                gdl_file_path = gr.State("")
                narrative_file_path = gr.State("")

                user_input = gr.Textbox(
                    placeholder="例如：为4人竞速推倒胡设计番型与流程；或“做一个双人合作麻将 roguelike” …",
                    show_label=False,
                    max_lines=5,
                    lines=3,
                    show_copy_button=True,
                    autofocus=True,
                )

                with gr.Row():
                    send_btn = gr.Button("发送", variant="primary")
                    stop_info = gr.Markdown("", visible=False)

                # ====== 核心：流式提交回调（生成器） ======
                def on_submit(user_text, history_msgs, files, custom_prompt, mode):
                    user_text = (user_text or "").strip()
                    if not user_text:
                        # 不提交空消息：输出不变
                        yield history_msgs, "", history_msgs, "", ""  # 🟩【修改】输出数量改为5个
                        return

                    # 立即显示“用户消息”
                    history = list(history_msgs or [])
                    history.append({"role": "user", "content": user_text})
                    yield history, "", history, "", ""

                    # 添加空的助手气泡，用于逐步填充
                    history.append({"role": "assistant", "content": ""})
                    yield history, "", history, "", ""

                    # 流式生成内容
                    tuples_hist = _messages_to_tuples(history)

                    if design_mahjong_game_stream is not None:
                        try:
                            for piece in design_mahjong_game_stream(user_text, tuples_hist, files, custom_prompt, mode):
                                if not piece:
                                    continue
                                history[-1]["content"] += str(piece)
                                yield history, "", history, "", ""
                        except Exception as e:
                            history[-1]["content"] += f"\n（流式出错）{type(e).__name__}: {e}"
                            yield history, "", history, "", ""
                    else:
                        try:
                            full = design_mahjong_game(user_text, tuples_hist, files, custom_prompt, mode)
                        except Exception as e:
                            full = f"（出错）{type(e).__name__}: {e}"
                        for piece in _chunk_fake_stream(str(full), step=40):
                            history[-1]["content"] += piece
                            yield history, "", history, "", ""

                    # 提取 GDL 和自然语言描述并保存
                    try:
                        gdl_content, narrative_content = extract_gdl_and_narrative(history[-1]["content"])
                        
                        # 保存 GDL 和自然语言文件
                        gdl_path, narrative_path = save_gdl_and_narrative(gdl_content, narrative_content)
                        
                        # 返回文件路径，以便下载
                        yield history, "", history, gdl_path, narrative_path  # 🟩【修改】返回文件路径
                    except Exception as e:
                        print(f"保存GDL和自然语言文件时出错: {e}")
                        yield history, "", history, "", ""  # 🟩【修改】返回空路径

                # 绑定：回车提交（Enter=提交；Shift+Enter=换行由浏览器处理）
                user_input.submit(
                    fn=on_submit,
                    inputs=[user_input, chat_state, file_uploader, custom_prompt_box, prompt_mode],
                    outputs=[chatbot, user_input, chat_state, gdl_file_path, narrative_file_path],  # 🟩【修改】输出改为5个
                    preprocess=True,
                )

                # 绑定：点击"发送"
                send_btn.click(
                    fn=on_submit,
                    inputs=[user_input, chat_state, file_uploader, custom_prompt_box, prompt_mode],
                    outputs=[chatbot, user_input, chat_state, gdl_file_path, narrative_file_path],  # 🟩【修改】输出改为5个
                    preprocess=True,
                )

                # 导出对话
                with gr.Row():
                    export_btn = gr.Button("导出对话（Markdown）", variant="secondary")
                    export_file = gr.File(label="点击下载导出文件", interactive=False)

                # 清空对话
                with gr.Row():
                    clear_dialog_btn = gr.Button("清空对话", variant="secondary")

                def _clear_chat():
                    return [], "", [], "", "" 

                clear_dialog_btn.click(
                    fn=_clear_chat,
                    inputs=None,
                    outputs=[chatbot, user_input, chat_state, gdl_file_path, narrative_file_path],  # 🟩【修改】输出改为5个
                )

                # ==================== 下载按钮部分 ====================
                with gr.Row():
                    gr.Markdown("### 下载 GDL 和自然语言描述")
                    download_gdl_btn = gr.Button("下载 GDL 文件", variant="secondary")  # 下载 GDL 按钮
                    download_narrative_btn = gr.Button("下载自然语言文件", variant="secondary")  # 下载自然语言按钮

                    download_gdl_file = gr.File(label="GDL 文件", interactive=False)  # 文件下载区域
                    download_narrative_file = gr.File(label="自然语言文件", interactive=False)  # 文件下载区域

                # 🟩【修改】绑定下载按钮与文件路径 - 修复版本
                def get_gdl_file(gdl_path):
                    if gdl_path and os.path.exists(gdl_path):
                        return gdl_path
                    return None

                def get_narrative_file(narrative_path):
                    if narrative_path and os.path.exists(narrative_path):
                        return narrative_path
                    return None

                # 绑定下载按钮与文件路径
                download_gdl_btn.click(
                    fn=get_gdl_file,
                    inputs=[gdl_file_path],  # 🟩【修改】接收State中的路径
                    outputs=[download_gdl_file]
                )

                download_narrative_btn.click(
                    fn=get_narrative_file,
                    inputs=[narrative_file_path],  # 🟩【修改】接收State中的路径
                    outputs=[download_narrative_file]
                )

    # ==================== 事件绑定（左侧） ====================
    clear_cache_btn.click(fn=clear_cache, outputs=[cache_status])
    clear_files_btn.click(fn=clear_files, outputs=[file_uploader])
    file_uploader.change(fn=update_file_status, inputs=[file_uploader], outputs=[file_status])

    # 导出按钮
    def _export_wrapper(chat_history):
        try:
            path = export_history_to_markdown(chat_history)
            return path
        except Exception as e:
            import tempfile
            tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".md")
            with open(tmp.name, 'w', encoding='utf-8') as f:
                f.write(f"导出失败：{type(e).__name__}: {e}\n")
            return tmp.name

    export_btn.click(fn=_export_wrapper, inputs=[chatbot], outputs=[export_file])


# ==================== 启动应用 ====================
if __name__ == "__main__":
    # 兼容不同 gradio 版本：
    # - 有的版本 queue() 不接受 concurrency_count/status_update_rate
    # - 有的版本甚至不需要手动 queue()
    app = demo
    try:
        app = demo.queue()   # 不带参数，启用队列以支持生成器流式
    except TypeError:
        # 某些版本 queue() 可能参数或签名不同，直接跳过即可
        app = demo
    app.launch(share=True, show_api=False)