Provides an API which can be used by pict-rs `external_validation` option
from dotenv import load_dotenv
import os
import logging
import sys

load_dotenv()

from pictrs_safety_api.argparser import args
from pictrs_safety_api.flask import APP
from loguru import logger

if __name__ == "__main__":
    if os.getenv("FEDIVERSE_SAFETY_WORKER_AUTH") is None:
        logger.critical("You must set the FEDIVERSE_SAFETY_WORKER_AUTH env var")
        sys.exit(1)
    if os.getenv("FEDIVERSE_SAFETY_IMGDIR") is None:
        logger.critical("You must set the FEDIVERSE_SAFETY_IMGDIR env var")
        sys.exit(1)

    # Only setting this for the WSGI logs
    logging.basicConfig(format='%(asctime)s - %(levelname)s - %(module)s:%(lineno)d - %(message)s', level=logging.DEBUG)
    from waitress import serve
    logger.init("WSGI Server", status="Starting")
    url_scheme = 'https'
    if args.insecure:
        os.environ['OAUTHLIB_INSECURE_TRANSPORT'] = '1'  # Disable this on prod
        url_scheme = 'http'
    allowed_host = "127.0.0.1"
    if args.insecure:
        allowed_host = "0.0.0.0"
        logger.init_warn("WSGI Mode", status="Insecure")
    if args.listen:
        allowed_host = args.listen
    serve(APP, host=allowed_host, port=args.port, url_scheme=url_scheme, threads=45, connection_limit=1024, asyncore_use_poll=True)
    logger.init("WSGI Server", status="Stopped")


api_server.py
db0/pictrs-safety
import os
import socket
from uuid import uuid4

from pictrs_safety_api.logger import logger
from pictrs_safety_api.flask import APP
from pictrs_safety_api.routes import * 
from pictrs_safety_api.apis import apiv1
from pictrs_safety_api.argparser import args
from pictrs_safety_api.consts import PICTRS_SAFETY_VERSION
import hashlib

APP.register_blueprint(apiv1)


@APP.after_request
def after_request(response):
    response.headers["Access-Control-Allow-Origin"] = "*"
    response.headers["Access-Control-Allow-Methods"] = "POST, GET, OPTIONS, PUT, DELETE, PATCH"
    response.headers["Access-Control-Allow-Headers"] = "Accept, Content-Type, Content-Length, Accept-Encoding, X-CSRF-Token, apikey, Client-Agent, X-Fields"
    response.headers["Fedi-Safety-API-Node"] = f"{socket.gethostname()}:{args.port}:{PICTRS_SAFETY_VERSION}"
    try:
        etag = hashlib.sha1(response.get_data()).hexdigest()
    except RuntimeError:
        etag = "Runtime Error"
    response.headers["ETag"] = etag
    return response

pictrs_safety_api/__init__.py
db0/pictrs-safety
from .apiv1 import blueprint as apiv1

pictrs_safety_api/apis/__init__.py
db0/pictrs-safety
from flask import Blueprint
from flask_restx import Api
from importlib import import_module

from pictrs_safety_api.apis.v1 import api as v1

blueprint = Blueprint('apiv1', __name__, url_prefix='/api')
api = Api(blueprint,
    version='1.0', 
    title=f'Pict-rs Safety',
    description=f'The API documentation for the Pict-rs Safety',
    contact_email="mail@dbzer0.com",
    default="v1",
    default_label="Latest Version",
    ordered=True,
)

api.add_namespace(v1)

pictrs_safety_api/apis/apiv1.py
db0/pictrs-safety
from flask_restx import fields

class Models:
    def __init__(self,api):
        self.response_model_error = api.model('RequestError', {
            'message': fields.String(description="The error message for this status code."),
        })
        self.response_model_simple_response = api.model('SimpleResponse', {
            "message": fields.String(default='OK',required=True, description="The result of this operation."),
        })
