app.py

import os
import threading
from datetime import date, timedelta

import streamlit as st
from dotenv import load_dotenv

load_dotenv(os.path.join(os.path.dirname(__file__), ".env"))

from db import get_connection, init_db

st.set_page_config(page_title="DeepL API Usage Logger", page_icon="🪵", layout="wide")

st.markdown("""
<link href="https://fonts.googleapis.com/css2?family=Press+Start+2P&display=swap" rel="stylesheet">
<style>
    /* 8-bit font for page title */
    .deepl-title {
        font-family: 'Press Start 2P', monospace !important;
        font-size: 1.2rem !important;
        color: rgb(15, 43, 70) !important;
        line-height: 1.4 !important;
        margin: 0 !important;
    }
    /* Brand navy for metric labels */
    [data-testid="stMetricLabel"] {
        color: rgb(15, 43, 70) !important;
        font-family: Roboto, sans-serif !important;
    }
    /* Metric values in action blue */
    [data-testid="stMetricValue"] {
        color: rgb(0, 112, 201) !important;
        font-weight: 600;
    }
    /* Download button styled as DeepL primary action */
    [data-testid="stDownloadButton"] button {
        background-color: rgb(0, 112, 201) !important;
        color: rgb(255, 255, 255) !important;
        border: none !important;
        border-radius: 4px !important;
    }
    [data-testid="stDownloadButton"] button:hover {
        background-color: rgb(0, 83, 169) !important;
    }
    /* Constrain content width */
    .block-container {
        max-width: 1100px !important;
        padding-left: 2rem !important;
        padding-right: 2rem !important;
    }
    /* Muted caption text */
    [data-testid="stCaptionContainer"] {
        color: rgb(111, 111, 111) !important;
    }
</style>
""", unsafe_allow_html=True)

st.markdown(
    '<div class="deepl-title">DeepL API Usage Logger</div>',
    unsafe_allow_html=True,
)

@st.cache_resource
def _get_conn():
    db_path = os.environ.get(
        "DEEPL_DB_PATH",
        os.path.join(os.path.dirname(__file__), "translation_usage.duckdb"),
    )
    if not os.path.exists(db_path):
        # Bootstrap an empty schema so the dashboard renders before any data exists.
        rw = get_connection()
        init_db(rw)
        rw.close()
    return get_connection(read_only=True)

conn = _get_conn()
_SQL_TIMEOUT = int(os.environ.get("SQL_TIMEOUT_SECONDS", "30"))


_PRESETS = ["Today", "Last 7 days", "Last 30 days", "This month", "This year", "All time", "Custom"]


def _date_range_picker(key: str) -> tuple[date, date]:
    preset = st.selectbox("Date range", _PRESETS, index=2, key=f"{key}_preset")
    if preset == "Custom":
        col1, col2 = st.columns(2)
        with col1:
            start = st.date_input("Start date", value=date.today() - timedelta(days=29), key=f"{key}_start")
        with col2:
            end = st.date_input("End date", value=date.today(), key=f"{key}_end")
    else:
        today = date.today()
        if preset == "Today":
            start = end = today
        elif preset == "Last 7 days":
            start, end = today - timedelta(days=6), today
        elif preset == "Last 30 days":
            start, end = today - timedelta(days=29), today
        elif preset == "This month":
            start, end = today.replace(day=1), today
        elif preset == "This year":
            start, end = today.replace(month=1, day=1), today
        else:
            start, end = date(2000, 1, 1), date(2099, 12, 31)
        st.caption(f"{start.strftime('%b %-d, %Y')}  →  {end.strftime('%b %-d, %Y')}")
    return start, end

tab_usage, tab_errors, tab_sql, tab_about = st.tabs(["Usage", "Errors", "SQL Explorer", "About"])

