app.py

import html
import json
import re
import tempfile
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

import gradio as gr
import pandas as pd
import plotly.express as px
import plotly.graph_objects as go
from docx import Document
from pypdf import PdfReader


STRUCTURED_EXTENSIONS = {".csv", ".tsv", ".xlsx", ".xls", ".json", ".jsonl"}
TEXT_EXTENSIONS = {".txt", ".md", ".log"}
DOCUMENT_EXTENSIONS = {".docx", ".pdf"}
SUPPORTED_EXTENSIONS = STRUCTURED_EXTENSIONS | TEXT_EXTENSIONS | DOCUMENT_EXTENSIONS
EXTENSION_HINTS = {
    "text/csv": ".csv",
    "text/plain": ".txt",
    "text/markdown": ".md",
    "application/json": ".json",
    "application/pdf": ".pdf",
    "application/vnd.openxmlformats-officedocument.wordprocessingml.document": ".docx",
    "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet": ".xlsx",
    "application/vnd.ms-excel": ".xls",
}
APP_CSS = """
.gradio-container { background: #0f172a; }
#hero {
  border: 1px solid rgba(148, 163, 184, 0.25);
  border-radius: 24px;
  padding: 24px;
  background: linear-gradient(135deg, rgba(37, 99, 235, 0.22), rgba(15, 23, 42, 0.9));
}
"""
PAGE_MARKER_PATTERN = re.compile(r"^---\s*Page\s+\d+\s*---$", re.IGNORECASE)
LIST_START_PATTERN = re.compile(r"^[\"'“”‘’]*\s*(?:[●•*-]|\d+[.)])\s+")


def _file_path(file_value: Any) -> Path:
    if isinstance(file_value, dict):
        for key in ("path", "name"):
            value = file_value.get(key)
            if value:
                return Path(value)
    if isinstance(file_value, (str, Path)):
        return Path(file_value)
    if hasattr(file_value, "name"):
        return Path(file_value.name)
    raise ValueError("Unsupported upload value.")


def _display_name(file_value: Any, path: Path) -> str:
    if isinstance(file_value, dict):
        return str(file_value.get("orig_name") or file_value.get("name") or path.name)
    return path.name


def _file_suffix(file_value: Any, path: Path) -> str:
    suffix = path.suffix.lower()
    display_suffix = Path(_display_name(file_value, path)).suffix.lower()
    if display_suffix:
        return display_suffix
    if suffix:
        return suffix
    if isinstance(file_value, dict):
        mime_type = str(file_value.get("mime_type") or file_value.get("type") or "").lower()
        return EXTENSION_HINTS.get(mime_type, "")
    return ""


def _read_structured_file(path: Path) -> pd.DataFrame:
    suffix = path.suffix.lower()

    if suffix == ".csv":
        return pd.read_csv(path)
    if suffix == ".tsv":
        return pd.read_csv(path, sep="\t")
    if suffix in {".xlsx", ".xls"}:
        return pd.read_excel(path)
    if suffix == ".jsonl":
        return pd.read_json(path, lines=True)
    if suffix == ".json":
        data = json.loads(path.read_text(encoding="utf-8"))
        if isinstance(data, list):
            return pd.json_normalize(data)
        if isinstance(data, dict):
            return pd.json_normalize(data)

    raise ValueError(f"{path.name} is not a supported structured data file.")


def _path_with_suffix(path: Path, suffix: str) -> Path:
    if suffix == path.suffix.lower():
        return path

    copied_path = Path(tempfile.mkdtemp(prefix="hf_app_upload_")) / f"upload{suffix}"
    copied_path.write_bytes(path.read_bytes())
    return copied_path


def _read_structured_file_by_suffix(path: Path, suffix: str) -> pd.DataFrame:
    return _read_structured_file(_path_with_suffix(path, suffix))


def _read_text_file(path: Path) -> str:
    return path.read_text(encoding="utf-8", errors="replace")


def _normalize_text_line(line: str) -> str:
    return " ".join(line.split())


def _dedupe_preserving_order(values: list[str]) -> list[str]:
    seen = set()
    deduped = []
    for value in values:
        if value in seen:
            continue
        seen.add(value)
        deduped.append(value)
    return deduped


