report_flags.py

"""
Report-guided rewriting with IMAGE PROCESSING.

Every page of an AI-detection or similarity report (Turnitin, iThenticate, GPTZero, ZeroGPT, Copyleaks, Academi,
our own report, ...) is rendered to an image. For every word on the page we look at the pixels inside the word's
box and decide whether it is flagged:

  * highlighted  - the BACKGROUND behind the letters is tinted (cyan, purple, yellow, pink, green ...). Even very
                   pale tints count (Turnitin-style "moderate" purple), plain white / grey does not.
  * coloured text - the LETTERS themselves are coloured (similarity reports colour matched text by source).

Word positions come from the PDF's text layer; for scanned / image-only reports they come from OCR (Tesseract).
Flagged words are grouped into the report's sentences; a report sentence is flagged when most of its words are.
Those sentences are then matched to the customer's document (word-trigram overlap), and Smart Humanize rewrites
exactly those sentences. The report's headline percentage is read as well, for the before / estimated-after score.
"""
import contextvars
import re

ACTIVE = contextvars.ContextVar("report_flags", default=None)   # set per job when a report is uploaded

DPI = 110                # render resolution for the pixel analysis (pages with a text layer)
OCR_DPI = 150            # render resolution for scanned pages (OCR needs more detail)
TINT_MIN = 9             # min (max-min) channel spread for a background pixel to count as tinted (pale purple ~ 12)
TINT_SHARE = 0.45        # share of a word's background pixels that must be tinted to call the word highlighted
INK_COLOUR = 60          # min channel spread for a letter pixel to count as coloured ink
INK_SHARE = 0.45         # share of a word's letter pixels that must be coloured to call the word coloured
SENT_SHARE = 0.5         # AI report: share of a report sentence's words that must be flagged
MATCH_SHARE = 0.5        # AI report: share of a document sentence's word trigrams found in flagged report text
SIM_SENT_SHARE = 0.25    # similarity report: matched phrases are often only part of a sentence
SIM_MATCH_SHARE = 0.25
KEEP_OVERLAP = 0.5       # wording overlap at/above which a rewritten sentence is assumed to stay flagged

_WORD = re.compile(r"[A-Za-z0-9]+(?:['’][A-Za-z]+)?")
_SENT_END = re.compile(r"(?<=[.!?])[\"')\]]*$")


class ReportError(Exception):
    """The report could not be used (wrong file, no text, nothing flagged)."""


def _norm(text):
    return [w.lower().replace("’", "'") for w in _WORD.findall(text or "")]


def _tri(words):
    return {tuple(words[i:i + 3]) for i in range(len(words) - 2)}


def _hue_name(rgb):
    r, g, b = [x / 255 for x in rgb]
    mx, mn = max(r, g, b), min(r, g, b)
    if mx - mn < 0.02:
        return "flagged"
    if mx == r:
        h = (60 * ((g - b) / (mx - mn)) + 360) % 360
    elif mx == g:
        h = 60 * ((b - r) / (mx - mn)) + 120
    else:
        h = 60 * ((r - g) / (mx - mn)) + 240
    for lim, name in ((15, "red"), (45, "orange"), (70, "yellow"), (160, "green"), (200, "cyan"), (250, "blue"),
                      (290, "purple"), (335, "pink"), (361, "red")):
        if h < lim:
            return name
    return "flagged"


# ------------------------------------------------------------------ page analysis
def _page_words(page, img, scale):
    """[(text, x0, y0, x1, y1)] in IMAGE pixels: from the text layer, else OCR."""
    words = [(w[4], w[0] * scale, w[1] * scale, w[2] * scale, w[3] * scale)
             for w in sorted(page.get_text("words"), key=lambda w: (w[5], w[6], w[7]))]
    if words:
        return words, False
    try:
        import pytesseract
        from PIL import Image
        data = pytesseract.image_to_data(Image.fromarray(img), output_type=pytesseract.Output.DICT)
    except Exception:
        return [], True
    out = []
    for i, t in enumerate(data["text"]):
        if t.strip() and float(data["conf"][i]) > 30:
            x, y, w, h = data["left"][i], data["top"][i], data["width"][i], data["height"][i]
            out.append((t, x, y, x + w, y + h))
    return out, True


