ScoDoc/app/scodoc/sco_photos.py

# -*- mode: python -*-
# -*- coding: utf-8 -*-

##############################################################################
#
# Gestion scolarite IUT
#
# Copyright (c) 1999 - 2024 Emmanuel Viennet.  All rights reserved.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
#
#   Emmanuel Viennet      emmanuel.viennet@gmail.com
#
##############################################################################

"""(Nouvelle) (Nouvelle) gestion des photos d'etudiants

Les images sont stockées dans .../var/scodoc/photos
L'attribut "photo_filename" de la table identite donne le nom du fichier image,
sans extension (e.g. "F44/RT_EID31545").
Toutes les images sont converties en jpg, et stockées dans photo_filename.jpg en taille originale.
Elles sont aussi réduites en 90 pixels de hauteur, et stockées dans photo_filename.h90.jpg

Les images sont servies par ScoDoc, via la méthode getphotofile?etudid=xxx


## Historique:
 - jusqu'à novembre 2009, les images étaient stockées dans Zope (ZODB).
 - jusqu'à v1908, stockées dans .../static/photos (et donc accessibles sans authentification).
 - support for legacy ZODB removed in v1909.

"""
import datetime
import glob
import io
import os
import random
import requests
import time
import traceback

import PIL
from PIL import Image as PILImage

from flask import abort, request, g, has_request_context
from flask.helpers import make_response, url_for

from app import log
from app import db
from app.models import Identite, Scolog
from app.scodoc import sco_etud
from app.scodoc import sco_portal_apogee
from app.scodoc import sco_preferences
from app.scodoc.sco_exceptions import ScoValueError
from app.scodoc.scolog import logdb
import app.scodoc.notesdb as ndb
import app.scodoc.sco_utils as scu
from config import Config

# Full paths on server's filesystem. Something like "/opt/scodoc-data/photos"
PHOTO_DIR = os.path.join(Config.SCODOC_VAR_DIR, "photos")
ICONS_DIR = os.path.join(Config.SCODOC_DIR, "app", "static", "icons")
UNKNOWN_IMAGE_PATH = os.path.join(ICONS_DIR, "unknown.jpg")

IMAGE_EXT = ".jpg"
JPG_QUALITY = 0.92
REDUCED_HEIGHT = 90  # pixels
MAX_FILE_SIZE = 4 * 1024 * 1024  # max allowed size for uploaded image, in bytes
H90 = ".h90"  # suffix for reduced size images


def unknown_image_url() -> str:
    "URL for 'unkwown' face image"
    return url_for("scolar.get_photo_image", scodoc_dept=g.scodoc_dept, etudid="")


def photo_portal_url(code_nip: str):
    """Returns external URL to retreive photo on portal,
    or None if no portal configured"""
    photo_url = sco_portal_apogee.get_photo_url()
    if photo_url and code_nip:
        return photo_url + "?nip=" + code_nip
    else:
        return None


def get_etud_photo_url(etudid, size="small"):
    return (
        url_for(
            "scolar.get_photo_image",
            scodoc_dept=g.scodoc_dept,
            etudid=etudid,
            size=size,
        )
        if has_request_context()
        else ""
    )


def etud_photo_url(etud: dict, size="small", fast=False) -> str:
    """url to the image of the student, in "small" size or "orig" size.
    If ScoDoc doesn't have an image and a portal is configured, link to it.

    """
    photo_url = get_etud_photo_url(etud["etudid"], size=size)
    if fast:
        return photo_url
    path = photo_pathname(etud["photo_filename"], size=size)
    if not path:
        # Portail ?
        ext_url = photo_portal_url(etud["code_nip"])
        if not ext_url:
            # fallback: Photo "unknown"
            photo_url = unknown_image_url()
        else:
            # essaie de copier la photo du portail
            new_path, _ = copy_portal_photo_to_fs(etud["etudid"])
            if not new_path:
                # copy failed, can we use external url ?
                # nb: rarement utile, car le portail est rarement accessible sans authentification
                if scu.CONFIG.PUBLISH_PORTAL_PHOTO_URL:
                    photo_url = ext_url
                else:
                    photo_url = unknown_image_url()
    return photo_url


def get_photo_image(etudid=None, size="small"):
    """Returns photo image (HTTP response)
    If not etudid, use "unknown" image
    """
    if not etudid:
        filename = UNKNOWN_IMAGE_PATH
    else:
        etud = Identite.get_etud(etudid)
        filename = photo_pathname(etud.photo_filename, size=size)
        if not filename:
            filename = UNKNOWN_IMAGE_PATH
    r = build_image_response(filename)
    return r