def _is_page_marker(line: str) -> bool:
    return bool(PAGE_MARKER_PATTERN.match(line))


def _is_list_start(line: str) -> bool:
    return bool(LIST_START_PATTERN.match(line))


def _is_heading(line: str) -> bool:
    return line.endswith(":") and len(line) <= 100 and bool(line[:1].isupper())


def _collapse_adjacent_duplicates(lines: list[str]) -> list[str]:
    collapsed = []
    previous_line = None
    for line in lines:
        if line == previous_line:
            continue
        collapsed.append(line)
        previous_line = line
    return collapsed


def _remove_repeated_extracted_lines(lines: list[str]) -> list[str]:
    repeated_candidates = {
        line
        for line in lines
        if lines.count(line) > 1 and len(line) >= 12 and len(line.split()) >= 2 and not _is_page_marker(line)
    }
    seen = set()
    cleaned = []
    for line in lines:
        if line in repeated_candidates:
            if line in seen:
                continue
            seen.add(line)
        cleaned.append(line)
    return cleaned


def _join_wrapped_document_lines(lines: list[str]) -> list[str]:
    blocks = []
    current = ""

    def flush_current() -> None:
        nonlocal current
        if current:
            blocks.append(current)
            current = ""

    for line in lines:
        if _is_page_marker(line):
            flush_current()
            blocks.append(line)
            continue

        if _is_heading(line):
            flush_current()
            blocks.append(line)
            continue

        if _is_list_start(line):
            flush_current()
            current = line
            continue

        if not current:
            current = line
            continue

        current = f"{current} {line}"

    flush_current()
    return blocks


def _clean_extracted_document_text(text: str) -> str:
    lines = [_normalize_text_line(line) for line in text.splitlines()]
    lines = [line for line in lines if line]
    lines = _collapse_adjacent_duplicates(lines)
    lines = _remove_repeated_extracted_lines(lines)

    # PDF extraction often returns one word or short fragment per line. Rebuild those
    # fragments into readable blocks while preserving page markers, headings, and bullets.
    return "\n".join(_join_wrapped_document_lines(lines))


def _read_document_file(path: Path) -> tuple[str, str]:
    suffix = path.suffix.lower()

    if suffix == ".docx":
        document = Document(path)
        parts = [paragraph.text for paragraph in document.paragraphs if paragraph.text.strip()]
        for table in document.tables:
            for row in table.rows:
                cells = _dedupe_preserving_order([cell.text.strip() for cell in row.cells if cell.text.strip()])
                if cells:
                    parts.append(" | ".join(cells))
        return _clean_extracted_document_text("\n".join(parts)), "Word document"

    if suffix == ".pdf":
        reader = PdfReader(str(path))
        pages = []
        for index, page in enumerate(reader.pages, start=1):
            page_text = page.extract_text() or ""
            if page_text.strip():
                pages.append(f"--- Page {index} ---\n{page_text.strip()}")
        page_label = "page" if len(reader.pages) == 1 else "pages"
        return _clean_extracted_document_text("\n\n".join(pages)), f"PDF document with {len(reader.pages)} {page_label}"

    raise ValueError(f"{path.name} is not a supported document file.")


def _read_document_file_by_suffix(path: Path, suffix: str) -> tuple[str, str]:
    return _read_document_file(_path_with_suffix(path, suffix))


def _text_analysis_outputs(path: Path, text: str, file_kind: str) -> tuple[str, pd.DataFrame, pd.DataFrame, go.Figure]:
    lines = text.splitlines()
    words = len(text.split())
    line_count = len(lines)
    preview = pd.DataFrame({"line": lines[:100]})
    unique_lines = int(preview["line"].nunique()) if not preview.empty else 0
    profile = pd.DataFrame(
        [
            {
                "column": "line",
                "type": "text",
                "filled": int(len(preview)),
                "missing": 0,
                "unique": unique_lines,
                "examples": " | ".join(lines[:3]),
            }
        ]
    )
    chart = px.bar(
        pd.DataFrame({"metric": ["lines", "words"], "count": [line_count, words]}),
        x="metric",
        y="count",
        title=f"{file_kind} summary for {path.name}",
        template="plotly_dark",
    )
    summary = f"{file_kind} with {line_count:,} lines and {words:,} words."
    return summary, preview, profile, chart