pictrs_safety_api/apis/models/v1.py
db0/pictrs-safety
import pictrs_safety_api.apis.v1.base as base
from pictrs_safety_api.apis.v1.base import api

api.add_resource(base.Scan, "/scan/<string:pictrs_id>")
api.add_resource(base.Pop, "/pop")

pictrs_safety_api/apis/v1/__init__.py
db0/pictrs-safety
import os
import time
import json
from uuid import uuid4
from PIL import Image
from io import BytesIO
from werkzeug.datastructures import FileStorage
from flask import request, send_file
from flask_restx import Namespace, Resource, reqparse
from pictrs_safety_api.flask import cache, db
from loguru import logger
from pictrs_safety_api.classes.request import ScanRequest
from pictrs_safety_api.database import functions as database
from pictrs_safety_api import exceptions as e
from pictrs_safety_api import enums
from pictrs_safety_api.classes.worker import worker

api = Namespace('v1', 'API Version 1' )

from pictrs_safety_api.apis.models.v1 import Models

models = Models(api)

handle_bad_request = api.errorhandler(e.BadRequest)(e.handle_bad_requests)
handle_forbidden = api.errorhandler(e.Forbidden)(e.handle_bad_requests)
handle_unauthorized = api.errorhandler(e.Unauthorized)(e.handle_bad_requests)
handle_not_found = api.errorhandler(e.NotFound)(e.handle_bad_requests)
handle_too_many_requests = api.errorhandler(e.TooManyRequests)(e.handle_bad_requests)
handle_internal_server_error = api.errorhandler(e.InternalServerError)(e.handle_bad_requests)
handle_service_unavailable = api.errorhandler(e.ServiceUnavailable)(e.handle_bad_requests)

# Used to for the flask limiter, to limit requests per url paths
def get_request_path():
    # logger.info(dir(request))
    return f"{request.remote_addr}@{request.method}@{request.path}"


