demo_app.py

"""Streamlit demo for runescroll.

Run:
    uv run streamlit run demo_app.py

Two tabs:
    🔮 Encode — point at a directory, get a rune PNG (download + preview).
    📜 Decode — drop a rune PNG, browse the recovered file tree (clickable),
        and read any file's contents inline.
"""

from __future__ import annotations

import io
import sys
import tempfile
import time
from pathlib import Path

# Make the package importable when running ``streamlit run`` from the
# project root without installing it.
sys.path.insert(0, str(Path(__file__).resolve().parent / "src"))

import streamlit as st
from PIL import Image

from runescroll import decoder, encoder, pack
from runescroll.spiral import Pattern, PatternParams


# ---------------------------------------------------------------------------
# Page config
# ---------------------------------------------------------------------------

st.set_page_config(
    page_title="压缩符文 · runescroll",
    page_icon="🔮",
    layout="wide",
)

st.markdown(
    """
    # 压缩符文 · `runescroll`
    把一整个项目目录"压"进一枚符文图。一根活着的笔触，多环交织、粗细变化——
    拿到图就能解出整个项目。
    """
)


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

LANGUAGE_BY_EXT = {
    ".py": "python", ".rb": "ruby", ".js": "javascript", ".ts": "typescript",
    ".tsx": "typescript", ".jsx": "javascript", ".go": "go", ".rs": "rust",
    ".java": "java", ".c": "c", ".h": "c", ".cpp": "cpp", ".hpp": "cpp",
    ".cs": "csharp", ".swift": "swift", ".kt": "kotlin", ".scala": "scala",
    ".sh": "bash", ".zsh": "bash", ".bash": "bash", ".fish": "bash",
    ".html": "html", ".css": "css", ".scss": "scss",
    ".json": "json", ".yaml": "yaml", ".yml": "yaml", ".toml": "toml",
    ".sql": "sql", ".md": "markdown", ".lua": "lua", ".php": "php",
    ".dockerfile": "dockerfile", ".rst": "text", ".txt": "text",
    ".gitignore": "text", ".env": "bash",
}

ICON_BY_EXT = {
    ".md": "📝", ".rst": "📝", ".txt": "📝",
    ".py": "🐍", ".rb": "💎", ".js": "📜", ".ts": "📘", ".tsx": "📘",
    ".jsx": "📜", ".go": "🐹", ".rs": "🦀", ".java": "☕",
    ".c": "🔧", ".h": "🔧", ".cpp": "🔧", ".hpp": "🔧",
    ".html": "🌐", ".css": "🎨", ".scss": "🎨",
    ".json": "📦", ".yaml": "📦", ".yml": "📦", ".toml": "📦",
    ".png": "🖼️", ".jpg": "🖼️", ".jpeg": "🖼️", ".gif": "🖼️", ".svg": "🖼️",
    ".sh": "💻", ".zsh": "💻", ".bash": "💻",
    ".lock": "🔒", ".gitignore": "🚫", ".env": "🔐",
}


def _icon(name: str, suffix: str) -> str:
    if name.lower() in {"dockerfile", "makefile"}:
        return "🐳" if name.lower() == "dockerfile" else "🛠️"
    return ICON_BY_EXT.get(suffix.lower(), "📄")


def _guess_language(path: Path) -> str:
    n = path.name.lower()
    if n in {"dockerfile", "makefile"}:
        return n
    return LANGUAGE_BY_EXT.get(path.suffix.lower(), "text")


def _thumb(img: Image.Image, max_dim: int = 1024) -> Image.Image:
    """Downsample big images for animation frames (the page column is
    bounded anyway; PNG-encoding a 3000-px frame per tick is what would
    actually slow the animation down)."""
    if max(img.size) <= max_dim:
        return img
    scale = max_dim / max(img.size)
    return img.resize(
        (int(img.size[0] * scale), int(img.size[1] * scale)),
        Image.LANCZOS,
    )


