catask/functions.py

from flask import url_for, request, jsonify
from markupsafe import Markup
from bleach.sanitizer import Cleaner
from datetime import datetime, timezone
from pathlib import Path
from mistune import HTMLRenderer, escape
from PIL import Image
import mistune
import humanize
import mysql.connector
import re
import os
import random
import json
import constants as const

# load json file
def loadJSON(file_path):
    # open the file
    path = Path.cwd() / file_path
    with open(path, 'r', encoding="utf-8") as file:
        # return loaded file
        return json.load(file)

# save json file
def saveJSON(dict, file_path):
    # open the file
    path = Path.cwd() / file_path
    with open(path, 'w', encoding="utf-8") as file:
        # dump the contents
        json.dump(dict, file, indent=4)

cfg = loadJSON(const.configFile)

def formatRelativeTime(date_str):
    date_format = "%Y-%m-%d %H:%M:%S"
    past_date = datetime.strptime(date_str, date_format)

    now = datetime.now()
    time_difference = now - past_date

    return humanize.naturaltime(time_difference)

def formatRelativeTime2(date_str):
    date_format = "%Y-%m-%dT%H:%M:%SZ"

    past_date = None
    try:
        if date_str:
            past_date = datetime.strptime(date_str, date_format)
        else:
            pass
    except ValueError:
        pass

    if past_date is None:
        return ''
        # raise ValueError("Date string does not match any supported format.")

    if past_date.tzinfo is None:
        past_date = past_date.replace(tzinfo=timezone.utc)

    now = datetime.now(timezone.utc)

    time_difference = now - past_date

    return humanize.naturaltime(time_difference)

dbHost = os.environ.get("DB_HOST")
dbUser = os.environ.get("DB_USER")
dbPass = os.environ.get("DB_PASS")
dbName = os.environ.get("DB_NAME")
dbPort = os.environ.get("DB_PORT")
if not dbPort:
    dbPort = 3306

def createDatabase(cursor, dbName):
	try:
		cursor.execute("CREATE DATABASE {} DEFAULT CHARACTER SET 'utf8'".format(dbName))
		print(f"Database {dbName} created successfully")
	except mysql.connector.Error as error:
		print("Failed to create database:", error)
		exit(1)

def connectToDb():
	conn = mysql.connector.connect(
		host=dbHost,
		user=dbUser,
		password=dbPass,
		database=dbName,
		port=dbPort,
		autocommit=True
	)
	return conn

def getQuestion(question_id: int):
    conn = connectToDb()
    cursor = conn.cursor(dictionary=True)
    cursor.execute("SELECT * FROM questions WHERE id=%s", (question_id,))
    question = cursor.fetchone()
    cursor.close()
    conn.close()
    return question

def getAnswer(question_id: int):
    conn = connectToDb()
    cursor = conn.cursor(dictionary=True)
    cursor.execute("SELECT * FROM answers WHERE question_id=%s", (question_id,))
    answer = cursor.fetchone()
    cursor.close()
    conn.close()
    return answer

def readPlainFile(file, split=False):
    if os.path.exists(file):
        with open(file, 'r', encoding="utf-8") as file:
            if split == False:
                return file.read()
            if split == True:
                return file.read().splitlines()
    else:
        return []

def savePlainFile(file, contents):
	with open(file, 'w') as file:
		file.write(contents)

def getRandomWord():
    items = readPlainFile(const.antiSpamFile, split=True)
    return random.choice(items)

def trimContent(var, trim):
    trim = int(trim)
    if trim > 0:
        trimmed = var[:trim] + '…' if len(var) >= trim else var
        trimmed = trimmed.rstrip()
        return trimmed
    else:
        return var

# mistune plugin
inlineBtnPattern = r'\[btn\](?P<button_text>.+?)\[/btn\]'

def parse_inline_button(inline, m, state):
    text = m.group("button_text")
    state.append_token({"type": "inline_button", "raw": text})
    return m.end()

def render_inline_button(renderer, text):
    return f"<button class='btn btn-secondary' type='button'>{text}</button>"

def button(md):
    md.inline.register('inline_button', inlineBtnPattern, parse_inline_button, before='link')
    if md.renderer and md.renderer.NAME == 'html':
        md.renderer.register('inline_button', render_inline_button)

# Base directory where emoji packs are stored
EMOJI_BASE_PATH = Path.cwd() / 'static' / 'emojis'

emoji_cache = {}

