app.py

from __future__ import annotations

import inspect
from pathlib import Path
from typing import Any

import gradio as gr

from src.space_service import (
    RefCheckOptions,
    RefCheckResult,
    apply_review_action,
    preview_review_action,
    run_refcheck_file,
)


def _uploaded_path(uploaded: Any) -> str | None:
    if not uploaded:
        return None
    if isinstance(uploaded, str):
        return uploaded
    if isinstance(uploaded, dict):
        return uploaded.get("path") or uploaded.get("name")
    name = getattr(uploaded, "name", None)
    if name:
        return str(name)
    return None


def process_bib(
    uploaded: Any,
    remove_unverified: bool,
    enable_google_scholar: bool,
    max_workers: int,
) -> tuple[str, str | None, str | None]:
    file_path = _uploaded_path(uploaded)
    if not file_path:
        return "## RefCheck Report\n\nNo BibTeX file was uploaded.", None, None

    try:
        options = RefCheckOptions(
            remove_unverified=remove_unverified,
            enable_google_scholar=enable_google_scholar,
            max_workers=int(max_workers),
        )
        result = run_refcheck_file(Path(file_path), options)
        return result.report_markdown, result.fixed_bib_path, result.report_path
    except Exception as exc:
        return f"## RefCheck Report\n\nProcessing failed: `{exc}`", None, None


def process_bib_for_ui(
    uploaded: Any,
    remove_unverified: bool,
    enable_google_scholar: bool,
    max_workers: int,
):
    file_path = _uploaded_path(uploaded)
    if not file_path:
        report = "## RefCheck Report\n\nNo BibTeX file was uploaded."
        return report, None, None, None, gr.update(choices=[], value=None), gr.update(choices=[], value=None), ""

    try:
        options = _options(remove_unverified, enable_google_scholar, max_workers)
        result = run_refcheck_file(Path(file_path), options)
        return _ui_outputs(result)
    except Exception as exc:
        report = f"## RefCheck Report\n\nProcessing failed: `{exc}`"
        return report, None, None, None, gr.update(choices=[], value=None), gr.update(choices=[], value=None), ""


def update_review_controls(review_label: str | None, result: RefCheckResult | None):
    action_choices = _action_choices(result, review_label)
    action_value = action_choices[0] if action_choices else None
    preview = _review_summary(result, review_label)
    return gr.update(choices=action_choices, value=action_value), preview


def preview_selected_review(
    review_label: str | None,
    action_label: str | None,
    result: RefCheckResult | None,
    remove_unverified: bool,
    enable_google_scholar: bool,
    max_workers: int,
) -> str:
    review_index = _parse_review_index(review_label)
    action, candidate_index = _parse_action(action_label)
    options = _options(remove_unverified, enable_google_scholar, max_workers)
    return preview_review_action(result, review_index, action, candidate_index, options)


def apply_selected_review(
    review_label: str | None,
    action_label: str | None,
    result: RefCheckResult | None,
    remove_unverified: bool,
    enable_google_scholar: bool,
    max_workers: int,
):
    review_index = _parse_review_index(review_label)
    action, candidate_index = _parse_action(action_label)
    options = _options(remove_unverified, enable_google_scholar, max_workers)
    try:
        updated = apply_review_action(result, review_index, action, candidate_index, options)
    except Exception as exc:
        preview = f"### Manual review\n\n{exc}"
        return (
            result.report_markdown if result else "",
            result.fixed_bib_path if result else None,
            result.report_path if result else None,
            result,
            gr.update(),
            gr.update(),
            preview,
        )
    return _ui_outputs(updated)


def _options(remove_unverified: bool, enable_google_scholar: bool, max_workers: int) -> RefCheckOptions:
    return RefCheckOptions(
        remove_unverified=remove_unverified,
        enable_google_scholar=enable_google_scholar,
        max_workers=int(max_workers),
    )


def _ui_outputs(result: RefCheckResult):
    review_choices = _review_choices(result)
    review_value = review_choices[0] if review_choices else None
    action_choices = _action_choices(result, review_value)
    action_value = action_choices[0] if action_choices else None
    preview = _review_summary(result, review_value)
    return (
        result.report_markdown,
        result.fixed_bib_path,
        result.report_path,
        result,
        gr.update(choices=review_choices, value=review_value),
        gr.update(choices=action_choices, value=action_value),
        preview,
    )


def _review_choices(result: RefCheckResult | None) -> list[str]:
    if not result:
        return []
    choices = []
    for idx, item in enumerate(result.review_items, 1):
        entry = item["entry"]
        title = entry.title or "[missing title]"
        if len(title) > 90:
            title = title[:87] + "..."
        choices.append(f"{idx}. {entry.key}: {title}")
    return choices