# ── Usage tab ─────────────────────────────────────────────────────────────────
with tab_usage:

    # --- Date range ---
    start_date, end_date = _date_range_picker("usage")

    # --- Summary metrics ---
    totals = conn.execute(
        """
        SELECT
            coalesce(sum(billed_characters), 0) as total_billed_chars,
            count(*)                            as total_requests,
            count(distinct source_lang || target_lang) as language_pairs
        FROM translation_usage
        WHERE (logged_date BETWEEN ? AND ?)
        AND is_success = true
        """,
        [start_date, end_date],
    ).fetchone()

    c1, c2, c3 = st.columns(3)
    c1.metric("Total Billed Characters", f"{totals[0]:,}")
    c2.metric("Total Requests", f"{totals[1]:,}")
    c3.metric("Language Pairs", totals[2])

    st.divider()

    # --- Group by ---
    DIMENSIONS = {
        "Date":               ("logged_date",                             "logged_date"),
        "Language pair":      ("coalesce(source_lang, 'N/A') || ' → ' || target_lang", "language_pair"),
        "Reporting tag":      ("coalesce(reporting_tag, '(no tag)')",     "tag"),
        "API key":            ("api_key_alias",                              "api_key_alias"),
        "Translation Type":   ("translation_type",                        "translation_type"),
    }

    selected = st.multiselect(
        "Group by",
        options=list(DIMENSIONS.keys()),
        default=["Language pair"],
    )

    if not selected:
        st.warning("Select at least one dimension to group by.")
        st.stop()

    if len(selected) == len(DIMENSIONS):
        st.caption("All dimensions selected.")
    else:
        remaining = [d for d in DIMENSIONS if d not in selected]
        st.caption(f"Also group by: {', '.join(remaining)}")

    st.divider()

    # --- Build and run query ---
    select_parts = [f"{expr} as {alias}" for dim in selected for expr, alias in [DIMENSIONS[dim]]]
    n_dims = len(select_parts)
    select_parts += ["sum(billed_characters) as billed_characters", "count(*) as requests"]

    df = conn.execute(
        f"""
        SELECT {', '.join(select_parts)}
        FROM translation_usage
        WHERE (logged_date BETWEEN ? AND ?)
        AND is_success = true
        GROUP BY {', '.join(str(i + 1) for i in range(n_dims))}
        ORDER BY billed_characters DESC
        """,
        [start_date, end_date],
    ).fetchdf()

    if df.empty:
        st.info("No data for the selected filters.")
    else:
        COLUMN_LABELS = {
            "logged_date":       "Date",
            "language_pair":     "Language Pair",
            "tag":               "Reporting Tag",
            "api_key_alias":        "API Key",
            "translation_type":  "Translation Type",
            "billed_characters": "Billed Characters",
            "requests":          "Requests",
        }
        display_df = df.rename(columns=COLUMN_LABELS)
        if "Date" in display_df.columns:
            display_df["Date"] = display_df["Date"].astype(str)
        display_df["Billed Characters"] = display_df["Billed Characters"].astype(int)
        st.download_button(
            label="Download as CSV",
            data=display_df.to_csv(index=False),
            file_name="translation_usage.csv",
            mime="text/csv",
        )
        st.dataframe(display_df, use_container_width=True, hide_index=True)