def find_emoji_path(emoji_name):
	head, sep, tail = emoji_name.partition('_')
	if any(Path(EMOJI_BASE_PATH).glob(f'{head}.json')):
		for json_file in Path(EMOJI_BASE_PATH).glob('*.json'):
			print("\n[CatAsk/functions/find_emoji_path] Using JSON meta file\n")
			pack_data = loadJSON(json_file)
			emojis = pack_data.get('emojis', [])

			for emoji in emojis:
				if emoji['name'] == emoji_name:
					rel_dir = json_file.stem
					emoji_path = os.path.join('static/emojis', rel_dir, emoji['file_name'])
					emoji_cache[emoji_name] = emoji_path
					return emoji_path

	else:
		for address, dirs, files in os.walk(EMOJI_BASE_PATH):
			print("\n[CatAsk/functions/find_emoji_path] Falling back to scanning directories\n")
			if f"{emoji_name}.png" in files:
				rel_dir = os.path.relpath(address, EMOJI_BASE_PATH)
				emoji_path = os.path.join("static/emojis", rel_dir, f"{emoji_name}.png")
				emoji_cache[emoji_name] = emoji_path
				return emoji_path

	return None

emojiPattern = r':(?P<emoji_name>[a-zA-Z0-9_]+):'

def parse_emoji(inline, m, state):
    emoji_name = m.group("emoji_name")
    state.append_token({"type": "emoji", "raw": emoji_name})
    return m.end()

def render_emoji(renderer, emoji_name):
    emoji_path = find_emoji_path(emoji_name)
    
    if emoji_path:
        absolute_emoji_path = url_for('static', filename=emoji_path.replace('static/', ''))
        return f"<img src='{absolute_emoji_path}' alt=':{emoji_name}:' title=':{emoji_name}:' class='emoji' loading='lazy' width='28' height='28' />"
    
    return f":{emoji_name}:"

def emoji(md):
    md.inline.register('emoji', emojiPattern, parse_emoji, before='link')

    if md.renderer and md.renderer.NAME == 'html':
        md.renderer.register('emoji', render_emoji)

def listEmojis():
    emojis = []
    emoji_base_path = Path.cwd() / 'static' / 'emojis'

    # Iterate over files that are directly in the emoji base path (not in subdirectories)
    for file in emoji_base_path.iterdir():
        # Only include files, not directories
        if file.is_file() and file.suffix in {'.png', '.jpg', '.jpeg', '.webp'}:
            # Get the relative path and name for the emoji
            relative_path = os.path.relpath(file, emoji_base_path)
            emojis.append({
                'name': file.stem,  # Get the file name without the extension
                'image': os.path.join('static/emojis', relative_path),  # Full relative path for image
                'relative_path': relative_path
            })

    return emojis

"""
def listEmojiPacks():
	emoji_packs = []
	emoji_base_path = Path.cwd() / 'static' / 'emojis'

	for pack_dir in emoji_base_path.iterdir():
		if pack_dir.is_dir():
			relative_path = os.path.relpath(pack_dir, emoji_base_path)
			json_file_path = const.emojiPath / f"{pack_dir.name}.json"
			if json_file_path.exists():
				pack_data = loadJSON(json_file_path)
				pack_info = {
					'name': pack_data['name'],
					'exportedAt': pack_data['exportedAt'],
					# 'website': pack_data['website'],
					'emojis': pack_data['emojis']
				}
				print(f"pack info: {pack_info}")
				print(f"pack name: {pack_info['name']}")
				return pack_info
			else:
				preview_image = None
				for file in pack_dir.iterdir():
					if file.suffix in {'.png', '.jpg', '.jpeg', '.webp'}:
						preview_image = os.path.join('static/emojis', relative_path, file.name)
						break

						emoji_packs.append({
							'name': pack_dir.name,
							'preview_image': preview_image,
							'relative_path': f'static/emojis/{relative_path}'
							})

				return emoji_packs
"""

def listEmojiPacks():
    emoji_packs = []
    emoji_base_path = const.emojiPath

    # Iterate through all directories in the emoji base path
    for pack_dir in emoji_base_path.iterdir():
        if pack_dir.is_dir():
            relative_path = os.path.relpath(pack_dir, emoji_base_path)

            # Check if a meta.json file exists in the directory
            meta_json_path = const.emojiPath / f"{pack_dir}.json"
            if meta_json_path.exists():
                print(f"[CatAsk/functions/listEmojiPacks] Using meta.json file ({meta_json_path})")
                # Load data from the meta.json file
                pack_data = loadJSON(meta_json_path)
                
                emoji_packs.append({
                    'name': pack_data.get('name', pack_dir.name).capitalize(),
                    'exportedAt': pack_data.get('exportedAt', 'Unknown'),
                    'preview_image': pack_data.get('preview_image', ''),
                    'website': pack_data.get('website', ''),
                    'relative_path': f'static/emojis/{relative_path}',
                    'emojis': pack_data.get('emojis', [])
                })
            else:
                print(f"[CatAsk/functions/listEmojiPacks] Falling back to directory scan ({pack_dir})")
                # If no meta.json is found, fall back to directory scan
                preview_image = None
                # Find the first image in the directory for preview
                for file in pack_dir.iterdir():
                    if file.suffix in {'.png', '.jpg', '.jpeg', '.webp'}:
                        preview_image = os.path.join('static/emojis', relative_path, file.name)
                        break
                
                # Append pack info without meta.json
                emoji_packs.append({
                    'name': pack_dir.name.capitalize(),
                    'preview_image': preview_image,
                    'relative_path': f'static/emojis/{relative_path}'
                })

    return emoji_packs