def build_image_response(filename):
    """returns an image as a Flask response"""
    st = os.stat(filename)
    last_modified = st.st_mtime  # float timestamp
    file_size = st.st_size
    header = request.headers.get("If-Modified-Since")
    if header is not None:
        header = header.split(";")[0]
        # Some proxies seem to send invalid date strings for this
        # header. If the date string is not valid, we ignore it
        # rather than raise an error to be generally consistent
        # with common servers such as Apache (which can usually
        # understand the screwy date string as a lucky side effect
        # of the way they parse it).
        try:
            dt = datetime.datetime.strptime(header, "%a, %d %b %Y %H:%M:%S GMT")
            mod_since = dt.timestamp()
        except ValueError:
            mod_since = None
        if (mod_since is not None) and last_modified <= mod_since:
            return make_response(b"", 304)  # not modified
    #
    last_modified_str = time.strftime(
        "%a, %d %b %Y %H:%M:%S GMT", time.gmtime(last_modified)
    )
    response = make_response(open(filename, mode="rb").read())
    response.headers["Content-Type"] = "image/jpeg"
    response.headers["Last-Modified"] = last_modified_str
    response.headers["Cache-Control"] = "max-age=3600"
    response.headers["Content-Length"] = str(file_size)
    return response


def etud_photo_is_local(photo_filename: str, size="small"):
    return photo_pathname(photo_filename, size=size)


def etud_photo_html(etud: dict = None, etudid=None, title=None, size="small") -> str:
    """HTML img tag for the photo, either in small size (h90)
    or original size (size=="orig")
    """
    if not etud:
        if etudid:
            etuds = sco_etud.get_etud_info(filled=True, etudid=etudid)
            if not etuds:
                return abort(404, "etudiant inconnu")
            etud = etuds[0]
        else:
            abort(404, "etud_photo_html: either etud or etudid must be specified")
    photo_url = etud_photo_url(etud, size=size)
    nom = etud.get("nomprenom", etud["nom_disp"])
    if title is None:
        title = nom
    if not etud_photo_is_local(etud["photo_filename"]):
        fallback = (
            f"""onerror='this.onerror = null; this.src="{unknown_image_url()}"'"""
        )
    else:
        fallback = ""
    if size == "small":
        height_attr = 'height="%s"' % REDUCED_HEIGHT
    else:
        height_attr = ""
    return '<img src="%s" alt="photo %s" title="%s" border="0" %s %s />' % (
        photo_url,
        nom,
        title,
        height_attr,
        fallback,
    )


def etud_photo_orig_html(etud=None, etudid=None, title=None) -> str:
    """HTML img tag for the photo, in full size.
    Full-size images are always stored locally in the filesystem.
    They are the original uploaded images, converted in jpeg.
    """
    return etud_photo_html(etud=etud, etudid=etudid, title=title, size="orig")


def photo_pathname(photo_filename: str, size="orig"):
    """Returns full path of image file if etud has a photo (in the filesystem),
    or False.
    Do not distinguish the cases: no photo, or file missing.
    Argument: photo_filename (Identite attribute)
    Resultat: False or str
    """
    if size == "small":
        version = H90
    elif size == "orig":
        version = ""
    else:
        abort(404, "invalid size parameter for photo")
    if not photo_filename:
        return False
    path = os.path.join(PHOTO_DIR, photo_filename) + version + IMAGE_EXT
    if os.path.exists(path):
        return path
    else:
        return False


def store_photo(etud: Identite, data, filename: str) -> tuple[bool, str]:
    """Store image for this etud.
    If there is an existing photo, it is erased and replaced.
    data is a bytes string with image raw data.

    Update database to store filename.

    Returns (status, err_msg)
    """
    # basic checks
    filesize = len(data)
    if filesize < 10 or filesize > MAX_FILE_SIZE:
        return False, f"Fichier image '{filename}' de taille invalide ! ({filesize})"
    try:
        saved_filename = save_image(etud, data)
    except (OSError, PIL.UnidentifiedImageError) as exc:
        raise ScoValueError(
            msg="Fichier d'image '{filename}' invalide ou format non supporté"
        ) from exc

    # update database:
    etud.photo_filename = saved_filename
    db.session.add(etud)
    Scolog.logdb(method="changePhoto", msg=saved_filename, etudid=etud.id)
    db.session.commit()
    #
    return True, "ok"