class Scan(Resource):
    post_parser = api.parser()
    post_parser.add_argument("Content-Type", type=str, required=False, help="The file's media type.", location="headers")
    post_parser.add_argument('file', location='files', type=FileStorage, required=False)

    @api.expect(post_parser)
    def post(self,pictrs_id):
        '''Scan an image
        '''
        # I don't get why this is not using the import from earlier...
        from pictrs_safety_api import exceptions as e
        if pictrs_id == "IPADDR":
            if request.remote_addr not in json.loads(os.getenv("KNOWN_PICTRS_IPS","[]"))\
                    and not self.is_private_ip(request.remote_addr):
                raise e.Unauthorized("You are not authorized to use this service", f"Unauthorized IP: {request.remote_addr}")
        elif pictrs_id not in json.loads(os.getenv("KNOWN_PICTRS_IDS", "[]")):
            raise e.Unauthorized("You are not authorized to use this service", f"Unauthorized ID: {pictrs_id}")
        if worker.is_stale():
            logger.warning(f"Returning OK due to stale worker. Last seen: {worker.last_check_in}")
            return {"message": "Worker Stale"}, 200 
        scan_threshold = int(os.getenv("SCAN_BYPASS_THRESHOLD", 10))
        if scan_threshold > 0 and database.count_waiting_scan_requests() > scan_threshold:
            logger.warning(f"Returning OK due to full queue. Queue: {database.count_waiting_scan_requests()}")
            return {"message": "Queue Full"}, 200 
        self.args = self.post_parser.parse_args()
        file = self.args["file"]
        if not file:
            img_data = BytesIO(request.data)
            filetext = request.headers["Content-Type"].split('/',1)[1]
            upload_filename = f"{uuid4()}.{filetext}"
            if not img_data:
                raise e.BadRequest("No file provided","Missing file")
        else:
            upload_filename = file.filename
            allowed_extensions = {'png', 'jpg', 'jpeg', 'gif', 'webp'}
            if not upload_filename.lower().endswith(tuple(allowed_extensions)):
                raise e.BadRequest("Invalid file format",f"Invalid file format: {upload_filename}")
            img_data = BytesIO(file.read())
        self.filename = f"{os.getenv('FEDIVERSE_SAFETY_IMGDIR')}/{upload_filename}"
        try:
            img = Image.open(img_data)
            img.save(self.filename)
            new_request = ScanRequest(
                image = upload_filename
            )
            db.session.add(new_request)
            db.session.commit()
            retries = 0
            while new_request.state in [enums.State.PROCESSING, enums.State.WAITING]:
                db.session.refresh(new_request)
                retries += 1
                logger.debug([new_request,new_request.state])
                if retries > int(os.getenv('FEDIVERSE_SAFETY_TIME_THRESHOLD', 120)):
                    os.remove(self.filename)
                    db.session.delete(new_request)
                    db.session.commit()
                    logger.warning("Scanning timeout")
                    return {"message": "Could not scan request in reasonable amount of time. Returning OK"}, 200
                time.sleep(1)
            os.remove(self.filename)
            if new_request.state == enums.State.FAULTED:
                db.session.delete(new_request)
                db.session.commit()
                logger.warning("Faulted request. Returning OK")
                return {"message": "Faulted request. Returning OK"}, 200 
            if new_request.state == enums.State.DONE:
                if new_request.is_csam == True:
                    db.session.delete(new_request)
                    db.session.commit()
                    logger.warning("Potential CSAM Image detected")
                    return {"message": "Potential CSAM Image detected"}, 406
                else: 
                    db.session.delete(new_request)
                    db.session.commit()
                    logger.debug("Image OK")
                    return {"message": "Image OK"}, 200 
            else:
                db.session.delete(new_request)
                db.session.commit()
                logger.error(f"Image with state {new_request.state} detected!")
                return {"message": "Should not be here. Returning OK"},200
        except Exception as err:
            logger.error(f"Exception while processing scan: {err}")
            db.session.delete(new_request)
            db.session.commit()
            return {"message": "Something went wrong internally. Returning OK"}, 200
    
    def is_private_ip(self,remote_addr):
        if remote_addr == "127.0.0.1":
            return True
        if remote_addr.startswith("10"):
            s = remote_addr.split('.')[1]
            if int(s) in range(0,255):
                return True
        if remote_addr.startswith("172"):
            s = remote_addr.split('.')[1]
            if int(s) in range(16,31):
                return True
        if remote_addr.startswith("192"):
            s = remote_addr.split('.')[1]
            if int(s) == 168:
                return True
        return False


class Pop(Resource):
    get_parser = api.parser()
    get_parser.add_argument("apikey", type=str, required=True, help="The auth api key", location="headers")

    @api.expect(get_parser)
    def get(self):
        '''Pick up an image to safety validate
        '''
        # I don't get why this is not using the import from earlier...
        # logger.debug(request.remote_addr)
        self.args = self.get_parser.parse_args()
        if os.getenv("FEDIVERSE_SAFETY_WORKER_AUTH") != self.args.apikey:
            raise e.Forbidden("Access Denied")
        worker.check_in()
        pop: ScanRequest = database.find_waiting_scan_request()
        if not pop:
            return {"message": "Nothing to do"},204
        pop.state = enums.State.PROCESSING
        db.session.commit()
        return send_file(f"{os.getenv('FEDIVERSE_SAFETY_IMGDIR')}/{pop.image}", as_attachment=True, download_name=pop.image)

    post_parser = api.parser()
    post_parser.add_argument("apikey", type=str, required=True, help="The auth api key", location="headers")
    post_parser.add_argument("image", type=str, required=True, help="The image filename", location="json")
    post_parser.add_argument("is_csam", type=bool, required=True, help="Is this image csam?", location="json")
    post_parser.add_argument("is_nsfw", type=bool, required=True, help="Is this image nsfw?", location="json")

    @api.expect(post_parser)
    def post(self):
        '''Submit an image result
        '''
        # I don't get why this is not using the import from earlier...
        self.args = self.post_parser.parse_args()
        if os.getenv("FEDIVERSE_SAFETY_WORKER_AUTH") != self.args.apikey:
            raise e.Forbidden("Access Denied")
        pop: ScanRequest = database.find_scan_request_by_name(self.args.image)
        if not pop:
            raise e.NotFound("No image found waiting for this result.")
        pop.is_csam = self.args.is_csam
        pop.is_nsfw = self.args.is_nsfw
        pop.state = enums.State.DONE
        db.session.commit()
        return {"message": "OK"},201