def _classify(img, box):
    """(flagged, colour_name) for one word box, from its pixels."""
    import numpy as np
    h, w = img.shape[:2]
    x0, y0, x1, y1 = max(0, int(box[0]) - 1), max(0, int(box[1])), min(w, int(box[2]) + 1), min(h, int(box[3]))
    if x1 - x0 < 2 or y1 - y0 < 2:
        return False, None
    px = img[y0:y1, x0:x1].reshape(-1, 3).astype(int)
    mx, mn = px.max(1), px.min(1)
    spread = mx - mn
    bright = mx >= 170
    bg = bright & (px.sum(1) >= 3 * 150)               # background (between and around the letters)
    ink = (px.sum(1) < 3 * 170) | ((spread >= INK_COLOUR) & (mn < 120))   # letters: dark OR strongly coloured
    if ink.sum() >= 4:
        coloured = spread[ink] >= INK_COLOUR
        if coloured.mean() >= INK_SHARE:
            return True, _hue_name(tuple(px[ink][coloured].mean(0)))
    if bg.sum() >= 6:
        tinted = spread[bg] >= TINT_MIN
        if tinted.mean() >= TINT_SHARE:
            return True, _hue_name(tuple(px[bg][tinted].mean(0)))
    return False, None


def _report_words(path, progress=None):
    """All words of the report in reading order: [(text, flagged, colour)], plus the first pages' text."""
    import fitz  # PyMuPDF
    import numpy as np
    try:
        pdf = fitz.open(path)
    except Exception as err:
        raise ReportError(f"Could not open the report: {err}")
    out, head, used_ocr, any_words = [], [], False, False
    n = len(pdf)
    pages = []                                         # (page, img, dpi)
    for i, page in enumerate(pdf):
        if progress is not None:
            progress(0.02 + 0.02 * i / max(1, n), desc=f"Reading report page {i + 1} of {n}")
        dpi = DPI if page.get_text("words") else OCR_DPI          # scanned page: render sharper for OCR
        pix = page.get_pixmap(dpi=dpi, alpha=False)
        pages.append((page, np.frombuffer(pix.samples, np.uint8).reshape(pix.h, pix.w, pix.n)[:, :, :3].copy(), dpi))
    # word positions: text layer, or OCR for scanned pages (several pages at once)
    from concurrent.futures import ThreadPoolExecutor
    import os
    with ThreadPoolExecutor(max_workers=max(1, min(4, os.cpu_count() or 1))) as pool:
        found = list(pool.map(lambda x: _page_words(x[0], x[1], x[2] / 72.0), pages))
    for i, ((page, img, dpi), (words, ocr)) in enumerate(zip(pages, found)):
        used_ocr |= ocr
        any_words |= bool(words)
        for t, x0, y0, x1, y1 in words:
            f, c = _classify(img, (x0, y0, x1, y1))
            out.append((t, f, c))
        if i < 3:
            head.append(" ".join(t for t, *_ in words) if ocr else page.get_text())
        out.append(("\n", False, None))                # page break
    if not any_words:
        raise ReportError("This report is a scanned image and text recognition (OCR) is not available on the "
                          "server. Download the report as a PDF from the checker's website and upload that file.")
    return out, "\n".join(head), used_ocr


def _report_sentences(words):
    """Group the report's words into sentences -> [(sentence, flagged_share, main_colour)]."""
    sents, cur = [], []

    def close():
        if cur:
            toks = [w for w in cur if w[0] != "\n"]
            if toks:
                text = re.sub(r"(\w)- (\w)", r"\1\2", " ".join(t for t, _, _ in toks))
                fl = [c for _, f, c in toks if f]
                share = len(fl) / len(toks)
                colour = max(set(fl), key=fl.count) if fl else None
                sents.append((text, share, colour))
            cur.clear()

    prev_flag = None
    for t, f, c in words:
        if t == "\n":
            continue
        # a change between flagged and unflagged text also ends a "sentence" (reports highlight whole sentences)
        if prev_flag is not None and f != prev_flag and len(cur) >= 3:
            close()
        cur.append((t, f, c))
        prev_flag = f
        if _SENT_END.search(t):
            close()
            prev_flag = None
    close()
    return sents


# ------------------------------------------------------------------ score
_TOOLS = [("turnitin", "Turnitin"), ("ithenticate", "iThenticate"), ("gptzero", "GPTZero"), ("zerogpt", "ZeroGPT"),
          ("copyleaks", "Copyleaks"), ("originality", "Originality.ai"), ("quillbot", "QuillBot"),
          ("grammarly", "Grammarly"), ("scribbr", "Scribbr"), ("winston", "Winston AI"), ("drillbit", "DrillBit"),
          ("ouriginal", "Ouriginal"), ("urkund", "Ouriginal"), ("academi", "Academi"),
          ("cs software solutions", "CS report")]
_PCT = re.compile(r"(\d{1,3}(?:\.\d+)?)\s?%")
_AI_KEYS = ("detected as ai", "ai-generated", "ai generated", "ai content", "likely ai", "written by ai",
            "ai writing", "% ai", "gpt", "artificial intelligence", "ai detection", "ai score", "probability")