def _readable_size(n: int) -> str:
    f = float(n)
    for unit in ("B", "KB", "MB"):
        if f < 1024:
            return f"{f:.0f} {unit}" if unit == "B" else f"{f:.1f} {unit}"
        f /= 1024
    return f"{f:.1f} GB"


def _build_tree(files: dict[str, bytes]) -> dict:
    """Return a nested dict: {dirname: {...}, filename: bytes}."""
    tree: dict = {}
    for path, blob in files.items():
        parts = path.split("/")
        node = tree
        for part in parts[:-1]:
            node = node.setdefault(part, {})
            if not isinstance(node, dict):
                # File and directory share a name — shouldn't happen, but
                # guard against it by demoting the file.
                node = {}
        node[parts[-1]] = blob
    return tree


def _render_tree(tree: dict, prefix: str = "", expanded_default: bool = True):
    """Recursively render a tree with expanders for dirs, buttons for files."""
    # Files first sorted alphabetically, then dirs (or vice-versa). Show
    # dirs first so the structure is clearer.
    dirs = sorted((k for k, v in tree.items() if isinstance(v, dict)))
    files = sorted((k for k, v in tree.items() if not isinstance(v, dict)))

    for name in dirs:
        with st.expander(f"📁 **{name}**", expanded=expanded_default):
            _render_tree(
                tree[name],
                prefix=f"{prefix}{name}/",
                expanded_default=expanded_default,
            )

    for name in files:
        full = f"{prefix}{name}"
        blob = tree[name]
        suffix = Path(name).suffix
        icon = _icon(name, suffix)
        size = _readable_size(len(blob))
        is_selected = st.session_state.get("selected_file") == full
        marker = "▶ " if is_selected else "  "
        label = f"{marker}{icon} {name}  · {size}"
        if st.button(label, key=f"file::{full}", use_container_width=True):
            st.session_state["selected_file"] = full


# ---------------------------------------------------------------------------
# Tabs
# ---------------------------------------------------------------------------

tab_encode, tab_decode, tab_about = st.tabs(["🔮 Encode", "📜 Decode", "ℹ️ About"])


# --- Encode -----------------------------------------------------------------