def _profile_dataframe(df: pd.DataFrame) -> pd.DataFrame:
    rows = []
    for column in df.columns:
        series = df[column]
        sample_values = [str(value) for value in series.dropna().head(3).tolist()]
        rows.append(
            {
                "column": str(column),
                "type": str(series.dtype),
                "filled": int(series.notna().sum()),
                "missing": int(series.isna().sum()),
                "unique": int(series.nunique(dropna=True)),
                "examples": ", ".join(sample_values),
            }
        )
    return pd.DataFrame(rows)


def _make_chart(df: pd.DataFrame) -> go.Figure:
    if df.empty:
        return go.Figure().update_layout(title="No rows to chart")

    numeric_columns = df.select_dtypes(include="number").columns.tolist()
    text_columns = df.select_dtypes(include=["object", "category", "bool"]).columns.tolist()

    if numeric_columns:
        column = numeric_columns[0]
        return px.histogram(df, x=column, title=f"Distribution of {column}", template="plotly_dark")

    if text_columns:
        column = text_columns[0]
        counts = df[column].astype(str).value_counts().head(12).reset_index()
        counts.columns = [column, "count"]
        return px.bar(counts, x=column, y="count", title=f"Top values in {column}", template="plotly_dark")

    return go.Figure().update_layout(title="No chartable columns found")


def _table_to_html(df: pd.DataFrame, max_rows: int = 20) -> str:
    return df.head(max_rows).to_html(index=False, escape=True, border=0, classes="data-table")


def _write_report(
    file_summaries: list[dict[str, Any]],
    preview: pd.DataFrame,
    profile: pd.DataFrame,
    chart: go.Figure,
) -> tuple[str, str]:
    output_dir = Path(tempfile.mkdtemp(prefix="hf_app_report_"))
    report_path = output_dir / "generated_report.html"
    profile_path = output_dir / "data_profile.json"

    profile_data = {
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "files": file_summaries,
        "columns": profile.to_dict(orient="records") if not profile.empty else [],
    }
    profile_path.write_text(json.dumps(profile_data, indent=2), encoding="utf-8")

    file_cards = "\n".join(
        f"""
        <article class="card">
          <h3>{html.escape(item["name"])}</h3>
          <p>{html.escape(item["summary"])}</p>
        </article>
        """
        for item in file_summaries
    )

    report_path.write_text(
        f"""i
        <!doctype html>
        <html lang="en">
          <head>
            <meta charset="utf-8" />
            <meta name="viewport" content="width=device-width, initial-scale=1" />
            <title>Generated App Report</title>
            <style>
              body {{
                margin: 0;
                font-family: Inter, Arial, sans-serif;
                background: #0f172a;
                color: #e2e8f0;
              }}
              main {{
                max-width: 1100px;
                margin: 0 auto;
                padding: 40px 20px;
              }}
              h1, h2, h3 {{ color: #ffffff; }}
              .grid {{
                display: grid;
                grid-template-columns: repeat(auto-fit, minmax(220px, 1fr));
                gap: 16px;
              }}
              .card {{
                padding: 18px;
                border: 1px solid rgba(148, 163, 184, 0.25);
                border-radius: 18px;
                background: rgba(15, 23, 42, 0.8);
              }}
              .data-table {{
                width: 100%;
                border-collapse: collapse;
                overflow: hidden;
                border-radius: 12px;
              }}
              .data-table th, .data-table td {{
                padding: 10px;
                border-bottom: 1px solid rgba(148, 163, 184, 0.2);
                text-align: left;
              }}
              .data-table th {{ background: rgba(37, 99, 235, 0.25); }}
            </style>
          </head>
          <body>
            <main>
              <h1>Generated App Report</h1>
              <p>Created from your uploaded files. No third-party API key was used.</p>
              <section>
                <h2>Files</h2>
                <div class="grid">{file_cards}</div>
              </section>
              <section>
                <h2>Preview</h2>
                {_table_to_html(preview)}
              </section>
              <section>
                <h2>Column Profile</h2>
                {_table_to_html(profile, max_rows=100)}
              </section>
              <section>
                <h2>Chart</h2>
                {chart.to_html(full_html=False, include_plotlyjs="cdn")}
              </section>
            </main>
          </body>
        </html>
        """,
        encoding="utf-8",
    )

    return str(report_path), str(profile_path)