_SIM_KEYS = ("overall similarity", "similarity index", "similarity", "plagiarism", "matched", "match")


def read_score(text, want=None):
    """(percent, kind, tool) from the report's first pages, e.g. (39.0, 'AI', 'Academi'); percent None if absent.
    want = 'AI' or 'Similarity' (the customer's choice) prefers that kind of score."""
    low = (text or "").lower()
    low = re.sub(r"\bal\b", "ai", low).replace("al-generated", "ai-generated")   # OCR reads "AI" as "Al"
    tool = next((name for key, name in _TOOLS if key in low), "Report")
    best = None
    for m in _PCT.finditer(low):
        pct = float(m.group(1))
        if pct > 100:
            continue
        ls, le = low.rfind("\n", 0, m.start()) + 1, low.find("\n", m.end())
        line = low[ls: le if le >= 0 else len(low)]
        if re.search(r"human|original\b|\bmixed", line) and not re.search(r"\bai\b|gpt|similar", line):
            continue
        ctx = low[max(0, m.start() - 90): m.end() + 90]
        near = low[max(0, m.start() - 40): m.end() + 40]
        if re.search(r"detected as ai|overall similarity|similarity index", near):
            pri = 0
        elif any(k in near for k in _AI_KEYS + _SIM_KEYS):
            pri = 1
        elif any(k in ctx for k in _AI_KEYS + _SIM_KEYS):
            pri = 2
        else:
            continue
        ai = any(k in ctx for k in _AI_KEYS) and not re.search(r"overall similarity|similarity index", near)
        kind = "AI" if ai else "Similarity"
        if want and kind != want:
            pri += 3                                   # the other kind of score: only if nothing else is found
        cand = (pri, m.start(), pct, kind)
        if best is None or cand[:2] < best[:2]:
            best = cand
    if best is None:
        return None, None, tool
    return best[2], best[3], tool


def estimate_after(score, pairs, kind="AI"):
    """Estimated report score after rewriting; pairs = [(flagged original sentence, final sentence)]. A sentence
    only starts to count as cleared once more than half of its wording changed (detectors also flag lightly
    reworded / AI-paraphrased text)."""
    if score is None or not pairs:
        return None
    from humanizer import similarity_to_source
    tot = kept = 0.0
    for o, n in pairs:
        w = max(1, len(o.split()))
        tot += w
        ov = 1.0 if n.strip() == o.strip() else similarity_to_source(o, n) / 100.0
        # similarity checkers match wording directly; AI detectors also flag lightly reworded text
        kept += w * (ov if kind == "Similarity" else min(1.0, ov / KEEP_OVERLAP))
    return round(score * kept / tot, 1) if tot else score


# ------------------------------------------------------------------ public API
class Flags:
    def __init__(self, sentences, head_text, used_ocr, kind="AI"):
        self.sim = kind == "Similarity"
        need = SIM_SENT_SHARE if self.sim else SENT_SHARE
        self.match_share = SIM_MATCH_SHARE if self.sim else MATCH_SHARE
        self.flagged = [(s, c) for s, sh, c in sentences if sh >= need and len(_norm(s)) >= 5]   # skip headings like "39% detected as AI"
        self.score, found_kind, self.tool = read_score(head_text, kind)
        self.kind = kind
        self.used_ocr = used_ocr
        self._tri = []
        for s, c in self.flagged:
            self._tri.append(_tri(_norm(s)))
        self.all_tri = set().union(*self._tri) if self._tri else set()

    def match(self, sentence):
        """(flagged?, report_sentence, colour) for one sentence of the customer's document."""
        t = _tri(_norm(sentence))
        if not t or not (t & self.all_tri):
            return False, None, None
        if sum(1 for x in t if x in self.all_tri) / len(t) < self.match_share:
            return False, None, None
        best, score = None, 0.0
        for (s, c), rt in zip(self.flagged, self._tri):
            ov = len(t & rt) / len(t)
            if ov > score:
                best, score = (s, c), ov
        return True, (best[0] if best else sentence), (best[1] if best else None)


def load(path, progress=None, kind="AI"):
    p = str(path or "")
    if not p.lower().endswith(".pdf"):
        raise ReportError("Please upload the report as a PDF, as downloaded from the checker.")
    words, head, ocr = _report_words(p, progress)
    k = str(kind).lower()
    if k.startswith("auto"):                           # what the report itself says it measures
        k = (read_score(head)[1] or "AI").lower()
    flags = Flags(_report_sentences(words), head, ocr, "Similarity" if k.startswith("sim") else "AI")
    if not flags.flagged:
        raise ReportError("No highlighted or coloured sentences were found in this report. Upload the full report "
                          "PDF (the version with the highlighted text).")
    return flags