pictrs_safety_api/apis/v1/base.py
db0/pictrs-safety
import argparse

arg_parser = argparse.ArgumentParser()
arg_parser.add_argument('-i', '--insecure', action="store_true", help="If set, will use http instead of https (useful for testing)")
arg_parser.add_argument('-l', '--listen', action="store", help="Specify which IP to listen to")
arg_parser.add_argument('-v', '--verbosity', action='count', default=0, help="The default logging level is ERROR or higher. This value increases the amount of logging seen in your screen")
arg_parser.add_argument('-q', '--quiet', action='count', default=0, help="The default logging level is ERROR or higher. This value decreases the amount of logging seen in your screen")
arg_parser.add_argument('-p', '--port', action='store', default=14051, required=False, type=int, help="Provide a different port to start with")
arg_parser.add_argument('--color', default=False, action="store_true", help="Enabled colorized logs")
args = arg_parser.parse_args()

pictrs_safety_api/argparser.py
db0/pictrs-safety
from pictrs_safety_api.flask import db, APP

# Importing for DB creation
from pictrs_safety_api.classes.request import ScanRequest

with APP.app_context():

    db.create_all()
pictrs_safety_api/classes/__init__.py
db0/pictrs-safety
from datetime import datetime
from sqlalchemy import Enum

from loguru import logger
from pictrs_safety_api.flask import db
from pictrs_safety_api import enums


class ScanRequest(db.Model):
    __tablename__ = "scan_requests"

    id = db.Column(db.Integer, primary_key=True)
    image = db.Column(db.Text, unique=False, nullable=False, index=True)
    created = db.Column(db.DateTime, default=datetime.utcnow, nullable=False)
    updated = db.Column(db.DateTime, default=datetime.utcnow, nullable=False)
    state = db.Column(Enum(enums.State), default=enums.State.WAITING, nullable=False)
    is_csam = db.Column(db.Boolean, nullable=True)
    is_nsfw = db.Column(db.Boolean, nullable=True)
    
    def create(self):
        db.session.add(self)
        db.session.commit()

pictrs_safety_api/classes/request.py
db0/pictrs-safety
import os
from datetime import datetime, timedelta

class Worker():
    last_check_in = None
    
    def __init__(self):
        self.check_in()

    def check_in(self):
        self.last_check_in = datetime.utcnow()
    
    def is_stale(self):
        threshold = int(os.getenv("MISSING_WORKER_THRESHOLD", 5))
        # When threshold is 0, the worker is never stale
        if threshold == 0:
            return False
        return self.last_check_in < datetime.utcnow() - timedelta(seconds=threshold)

worker = Worker()
pictrs_safety_api/classes/worker.py
db0/pictrs-safety
PICTRS_SAFETY_VERSION = "v1.2.2"

pictrs_safety_api/consts.py
db0/pictrs-safety
from loguru import logger
from sqlalchemy import func, or_, and_, not_, Boolean
from sqlalchemy.orm import noload
from pictrs_safety_api.flask import db
from pictrs_safety_api.utils import hash_api_key
from sqlalchemy.orm import joinedload
from pictrs_safety_api.classes.request import ScanRequest
from pictrs_safety_api import enums

def find_waiting_scan_request():
    query = ScanRequest.query.filter(
        ScanRequest.state == enums.State.WAITING,
    ).order_by(
        ScanRequest.created.asc()
    ).with_for_update(
        skip_locked=True, 
        of=ScanRequest
    )
    return query.first()