with tab_encode:
    st.subheader("把一个项目目录打成符文")

    col_input, col_output = st.columns([2, 3])

    with col_input:
        src_input = st.text_input(
            "项目目录路径",
            value=str(Path("~/Desktop/cc-learning-hooks").expanduser()),
            help="本地任何目录的绝对路径。.git / node_modules / .venv 会自动跳过。",
        )
        do_pack = st.button("🔮 打包成符文", type="primary", use_container_width=True)

    def _render_encode_panel(img_bytes: bytes, name: str, blob_bytes: int,
                              t_pack_ms: float, t_render_ms: float,
                              animated_already: bool = False):
        """Show the metric strip + download. Image is shown only if we
        haven't already animated it into place above."""
        img = Image.open(io.BytesIO(img_bytes))
        if not animated_already:
            st.image(
                img,
                caption=f"{name} · {img.size[0]}×{img.size[1]}",
                use_container_width=True,
            )
        pattern = Pattern.from_image_size(img.size[0], PatternParams())
        stat_cols = st.columns(4)
        stat_cols[0].metric("压缩后", _readable_size(blob_bytes))
        stat_cols[1].metric("符文边长", f"{img.size[0]} px")
        stat_cols[2].metric("环数", pattern.n_rings)
        stat_cols[3].metric("总笔尖数", f"{pattern.total_atoms:,}")
        t_cols = st.columns(2)
        t_cols[0].caption(f"打包 · {t_pack_ms:.0f} ms")
        t_cols[1].caption(f"渲染 · {t_render_ms:.0f} ms")
        st.download_button(
            "📥 下载符文 PNG",
            data=img_bytes,
            file_name=f"{name}.rune.png",
            mime="image/png",
            use_container_width=True,
        )

    if do_pack:
        src = Path(src_input).expanduser()
        if not src.is_dir():
            with col_input:
                st.error(f"不是目录: {src}")
        else:
            with st.spinner("打包 + zstd 压缩中……"):
                t0 = time.time()
                blob = pack.pack_dir(src)
                t_pack = time.time() - t0

            with col_output:
                placeholder = st.empty()
                t0 = time.time()
                final = None
                # 2 rings per frame ≈ 25–30 frames for typical projects;
                # 50 ms per frame ≈ 1.5 s total animation.
                for frame in encoder.render_frames(blob, rings_per_frame=2):
                    final = frame
                    placeholder.image(_thumb(frame, 1024), use_container_width=True)
                    time.sleep(0.05)
                t_render = time.time() - t0
                # Final frame at full resolution.
                placeholder.image(final, use_container_width=True)

            with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as f:
                out_path = Path(f.name)
            assert final is not None
            final.save(out_path, format="PNG", optimize=True)
            rune_bytes = out_path.read_bytes()

            st.session_state["last_rune_bytes"] = rune_bytes
            st.session_state["last_rune_name"] = src.name
            st.session_state["last_rune_size"] = final.size
            st.session_state["last_blob_bytes"] = len(blob)
            st.session_state["last_pack_ms"] = t_pack * 1000
            st.session_state["last_render_ms"] = t_render * 1000

            with col_output:
                _render_encode_panel(
                    rune_bytes, src.name, len(blob),
                    t_pack * 1000, t_render * 1000,
                    animated_already=True,
                )

    elif "last_rune_bytes" in st.session_state:
        with col_output:
            _render_encode_panel(
                st.session_state["last_rune_bytes"],
                st.session_state["last_rune_name"],
                st.session_state["last_blob_bytes"],
                st.session_state["last_pack_ms"],
                st.session_state["last_render_ms"],
                animated_already=False,
            )


# --- Decode -----------------------------------------------------------------

