app.py

import ast
import base64
import binascii
import html
import io
import os
import re
import tempfile
from pathlib import Path

import gradio as gr
import pandas as pd
from openai import OpenAI
from PIL import Image, ImageOps

DEFAULT_MODEL = os.getenv("VISION_MODEL", "qwen/qwen3.6-27b")
DEFAULT_PROMPT = (
"""
You create accessibility alt text for images.

Return only the finished alt text.

Requirements:
- Write one concise sentence.
- Describe the image's essential meaning and purpose.
- Do not describe every minor visual detail.
- Do not begin with "Image of", "Picture of", or "This image shows".
- Do not include analysis, reasoning, headings, XML tags, or bullet points.
- Do not include <think> tags.
- Do not identify people unless their identity is explicitly provided.
- For diagrams, summarize the main process or relationship.
- For charts, state the chart type and principal trend or conclusion.
- Keep the response under 250 characters when practical.
"""
)

TFLIVE_RE = re.compile(
    br'^<TFLiveData\s+(?P<attrs>[^>]*)>',
    flags=re.IGNORECASE | re.DOTALL,
)
ATTR_RE = re.compile(
    r'([A-Za-z_][\w.-]*)\s*=\s*"([^"]*)"'
)


def _read_table(path: str) -> pd.DataFrame:
    suffix = Path(path).suffix.lower()
    if suffix == ".csv":
        return pd.read_csv(path, dtype=object, keep_default_na=False)
    if suffix in {".tsv", ".txt"}:
        return pd.read_csv(path, sep="\t", dtype=object, keep_default_na=False)
    if suffix in {".xlsx", ".xlsm"}:
        return pd.read_excel(path, dtype=object, keep_default_na=False)
    raise ValueError("Upload a CSV, TSV, XLSX, or XLSM file.")


def inspect_table(path: str):
    if not path:
        return gr.update(choices=[], value=None), gr.update(choices=[], value=None), pd.DataFrame()
    df = _read_table(path)
    cols = [str(c) for c in df.columns]
    preview = df.head(10).copy()
    for c in preview.columns:
        preview[c] = preview[c].map(lambda x: _truncate_cell(x, 180))
    id_default = cols[0] if cols else None
    binary_default = cols[1] if len(cols) > 1 else (cols[0] if cols else None)
    return (
        gr.update(choices=cols, value=binary_default),
        gr.update(choices=["(row number)"] + cols, value=id_default),
        preview,
    )


def _truncate_cell(value, limit=180):
    text = repr(value) if isinstance(value, (bytes, bytearray)) else str(value)
    return text if len(text) <= limit else text[:limit] + "…"


def _to_candidate_bytes(value, encoding_mode: str) -> bytes:
    if isinstance(value, bytes):
        return value
    if isinstance(value, bytearray):
        return bytes(value)
    if isinstance(value, memoryview):
        return value.tobytes()
    if value is None:
        raise ValueError("empty value")

    text = str(value).strip()
    if not text:
        raise ValueError("empty value")

    if encoding_mode == "base64":
        return base64.b64decode(re.sub(r"\s+", "", text), validate=True)
    if encoding_mode == "hex":
        return bytes.fromhex(re.sub(r"\s+", "", text).replace("0x", ""))
    if encoding_mode == "escaped":
        return text.encode("utf-8").decode("unicode_escape").encode("latin-1")
    if encoding_mode == "latin1":
        return text.encode("latin-1")

    # Auto detection
    if text.startswith(("b'", 'b"')):
        parsed = ast.literal_eval(text)
        if isinstance(parsed, bytes):
            return parsed

    if "\\x" in text:
        try:
            return text.encode("utf-8").decode("unicode_escape").encode("latin-1")
        except Exception:
            pass

    compact = re.sub(r"\s+", "", text)
    if len(compact) >= 16 and len(compact) % 2 == 0 and re.fullmatch(r"(?:0x)?[0-9A-Fa-f]+", compact):
        try:
            return bytes.fromhex(compact.replace("0x", ""))
        except ValueError:
            pass

    if len(compact) >= 16 and re.fullmatch(r"[A-Za-z0-9+/]*={0,2}", compact):
        try:
            decoded = base64.b64decode(compact, validate=True)
            if decoded:
                return decoded
        except (binascii.Error, ValueError):
            pass

    # This preserves byte values when the database export was decoded as Latin-1.
    try:
        return text.encode("latin-1")
    except UnicodeEncodeError:
        return text.encode("utf-8")