def _action_choices(result: RefCheckResult | None, review_label: str | None) -> list[str]:
    review_index = _parse_review_index(review_label)
    if not result or review_index < 0 or review_index >= len(result.review_items):
        return []

    item = result.review_items[review_index]
    choices = []
    for idx, candidate in enumerate(item.get("candidates", []), 1):
        fetched = candidate.fetched_data
        title = getattr(fetched, "title", "") or "[missing title]"
        if len(title) > 72:
            title = title[:69] + "..."
        choices.append(f"Candidate {idx}: {candidate.source} ({candidate.confidence:.2f}) {title}")
    choices.extend(["Keep original", "Remove entry"])
    return choices


def _review_summary(result: RefCheckResult | None, review_label: str | None) -> str:
    review_index = _parse_review_index(review_label)
    if not result or not result.review_items:
        return "### Manual review\n\nNo unresolved entries."
    if review_index < 0 or review_index >= len(result.review_items):
        return "### Manual review\n\nSelect an unresolved entry."

    item = result.review_items[review_index]
    entry = item["entry"]
    best = item.get("best_result")
    reason = "; ".join(best.issues) if best and best.issues else "Ambiguous match"
    return (
        "### Manual review\n\n"
        f"**Key:** `{entry.key}`\n\n"
        f"**Title:** {entry.title or '[missing]'}\n\n"
        f"**Authors:** {entry.author or '[missing]'}\n\n"
        f"**Year:** {entry.year or '[missing]'}\n\n"
        f"**Reason:** {reason}\n\n"
        f"**Candidates:** {len(item.get('candidates', []))}"
    )


def _parse_review_index(review_label: str | None) -> int:
    if not review_label:
        return -1
    try:
        return int(review_label.split(".", 1)[0]) - 1
    except Exception:
        return -1


def _parse_action(action_label: str | None) -> tuple[str, int | None]:
    if not action_label:
        return "", None
    if action_label.startswith("Candidate "):
        try:
            number = int(action_label.split(":", 1)[0].replace("Candidate", "").strip())
        except Exception:
            return "", None
        return "candidate", number - 1
    if action_label == "Keep original":
        return "keep", None
    if action_label == "Remove entry":
        return "remove", None
    return "", None


with gr.Blocks(title="RefCheck") as demo:
    gr.Markdown("# RefCheck")
    session_state = gr.State(None)

    with gr.Row():
        with gr.Column(scale=1):
            bib_file = gr.File(
                label="BibTeX file",
                file_types=[".bib", ".txt"],
                type="filepath",
            )
            remove_unverified = gr.Checkbox(
                label="Remove entries with no candidates",
                value=False,
            )
            enable_google_scholar = gr.Checkbox(
                label="Google Scholar fallback",
                value=False,
            )
            max_workers = gr.Slider(
                label="Parallel lookups",
                minimum=1,
                maximum=8,
                step=1,
                value=4,
            )
            run_button = gr.Button("Run RefCheck", variant="primary")
            api_button = gr.Button(visible=False)

        with gr.Column(scale=2):
            report = gr.Markdown(label="Report")
            fixed_bib = gr.File(label="Fixed BibTeX")
            report_file = gr.File(label="Markdown report")
            with gr.Accordion("Manual review", open=True):
                review_entry = gr.Dropdown(label="Unresolved entry", choices=[])
                review_action = gr.Radio(label="Candidate/action", choices=[])
                with gr.Row():
                    test_button = gr.Button("Test selected")
                    apply_button = gr.Button("Apply exact selected", variant="primary")
                review_preview = gr.Markdown()

    run_button.click(
        fn=process_bib_for_ui,
        inputs=[bib_file, remove_unverified, enable_google_scholar, max_workers],
        outputs=[report, fixed_bib, report_file, session_state, review_entry, review_action, review_preview],
        api_visibility="private",
    )
    api_button.click(
        fn=process_bib,
        inputs=[bib_file, remove_unverified, enable_google_scholar, max_workers],
        outputs=[report, fixed_bib, report_file],
        api_name="refcheck",
    )
    review_entry.change(
        fn=update_review_controls,
        inputs=[review_entry, session_state],
        outputs=[review_action, review_preview],
        api_visibility="private",
    )
    test_button.click(
        fn=preview_selected_review,
        inputs=[
            review_entry,
            review_action,
            session_state,
            remove_unverified,
            enable_google_scholar,
            max_workers,
        ],
        outputs=review_preview,
        api_visibility="private",
    )
    apply_button.click(
        fn=apply_selected_review,
        inputs=[
            review_entry,
            review_action,
            session_state,
            remove_unverified,
            enable_google_scholar,
            max_workers,
        ],
        outputs=[report, fixed_bib, report_file, session_state, review_entry, review_action, review_preview],
        api_visibility="private",
    )


def launch() -> None:
    launch_kwargs = {}
    if "ssr_mode" in inspect.signature(demo.launch).parameters:
        launch_kwargs["ssr_mode"] = False
    demo.queue(default_concurrency_limit=2).launch(**launch_kwargs)


if __name__ == "__main__":
    launch()