def suppress_photo(etud: Identite) -> None:
    """Suppress a photo"""
    log(f"suppress_photo {etud}")
    rel_path = photo_pathname(etud.photo_filename)
    # 1- remove ref. from database
    etud.photo_filename = None
    db.session.add(etud)

    # 2- erase images files
    if rel_path:
        # remove extension and glob
        rel_path = rel_path[: -len(IMAGE_EXT)]
        filenames = glob.glob(rel_path + "*" + IMAGE_EXT)
        for filename in filenames:
            log(f"removing file {filename}")
            os.remove(filename)
    db.session.commit()
    # 3- log
    cnx = ndb.GetDBConnexion()
    logdb(cnx, method="changePhoto", msg="suppression", etudid=etud.id)


# ---------------------------------------------------------------------------
# Internal functions


def save_image(etud: Identite, data: bytes):
    """data is a bytes string.
    Save image in JPEG in 2 sizes (original and h90).
    Returns filename (relative to PHOTO_DIR), without extension
    """
    data_file = io.BytesIO()
    data_file.write(data)
    data_file.seek(0)
    img = PILImage.open(data_file)
    filename = get_new_filename(etud)
    path = os.path.join(PHOTO_DIR, filename)
    log("saving %dx%d jpeg to %s" % (img.size[0], img.size[1], path))
    img = img.convert("RGB")
    img.save(path + IMAGE_EXT, format="JPEG", quality=92)
    # resize:
    img = scale_height(img)
    log("saving %dx%d jpeg to %s.h90" % (img.size[0], img.size[1], filename))
    img.save(path + H90 + IMAGE_EXT, format="JPEG", quality=92)
    return filename


def scale_height(img, W=None, H=REDUCED_HEIGHT):
    if W is None:
        # keep aspect
        W = int((img.size[0] * H) / img.size[1])
    img.thumbnail((W, H), PILImage.LANCZOS)
    return img


def get_new_filename(etud: Identite):
    """Constructs a random filename to store a new image.
    The path is constructed as: Fxx/etudid
    """
    dept = etud.departement.acronym
    return find_new_dir() + dept + "_" + str(etud.id)


def find_new_dir():
    """select randomly a new subdirectory to store a new file.
    We define 100 subdirectories named from F00 to F99.
    Returns a path relative to the PHOTO_DIR.
    """
    d = "F" + "%02d" % random.randint(0, 99)
    path = os.path.join(PHOTO_DIR, d)
    if not os.path.exists(path):
        # ensure photos directory exists
        if not os.path.exists(PHOTO_DIR):
            os.mkdir(PHOTO_DIR)
        # create subdirectory
        log(f"creating directory {path}")
        os.mkdir(path)
    return d + "/"


def copy_portal_photo_to_fs(etudid: int):
    """Copy the photo from portal (distant website) to local fs.
    Returns rel. path or None if copy failed, with a diagnostic message
    """
    etud: Identite = Identite.query.get_or_404(etudid)
    url = photo_portal_url(etud.code_nip)
    if not url:
        return None, f"""{etud.nomprenom}: pas de code NIP"""
    portal_timeout = sco_preferences.get_preference("portal_timeout")
    error_message = None
    try:
        r = requests.get(url, timeout=portal_timeout)
    except requests.ConnectionError:
        error_message = "ConnectionError"
    except requests.Timeout:
        error_message = "Timeout"
    except requests.TooManyRedirects:
        error_message = "TooManyRedirects"
    except requests.RequestException:
        error_message = "unknown requests error"
    if error_message is not None:
        log("sco_photos: download failed")
        log(traceback.format_exc())
        log(f"copy_portal_photo_to_fs: {error_message}")
        return (
            None,
            f"""{etud.nomprenom}: erreur chargement de {url}\n{error_message}""",
        )
    if r.status_code != 200:
        log(f"copy_portal_photo_to_fs: download failed {r.status_code }")
        return None, f"""{etud.nomprenom}: erreur chargement de {url}"""

    data = r.content  # image bytes
    try:
        status, error_message = store_photo(etud, data, "(inconnue)")
    except Exception:
        status = False
        error_message = "Erreur chargement photo du portail"
        log("copy_portal_photo_to_fs: failure (exception in store_photo)!")
    if status:
        log(f"copy_portal_photo_to_fs: copied {url}")
        return (
            photo_pathname(etud.photo_filename),
            f"{etud.nomprenom}: photo chargée",
        )
    else:
        return None, f"{etud.nomprenom}: <b>{error_message}</b>"