def find_scan_request_by_name(image_name: str):
    query = ScanRequest.query.filter(
        ScanRequest.image == image_name,
    )
    return query.first()

def find_scan_request_by_id(scan_id):
    query = ScanRequest.query.filter(
        ScanRequest.id == scan_id,
    )
    return query.first()


def count_waiting_scan_requests():
    query = ScanRequest.query.filter(
        ScanRequest.state == enums.State.WAITING,
    )
    return query.count()

pictrs_safety_api/database/functions.py
db0/pictrs-safety
import enum

class State(enum.Enum):
    WAITING = 0
    PROCESSING = 1
    DONE = 2
    CANCELLED = 3
    FAULTED = 4
    PARTIAL = 5
pictrs_safety_api/enums.py
db0/pictrs-safety
from werkzeug import exceptions as wze
from loguru import logger

class BadRequest(wze.BadRequest):
    def __init__(self, message, log="BadRequest"):
        self.specific = message
        self.log = log

class Forbidden(wze.Forbidden):
    def __init__(self, message, log="Forbidden"):
        self.specific = message
        self.log = log

class Unauthorized(wze.Unauthorized):
    def __init__(self, message, log="Unauthorized"):
        self.specific = message
        self.log = log

class NotFound(wze.NotFound):
    def __init__(self, message, log="NotFound"):
        self.specific = message
        self.log = log

class Locked(wze.Locked):
    def __init__(self, message, log="Locked"):
        self.specific = message
        self.log = log

class TooManyRequests(wze.TooManyRequests):
    def __init__(self, message, log="TooManyRequests"):
        self.specific = message
        self.log = log

class InternalServerError(wze.InternalServerError):
    def __init__(self, message, log="InternalServerError"):
        self.specific = message
        self.log = log
        
class ServiceUnavailable(wze.ServiceUnavailable):
    def __init__(self, message, log="ServiceUnavailable"):
        self.specific = message
        self.log = log
        
def handle_bad_requests(error):
    '''Namespace error handler'''
    if error.log:
        logger.debug(error.log)
    return({'message': error.specific}, error.code)

pictrs_safety_api/exceptions.py
db0/pictrs-safety
import os
from flask import Flask, redirect
from flask_caching import Cache
from werkzeug.middleware.proxy_fix import ProxyFix
from flask_sqlalchemy import SQLAlchemy
from loguru import logger
from sqlalchemy import event
from sqlalchemy.engine import Engine

os.makedirs(os.getenv("FEDIVERSE_SAFETY_IMGDIR"), exist_ok=True)
cache = None
APP = Flask(__name__)
APP.wsgi_app = ProxyFix(APP.wsgi_app, x_for=1)

SQLITE_MODE = os.getenv("USE_SQLITE", "0") == "1"

if SQLITE_MODE:
    logger.warning("Using SQLite for database")
    APP.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///fedi_safety_api.db"
    APP.config['SQLALCHEMY_ENGINE_OPTIONS'] = {
        "isolation_level": None,
    }
else:
    if os.getenv('POSTGRES_URI') is None:
        raise Exception("Please set the POSTGRES_URI env var")
    APP.config["SQLALCHEMY_DATABASE_URI"] = os.getenv('POSTGRES_URI')
    APP.config['SQLALCHEMY_ENGINE_OPTIONS'] = {
        "pool_size": 50,
        "max_overflow": -1,
    }

APP.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(APP)
db.init_app(APP)

if not SQLITE_MODE:
    with APP.app_context():
        logger.debug("pool size = {}".format(db.engine.pool.size()))
else:
    @event.listens_for(Engine, "connect")
    def set_sqlite_pragma(dbapi_connection, connection_record):
        cursor = dbapi_connection.cursor()
        cursor.execute("PRAGMA journal_mode=WAL;")
        cursor.close()
        logger.debug("Set pragma to wal")

logger.init_ok("Fedi Safety Database", status="Started")