with tab_decode:
    st.subheader("把符文还原成项目")

    upload_col, browse_col = st.columns([2, 3])

    with upload_col:
        uploaded = st.file_uploader(
            "拖一张符文 PNG 进来", type=["png"], key="decode_upload"
        )
        st.caption("或者用刚刚 Encode 出来的那张：")
        use_last = st.button(
            "↩️ 用刚才生成的",
            use_container_width=True,
            disabled="last_rune_bytes" not in st.session_state,
        )

        # Decide which bytes to decode (if any), and only re-run the decode
        # when the bytes actually change. The result is cached in
        # session_state so clicking files in the tree doesn't re-decode.
        new_bytes: bytes | None = None
        new_label: str | None = None
        if use_last and "last_rune_bytes" in st.session_state:
            new_bytes = st.session_state["last_rune_bytes"]
            new_label = "from-encoder.png"
        elif uploaded is not None:
            new_bytes = uploaded.getvalue()
            new_label = uploaded.name

        if new_bytes is not None and new_bytes != st.session_state.get("decoded_bytes"):
            with st.spinner("正在测量笔触粗细，校准 bins，恢复字节流……"):
                t0 = time.time()
                with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as f:
                    f.write(new_bytes)
                    rune_path = Path(f.name)
                try:
                    blob = decoder.decode_from_image(rune_path)
                    elapsed = time.time() - t0
                    with tempfile.TemporaryDirectory() as td:
                        dest = Path(td) / "extracted"
                        pack.unpack_to_dir(blob, dest)
                        files = sorted(p for p in dest.rglob("*") if p.is_file())
                        file_blobs = {
                            p.relative_to(dest).as_posix(): p.read_bytes()
                            for p in files
                        }
                    st.session_state["decoded_bytes"] = new_bytes
                    st.session_state["decoded_label"] = new_label
                    st.session_state["decoded_files"] = file_blobs
                    st.session_state["decoded_elapsed_ms"] = elapsed * 1000
                    st.session_state["selected_file"] = (
                        next(iter(file_blobs), None) if file_blobs else None
                    )
                except Exception as exc:
                    st.session_state["decode_error"] = str(exc)
                    st.session_state.pop("decoded_files", None)

        if "decoded_bytes" in st.session_state:
            img = Image.open(io.BytesIO(st.session_state["decoded_bytes"]))
            st.image(
                img,
                caption=f"{st.session_state['decoded_label']} · {img.size[0]}×{img.size[1]}",
                use_container_width=True,
            )

        if "decode_error" in st.session_state and "decoded_files" not in st.session_state:
            st.error(f"解码失败: {st.session_state['decode_error']}")

    with browse_col:
        if "decoded_files" not in st.session_state:
            st.info("解码完成后，文件树会出现在这里。")
        else:
            files = st.session_state["decoded_files"]
            total_size = sum(len(v) for v in files.values())
            st.success(
                f"解码 + 解压 · {st.session_state['decoded_elapsed_ms']:.0f} ms · "
                f"**{len(files)}** 个文件 · 共 **{_readable_size(total_size)}**"
            )

            tree_col, view_col = st.columns([2, 3])
            with tree_col:
                st.markdown("**📂 目录结构**")
                tree = _build_tree(files)
                _render_tree(tree, expanded_default=True)

            with view_col:
                selected = st.session_state.get("selected_file")
                if selected and selected in files:
                    content = files[selected]
                    st.markdown(f"**`{selected}`** · {_readable_size(len(content))}")
                    try:
                        text = content.decode("utf-8")
                    except UnicodeDecodeError:
                        st.warning("二进制文件，下面是前 256 字节的 hex dump。")
                        chunk = content[:256]
                        lines = []
                        for i in range(0, len(chunk), 16):
                            row = chunk[i : i + 16]
                            hex_part = " ".join(f"{b:02x}" for b in row)
                            lines.append(f"{i:08x}  {hex_part}")
                        st.code("\n".join(lines), language="text")
                    else:
                        # Long markdown sometimes hangs Streamlit's syntax
                        # highlighter; render as plain text past a threshold.
                        lang = _guess_language(Path(selected))
                        if len(text) > 200_000:
                            st.text(text[:200_000])
                            st.caption(
                                f"显示了前 200 KB，文件总长 {_readable_size(len(content))}"
                            )
                        else:
                            st.code(text, language=lang)

                    st.download_button(
                        "📥 单独下载这个文件",
                        data=content,
                        file_name=Path(selected).name,
                        use_container_width=True,
                    )
                else:
                    st.caption("← 在左边点一个文件来查看")


# --- About ------------------------------------------------------------------

with tab_about:
    st.subheader("怎么工作的")
    st.markdown("""
    1. **打包**：tar 整个目录 + zstd 压缩 → 一个字节流
    2. **加冗余**：Reed-Solomon 给字节流加上奇偶校验，header 写 3 份
    3. **比特流 → 笔尖**：每 2 比特 = 一个笔尖等级（4 档粗细：细丝、半细、半粗、粗墨）
    4. **画**：在数十圈同心环上沿环走，每个采样点用 polygon 画一个粗细对应的小矩形；
       每环各自带一点正弦扰动（不同相位错开），视觉上看起来像编织
    5. **装饰**：中央八角红印，外圈 24 个程序化卢恩字符（其中四个固定的"finder rune"是
       未来摄像头扫码用的方位锚点）

    解码反过来：从图重建螺旋几何 → 实测每个采样点的笔触粗细 → 校准 bins 吃掉
    PIL 多边形光栅化偏差 → 量化回比特 → ECC 修复 → CRC 验证 → zstd 解压 → tar 还原。
    """)


# ---------------------------------------------------------------------------
# Footer
# ---------------------------------------------------------------------------

st.markdown("---")
st.caption(
    "Pre-loaded examples that work out of the box: "
    "`~/Desktop/cc-learning-hooks`, `~/Desktop/gintemplate/repo`."
)