def processEmojis(meta_json_path):
    emoji_metadata = loadJSON(meta_json_path)
    emojis = emoji_metadata.get('emojis', [])
    pack_name = emoji_metadata['emojis'][0]['emoji']['category'].capitalize()
    exported_at = emoji_metadata.get('exportedAt', 'Unknown')
    website = emoji_metadata.get('host', '')
    preview_image = os.path.join('static/emojis', pack_name.lower(), emoji_metadata['emojis'][0]['fileName'])
    relative_path = os.path.join('static/emojis', pack_name.lower())

    processed_emojis = []
    for emoji in emojis:
        emoji_info = {
            'name': emoji['emoji']['name'],
            'file_name': emoji['fileName'],
        }
        processed_emojis.append(emoji_info)
        print(f"[CatAsk/API/upload_emoji_pack] Processed emoji: {emoji_info['name']}\t(File: {emoji_info['file_name']})")

    # Create the pack info structure
    pack_info = {
        'name': pack_name,
        'exportedAt': exported_at,
        'preview_image': preview_image,
        'relative_path': relative_path,
        'website': website,
        'emojis': processed_emojis
    }

    # Save the combined pack info to <pack_name>.json
    pack_json_name = const.emojiPath / f"{pack_name.lower()}.json"
    saveJSON(pack_info, pack_json_name)

    return processed_emojis

def renderMarkdown(text):
    plugins = [
        'strikethrough',
        button,
        emoji
    ]
    allowed_tags = [
        'p',
        'em',
        'b',
        'strong',
        'i',
        'br',
        's',
        'del',
        'a',
        'button',
        'ol',
        'li',
        'hr',
        'img'
    ]
    allowed_attrs = {
        'a': 'href',
        'button': 'class',
        'img': ['src', 'width', 'height', 'alt', 'class', 'loading', 'title']
    }
    # hard_wrap=True means that newlines will be
    # converted into <br> tags
    #
    # yes, markdown usually lets you make line breaks only
    # with 2 spaces or <br> tag, but hard_wrap is enabled to keep
    # sanity of whoever will use this software
    # (after all, not everyone knows markdown syntax)
    md = mistune.create_markdown(
        escape=True,
        plugins=plugins,
        hard_wrap=True
    )
    html = md(text)
    cleaner = Cleaner(tags=allowed_tags, attributes=allowed_attrs)
    clean_html = cleaner.clean(html)
    return Markup(clean_html)

def generateMetadata(question=None, answer=None):
    metadata = {
        'title': cfg['instance']['title'],
        'description': cfg['instance']['description'],
        'url': cfg['instance']['fullBaseUrl'],
        'image': cfg['instance']['image']
    }

    # if question is specified, generate metadata for that question
    if question and answer:
        metadata.update({
            'title': trimContent(question['content'], 150) + " | " + cfg['instance']['title'],
            'description': trimContent(answer['content'], 150),
            'url': cfg['instance']['fullBaseUrl'] + url_for('viewQuestion', question_id=question['id']),
            'image': cfg['instance']['image']
        })

    # return 'metadata' dictionary
    return metadata

allowedFileExtensions = {'png', 'jpg', 'jpeg', 'webp', 'bmp', 'jxl'}
allowedArchiveExtensions = {'zip', 'tar', 'gz', 'bz2', 'xz'}

def allowedFile(filename):
    return '.' in filename and filename.rsplit('.', 1)[1].lower() in allowedFileExtensions

def allowedArchive(filename):
    return '.' in filename and filename.rsplit('.', 1)[1].lower() in allowedArchiveExtensions

def stripArchExtension(filename):
    if filename.endswith(('.tar.gz', '.tar.bz2', '.tar.xz')):
        filename = filename.rsplit('.', 2)[0]
    else:
        filename = filename.rsplit('.', 1)[0]
    return filename

def generateFavicon(file_name):
	sizes = {
		'apple-touch-icon.png': (180, 180),
        'android-chrome-192x192.png': (192, 192),
        'android-chrome-512x512.png': (512, 512),
		'favicon-32x32.png': (32, 32),
		'favicon-16x16.png': (16, 16),
		'favicon.ico': (16, 16)
	}

	img = Image.open(const.faviconDir / file_name)

	if not os.path.exists(const.faviconDir):
		os.makedirs(const.faviconDir)

	for filename, size in sizes.items():
		resized_img = img.resize(size)
		resized_img_absolute_path = const.faviconDir / filename
		resized_img.save(resized_img_absolute_path)

# reserved for 1.7.0 or later
"""
def getUserIp():
	if request.environ.get('HTTP_X_FORWARDED_FOR') is None:
	    return request.environ['REMOTE_ADDR']
	else:
	    return request.environ['HTTP_X_FORWARDED_FOR']

def isIpBlacklisted(user_ip):
    blacklist = readPlainFile(const.ipBlacklistFile, split=True)
    return user_ip in blacklist
"""