# Allow local workstation run
if cache is None:
    cache_config = {
        "CACHE_TYPE": "SimpleCache",
        "CACHE_DEFAULT_TIMEOUT": 300
    }
    cache = Cache(config=cache_config)
    cache.init_app(APP)
    logger.init_warn("Flask Cache", status="SimpleCache")

pictrs_safety_api/flask.py
db0/pictrs-safety
import sys
from functools import partialmethod
from loguru import logger
from pictrs_safety_api.argparser import args

STDOUT_LEVELS = ["GENERATION", "PROMPT"]
INIT_LEVELS = ["INIT", "INIT_OK", "INIT_WARN", "INIT_ERR"]
MESSAGE_LEVELS = ["MESSAGE"]
# By default we're at error level or higher
verbosity = 40
quiet = 0

def set_logger_verbosity(count):
    global verbosity
    # The count comes reversed. So count = 0 means minimum verbosity
    # While count 5 means maximum verbosity
    # So the more count we have, the lowe we drop the versbosity maximum
    verbosity = 40 - (count * 10)

def quiesce_logger(count):
    global quiet
    # The bigger the count, the more silent we want our logger
    quiet = count * 10

def is_stdout_log(record):
    if record["level"].name not in STDOUT_LEVELS:
        return(False)
    if record["level"].no < verbosity + quiet:
        return(False)
    return(True)

def is_init_log(record):
    if record["level"].name not in INIT_LEVELS:
        return(False)
    if record["level"].no < verbosity + quiet:
        return(False)
    return(True)

def is_msg_log(record):
    if record["level"].name not in MESSAGE_LEVELS:
        return(False)
    if record["level"].no < verbosity + quiet:
        return(False)
    return(True)

def is_stderr_log(record):
    if record["level"].name in STDOUT_LEVELS + INIT_LEVELS + MESSAGE_LEVELS:
        return(False)
    if record["level"].no < verbosity + quiet:
        return(False)
    return(True)

def test_logger():
    logger.generation("This is a generation message\nIt is typically multiline\nThee Lines".encode("unicode_escape").decode("utf-8"))
    logger.prompt("This is a prompt message")
    logger.debug("Debug Message")
    logger.info("Info Message")
    logger.warning("Info Warning")
    logger.error("Error Message")
    logger.critical("Critical Message")
    logger.init("This is an init message", status="Starting")
    logger.init_ok("This is an init message", status="OK")
    logger.init_warn("This is an init message", status="Warning")
    logger.init_err("This is an init message", status="Error")
    logger.message("This is user message")
    sys.exit()


logfmt = "<level>{level: <10}</level> | <green>{time:YYYY-MM-DD HH:mm:ss}</green> | <green>{name}</green>:<green>{function}</green>:<green>{line}</green> - <level>{message}</level>"
genfmt = "<level>{level: <10}</level> @ <green>{time:YYYY-MM-DD HH:mm:ss}</green> | <level>{message}</level>"
initfmt = "<magenta>INIT      </magenta> | <level>{extra[status]: <14}</level> | <magenta>{message}</magenta>"
msgfmt = "<level>{level: <10}</level> | <level>{message}</level>"

logger.level("GENERATION", no=24, color="<cyan>")
logger.level("PROMPT", no=23, color="<yellow>")
logger.level("INIT", no=31, color="<white>")
logger.level("INIT_OK", no=31, color="<green>")
logger.level("INIT_WARN", no=31, color="<yellow>")
logger.level("INIT_ERR", no=31, color="<red>")
logger.level("AUDIT", no=19, color="<blue>")
# Messages contain important information without which this application might not be able to be used
# As such, they have the highest priority
logger.level("MESSAGE", no=61, color="<green>")

