iserv-filter/taskfilter.py

"""filter_task: the one place that decides what leaves the server. Explicit whitelist, free text always filtered."""
import base64
import io
import mimetypes
import zipfile
from collections import Counter

from iserv import html_text
from nerfilter import filter_filename, filter_text, split_ext

READ_LIMIT = 20_000  # chars read from a file; bounds CPU time on huge PDFs
MAX_TEXT = 3_000  # chars of a file that go out: worksheets are long and Muse needs the gist
NOT_READABLE = "Inhalt nicht maschinell lesbar"
FILE_WITHHELD = "Originaldatei nicht herausgegeben (Namen enthalten oder nicht prüfbar)"
_MIME = {"pdf": "application/pdf", "txt": "text/plain", "md": "text/markdown",
         "docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document"}
_MEDIA = {"jpg": "bild", "jpeg": "bild", "png": "bild", "gif": "bild", "webp": "bild", "bmp": "bild", "heic": "bild",
          "svg": "bild", "mp3": "audio", "wav": "audio", "m4a": "audio", "ogg": "audio", "mp4": "video",
          "mov": "video", "avi": "video", "mkv": "video", "webm": "video"}


def _decode(data):
    try:
        return data.decode("utf-8")
    except UnicodeDecodeError:
        return data.decode("cp1252", errors="replace")


def _pdf(data):
    from pypdf import PdfReader

    reader = PdfReader(io.BytesIO(data))
    if reader.is_encrypted:
        return ""
    text = ""
    for page in reader.pages:
        text += (page.extract_text() or "") + "\n"
        if len(text) > READ_LIMIT:
            break
    return text


def _docx(data):
    with zipfile.ZipFile(io.BytesIO(data)) as z:
        if sum(i.file_size for i in z.infolist()) > 50 * 1024 * 1024:  # zip bomb
            return ""
    from docx import Document

    doc = Document(io.BytesIO(data))
    cells = {}  # a merged cell is returned once per grid column it spans: take each one only once
    for t in doc.tables:
        for r in t.rows:
            for c in r.cells:
                cells.setdefault(c._tc, c.text)
    return "\n".join([p.text for p in doc.paragraphs] + list(cells.values()))


def _html(data):
    return html_text(_decode(data))


_EXTRACT = {"pdf": _pdf, "docx": _docx, "txt": _decode, "md": _decode, "html": _html, "htm": _html}


def _tidy(text):
    """Drops what says nothing: blank lines, page numbers, a line identical to the one before (running headers)."""
    out = []
    for line in (l.strip() for l in text.splitlines()):
        if line and not line.isdigit() and (not out or line != out[-1]):
            out.append(line)
    return "\n".join(out)


def _docx_checked(data):
    """(text of headers and footers, cleaned bytes) of a docx, or None if it has comments (author names)."""
    with zipfile.ZipFile(io.BytesIO(data)) as z:
        if "word/comments.xml" in z.namelist():
            return None
    from docx import Document

    doc = Document(io.BytesIO(data))
    extra = "\n".join(p.text for s in doc.sections for part in (s.header, s.footer) for p in part.paragraphs)
    props = doc.core_properties
    for field in ("author", "last_modified_by", "title", "subject", "comments", "keywords", "category"):
        setattr(props, field, "")
    out = io.BytesIO()
    doc.save(out)
    return extra, out.getvalue()


def _pdf_clean(data):
    """The pages only: no document info, bookmarks, forms or annotations (links, comments)."""
    from pypdf import PdfReader, PdfWriter

    writer = PdfWriter()
    for page in PdfReader(io.BytesIO(data)).pages:
        page.pop("/Annots", None)
        writer.add_page(page)
    out = io.BytesIO()
    writer.write(out)
    return out.getvalue()