def analyze_files(files: list[Any] | None) -> tuple[str, pd.DataFrame, pd.DataFrame, go.Figure, str | None, str | None]:
    if not files:
        empty_chart = go.Figure().update_layout(title="Upload files to generate a dashboard")
        return "Upload at least one file to begin.", pd.DataFrame(), pd.DataFrame(), empty_chart, None, None

    file_summaries: list[dict[str, Any]] = []
    selected_preview = pd.DataFrame()
    selected_profile = pd.DataFrame()
    selected_chart = go.Figure().update_layout(title="No chart generated")

    for file_value in files:
        path = _file_path(file_value)
        name = _display_name(file_value, path)
        suffix = _file_suffix(file_value, path)

        if suffix not in SUPPORTED_EXTENSIONS:
            file_summaries.append({"name": name, "summary": f"Skipped unsupported file type: {suffix or 'unknown'}"})
            continue

        try:
            if suffix in TEXT_EXTENSIONS:
                text = _read_text_file(path)
                summary, preview_df, profile_df, chart = _text_analysis_outputs(path, text, "Text file")
                file_summaries.append({"name": name, "summary": summary})
                if selected_preview.empty:
                    selected_preview = preview_df
                    selected_profile = profile_df
                    selected_chart = chart
                continue

            if suffix in DOCUMENT_EXTENSIONS:
                text, document_kind = _read_document_file_by_suffix(path, suffix)
                summary, preview_df, profile_df, chart = _text_analysis_outputs(path, text, document_kind)
                file_summaries.append({"name": name, "summary": summary})
                if selected_preview.empty:
                    selected_preview = preview_df
                    selected_profile = profile_df
                    selected_chart = chart
                continue

            df = _read_structured_file_by_suffix(path, suffix)
            file_summaries.append({"name": name, "summary": f"Structured data with {len(df):,} rows and {len(df.columns):,} columns."})
            if selected_preview.empty:
                selected_preview = df.head(100)
                selected_profile = _profile_dataframe(df)
                selected_chart = _make_chart(df)
        except Exception as exc:  # Show readable upload errors instead of crashing the Space.
            file_summaries.append({"name": name, "summary": f"Could not read file: {exc}"})

    if selected_preview.empty:
        selected_preview = pd.DataFrame(file_summaries)
        selected_profile = pd.DataFrame()

    report_path, profile_path = _write_report(file_summaries, selected_preview, selected_profile, selected_chart)

    summary_lines = ["## Generated dashboard", ""]
    summary_lines.extend(f"- **{item['name']}**: {item['summary']}" for item in file_summaries)
    summary_lines.append("")
    summary_lines.append("Download the generated HTML report or JSON profile below.")

    return "\n".join(summary_lines), selected_preview, selected_profile, selected_chart, report_path, profile_path


def build_app() -> gr.Blocks:
    with gr.Blocks(title="File App Generator") as demo:
        gr.Markdown(
            """
            <div id="hero">

            # File App Generator

            Upload documents and data files: PDF, DOCX, CSV, Excel, JSON, JSONL, TXT, MD, or LOG.
            The app reads the contents, creates a preview, column/text profile, chart, and
            downloadable HTML report without using a paid API key.

            </div>
            """
        )

        with gr.Row():
            uploads = gr.File(
                label="Upload files",
                file_count="multiple",
                type="filepath",
            )
            run_button = gr.Button("Generate dashboard", variant="primary")

        summary = gr.Markdown()
        preview = gr.Dataframe(label="Preview", interactive=False)
        profile = gr.Dataframe(label="Column profile", interactive=False)
        chart = gr.Plot(label="Auto chart")

        with gr.Row():
            report_file = gr.File(label="Download HTML report")
            profile_file = gr.File(label="Download JSON profile")

        run_button.click(
            analyze_files,
            inputs=[uploads],
            outputs=[summary, preview, profile, chart, report_file, profile_file],
        )

    return demo