logger.__class__.generation = partialmethod(logger.__class__.log, "GENERATION")
logger.__class__.prompt = partialmethod(logger.__class__.log, "PROMPT")
logger.__class__.init = partialmethod(logger.__class__.log, "INIT")
logger.__class__.init_ok = partialmethod(logger.__class__.log, "INIT_OK")
logger.__class__.init_warn = partialmethod(logger.__class__.log, "INIT_WARN")
logger.__class__.init_err = partialmethod(logger.__class__.log, "INIT_ERR")
logger.__class__.message = partialmethod(logger.__class__.log, "MESSAGE")
logger.__class__.audit = partialmethod(logger.__class__.log, "AUDIT")

config = {
    "handlers": [
        {"sink": sys.stderr, "format": logfmt, "colorize":args.color, "filter": is_stderr_log},
        {"sink": sys.stdout, "format": genfmt, "level": "PROMPT", "colorize":args.color, "filter": is_stdout_log},
        {"sink": sys.stdout, "format": initfmt, "level": "INIT", "colorize":args.color, "filter": is_init_log},
        {"sink": sys.stdout, "format": msgfmt, "level": "MESSAGE", "colorize":args.color, "filter": is_msg_log}
    ],
}
logger.configure(**config)
logger.add("fedi_safety_api.log", retention="7 days", rotation="1d", compression="bz2", level=19)
logger.disable("__main__")
logger.warning("disabled")
logger.enable("")
logger.enable(None)


set_logger_verbosity(args.verbosity)
quiesce_logger(args.quiet)

pictrs_safety_api/logger.py
db0/pictrs-safety
from flask import render_template, redirect, url_for, request
from markdown import markdown
from loguru import logger
from pictrs_safety_api.flask import APP
import pictrs_safety_api.exceptions as e

@logger.catch(reraise=True)
@APP.route('/')
# @cache.cached(timeout=300)
def index():
    with open(f'fedi_safety_api/templates/index.md') as index_file:
        index = index_file.read()
    findex = index.format()

    style = """<style>
        body {
            max-width: 120ex;
            margin: 0 auto;
            color: #333333;
            line-height: 1.4;
            font-family: sans-serif;
            padding: 1em;
        }
        </style>
    """

    head = f"""<head>
    <title>Pict-rs Safety</title>
    <meta name="viewport" content="width=device-width, initial-scale=1" />
    {style}
    </head>
    """
    return(head + markdown(findex))

pictrs_safety_api/routes.py
db0/pictrs-safety
import uuid
import bleach
import secrets
import hashlib
import os
import random
import regex as re
import json
from datetime import datetime
import dateutil.relativedelta
from loguru import logger
from pictrs_safety_api.flask import SQLITE_MODE

random.seed(random.SystemRandom().randint(0, 2**32 - 1))


def get_db_uuid():
    if SQLITE_MODE:
        return str(uuid.uuid4())
    else: 
        return uuid.uuid4()

def generate_client_id():
    return secrets.token_urlsafe(16)

def sanitize_string(text):
    santxt = bleach.clean(text).strip()
    return santxt

def hash_api_key(unhashed_api_key):
    salt = os.getenv("secret_key", "s0m3s3cr3t") # Note default here, just so it can run without env file
    hashed_key = hashlib.sha256(salt.encode() + unhashed_api_key.encode()).hexdigest()
    # logger.warning([os.getenv("secret_key", "s0m3s3cr3t"), hashed_key,unhashed_api_key])
    return hashed_key


def hash_dictionary(dictionary):
    # Convert the dictionary to a JSON string
    json_string = json.dumps(dictionary, sort_keys=True)
    # Create a hash object
    hash_object = hashlib.sha256(json_string.encode())
    # Get the hexadecimal representation of the hash
    hash_hex = hash_object.hexdigest()
    return hash_hex

def get_expiry_date():
    return datetime.utcnow() + dateutil.relativedelta.relativedelta(minutes=+20)

def get_random_seed(start_point=0):
    '''Generated a random seed, using a random number unique per node'''
    return random.randint(start_point, 2**32 - 1)

def validate_regex(regex_string):
    try:
        re.compile(regex_string, re.IGNORECASE)
    except:
        return False
    return True

pictrs_safety_api/utils.py
db0/pictrs-safety