def _strip_tflive_header(data: bytes):
    metadata = {}
    match = TFLIVE_RE.match(data)
    if not match:
        return data, metadata
    attrs = match.group("attrs").decode("latin-1", errors="replace")
    metadata = {k.lower(): html.unescape(v) for k, v in ATTR_RE.findall(attrs)}
    return data[match.end():], metadata


def decode_image(value, encoding_mode="auto", max_dimension=1280, jpeg_quality=88):
    raw = _to_candidate_bytes(value, encoding_mode)
    image_bytes, metadata = _strip_tflive_header(raw)
    if not image_bytes:
        raise ValueError("no image bytes remained after removing the TFLiveData header")

    try:
        with Image.open(io.BytesIO(image_bytes)) as opened:
            opened.seek(0)
            image = ImageOps.exif_transpose(opened).convert("RGB")
    except Exception as exc:
        raise ValueError(f"Pillow could not decode the image: {exc}") from exc

    if max(image.size) > max_dimension:
        image.thumbnail((max_dimension, max_dimension), Image.Resampling.LANCZOS)

    out = io.BytesIO()
    # PNG avoids issues with ICO/GIF/TIFF support at inference providers.
    image.save(out, format="PNG", optimize=True)
    return out.getvalue(), metadata, image.size


def _data_uri(png_bytes: bytes) -> str:
    return "data:image/png;base64," + base64.b64encode(png_bytes).decode("ascii")


ALT_TEXT_SYSTEM_PROMPT = """
You create accessibility alt text for images.

Return only the finished alt text.

Requirements:
- Write one concise sentence.
- Describe the image's essential meaning and purpose.
- Do not describe every minor visual detail.
- Do not begin with "Image of", "Picture of", or "This image shows".
- Do not include analysis, reasoning, headings, XML tags, or bullet points.
- Do not include <think> tags.
- Do not identify people unless their identity is explicitly provided.
- For diagrams, summarize the main process or relationship.
- For charts, state the chart type and principal trend or conclusion.
- Keep the response under 250 characters when practical.
"""

def generate_alt_text(client, model, prompt, png_bytes, max_tokens):
    completion = client.chat.completions.create(
        model=model,
        reasoning_effort="none",
        messages=[
            {
                "role": "user",
                "content": [
                    {"type": "text", "text": prompt},
                    {
                        "type": "image_url",
                        "image_url": {"url": _data_uri(png_bytes)},
                    },
                ],
            }
        ],
        max_tokens=int(max_tokens),
        temperature=0,
    )
    return (completion.choices[0].message.content or "").strip()