def _file_out(ext, data, text, stats=None):
    """(datei, filtered): {mime, base64} of the original file if nothing in it is a name (else None), and the filtered
    text, so the NER reads the file only once. Only what the NER can read is checked, so scans, images and audio never
    go out; a file cut at READ_LIMIT was not read completely (filtered is None: the caller filters the part it shows)."""
    if ext not in _MIME or not text or len(text) > READ_LIMIT:
        return None, None
    found = Counter()
    filtered = filter_text(text, found)
    if stats is not None:
        stats["replacements"] += found["replacements"]
    try:
        if ext == "docx":
            checked = _docx_checked(data)
            if checked is None:
                return None, filtered
            if checked[0].strip():
                filter_text(checked[0], found)
            data = checked[1]
        elif ext == "pdf":
            data = _pdf_clean(data)
    except Exception:
        return None, filtered
    if found["replacements"]:
        return None, filtered
    return {"mime": _MIME[ext], "base64": base64.b64encode(data).decode()}, filtered


def _unchecked(out, att, fid, release):
    """A file Muse may not see by itself: it gets an id to ask the user for. Only an id the user released
    (via the backend's /iserv/freigabe) hands out the original, as it is: it was not checked for names."""
    out["datei_hinweis"] = FILE_WITHHELD
    out["datei"] = None
    if fid in release:
        out["datei"] = {"mime": mimetypes.guess_type(att.get("name") or "")[0] or "application/octet-stream",
                        "base64": base64.b64encode(att["data"]).decode()}
        out["datei_hinweis"] = "Original auf Freigabe herausgegeben, nicht auf Namen geprüft"
    else:
        out["freigabe"] = fid
    return out


def describe_attachment(att, stats=None, files=False, fid=None, release=()):
    """{name, typ, groesse, text | hinweis}; with files also datei (original file) or datei_hinweis + freigabe id.
    The file bytes leave this function only through _file_out and _unchecked."""
    if att.get("skipped"):
        return {"hinweis": "weiterer Anhang nicht abgerufen"}
    name = att.get("name") or ""
    ext = split_ext(name)[1]
    out = {"name": filter_filename(name, stats), "typ": _MEDIA.get(ext) or ext or "unbekannt", "groesse": att.get("size")}
    if att.get("failed"):
        if stats is not None:
            stats["attachments_failed"] += 1
        return {**out, "hinweis": "Datei nicht abrufbar"}
    if att.get("data") is None:
        return {**out, "hinweis": "Datei größer als 10 MB, nicht abgerufen"}
    if ext not in _EXTRACT:
        out["hinweis"] = NOT_READABLE
        return _unchecked(out, att, fid, release) if files else out
    try:
        text = _tidy(_EXTRACT[ext](att["data"]))
    except Exception:  # corrupt/encrypted file; the message may quote its content
        text = ""
    if not text:
        out["hinweis"] = NOT_READABLE
        return _unchecked(out, att, fid, release) if files else out
    filtered = None
    if files:
        datei, filtered = _file_out(ext, att["data"], text, stats)
        if datei:
            out["datei"] = datei
        else:
            _unchecked(out, att, fid, release)
    shown = filtered if filtered is not None else text
    if len(shown) > MAX_TEXT:  # cut at a space so no name is split in half
        shown, out["gekuerzt"] = shown[:MAX_TEXT].rsplit(None, 1)[0], True
    return {**out, "text": shown if filtered is not None else filter_text(shown, stats)}


def filter_task(task, stats: Counter | None = None, files=False, release=()):
    """Whitelist: anything not named here (creator, groups, feedback, ...) cannot appear in the output."""
    return {
        "id": int(task["id"]),
        "fach": filter_text(task.get("tags", ""), stats),
        "titel": filter_text(task.get("title", ""), stats),
        "beschreibung": filter_text(task.get("description", ""), stats),
        "start": task.get("start"),
        "abgabe": task.get("due"),
        "status": None if "done" not in task else "erledigt" if task["done"] else "offen",  # None: source cannot tell
        "anhaenge": [describe_attachment(a, stats, files, f"{int(task['id'])}-{i}", release) for i, a in enumerate(task.get("attachments", []))],
    }