# ── Errors tab ────────────────────────────────────────────────────────────────
with tab_errors:

    # --- Date range ---
    start_date_err, end_date_err = _date_range_picker("err")

    # --- Summary metric ---
    total_errors = conn.execute(
        """
        SELECT count(*) FROM translation_usage
        WHERE (logged_date BETWEEN ? AND ?)
        AND is_success = false
        """,
        [start_date_err, end_date_err],
    ).fetchone()[0]

    st.metric("Total Errors", f"{total_errors:,}")

    st.divider()

    # --- Group by ---
    ERROR_DIMENSIONS = {
        "Date":             ("logged_date",                                                                                    "logged_date"),
        "API Key":          ("api_key_alias",                                                                                    "api_key_alias"),
        "HTTP Status":      ("error_http_status",                                                                             "error_http_status"),
        "Document Type":    ("CASE WHEN translation_type = 'Document' THEN coalesce(document_type, '(unknown)') ELSE '(text)' END", "document_type"),
        "Error Code":       ("error_code",                                                                                    "error_code"),
        "Error Message":    ("error_message",                                                                                 "error_message"),
    }

    selected_err = st.multiselect(
        "Group by",
        options=list(ERROR_DIMENSIONS.keys()),
        default=list(ERROR_DIMENSIONS.keys()),
        key="err_group",
    )

    if not selected_err:
        st.warning("Select at least one dimension to group by.")
        st.stop()

    if len(selected_err) == len(ERROR_DIMENSIONS):
        st.caption("All dimensions selected.")
    else:
        remaining_err = [d for d in ERROR_DIMENSIONS if d not in selected_err]
        st.caption(f"Also group by: {', '.join(remaining_err)}")

    st.divider()

    # --- Build and run query ---
    err_select_parts = [f"{expr} as {alias}" for dim in selected_err for expr, alias in [ERROR_DIMENSIONS[dim]]]
    n_err_dims = len(err_select_parts)
    err_select_parts += ["count(*) as errors"]

    err_df = conn.execute(
        f"""
        SELECT {', '.join(err_select_parts)}
        FROM translation_usage
        WHERE (logged_date BETWEEN ? AND ?)
        AND is_success = false
        GROUP BY {', '.join(str(i + 1) for i in range(n_err_dims))}
        ORDER BY errors DESC
        """,
        [start_date_err, end_date_err],
    ).fetchdf()

    if err_df.empty:
        st.info("No errors for the selected date range.")
    else:
        ERROR_COLUMN_LABELS = {
            "logged_date":        "Date",
            "api_key_alias":         "API Key",
            "error_http_status":  "HTTP Status",
            "document_type":      "Document Type",
            "error_code":         "Error Code",
            "error_message":      "Error Message",
            "errors":             "Errors",
        }
        display_err_df = err_df.rename(columns=ERROR_COLUMN_LABELS)
        if "Date" in display_err_df.columns:
            display_err_df["Date"] = display_err_df["Date"].astype(str)
        st.download_button(
            label="Download as CSV",
            data=display_err_df.to_csv(index=False),
            file_name="translation_errors.csv",
            mime="text/csv",
        )
        st.dataframe(display_err_df, use_container_width=True, hide_index=True)

# ── SQL Explorer tab ───────────────────────────────────────────────────────────
# Wrapped in a fragment so the "Run" click reruns only this block, preserving
# the active tab. Without this, clicking Run triggers a full app rerun and
# Streamlit resets the visible tab to the first one.
@st.fragment
def _sql_explorer():
    st.caption("Query the `translation_usage` table directly using DuckDB SQL.")

    default_query = "SELECT *\nFROM translation_usage\nLIMIT 50"

    query = st.text_area("SQL", value=default_query, height=150, label_visibility="collapsed")

    if st.button("Run", type="primary"):
        box: dict = {}

        def _run_query():
            try:
                box["df"] = conn.execute(query).fetchdf()
            except Exception as e:
                box["err"] = e

        t = threading.Thread(target=_run_query, daemon=True)
        t.start()
        t.join(timeout=_SQL_TIMEOUT)

        if t.is_alive():
            conn.interrupt()
            t.join()
            st.error(f"Query cancelled: exceeded the {_SQL_TIMEOUT}s time limit.")
        elif "err" in box:
            st.error(str(box["err"]))
        else:
            result = box["df"]
            st.caption(f"{len(result)} row{'s' if len(result) != 1 else ''} returned")
            st.download_button(
                label="Download as CSV",
                data=result.to_csv(index=False),
                file_name="query_result.csv",
                mime="text/csv",
            )
            st.dataframe(result, use_container_width=True, hide_index=True)


with tab_sql:
    _sql_explorer()

# ── About tab ──────────────────────────────────────────────────────────────────
with tab_about:

    st.markdown("""
This demo shows how to log per-request usage data for the DeepL API (billed characters,
language pairs, reporting tags, and API key identifiers) and explore it through a local dashboard.
It covers both text and document translation, and captures errors alongside successful requests so
teams can track usage and reliability in one place.

The stack for this demo is lightweight and easy to run: the DeepL Python client is wrapped to write
to a local DuckDB file, and Streamlit serves the dashboard with no external dependencies.

The pattern is meant to show what types of reporting users can build themselves when they need
more flexibility than DeepL offers via the API Usage tab, admin analytics API, and /usage endpoint.

The project is open source under the MIT license and can be found here in GitHub: https://github.com/DeepLcom/deepl-api-usage-logger
    """)