def process_table(
    path,
    binary_column,
    id_column,
    encoding_mode,
    prompt,
    model,
    max_tokens,
    max_dimension,
    output_format,
    progress=gr.Progress(track_tqdm=False),
):
    if not path:
        raise gr.Error("Upload a table first.")
    token = os.getenv("HF_TOKEN")
    if not token:
        raise gr.Error(
            "HF_TOKEN is not configured. Add it as a private Space secret with Inference Providers permission."
        )

    df = _read_table(path)
    if binary_column not in df.columns:
        raise gr.Error("Select the column containing the image data.")

    # client = OpenAI(base_url="https://router.huggingface.co/v1", api_key=token)
    client = OpenAI(base_url="https://api.groq.com/openai/v1", api_key=os.environ["GROQ_API_KEY"])
    results = []
    total = len(df)

    for position, (_, row) in enumerate(df.iterrows(), start=1):
        progress((position - 1) / max(total, 1), desc=f"Processing {position} of {total}")
        record_id = position if id_column == "(row number)" else row.get(id_column, position)
        status = "ok"
        alt_text = ""
        detected_name = ""
        detected_mime = ""
        dimensions = ""
        error = ""

        try:
            png_bytes, metadata, size = decode_image(
                row[binary_column],
                encoding_mode=encoding_mode,
                max_dimension=int(max_dimension),
            )
            detected_name = metadata.get("name", "")
            detected_mime = metadata.get("mime", "")
            dimensions = f"{size[0]}x{size[1]}"
            alt_text = generate_alt_text(
                client=client,
                model=model.strip(),
                prompt=prompt.strip() or DEFAULT_PROMPT,
                png_bytes=png_bytes,
                max_tokens=max_tokens,
            )
        except Exception as exc:
            status = "error"
            error = str(exc)

        results.append(
            {
                "record_id": record_id,
                "alt_text": alt_text,
                "status": status,
                "error": error,
                "source_name": detected_name,
                "source_mime": detected_mime,
                "processed_dimensions": dimensions,
            }
        )

    progress(1, desc="Preparing download")
    result_df = pd.DataFrame(results)
    combined = df.copy()
    for c in result_df.columns:
        combined[c] = result_df[c].values

    temp_dir = tempfile.mkdtemp(prefix="alt_text_results_")
    if output_format == "Excel (.xlsx)":
        output_path = os.path.join(temp_dir, "alt_text_results.xlsx")
        combined.to_excel(output_path, index=False)
    else:
        output_path = os.path.join(temp_dir, "alt_text_results.csv")
        combined.to_csv(output_path, index=False)

    preview_cols = [c for c in [id_column if id_column != "(row number)" else None, "alt_text", "status", "error"] if c]
    preview = combined[preview_cols].head(25)
    return preview, output_path, f"Completed {total} rows: {(result_df.status == 'ok').sum()} succeeded, {(result_df.status == 'error').sum()} failed."


with gr.Blocks(title="Database Image Alt-Text Generator") as demo:
    gr.Markdown(
        """
# Database Image Alt-Text Generator
Upload a CSV, TSV, or Excel table containing one image value per row. The app recognizes the uploaded sample's `TFLiveData` wrapper, as well as base64, hexadecimal, escaped `\\xNN`, and Latin-1-preserved binary strings. Images are decoded and normalized to PNG before inference.

**Privacy:** Uploaded image contents are sent to the selected Hugging Face Inference Provider model. Use a private Space and an approved provider for confidential data.
"""
    )

    with gr.Row():
        table_file = gr.File(label="Input table", file_types=[".csv", ".tsv", ".txt", ".xlsx", ".xlsm"], type="filepath")
        with gr.Column():
            binary_column = gr.Dropdown(label="Image/binary column", choices=[])
            id_column = gr.Dropdown(label="Record ID column", choices=[])
            encoding_mode = gr.Dropdown(
                label="Binary representation",
                choices=["auto", "base64", "hex", "escaped", "latin1"],
                value="auto",
            )

    table_preview = gr.Dataframe(label="Input preview", interactive=False)
    table_file.change(inspect_table, inputs=table_file, outputs=[binary_column, id_column, table_preview])

    with gr.Accordion("Model and processing settings", open=True):
        model = gr.Textbox(label="Inference model", value=DEFAULT_MODEL)
        prompt = gr.Textbox(label="Alt-text prompt", value=DEFAULT_PROMPT, lines=5)
        with gr.Row():
            max_tokens = gr.Slider(20, 200, value=80, step=5, label="Maximum output tokens")
            max_dimension = gr.Slider(256, 2048, value=1280, step=128, label="Maximum image dimension")
            output_format = gr.Radio(["CSV", "Excel (.xlsx)"], value="Excel (.xlsx)", label="Download format")

    run_button = gr.Button("Generate alt text", variant="primary")
    status = gr.Markdown()
    results_preview = gr.Dataframe(label="Results preview", interactive=False)
    download = gr.File(label="Download completed table")

    run_button.click(
        process_table,
        inputs=[
            table_file,
            binary_column,
            id_column,
            encoding_mode,
            prompt,
            model,
            max_tokens,
            max_dimension,
            output_format,
        ],
        outputs=[results_preview, download, status],
    )

if __name__ == "__main__":
    demo.queue(default_concurrency_limit=2).launch()