app.py

# -*- coding: utf-8 -*-
"""
Gradio Web UI for GDL -> IR(v0.95) pipeline

功能：
- 上传 GDL 文本文件（.txt/.gdl）与可选 Schema（默认自动发现 poker_gdl_ir.schema.v0.95.zh.json）
- 运行 parse -> normalize -> map_to_v095 -> validate -> report 全流程
- 在线预览 IR(JSON)、Issues(JSON)、Report(Markdown)
- 支持将结果作为文件下载
"""
from __future__ import annotations

import io
import json
import pathlib
from typing import Tuple, Any
import tempfile

import gradio as gr

from gdl_parser_v2 import parse_gdl
from normalizer_v2 import normalize_ir
from mapper_v2 import map_to_v095
from validator_v2 import validate_with_schema
from gap_detector_v2 import make_report


def _auto_discover_schema(cli_arg: str | None) -> pathlib.Path:
    candidates = []
    if cli_arg:
        candidates.append(cli_arg)
    here = pathlib.Path(__file__).resolve().parent
    candidates.append(str(here / "poker_gdl_ir.schema.v0.95.zh.json"))
    candidates.append("poker_gdl_ir.schema.v0.95.zh.json")
    candidates.append("/mnt/data/poker_gdl_ir.schema.v0.95.zh.json")
    for c in candidates:
        try:
            p = pathlib.Path(c)
            if p.exists():
                return p
        except Exception:
            pass
    raise FileNotFoundError("未找到 Schema。请上传或将 poker_gdl_ir.schema.v0.95.zh.json 放在程序目录/当前目录。")


def _read_as_bytes(maybe_file: Any) -> bytes:
    # Gradio File with type="binary" returns bytes directly.
    if maybe_file is None:
        raise gr.Error("未提供文件")
    if isinstance(maybe_file, (bytes, bytearray)):
        return bytes(maybe_file)
    if isinstance(maybe_file, str):  # filepath
        return pathlib.Path(maybe_file).read_bytes()
    if hasattr(maybe_file, "read") and callable(getattr(maybe_file, "read")):
        return maybe_file.read()
    if isinstance(maybe_file, dict):
        data = maybe_file.get("data")
        if isinstance(data, (bytes, bytearray)):
            return bytes(data)
    raise gr.Error("无法读取上传的文件内容")


def run_pipeline(gdl_file: Any, schema_file: Any | None) -> Tuple[str, str, str, bytes, bytes, bytes]:
    if gdl_file is None:
        raise gr.Error("请先上传 GDL 文件")

    # 读取 GDL 文本
    gdl_bytes = _read_as_bytes(gdl_file)
    try:
        gdl_txt = gdl_bytes.decode("utf-8")
    except Exception:
        # 尝试 gbk / latin-1 以容错
        for enc in ("utf-8-sig", "gbk", "latin-1"):
            try:
                gdl_txt = gdl_bytes.decode(enc)
                break
            except Exception:
                continue
        else:
            raise gr.Error("无法解码 GDL 文本，请确认编码为 UTF-8")

    # 解析 -> 规整 -> 映射
    gdl_ast = parse_gdl(gdl_txt)
    nz = normalize_ir(gdl_ast, gdl_txt)
    ir = map_to_v095(nz, gdl_txt)

    # Schema 路径：优先使用上传，其次自动发现
    if schema_file is not None:
        try:
            schema_bytes = _read_as_bytes(schema_file)
            tmp = tempfile.NamedTemporaryFile(suffix=".json", delete=False)
            tmp.write(schema_bytes)
            tmp.flush()
            tmp.close()
            schema_path = pathlib.Path(tmp.name)
        except Exception:
            raise gr.Error("读取 Schema 失败，请确认文件格式正确")
    else:
        try:
            schema_path = _auto_discover_schema(None)
        except FileNotFoundError as e:
            raise gr.Error(str(e))

    # 校验（兼容 2/3 元组）
    ret = validate_with_schema(ir, str(schema_path))
    if isinstance(ret, tuple) and len(ret) == 3:
        ok, issues, schema_msg = ret
    elif isinstance(ret, tuple) and len(ret) == 2:
        ok, issues = ret
        schema_msg = f"Schema: {'OK' if ok else 'FAIL'}; issues={len(issues)}"
    else:
        ok = bool(ret)
        issues = []
        schema_msg = f"Schema: {'OK' if ok else 'FAIL'}"

    report_md = make_report(ir)

    ir_json_str = json.dumps(ir, ensure_ascii=False, indent=2)
    issues_json_str = json.dumps(issues, ensure_ascii=False, indent=2)

    # 准备下载的文件内容
    ir_bytes = ir_json_str.encode("utf-8")
    issues_bytes = issues_json_str.encode("utf-8")
    report_bytes = report_md.encode("utf-8")

    return (
        ir_json_str,
        issues_json_str,
        report_md,
        ir_bytes,
        issues_bytes,
        report_bytes,
    )


def _save_temp(name_hint: str, data: bytes) -> str:
    suffix = pathlib.Path(name_hint).suffix or ""
    tmp = tempfile.NamedTemporaryFile(delete=False, suffix=suffix)
    tmp.write(data)
    tmp.flush()
    tmp.close()
    return tmp.name


with gr.Blocks(title="GDL → IR(v0.95) Web UI") as demo:
    gr.Markdown("""
    **GDL → IR(v0.95) 一键转换与校验**  
    - 上传 GDL 文本文件（.txt/.gdl）  
    - 可选上传 Schema（若不上传，将自动发现同目录中的 `poker_gdl_ir.schema.v0.95.zh.json`）  
    - 生成：IR(JSON)、Issues(JSON)、自检报告(Markdown)
    """
    )

    with gr.Row():
        gdl_in = gr.File(label="GDL 文件 (.txt/.gdl)", file_types=[".txt", ".gdl"], type="binary")
        schema_in = gr.File(label="Schema 文件 (可选)", file_types=[".json"], type="binary")

    run_btn = gr.Button("运行转换与校验")

    with gr.Tab("IR 预览"):
        ir_json = gr.Code(label="IR (v0.95)", language="json")
        ir_dl = gr.File(label="下载 IR JSON", interactive=False)

    with gr.Tab("Issues 预览"):
        issues_json = gr.Code(label="Schema 校验问题", language="json")
        issues_dl = gr.File(label="下载 Issues JSON", interactive=False)

    with gr.Tab("自检报告"):
        report_md = gr.Markdown()
        report_dl = gr.File(label="下载 自检报告.md", interactive=False)

    def _on_click(gdl_file, schema_file):
        ir_str, issues_str, report_str, ir_b, issues_b, report_b = run_pipeline(gdl_file, schema_file)
        ir_path = _save_temp("ir.v0.95.json", ir_b)
        issues_path = _save_temp("issues.json", issues_b)
        report_path = _save_temp("selfcheck.md", report_b)
        return ir_str, ir_path, issues_str, issues_path, report_str, report_path

    run_btn.click(
        _on_click,
        inputs=[gdl_in, schema_in],
        outputs=[ir_json, ir_dl, issues_json, issues_dl, report_md, report_dl]
    )

if __name__ == "__main__":
    demo.launch(share=True)