"""Prismarkedsføring: finds price and discount claims in Norwegian marketing text.

The model is ltg/norbert3-base, fine-tuned on sentences labelled against
Forbrukertilsynet's guidance on price marketing. It reads an advert one sentence at a
time, and each pair of neighbouring sentences, and shows which piece decided the answer.
"""

import io
import json
import os
import re
import tempfile
from pathlib import Path

import gradio as gr
import pandas as pd
import spaces
import torch
from transformers import AutoModelForSequenceClassification, AutoTokenizer

MODEL_ID = os.environ.get("MODEL_ID", "")
MODEL_DIR = Path(__file__).parent / "model"
# CPU on purpose: the free tier on Hugging Face Spaces has no GPU, and this model
# classifies a few hundred short texts a second on a CPU anyway.
# On ZeroGPU a GPU is lent to each request; locally this falls back to the CPU.
DEVICE = os.environ.get("DEVICE", "cuda" if torch.cuda.is_available() else "cpu")
torch.set_num_threads(max(1, (os.cpu_count() or 4) // 2))
# Between these the model is guessing, and a person should read the text.
UNCERTAIN = (0.35, 0.65)

DEFINITION = """
### Hva teller som en prispåstand

Dette er ikke vår definisjon. Den er hentet fra Forbrukertilsynets egen veiledning
*Salg og bruk av førpriser*, og fra vedtakene.

Det holder at teksten gir inntrykk av at prisen er satt ned eller er gunstig. Det trengs
ikke noe tall. Veiledningen nevner selv «-50 %», «Få 200 kr i avslag på alle bukser»,
«Black Friday-priser» og «SALG» som eksempler på salgsmarkedsføring. Det samme gjelder
påstander om å være billigst, og tilbud om gratis vare, prøve eller rentefri periode.
«Rekordbillig» ga overtredelsesgebyr i saken mot Stayclassy (FOV-2022-1369).

Navnet på et produkt teller ikke, selv om det inneholder ordet pris. «Medlemspris» og
«Spotpris med Elkjøprabatt» er avtalenavn. I de sakene var problemet manglende
opplysninger, ikke navnet. Det samme gjelder nøytrale prisopplysninger, avtalevilkår og
juridisk tekst om prismarkedsføring.
"""

# Written by hand; which answer each one gets is the model's call.
EXAMPLES_FILE = MODEL_DIR / "examples.json"
EXAMPLES = (
    json.loads(EXAMPLES_FILE.read_text())
    if EXAMPLES_FILE.exists()
    else [
        "Black Friday-priser på alle jakker denne helgen",
        "Norges billigste strømavtale, garantert",
        "Prøv gratis i 30 dager, betal kun frakt 79,-",
        "Strøm til en grei pris",
        "Medlemspris",
        "Spotpris med Elkjøprabatt",
        "Vi sier opp den gamle avtalen din for deg, helt uten styr",
        "Avtalen har ingen bindingstid og faktureres etterskuddsvis hver måned",
    ]
)
SPLIT = re.compile(r"(?<=[.!?:])\s+|\n+")


def load_model():
    source = MODEL_ID or (str(MODEL_DIR) if MODEL_DIR.exists() else "")
    if not source:
        return None, None
    # NorBERT ships its own model code, so it has to be allowed to run it.
    tok = AutoTokenizer.from_pretrained(source, trust_remote_code=True)
    model = AutoModelForSequenceClassification.from_pretrained(source, trust_remote_code=True)
    return tok, model.to(DEVICE).eval()


def load_temperature() -> float:
    # How much to soften the scores so that "70% sure" is right about 70% of the time.
    path = MODEL_DIR / "calibration.json"
    return json.loads(path.read_text())["temperature"] if path.exists() else 1.0


TOKENIZER, MODEL = load_model()
TEMPERATURE = load_temperature()


def gpu_seconds(texts: list[str], batch: int = 32) -> int:
    # Shorter requests get better queue priority, so ask only for what the batch needs.
    return min(120, 10 + len(texts) // 50)


@spaces.GPU(duration=gpu_seconds)
def score_sentences(texts: list[str], batch: int = 32) -> list[float]:
    """Calibrated probability that each sentence contains a price or discount claim."""
    out: list[float] = []
    with torch.no_grad():
        for i in range(0, len(texts), batch):
            enc = TOKENIZER(
                texts[i : i + batch],
                truncation=True,
                max_length=256,
                padding=True,
                return_tensors="pt",
            ).to(DEVICE)
            probs = torch.softmax(MODEL(**enc).logits / TEMPERATURE, dim=1)[:, 1]
            out.extend(probs.cpu().tolist())
    return out


def sentences(text: str) -> list[str]:
    return [s.strip() for s in SPLIT.split(text) if len(s.strip()) >= 3] or [text.strip()]


def pieces(text: str) -> list[str]:
    """Each sentence, and each pair of neighbouring sentences.

    Scoring sentences alone read «169,90 kr per kilo. Før 229,90.» as two harmless
    fragments and gave the advert 44%. Together they are a textbook before-price claim.
    """
    s = sentences(text)
    return s + [f"{a} {b}" for a, b in zip(s, s[1:], strict=False)]


def predict(texts: list[str]) -> tuple[list[float], list[str]]:
    """For each advert: the highest score of any piece, and the piece that got it."""
    flat = [(i, s) for i, t in enumerate(texts) for s in pieces(t)]
    scores = score_sentences([s for _, s in flat])
    best = [(0.0, "")] * len(texts)
    for (i, s), p in zip(flat, scores, strict=True):
        if p >= best[i][0]:
            best[i] = (p, s)
    return [p for p, _ in best], [s for _, s in best]


CUES = {
    "prosent": re.compile(r"\d+\s*%|prosent", re.I),
    "beløp": re.compile(r"\bkr\b|kroner|,-|øre\b", re.I),
    "sammenligning": re.compile(r"billigst|rimeligst|lavest|best\w* pris|mest for pengene", re.I),
    "gratis": re.compile(r"gratis|kostnadsfri\w*|uten ekstra kostnad|på kjøpet", re.I),
    "salgsord": re.compile(r"salg|tilbud|rabatt|black \w+|kampanje|kupp|spar\b", re.I),
    "førpris": re.compile(r"førpris|før kr|veil\.? pris|ordinær pris", re.I),
}


def cues_in(text: str) -> str:
    hits = [name for name, pat in CUES.items() if pat.search(text)]
    return ", ".join(hits) if hits else "ingen tydelige signalord"


def verdict(p: float) -> str:
    if p >= UNCERTAIN[1]:
        return "Prispåstand"
    if p <= UNCERTAIN[0]:
        return "Ingen prispåstand"
    return "Usikker, les den selv"


COLOURS = {"Prispåstand": "#e8590c", "Ingen prispåstand": "#2f9e44"}


def meter(p: float, call: str) -> str:
    """One bar that never moves: 0 to 100% chance of a price claim, usikker zone shaded.

    gr.Label sorted its bars by score, so the winner jumped to the top with a full bar
    every time. This keeps the same scale in the same place, so 72% looks like 72%.
    """
    colour = COLOURS.get(call, "#f59f00")
    lo, hi = UNCERTAIN[0] * 100, UNCERTAIN[1] * 100
    return f"""
<div style="padding:12px 4px">
  <div style="font-size:1.6em;font-weight:700;text-align:center;margin-bottom:14px">{call}</div>
  <div style="position:relative;height:18px;border-radius:9px;background:rgba(128,128,128,.25)">
    <div style="position:absolute;left:{lo}%;width:{hi - lo}%;top:0;bottom:0;
                background:rgba(245,159,0,.25)"></div>
    <div style="position:absolute;left:0;top:0;bottom:0;width:{p * 100:.1f}%;
                border-radius:9px;background:{colour}"></div>
  </div>
  <div style="display:flex;justify-content:space-between;font-size:.85em;margin-top:6px;
              opacity:.8">
    <span>0% · ingen prispåstand</span><span>usikker</span><span>prispåstand · 100%</span>
  </div>
  <div style="text-align:center;margin-top:10px;font-size:1.1em">
    Sannsynlighet for prispåstand: <b>{p:.0%}</b>
  </div>
</div>"""


def classify_one(text: str):
    if MODEL is None:
        return "", "Modellen er ikke trent ennå. Modellfilene mangler."
    text = (text or "").strip()
    if not text:
        return "", ""
    ps, triggers = predict([text])
    p, trigger = ps[0], triggers[0]
    call = verdict(p)
    label = meter(p, call)
    lines = []
    several = len(sentences(text)) > 1
    if call == "Prispåstand" and several:
        lines.append(f"Det som avgjør: «{trigger}»")
    elif call == "Usikker, les den selv":
        lines.append("Den ligger i grenseland, så her bør et menneske lese teksten.")
        if several:
            lines.append(f"Det den er mest i tvil om: «{trigger}»")
    cues = cues_in(text)
    if cues == "ingen tydelige signalord":
        lines.append(
            "Det er ingen typiske signalord i teksten, som *rabatt*, *kr* eller *billigst*."
        )
    else:
        lines.append(f"Signalord i teksten: {cues}.")
    return label, "\n\n".join(lines)


def classify_file(file, column: str, pasted: str):
    if MODEL is None:
        return None, "Modellen er ikke trent ennå. Modellfilene mangler.", None

    if file is not None:
        df = pd.read_csv(file.name)
        col = column.strip() or df.columns[0]
        if col not in df.columns:
            return (
                None,
                f"Finner ingen kolonne som heter «{col}». Filen har: {', '.join(df.columns)}",
                None,
            )
        texts = df[col].astype(str).tolist()
    elif pasted.strip():
        texts = [ln.strip() for ln in pasted.splitlines() if ln.strip()]
        df = pd.DataFrame({"tekst": texts})
        col = "tekst"
    else:
        return None, "Legg inn noe først: enten en CSV, eller tekst med én annonse per linje.", None

    probs, triggers = predict(texts)
    verdicts = [verdict(p) for p in probs]
    df = df.assign(
        sannsynlighet=[round(p, 3) for p in probs],
        vurdering=verdicts,
        avgjørende_setning=triggers,
        signalord=[cues_in(t) for t in texts],
    )
    df = df.sort_values("sannsynlighet", ascending=False)

    n = len(df)
    flagged = sum(v == "Prispåstand" for v in verdicts)
    unsure = sum(v.startswith("Usikker") for v in verdicts)
    summary = (
        f"**{n} tekster gjennomgått.** {flagged} ser ut til å inneholde en prispåstand. "
        f"{unsure} ligger i grenseland og bør noen lese selv. "
        f"De siste {n - flagged - unsure} ser rene ut."
    )

    buf = io.StringIO()
    df.to_csv(buf, index=False)
    path = Path(tempfile.mkdtemp()) / "vurdert.csv"
    path.write_text(buf.getvalue())
    return df.head(200), summary, str(path)


with gr.Blocks(title="Prismarkedsføring") as demo:
    gr.Markdown(
        "# Prismarkedsføring\n"
        "Lim inn en annonsetekst, så sier verktøyet om den inneholder en pris- eller "
        "rabattpåstand. Modellen er lært opp på sitater fra "
        "Forbrukertilsynets egne vedtak og på setninger fra norske nettbutikker. "
        "Definisjonen er tilsynets egen, ikke vår. Dette er et uavhengig prosjekt, "
        "ikke laget av eller tilknyttet Forbrukertilsynet."
    )

    with gr.Tab("Én tekst"):
        with gr.Row():
            with gr.Column():
                text_in = gr.Textbox(
                    label="Markedsføringstekst", lines=4, placeholder="Lim inn en annonsetekst"
                )
                gr.Examples(EXAMPLES, inputs=text_in)
                run_one = gr.Button("Sjekk teksten", variant="primary")
            with gr.Column():
                label_out = gr.HTML()
                note_out = gr.Markdown()
        run_one.click(classify_one, inputs=text_in, outputs=[label_out, note_out])
        text_in.submit(classify_one, inputs=text_in, outputs=[label_out, note_out])

    with gr.Tab("Mange tekster"):
        gr.Markdown(
            "Har du mange tekster? Last opp en CSV, eller lim inn én per linje. "
            "Du får dem tilbake sortert, med de mest sannsynlige først, så du kan lese "
            "toppen og la resten ligge."
        )
        with gr.Row():
            file_in = gr.File(label="CSV-fil", file_types=[".csv"])
            col_in = gr.Textbox(label="Hvilken kolonne?", placeholder="tom = første kolonne")
        paste_in = gr.Textbox(
            label="…eller bare lim inn teksten", lines=6, placeholder="én annonse per linje"
        )
        run_many = gr.Button("Sjekk alle", variant="primary")
        summary_out = gr.Markdown()
        table_out = gr.Dataframe(label="De 200 øverste", wrap=True)
        download_out = gr.File(label="Last ned hele lista")
        run_many.click(
            classify_file,
            inputs=[file_in, col_in, paste_in],
            outputs=[table_out, summary_out, download_out],
        )

    with gr.Accordion("Hva regnes egentlig som en prispåstand?", open=False):
        gr.Markdown(DEFINITION)

    gr.Markdown(
        "---\nVerktøyet bruker én modell, en norsk språkmodell som er trent spesielt på denne "
        "oppgaven. Når den er usikker, sier den det, og da bør et menneske lese teksten. "
        "Verktøyet foreslår, det avgjør ingenting."
    )

if __name__ == "__main__":
    demo.launch(server_name="0.0.0.0", server_port=int(os.environ.get("PORT", "7860")))