from flask import url_for, request, jsonify, Flask 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 time import zipfile import shutil import subprocess import mistune import humanize import mysql.connector import re import os import random import json import requests import constants as const app = Flask(const.appName) # 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) # append to a json file def appendToJSON(new_data, file_path): try: # open the file path = Path(file_path) if not path.is_file(): with open(path, 'w', encoding="utf-8") as file: json.dump([], file) with open(path, 'r+', encoding="utf-8") as file: file_data = json.load(file) file_data.append(new_data) file.seek(0) json.dump(file_data, file, indent=4) return True except Exception as e: app.logger.error(str(e)) return False 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 addQuestion(from_who, question, cw, noAntispam=False): if cfg['antispam']['type'] == 'basic': antispam = request.form.get('antispam', '') elif cfg['antispam']['type'] == 'recaptcha': antispam = request.form.get('g-recaptcha-response', '') elif cfg['antispam']['type'] == 'turnstile': antispam = request.form.get('cf-turnstile-response', '') elif cfg['antispam']['type'] == 'frc': antispam = request.form.get('frc-captcha-response', '') if cfg['antispam']['enabled'] and not noAntispam: if cfg['antispam']['type'] == 'basic': if not antispam: abort(400, "Anti-spam word must not be empty") antispam_wordlist = readPlainFile(const.antiSpamFile, split=True) antispam_valid = antispam in antispam_wordlist if not antispam_valid: # return a generic error message so bad actors wouldn't figure out the antispam list return {'error': 'An error has occurred'}, 500 # it's probably bad to hardcode the siteverify urls, but meh, that will do for now elif cfg['antispam']['type'] == 'recaptcha': r = requests.post( 'https://www.google.com/recaptcha/api/siteverify', data={'response': antispam, 'secret': cfg['antispam']['recaptcha']['secretkey']} ) json_r = r.json() success = json_r['success'] if not success: return {'error': 'An error has occurred'}, 500 elif cfg['antispam']['type'] == 'turnstile': r = requests.post( 'https://challenges.cloudflare.com/turnstile/v0/siteverify', data={'response': antispam, 'secret': cfg['antispam']['turnstile']['secretkey']} ) json_r = r.json() success = json_r['success'] if not success: return {'error': 'An error has occurred'}, 500 elif cfg['antispam']['type'] == 'frc': url = 'https://global.frcapi.com/api/v2/captcha/siteverify' headers = {'X-API-Key': cfg['antispam']['frc']['apikey']} data = {'response': antispam, 'sitekey': cfg['antispam']['frc']['sitekey']} r = requests.post(url, data=data, headers=headers) json_r = r.json() success = json_r['success'] if not success: return {'error': 'An error has occurred'}, 500 blacklist = readPlainFile(const.blacklistFile, split=True) for bad_word in blacklist: if bad_word in question or bad_word in from_who: # return a generic error message so bad actors wouldn't figure out the blacklist return {'error': 'An error has occurred'}, 500 conn = connectToDb() cursor = conn.cursor() app.logger.debug("[CatAsk/API/add_question] INSERT'ing new question into database") cursor.execute("INSERT INTO questions (from_who, content, answered, cw) VALUES (%s, %s, %s, %s)", (from_who, question, False, cw)) cursor.close() conn.close() return {'message': 'Question asked successfully!'}, 201 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 addAnswer(question_id, answer, cw): conn = connectToDb() try: cursor = conn.cursor() app.logger.debug("[CatAsk/API/add_answer] INSERT'ing an answer into database") cursor.execute("INSERT INTO answers (question_id, content, cw) VALUES (%s, %s, %s)", (question_id, answer, cw)) answer_id = cursor.lastrowid app.logger.debug("[CatAsk/API/add_answer] UPDATE'ing question to set answered and answer_id") cursor.execute("UPDATE questions SET answered=%s, answer_id=%s WHERE id=%s", (True, answer_id, question_id)) conn.commit() except Exception as e: conn.rollback() return jsonify({'error': str(e)}), 500 finally: cursor.close() conn.close() return jsonify({'message': 'Answer added successfully!'}), 201 def readPlainFile(file, split=False): if os.path.exists(file): with open(file, 'r', encoding="utf-8") as file: if split: return file.read().splitlines() else: return file.read() 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.+?)\[/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"" 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): app.logger.debug("\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[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":{emoji_name}:" 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 = 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(): app.logger.debug(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: app.logger.debug(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) app.logger.debug(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 .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
tags # # yes, markdown usually lets you make line breaks only # with 2 spaces or
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 ) cleaner = Cleaner(tags=allowed_tags) clean_text = cleaner.clean(text) html = md(clean_text) return Markup(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 """