import os import sqlite3 import time as _time import json as _json import re as _re import glob as _glob import random import string as _string from datetime import datetime, timezone as _tz from dotenv import load_dotenv from functools import wraps from flask import session, request, g, redirect, abort, url_for, jsonify from werkzeug.security import generate_password_hash, check_password_hash load_dotenv() # Read website_url directly from .env — single source of truth, no circular import website_url = os.getenv("website_url", "localhost") def _(text): return text ALLOWED_DOMAIN = website_url dir_path = os.path.dirname(os.path.realpath(__file__)) def _db(): db_path = os.path.join(dir_path, "storage", "sqlite.db") c = sqlite3.connect(db_path) c.row_factory = sqlite3.Row return c def _hash_password(password): return generate_password_hash(password) def _check_password(stored, provided): return check_password_hash(stored, provided) def url(path): if g.lang_code == "en": return path return "/" + g.lang_code + path def static_url(file_url): if file_url.startswith("http"): return file_url from flask import current_app rel_path = os.path.relpath(file_url, current_app.config['UPLOAD_DIR']) rel_path = rel_path.replace(os.sep, '/') main_domain = website_url scheme = "https" return f"{scheme}://{main_domain}/download/{rel_path}" def get_file_size(file_path): file_size = os.path.getsize(file_path) units = ['B', 'KB', 'MB', 'GB', 'TB'] unit_index = 0 while file_size >= 1024 and unit_index < len(units) - 1: file_size /= 1024 unit_index += 1 return f'{file_size:.2f} {units[unit_index]}' def authenticate(): """Return (True, None) if admin or editor is logged in, else (False, redirect).""" if session.get('login') or session.get('editor_login'): return True, None request_xhr_key = request.headers.get('X-Requested-With') if request_xhr_key and request_xhr_key == 'XMLHttpRequest': return False, abort(403) return False, redirect(url("/admin/login")) def is_admin(): """True only for the full super-admin (not editor).""" return bool(session.get('login')) def is_editor(): """True for restricted editor accounts.""" return bool(session.get('editor_login')) and not session.get('login') def get_editor_info(): """Return the editor's DB row dict, or None if not an editor session.""" editor_id = session.get('editor_id') if not editor_id: return None with sqlite3.connect(os.path.join(dir_path, "storage", "sqlite.db")) as conn: conn.row_factory = sqlite3.Row row = conn.execute("SELECT * FROM editors WHERE id=?", (editor_id,)).fetchone() return dict(row) if row else None def get_editor_langs(): """Return list of allowed lang codes for the current editor session.""" raw = session.get('editor_langs', '["en"]') try: return _json.loads(raw) if isinstance(raw, str) else list(raw) except Exception: return ['en'] def is_human(captcha_response): import requests settings = get_site_settings() secret = settings.get('recaptcha_secret_key', '') or os.environ.get('RECAPTCHA_SECRET_KEY', '') if not secret: return True # reCAPTCHA not configured — skip verification payload = {'response': captcha_response, 'secret': secret} response = requests.post("https://www.google.com/recaptcha/api/siteverify", payload) response_text = _json.loads(response.text) return response_text['success'] def log_event(event_type, description, user_id=None, username='', meta=''): try: with _db() as c: c.execute( "INSERT INTO activity_log (event_type, user_id, username, description, meta) VALUES (?,?,?,?,?)", (event_type, user_id, username, description, meta) ) c.commit() except Exception: pass def log_deletion(entity_type, entity_id='', entity_name='', actor_user_id=None, actor_username='', actor_role='user', file_size_bytes=None, team_id=None, team_name='', extra_meta=None, entity_owner_user_id=None): """Record a deletion event in the deletion_log table. Canonical actor_role values: 'user' — a logged-in user deleted their own content via dashboard 'admin' — an admin deleted content via the admin panel 'scheduler' — the auto-expiry scheduler (actor_username='Auto-Delete') 'api' — a programmatic/API caller (reserved for future use) 'team_member'— a team member acting on team-owned content entity_type examples: 'file', 'conversion', 'share', 'share_grant', 'team', 'team_member', 'user', 'plan', 'blog', 'message', 'editor', 'admin_account', 'promo_code', 'broadcast', 'backup', 'review' entity_owner_user_id (intentional extension beyond the base schema): The user_id of the account that owned the deleted entity, regardless of who performed the deletion. Used by /dashboard/deletion-history to show users entries about their content that was removed by admin/scheduler. Indexed via idx_deletion_log_owner for efficient owner-based queries. """ try: meta_str = _json.dumps(extra_meta or {}) with _db() as c: c.execute( "INSERT INTO deletion_log " "(actor_user_id, actor_username, actor_role, entity_type, entity_id, " "entity_name, file_size_bytes, team_id, team_name, extra_meta, entity_owner_user_id) " "VALUES (?,?,?,?,?,?,?,?,?,?,?)", (actor_user_id, actor_username or '', actor_role, entity_type, str(entity_id), entity_name or '', file_size_bytes, team_id, team_name or '', meta_str, entity_owner_user_id) ) c.commit() except Exception: pass def _notify_admin_pending_blog(title, username): log_event('blog_pending', f"User \"{username}\" submitted blog for review: \"{title}\"", username=username, meta=f"title={title}") def send_email(to_email, subject, html_body, text_body=None): import smtplib from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText from email.utils import make_msgid settings = get_site_settings() host = settings.get('smtp_host', '').strip() port = int(settings.get('smtp_port', '587') or '587') username = settings.get('smtp_username', '').strip() password = settings.get('smtp_password', '').strip() from_email = settings.get('smtp_from_email', '').strip() or username from_name = settings.get('smtp_from_name', 'OnlineConvert').strip() use_tls = settings.get('smtp_use_tls', '1') == '1' if not host or not from_email: return False, "SMTP not configured. Please set SMTP settings in admin." try: msg = MIMEMultipart('alternative') msg['Subject'] = subject msg['From'] = f"{from_name} <{from_email}>" msg['To'] = to_email msg['Message-ID'] = make_msgid(domain=website_url) if text_body: msg.attach(MIMEText(text_body, 'plain')) msg.attach(MIMEText(html_body, 'html')) if use_tls: server = smtplib.SMTP(host, port, timeout=10) server.starttls() else: server = smtplib.SMTP_SSL(host, port, timeout=10) if username and password: server.login(username, password) server.sendmail(from_email, [to_email], msg.as_string()) server.quit() return True, None except Exception as e: return False, str(e) def send_job_notification_email(to_email, subject, html_body, text_body=None): import smtplib from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText from email.utils import make_msgid settings = get_site_settings() host = settings.get('job_notify_smtp_host', '').strip() if not host: return send_email(to_email, subject, html_body, text_body) port = int(settings.get('job_notify_smtp_port', '587') or '587') username = settings.get('job_notify_smtp_username', '').strip() password = settings.get('job_notify_smtp_password', '').strip() from_email = settings.get('job_notify_from_email', '').strip() or username from_name = settings.get('job_notify_from_name', '').strip() or settings.get('smtp_from_name', 'OnlineConvert').strip() use_tls = settings.get('job_notify_smtp_use_tls', '1') == '1' if not from_email: return False, "Notification SMTP not configured." try: msg = MIMEMultipart('alternative') msg['Subject'] = subject msg['From'] = f"{from_name} <{from_email}>" msg['To'] = to_email msg['Message-ID'] = make_msgid(domain=website_url) if text_body: msg.attach(MIMEText(text_body, 'plain')) msg.attach(MIMEText(html_body, 'html')) if use_tls: server = smtplib.SMTP(host, port, timeout=10) server.starttls() else: server = smtplib.SMTP_SSL(host, port, timeout=10) if username and password: server.login(username, password) server.sendmail(from_email, [to_email], msg.as_string()) server.quit() return True, None except Exception as e: return False, str(e) def get_vapid_keys(): """Return (private_key_pem, public_key_b64). Auto-generates and persists if not set. Raises ImportError (with a clear message) if py_vapid is not installed.""" try: from py_vapid import Vapid from cryptography.hazmat.primitives.serialization import ( Encoding, PublicFormat, PrivateFormat, NoEncryption ) except ImportError: raise ImportError( "py_vapid is not installed. Add 'py_vapid' to requirements.txt and reinstall." ) import base64 as _b64 from cryptography.hazmat.primitives.serialization import load_pem_private_key as _load_pem _db_path = os.path.join(dir_path, "storage", "sqlite.db") def _generate_and_store(): """Generate a fresh VAPID key pair, persist it, and return (priv_pem, pub_b64).""" _v = Vapid() _v.generate_keys() _priv_pem = _v.private_key.private_bytes( Encoding.PEM, PrivateFormat.TraditionalOpenSSL, NoEncryption() ).decode() _pub_raw = _v.public_key.public_bytes(Encoding.X962, PublicFormat.UncompressedPoint) _pub_b64 = _b64.urlsafe_b64encode(_pub_raw).rstrip(b'=').decode() try: set_site_setting('vapid_private_key', _priv_pem) set_site_setting('vapid_public_key', _pub_b64) except Exception: pass return _priv_pem, _pub_b64 def _repair_and_validate_pem(pem_str): """Try to return a valid, normalised PEM string. Returns (pem, repaired) or raises ValueError.""" import logging as _lg _log = _lg.getLogger(__name__) # Step 1: try as-is try: _load_pem(pem_str.encode(), password=None) return pem_str, False except Exception: pass # Step 2: normalise escaped newlines (literal \n → real newline) then retry fixed = pem_str.replace('\\n', '\n').strip() try: _load_pem(fixed.encode(), password=None) _log.warning("VAPID private key had escaped newlines — repaired in-place (no subscriber loss).") return fixed, True except Exception: pass # Step 3: truly unrecoverable raise ValueError( "Stored VAPID private key is corrupted and cannot be repaired automatically. " "Use 'Regenerate Keys' in Admin → General → VAPID Diagnostics, then ask your users to re-subscribe." ) try: with sqlite3.connect(_db_path) as _c: _priv = _c.execute("SELECT value FROM site_settings WHERE key='vapid_private_key'").fetchone() _pub = _c.execute("SELECT value FROM site_settings WHERE key='vapid_public_key'").fetchone() if _priv and _priv[0] and _pub and _pub[0]: _good_pem, _was_repaired = _repair_and_validate_pem(_priv[0]) if _was_repaired: # Save the fixed key back so it doesn't need repairing every call _c.execute( "UPDATE site_settings SET value=? WHERE key='vapid_private_key'", (_good_pem,) ) _c.commit() return _good_pem, _pub[0] except ValueError: raise # surface the human-readable error to the caller except Exception: pass return _generate_and_store() def send_push_notification(endpoint, p256dh, auth, title, body, url='', icon='', image='', click_beacon=''): """Send a single web-push notification via VAPID. Args: url: The URL opened when the user clicks the notification. icon: Small icon URL shown beside the notification title. Falls back to the site favicon in the service worker. image: Large banner image URL displayed below the body text. click_beacon: Silent tracking URL fired by the service worker on click (returns 204 – not opened as a page). Returns: True – notification delivered successfully False – delivery failed (network error, server error, etc.) 'gone' – endpoint is expired/invalid (HTTP 404 or 410); caller should delete the subscription record """ import json as _json import logging as _logging _log = _logging.getLogger(__name__) try: from pywebpush import webpush, WebPushException import base64 as _b64 from cryptography.hazmat.primitives.serialization import ( load_pem_private_key as _lpk, Encoding as _Enc, PrivateFormat as _PF, NoEncryption as _NE, ) priv_pem, _ = get_vapid_keys() # py_vapid 1.9.x from_string() expects base64url-encoded DER bytes, NOT a # PEM string. Convert PEM → key object → DER → base64url so py_vapid can # decode and load it correctly via load_der_private_key. _key_obj = _lpk(priv_pem.encode(), password=None) _der = _key_obj.private_bytes(_Enc.DER, _PF.TraditionalOpenSSL, _NE()) _b64url_key = _b64.urlsafe_b64encode(_der).rstrip(b'=').decode() _claim_domain = website_url.replace('https://', '').replace('http://', '').split('/')[0] _payload = {"title": title, "body": body, "url": url or '/'} if icon: _payload['icon'] = icon if image: _payload['image'] = image if click_beacon: _payload['click_beacon'] = click_beacon webpush( subscription_info={"endpoint": endpoint, "keys": {"p256dh": p256dh, "auth": auth}}, data=_json.dumps(_payload), vapid_private_key=_b64url_key, vapid_claims={"sub": f"mailto:admin@{_claim_domain}"}, ttl=86400, ) return True except Exception as _e: # Detect expired / unregistered subscriptions (push service returns 404 or 410) _status = getattr(getattr(_e, 'response', None), 'status_code', None) if _status in (404, 410): _log.warning("push subscription gone (HTTP %s) for endpoint %s", _status, endpoint[:60]) return 'gone' _log.error("send_push_notification failed for endpoint %s: %s", endpoint[:60], _e) return False _NOTIF_LABELS = { 'login': ('Login Alert', 'fa-right-to-bracket', '#6366f1'), 'file_uploaded': ('File Uploaded', 'fa-file-arrow-up', '#10b981'), 'blog_submitted': ('Blog Submitted', 'fa-blog', '#f59e0b'), 'blog_reviewed': ('Blog Reviewed', 'fa-check-circle', '#3b82f6'), 'blog_published': ('Blog Published', 'fa-newspaper', '#8b5cf6'), 'plan_changed': ('Plan Changed', 'fa-credit-card', '#06b6d4'), 'payment_received': ('Payment Received', 'fa-money-bill-wave', '#22c55e'), 'referral_signup': ('Referral Signup', 'fa-user-plus', '#f97316'), 'password_changed': ('Password Changed', 'fa-lock', '#ef4444'), } def get_user_notif_prefs(user_id): with _db() as c: row = c.execute("SELECT * FROM user_notification_prefs WHERE user_id=?", (user_id,)).fetchone() if row: return dict(row) return {k: 1 for k in _NOTIF_LABELS} def send_user_notification(user_id, event_type, subject, html_body, text_body=None): try: with _db() as c: user = c.execute("SELECT email, username FROM users WHERE id=?", (user_id,)).fetchone() if not user: return try: with _db() as _ic: _ic.execute( "INSERT INTO user_inbox (user_id, event_type, subject, body_html) VALUES (?,?,?,?)", (user_id, event_type, subject, html_body) ) _ic.commit() except Exception: pass prefs = get_user_notif_prefs(user_id) if not prefs.get(event_type, 1): return settings = get_site_settings() site_name = settings.get('site_name', 'OnlineConvert') label, icon, color = _NOTIF_LABELS.get(event_type, ('Notification', 'fa-bell', '#6366f1')) full_html = f"""

{site_name}

{label}

{html_body}

You're receiving this because you have {label} notifications enabled. Manage preferences

""" send_email(user['email'], f"[{site_name}] {subject}", full_html, text_body) except Exception: pass def get_reviews(page_id, approved_only=True, limit=50, offset=0): with sqlite3.connect("storage/sqlite.db") as conn: conn.row_factory = sqlite3.Row if approved_only: rows = conn.execute( "SELECT * FROM reviews WHERE page_id = ? AND approved = 1 ORDER BY created_at DESC LIMIT ? OFFSET ?", (page_id, limit, offset)).fetchall() else: rows = conn.execute( "SELECT * FROM reviews WHERE page_id = ? ORDER BY created_at DESC LIMIT ? OFFSET ?", (page_id, limit, offset)).fetchall() return [dict(r) for r in rows] def get_review_summary(page_id): with sqlite3.connect("storage/sqlite.db") as conn: row = conn.execute( "SELECT COUNT(*) as cnt, COALESCE(AVG(rating), 0) as avg_rating FROM reviews WHERE page_id = ?", (page_id,)).fetchone() count = row[0] average = round(row[1], 1) if row[1] else 0 dist = {} for i in range(1, 6): r = conn.execute( "SELECT COUNT(*) FROM reviews WHERE page_id = ? AND rating = ?", (page_id, i)).fetchone() dist[i] = r[0] return {"count": count, "average": average, "distribution": dist} def add_review(page_id, reviewer_name, rating, review_text='', featured=0): with sqlite3.connect("storage/sqlite.db") as conn: conn.execute( "INSERT INTO reviews (page_id, reviewer_name, rating, review_text, approved) VALUES (?, ?, ?, ?, ?)", (page_id, reviewer_name, int(rating), review_text, int(featured))) conn.commit() def delete_review(review_id): with sqlite3.connect("storage/sqlite.db") as conn: conn.execute("DELETE FROM reviews WHERE id = ?", (review_id,)) conn.commit() def toggle_review_approval(review_id): with sqlite3.connect("storage/sqlite.db") as conn: conn.execute("UPDATE reviews SET approved = CASE WHEN approved = 1 THEN 0 ELSE 1 END WHERE id = ?", (review_id,)) conn.commit() def get_all_reviews(status=None, search=None, limit=50, offset=0): with sqlite3.connect("storage/sqlite.db") as conn: conn.row_factory = sqlite3.Row query = "SELECT * FROM reviews WHERE 1=1" params = [] if status == 'approved': query += " AND approved = 1" elif status == 'pending': query += " AND approved = 0" if search: query += " AND (page_id LIKE ? OR reviewer_name LIKE ? OR review_text LIKE ?)" params.extend([f"%{search}%", f"%{search}%", f"%{search}%"]) query += " ORDER BY created_at DESC LIMIT ? OFFSET ?" params.extend([limit, offset]) rows = conn.execute(query, params).fetchall() return [dict(r) for r in rows] def get_reviews_count(status=None, search=None): with sqlite3.connect("storage/sqlite.db") as conn: query = "SELECT COUNT(*) FROM reviews WHERE 1=1" params = [] if status == 'approved': query += " AND approved = 1" elif status == 'pending': query += " AND approved = 0" if search: query += " AND (page_id LIKE ? OR reviewer_name LIKE ? OR review_text LIKE ?)" params.extend([f"%{search}%", f"%{search}%", f"%{search}%"]) return conn.execute(query, params).fetchone()[0] def get_site_settings(): """Get all site settings from SQLite with defaults""" _db_path = os.path.join(dir_path, "storage", "sqlite.db") try: with sqlite3.connect(_db_path) as conn: cursor = conn.execute("SELECT key, value FROM site_settings") settings = {row[0]: row[1] for row in cursor.fetchall()} except Exception: settings = {} # Define all default settings defaults = { # Existing settings 'site_logo': '', 'site_favicon': '', 'subdomains_enabled': '0', 'global_head_html': '', 'global_body_html': '', 'footer_text': '', 'social_buttons_enabled': '1', 'social_bookmark_enabled': '1', 'social_like_enabled': '1', 'social_share_enabled': '1', 'social_tweet_enabled': '1', 'social_like_count': '106k', 'social_share_count': '2k', 'social_bookmark_url': '', 'social_tweet_text': '', 'recaptcha_enabled': '1', 'recaptcha_site_key': '', 'recaptcha_secret_key': '', 'smtp_host': '', 'smtp_port': '587', 'smtp_username': '', 'smtp_password': '', 'smtp_from_email': '', 'smtp_from_name': 'OnlineConvert', 'smtp_use_tls': '1', # Logo type and text settings 'logo_type': 'image', 'logo_text': 'OnlineConvert', 'logo_text_font': 'font-sans', 'logo_text_custom_font': '', 'logo_text_font_url': '', 'logo_text_size': 'text-2xl', 'logo_text_weight': 'font-bold', 'logo_text_color_light': '#000000', 'logo_text_color_dark': '#FFFFFF', 'logo_text_italic': '0', 'logo_text_underline': '0', 'logo_text_uppercase': '0', 'logo_text_letter_spacing': '0', # Server / performance settings 'server_max_upload_mb': '500', 'session_lifetime_days': '31', 'compress_level': '6', 'compress_min_size': '500', # Job notification settings 'job_notify_browser_enabled': '0', 'job_notify_email_enabled': '0', 'job_notify_push_enabled': '0', 'job_notify_from_email': '', 'job_notify_from_name': '', 'job_notify_smtp_host': '', 'job_notify_smtp_port': '587', 'job_notify_smtp_username': '', 'job_notify_smtp_password': '', 'job_notify_smtp_use_tls': '1', # Broadcast tracking 'broadcast_secret_key': '', # Brand replacement (white-label / rebranding) 'brand_replacement': '', 'brand_domain': '', # e.g. mybrand.io — if empty, uses {brand_replacement}.cc # Brand / primary color 'primary_color': '#3B82F6', 'primary_dark_color': '#2563EB', 'primary_light_color': '#60A5FA', # Scrollbar 'scrollbar_thumb': '#3B82F6', 'scrollbar_track_dark': '#0f172a', 'scrollbar_track_light': '#f1f5f9', # Hero appearance 'hero_bg_type': 'gradient', 'hero_grad_from_light': '#1e3a5f', 'hero_grad_to_light': '#8B5CF6', 'hero_grad_from_dark': '#0f172a', 'hero_grad_to_dark': '#5b21b6', 'hero_grad_dir': '135deg', 'hero_solid_light': '#1e3a5f', 'hero_solid_dark': '#0f172a', 'hero_bg_image': '', 'hero_overlay_opacity': '40', 'hero_text_color_light': '#ffffff', 'hero_text_color_dark': '#ffffff', } # Apply defaults for any missing keys for key, default_value in defaults.items(): if key not in settings: settings[key] = default_value try: set_site_setting(key, default_value) except Exception: pass return settings def set_site_setting(key, value): """Save a single site setting to SQLite with upsert""" _db_path = os.path.join(dir_path, "storage", "sqlite.db") with sqlite3.connect(_db_path) as conn: conn.execute( "INSERT INTO site_settings (key, value) VALUES (?, ?) ON CONFLICT(key) DO UPDATE SET value = ?", (key, value, value)) conn.commit() def load_server_settings(app): """Read server-level settings from site_settings DB and apply to Flask app.config. Safe to call at startup and also live when admin saves changes.""" from datetime import timedelta site = get_site_settings() try: max_mb = int(site.get('server_max_upload_mb', '500')) if max_mb < 1: max_mb = 500 except (ValueError, TypeError): max_mb = 500 app.config['MAX_CONTENT_LENGTH'] = max_mb * 1024 * 1024 app.config['SEND_FILE_MAX_AGE_DEFAULT'] = 0 try: sess_days = int(site.get('session_lifetime_days', '31')) if sess_days < 1: sess_days = 1 except (ValueError, TypeError): sess_days = 31 app.config['PERMANENT_SESSION_LIFETIME'] = timedelta(days=sess_days) try: c_level = int(site.get('compress_level', '6')) c_level = max(1, min(9, c_level)) except (ValueError, TypeError): c_level = 6 app.config['COMPRESS_LEVEL'] = c_level try: c_min = int(site.get('compress_min_size', '500')) if c_min < 0: c_min = 0 except (ValueError, TypeError): c_min = 500 app.config['COMPRESS_MIN_SIZE'] = c_min def get_payment_setting(key, env_var=''): try: with sqlite3.connect("storage/sqlite.db") as conn: row = conn.execute("SELECT value FROM site_settings WHERE key=?", (key,)).fetchone() if row and row[0] and row[0].strip(): return row[0].strip() except Exception: pass return os.environ.get(env_var or key.upper(), '') def get_client_ip(): """Return the real client IP, honouring proxy headers.""" from flask import request as _req for header in ('X-Forwarded-For', 'X-Real-IP', 'CF-Connecting-IP'): val = _req.headers.get(header) if val: return val.split(',')[0].strip() return _req.remote_addr or '' # Maps tool_type → plans column that holds its daily limit _TOOL_LIMIT_COL = { 'converter': 'max_files_per_day', 'pdf': 'max_pdf_per_day', 'hash': 'max_hash_per_day', 'screenshot': 'max_screenshot_per_day', 'bg_remove': 'max_bg_remove_per_day', } # Human-readable labels used in templates TOOL_LABELS = { 'converter': 'Converter', 'pdf': 'PDF', 'hash': 'Hash', 'screenshot': 'Screenshot', 'bg_remove': 'BG Remove', } def check_conversion_limit(user_id=None, ip_address='', tool_type='converter', plan_id_override=None): """ Returns a dict: { 'allowed': bool, 'used': int, 'limit': int, 'plan_name': str, 'is_unlimited': bool } - Logged-in user → plan's limit column, counts today's rows by user_id + tool_type - Guest → free-plan limit, counts today's rows by ip_address + tool_type - limit == -1 → unlimited - plan_id_override → use this plan instead of user's personal plan (for team-context uploads) """ limit_col = _TOOL_LIMIT_COL.get(tool_type, 'max_files_per_day') try: with sqlite3.connect("storage/sqlite.db") as conn: conn.row_factory = sqlite3.Row if user_id: if plan_id_override: plan_id = plan_id_override else: user_row = conn.execute( "SELECT plan_id FROM users WHERE id=?", (user_id,) ).fetchone() plan_id = (user_row['plan_id'] if user_row else None) or 1 plan = conn.execute( f"SELECT name, {limit_col} AS day_limit FROM plans WHERE id=?", (plan_id,) ).fetchone() max_per_day = plan['day_limit'] if plan else 5 plan_name = plan['name'] if plan else 'Free' if max_per_day == -1: return {'allowed': True, 'used': 0, 'limit': -1, 'plan_name': plan_name, 'is_unlimited': True} used = conn.execute( "SELECT COUNT(*) FROM conversions " "WHERE user_id=? AND tool_type=? AND date(created_at)=date('now')", (user_id, tool_type) ).fetchone()[0] else: free_plan = conn.execute( f"SELECT name, {limit_col} AS day_limit FROM plans " "WHERE price_monthly=0 ORDER BY id LIMIT 1" ).fetchone() max_per_day = free_plan['day_limit'] if free_plan else 5 plan_name = 'Free' if max_per_day == -1: return {'allowed': True, 'used': 0, 'limit': -1, 'plan_name': plan_name, 'is_unlimited': True} used = 0 if ip_address: used = conn.execute( "SELECT COUNT(*) FROM conversions " "WHERE ip_address=? AND user_id IS NULL AND tool_type=? AND date(created_at)=date('now')", (ip_address, tool_type) ).fetchone()[0] allowed = used < max_per_day return { 'allowed': allowed, 'used': used, 'limit': max_per_day, 'plan_name': plan_name, 'is_unlimited': False, } except Exception: return {'allowed': True, 'used': 0, 'limit': -1, 'plan_name': 'Free', 'is_unlimited': True} def record_tool_usage(tool_type, user_id=None, ip_address='', file_name='', status='finished'): """Insert a lightweight usage record into conversions for non-converter tools.""" try: with sqlite3.connect("storage/sqlite.db") as conn: conn.execute( "INSERT INTO conversions " "(user_id, tool_type, input_format, output_format, job_id, status, file_name, ip_address, completed_at) " "VALUES (?,?,?,?,?,?,?,?,datetime('now'))", (user_id, tool_type, tool_type, tool_type, f"{tool_type}-{__import__('uuid').uuid4().hex[:8]}", status, file_name, ip_address) ) conn.commit() except Exception: pass def get_today_usage(user_id=None, ip_address=''): """Return today's usage count per tool type for a user or IP.""" result = {} try: with sqlite3.connect("storage/sqlite.db") as conn: if user_id: rows = conn.execute( "SELECT tool_type, COUNT(*) as cnt FROM conversions " "WHERE user_id=? AND date(created_at)=date('now') " "GROUP BY tool_type", (user_id,) ).fetchall() else: rows = conn.execute( "SELECT tool_type, COUNT(*) as cnt FROM conversions " "WHERE ip_address=? AND user_id IS NULL AND date(created_at)=date('now') " "GROUP BY tool_type", (ip_address,) ).fetchall() for row in rows: result[row[0]] = row[1] except Exception: pass return result def get_plan_tool_limits(plan_id): """Return the per-tool daily limits for a given plan id.""" try: with sqlite3.connect("storage/sqlite.db") as conn: conn.row_factory = sqlite3.Row cols = ', '.join( [f'{col} AS {col}' for col in _TOOL_LIMIT_COL.values()] ) row = conn.execute(f"SELECT {cols} FROM plans WHERE id=?", (plan_id,)).fetchone() if row: return dict(row) except Exception: pass return {col: 5 for col in _TOOL_LIMIT_COL.values()} def get_admin_profile(): with sqlite3.connect("storage/sqlite.db") as conn: conn.row_factory = sqlite3.Row row = conn.execute("SELECT * FROM admin_profiles LIMIT 1").fetchone() if row: return dict(row) default_username = os.getenv("admin_username", "admin") default_password = os.getenv("admin_password", "admin123") conn.execute( "INSERT INTO admin_profiles (username, password_hash) VALUES (?, ?)", (default_username, generate_password_hash(default_password)) ) conn.commit() conn.row_factory = sqlite3.Row row = conn.execute("SELECT * FROM admin_profiles LIMIT 1").fetchone() return dict(row) def update_admin_profile(data): allowed = ['username', 'first_name', 'last_name', 'email', 'password_hash', 'profile_picture', 'profile_description', 'totp_secret', 'totp_enabled'] fields = [] values = [] for key, value in data.items(): if key in allowed: fields.append(f"{key} = ?") values.append(value) if not fields: return fields.append("updated_at = datetime('now')") with sqlite3.connect("storage/sqlite.db") as conn: conn.execute(f"UPDATE admin_profiles SET {', '.join(fields)} WHERE id = (SELECT id FROM admin_profiles LIMIT 1)", values) conn.commit() import re as _re import glob as _glob import sqlite3 import time as _time # ───────────────────────────────────────────────────────────────────────────── # Language management helpers # ───────────────────────────────────────────────────────────────────────────── # Load from configs/languages.json (canonical single source of truth). # Falls back to a hardcoded set if the file is unavailable. try: _langs_json_path = os.path.join(dir_path, 'configs', 'languages.json') with open(_langs_json_path, encoding='utf-8') as _lf: _KNOWN_LANGUAGES = { code: { "name": meta.get("name", code), "native": meta.get("native", code), } for code, meta in json.loads(_lf.read()).items() } except Exception: _KNOWN_LANGUAGES = { "en": {"name": "English", "native": "English"}, "zh": {"name": "Chinese", "native": "中文"}, "ar": {"name": "Arabic", "native": "العربية"}, "de": {"name": "German", "native": "Deutsch"}, "es": {"name": "Spanish", "native": "Español"}, "fr": {"name": "French", "native": "Français"}, "id": {"name": "Indonesian", "native": "Bahasa Indonesia"}, "it": {"name": "Italian", "native": "Italiano"}, "pt": {"name": "Portuguese", "native": "Português"}, "ru": {"name": "Russian", "native": "Pусский"}, "tr": {"name": "Turkish", "native": "Türkçe"}, "ur": {"name": "Urdu", "native": "اردو"}, "vi": {"name": "Vietnamese", "native": "Tiếng Việt"}, } def get_all_languages_with_status(): """Return list of dicts with lang info + is_active flag from DB.""" try: with sqlite3.connect(os.path.join(dir_path, "storage", "sqlite.db")) as conn: rows = conn.execute( "SELECT lang_code, is_active FROM active_languages" ).fetchall() db_status = {r[0]: bool(r[1]) for r in rows} except Exception: db_status = {} result = [] for code, meta in _KNOWN_LANGUAGES.items(): result.append({ "code": code, "name": meta["name"], "native": meta["native"], "is_active": db_status.get(code, True), }) return result _active_lang_cache = None _active_lang_cache_time = 0.0 _ACTIVE_LANG_CACHE_TTL = 60 # seconds def get_active_language_codes(): """Return list of active lang codes from DB (always includes 'en'). Cached for _ACTIVE_LANG_CACHE_TTL seconds to avoid a DB hit on every request. Call _invalidate_active_lang_cache() after saving language settings. """ global _active_lang_cache, _active_lang_cache_time import time as _time_mod now = _time_mod.time() if _active_lang_cache is not None and now - _active_lang_cache_time < _ACTIVE_LANG_CACHE_TTL: return list(_active_lang_cache) try: with sqlite3.connect(os.path.join(dir_path, "storage", "sqlite.db")) as conn: rows = conn.execute( "SELECT lang_code FROM active_languages WHERE is_active = 1" ).fetchall() codes = [r[0] for r in rows] except Exception: codes = ['en'] if 'en' not in codes: codes.insert(0, 'en') _active_lang_cache = codes _active_lang_cache_time = now return list(codes) def _invalidate_active_lang_cache(): """Force the next call to get_active_language_codes() to re-read from DB.""" global _active_lang_cache, _active_lang_cache_time _active_lang_cache = None _active_lang_cache_time = 0.0 def save_active_languages(active_codes): """Enable the given language codes, disable all others. 'en' is always active.""" if 'en' not in active_codes: active_codes = ['en'] + list(active_codes) try: with sqlite3.connect(os.path.join(dir_path, "storage", "sqlite.db")) as conn: for code in _KNOWN_LANGUAGES: is_active = 1 if code in active_codes else 0 conn.execute( "INSERT INTO active_languages (lang_code, is_active) VALUES (?, ?) " "ON CONFLICT(lang_code) DO UPDATE SET is_active=?", (code, is_active, is_active) ) conn.commit() except Exception: pass # ───────────────────────────────────────────────────────────────────────────── # Editor CRUD helpers # ───────────────────────────────────────────────────────────────────────────── def get_all_editors(): with sqlite3.connect(os.path.join(dir_path, "storage", "sqlite.db")) as conn: conn.row_factory = sqlite3.Row rows = conn.execute( "SELECT id, username, allowed_languages, is_active, created_at FROM editors ORDER BY id" ).fetchall() result = [] for r in rows: d = dict(r) try: d['allowed_languages'] = _json.loads(d['allowed_languages']) except Exception: d['allowed_languages'] = ['en'] result.append(d) return result def get_editor_by_id(editor_id): with sqlite3.connect(os.path.join(dir_path, "storage", "sqlite.db")) as conn: conn.row_factory = sqlite3.Row row = conn.execute("SELECT * FROM editors WHERE id=?", (editor_id,)).fetchone() if not row: return None d = dict(row) try: d['allowed_languages'] = _json.loads(d['allowed_languages']) except Exception: d['allowed_languages'] = ['en'] return d def create_editor(username, password, allowed_languages): langs_json = _json.dumps(list(allowed_languages)) pw_hash = generate_password_hash(password) with sqlite3.connect(os.path.join(dir_path, "storage", "sqlite.db")) as conn: try: conn.execute( "INSERT INTO editors (username, password_hash, allowed_languages) VALUES (?,?,?)", (username.strip(), pw_hash, langs_json) ) conn.commit() return True, None except sqlite3.IntegrityError: return False, "Username already exists" def update_editor(editor_id, data): """Update editor fields. data may have: username, password, allowed_languages, is_active.""" sets, vals = [], [] if 'username' in data and data['username']: sets.append("username=?"); vals.append(data['username'].strip()) if 'password' in data and data['password']: sets.append("password_hash=?"); vals.append(generate_password_hash(data['password'])) if 'allowed_languages' in data: sets.append("allowed_languages=?"); vals.append(_json.dumps(list(data['allowed_languages']))) if 'is_active' in data: sets.append("is_active=?"); vals.append(int(data['is_active'])) if not sets: return True, None vals.append(editor_id) with sqlite3.connect(os.path.join(dir_path, "storage", "sqlite.db")) as conn: try: conn.execute(f"UPDATE editors SET {', '.join(sets)} WHERE id=?", vals) conn.commit() return True, None except sqlite3.IntegrityError: return False, "Username already exists" def delete_editor(editor_id): with sqlite3.connect(os.path.join(dir_path, "storage", "sqlite.db")) as conn: conn.execute("DELETE FROM editors WHERE id=?", (editor_id,)) conn.commit() def verify_editor_credentials(username, password): """Check editor credentials. Returns editor dict or None.""" with sqlite3.connect(os.path.join(dir_path, "storage", "sqlite.db")) as conn: conn.row_factory = sqlite3.Row row = conn.execute( "SELECT * FROM editors WHERE username=? AND is_active=1", (username,) ).fetchone() if not row: return None if check_password_hash(row['password_hash'], password): d = dict(row) try: d['allowed_languages'] = _json.loads(d['allowed_languages']) except Exception: d['allowed_languages'] = ['en'] return d return None # Caches _text_overrides_cache = {} # {lang: {original_text: override_text}} _text_overrides_cache_time = {} # {lang: float} _translatable_cache = None def get_text_overrides(lang='en'): """Return {original_text: override_text} for the given language. Looks up language-specific overrides first; if none found for a non-English language, falls through to the caller (no implicit English fallback here — that is handled in the app's _() wrapper so the caller controls fallback). Results are cached per-language with a 30-second TTL. """ global _text_overrides_cache, _text_overrides_cache_time now = _time.time() if now - _text_overrides_cache_time.get(lang, 0) > 30: with sqlite3.connect(os.path.join(dir_path, "storage", "sqlite.db")) as conn: cursor = conn.execute( "SELECT original_text, override_text FROM text_overrides " "WHERE lang = ? AND override_text != ''", (lang,) ) _text_overrides_cache[lang] = {row[0]: row[1] for row in cursor.fetchall()} _text_overrides_cache_time[lang] = now return _text_overrides_cache.get(lang, {}) def clear_text_overrides_cache(): global _text_overrides_cache, _text_overrides_cache_time, _translatable_cache _text_overrides_cache = {} _text_overrides_cache_time = {} _translatable_cache = None _PAGE_CACHE_TTL = 300 # 5 minutes # Page-level cache: {(page_id, lang): {field: value}} # One entry covers ALL fields for a page+lang, loaded in a single DB query. _page_content_cache = {} _page_content_cache_time = 0 # FAQ cache: {(page_id, lang): [faq_dicts]} _page_faqs_cache = {} _page_faqs_cache_time = 0 def _invalidate_page_cache(): """Wipe both caches immediately (called after any write).""" global _page_content_cache_time, _page_faqs_cache_time _page_content_cache_time = 0 _page_faqs_cache_time = 0 def _load_page_fields(page_id, lang, conn): """Fetch ALL non-empty fields for (page_id, lang) in one query.""" cursor = conn.execute( "SELECT field, value FROM page_content WHERE page_id = ? AND lang = ? AND value != ''", (page_id, lang)) return {row[0]: row[1] for row in cursor.fetchall()} def _ensure_page_cached(page_id, lang): """Make sure (page_id, lang) and its English fallback are in the cache.""" global _page_content_cache, _page_content_cache_time now = _time.time() if now - _page_content_cache_time > _PAGE_CACHE_TTL: _page_content_cache = {} _page_content_cache_time = now if (page_id, lang) not in _page_content_cache: with sqlite3.connect("storage/sqlite.db") as conn: _page_content_cache[(page_id, lang)] = _load_page_fields(page_id, lang, conn) if lang != 'en' and (page_id, 'en') not in _page_content_cache: _page_content_cache[(page_id, 'en')] = _load_page_fields(page_id, 'en', conn) def get_page_content(page_id, field, lang='en', default=''): if not page_id: return default _ensure_page_cached(page_id, lang) result = _page_content_cache.get((page_id, lang), {}).get(field, '') return result if result else default def get_page_all_fields(page_id, lang='en'): """Return all saved fields for a page — uses the batch cache.""" if not page_id: return {} _ensure_page_cached(page_id, lang) return dict(_page_content_cache.get((page_id, lang), {})) def set_page_content(page_id, field, value, lang='en'): with sqlite3.connect("storage/sqlite.db") as conn: if value.strip(): conn.execute( "INSERT INTO page_content (page_id, field, lang, value) VALUES (?, ?, ?, ?) ON CONFLICT(page_id, field, lang) DO UPDATE SET value = ?", (page_id, field, lang, value.strip(), value.strip())) else: conn.execute( "DELETE FROM page_content WHERE page_id = ? AND field = ? AND lang = ?", (page_id, field, lang)) conn.commit() _invalidate_page_cache() def get_page_faqs(page_id, lang='en'): global _page_faqs_cache, _page_faqs_cache_time if not page_id: return [] now = _time.time() if now - _page_faqs_cache_time > _PAGE_CACHE_TTL: _page_faqs_cache = {} _page_faqs_cache_time = now cache_key = (page_id, lang) if cache_key in _page_faqs_cache: return _page_faqs_cache[cache_key] with sqlite3.connect("storage/sqlite.db") as conn: cursor = conn.execute( "SELECT id, question, answer, position FROM page_faqs WHERE page_id = ? AND lang = ? ORDER BY position", (page_id, lang)) faqs = [{"id": r[0], "question": r[1], "answer": r[2], "position": r[3]} for r in cursor.fetchall()] if not faqs and lang != 'en': ck_en = (page_id, 'en') if ck_en in _page_faqs_cache: faqs = _page_faqs_cache[ck_en] else: cursor = conn.execute( "SELECT id, question, answer, position FROM page_faqs WHERE page_id = ? AND lang = 'en' ORDER BY position", (page_id, )) faqs = [{"id": r[0], "question": r[1], "answer": r[2], "position": r[3]} for r in cursor.fetchall()] _page_faqs_cache[ck_en] = faqs _page_faqs_cache[cache_key] = faqs return faqs def save_page_faqs(page_id, questions, answers, lang='en'): with sqlite3.connect("storage/sqlite.db") as conn: conn.execute("DELETE FROM page_faqs WHERE page_id = ? AND lang = ?", (page_id, lang)) for i, (q, a) in enumerate(zip(questions, answers)): if q.strip() or a.strip(): conn.execute( "INSERT INTO page_faqs (page_id, lang, position, question, answer) VALUES (?, ?, ?, ?, ?)", (page_id, lang, i, q.strip(), a.strip())) conn.commit() _invalidate_page_cache() def normalize_page_id(filetype=None, fileformat=None, hashformat=None, static_page=None, currency=None, to_currency=None, generator_type=None, source_format=None): if static_page: return "page:" + static_page if hashformat: return "hash:" + hashformat if currency and to_currency: return "currency:" + currency.lower() + ":" + to_currency.lower() if currency: return "currency:" + currency.lower() if generator_type and fileformat: return "generator:" + generator_type + ":" + fileformat if generator_type: return "generator:" + generator_type if filetype and fileformat and source_format: return "converter:" + filetype + ":" + source_format.lower() + ":" + fileformat if filetype and fileformat: return "converter:" + filetype + ":" + fileformat if filetype: return "converter:" + filetype return "page:index" STATIC_PAGES = [ {"id": "page:index", "label": "Homepage", "icon": "fa-home"}, {"id": "page:about", "label": "About Us", "icon": "fa-info-circle"}, {"id": "page:faq", "label": "FAQ", "icon": "fa-question-circle"}, {"id": "page:contact", "label": "Contact", "icon": "fa-envelope"}, {"id": "page:privacy", "label": "Privacy Policy", "icon": "fa-shield"}, {"id": "page:terms", "label": "Terms of Use", "icon": "fa-file-contract"}, {"id": "page:disclaimer", "label": "Disclaimer", "icon": "fa-exclamation-triangle"}, {"id": "page:imprint", "label": "Imprint", "icon": "fa-stamp"}, {"id": "page:blogs", "label": "Blogs", "icon": "fa-blog"}, {"id": "page:formats", "label": "All Formats", "icon": "fa-file-alt"}, {"id": "page:hash-generators", "label": "All Hash Generators", "icon": "fa-hashtag"}, {"id": "page:currencies", "label": "All Currencies", "icon": "fa-coins"}, {"id": "page:currency-rates", "label": "Currency Exchange Rates", "icon": "fa-chart-line"}, {"id": "page:currency-single", "label": "Currency Converter (base-to-target)", "icon": "fa-exchange-alt"}, {"id": "page:screenshot", "label": "Screenshot Tool", "icon": "fa-camera"}, {"id": "page:image-bg-remove", "label": "Image Background Remove", "icon": "fa-image"}, ] PAGE_FIELDS = [ # SEO Fields {"key": "title", "label": _("Meta Title"), "type": "text", "placeholder": _("Enter SEO title...")}, {"key": "description", "label": _("Meta Description"), "type": "textarea", "placeholder": _("Enter meta description...")}, {"key": "head_html", "label": _("Custom Head HTML"), "type": "textarea", "placeholder": _("")}, {"key": "og_image", "label": _("Open Graph Image"), "type": "text", "placeholder": _("https://example.com/image.png")}, # Page Content Fields {"key": "page_heading", "label": _("Page Heading (H1)"), "type": "text", "placeholder": _("Enter page heading...")}, {"key": "page_description", "label": _("Page Description"), "type": "textarea", "placeholder": _("Enter page description...")}, {"key": "page_body", "label": _("Page Body Content (HTML)"), "type": "textarea", "placeholder": _("Enter full page body HTML (overrides default content when set)...")}, {"key": "before_faq_title", "label": _("Before FAQ Title"), "type": "text", "placeholder": _("Enter title for section before FAQs...")}, {"key": "before_faq_content", "label": _("Before FAQ Content"), "type": "textarea", "placeholder": _("Enter content to show before FAQs... (HTML allowed)")}, # Filetype Description Fields {"key": "filetype_title", "label": _("Filetype Title"), "type": "text", "placeholder": _("Enter filetype title...")}, {"key": "filetype_description", "label": _("Filetype Short Description"), "type": "textarea", "placeholder": _("Enter short description...")}, {"key": "filetype_long_description", "label": _("Filetype Long Description"), "type": "textarea", "placeholder": _("Enter long description... (HTML allowed)")}, {"key": "extension_box_text", "label": _("Extension Box Text"), "type": "textarea", "placeholder": _("Enter text for extension cards...")}, # Format Definition Field {"key": "format_definition", "label": _("Format Definition"), "type": "textarea", "placeholder": _("Enter format definition...")}, # Filetype Icon Override (for converter category cards on homepage) {"key": "filetype_icon", "label": _("Filetype Icon Class"), "type": "text", "placeholder": _("e.g. fas fa-video (Font Awesome class)")}, # Homepage Section Fields {"key": "converter_section_heading", "label": _("Converter Section Heading"), "type": "text", "placeholder": _("e.g. Choose Your Converter")}, {"key": "converter_section_description", "label": _("Converter Section Description"), "type": "textarea", "placeholder": _("e.g. Select a file type below to get started...")}, {"key": "tools_section_heading", "label": _("Tools Section Heading"), "type": "text", "placeholder": _("e.g. More Useful Tools")}, {"key": "tools_section_description", "label": _("Tools Section Description"), "type": "textarea", "placeholder": _("e.g. Additional tools to help with your everyday tasks")}, # Homepage — Currency Converter card {"key": "tool_currency_icon", "label": _("Currency Card Icon"), "type": "text", "placeholder": _("e.g. fas fa-exchange-alt")}, {"key": "tool_currency_title", "label": _("Currency Card Title"), "type": "text", "placeholder": _("e.g. Currency Converter")}, {"key": "tool_currency_desc", "label": _("Currency Card Description"), "type": "textarea", "placeholder": _("e.g. Convert between 150+ world currencies...")}, # Homepage — Screenshot Tool card {"key": "tool_screenshot_icon", "label": _("Screenshot Card Icon"), "type": "text", "placeholder": _("e.g. fas fa-camera")}, {"key": "tool_screenshot_title", "label": _("Screenshot Card Title"), "type": "text", "placeholder": _("e.g. Screenshot Tool")}, {"key": "tool_screenshot_desc", "label": _("Screenshot Card Description"), "type": "textarea", "placeholder": _("e.g. Capture, edit, and annotate screenshots...")}, # Homepage — Background Remover card {"key": "tool_bgremove_icon", "label": _("BG Remover Card Icon"), "type": "text", "placeholder": _("e.g. fas fa-magic")}, {"key": "tool_bgremove_title", "label": _("BG Remover Card Title"), "type": "text", "placeholder": _("e.g. Background Remover")}, {"key": "tool_bgremove_desc", "label": _("BG Remover Card Description"), "type": "textarea", "placeholder": _("e.g. Remove image backgrounds automatically...")}, ] import re as _re import glob as _glob _translatable_cache = None def _collect_translatable_strings(): """ Collect all translatable strings from templates AND Python route files. Returns a list of dicts: {"text": ..., "file": ..., "line": ...} Scans templates for: - {% trans %} ... {% endtrans %} - _('string') / _("string") - gettext('string') - site.get('key', 'default') - placeholder="..." / title="..." / alt="..." attributes containing plain text - {{ _('...') }} inline expressions Scans Python source files (routes/, helpers.py) for: - _('string') / gettext('string') - flash('string') messages Uses in-memory caching (cleared on override save). """ global _translatable_cache if _translatable_cache is not None: return _translatable_cache results = [] # ── compiled patterns ────────────────────────────────────────────────── # {% trans %} ... {% endtrans %} (simple, no args) _pat_trans_simple = _re.compile( r'\{%-?\s*trans\s*-?%\}(.*?)\{%-?\s*endtrans\s*-?%\}', _re.DOTALL) # {% trans var=val %} ... {% endtrans %} (with variables) # The opening-tag inner part uses (?:[^%}]|%(?!\}))+ so it cannot cross # a %} boundary — prevents the old pattern from tunnelling across many # lines and treating an entire {% trans %}...{% endtrans %} block as the # opening tag. The lookahead (?=[^}]*[a-zA-Z_=]) ensures there is at # least one real variable character (letter / _ / =) before the %}, # which excludes bare dash-only tags like {%- trans -%}. _pat_trans_vars = _re.compile( r'\{%-?\s*trans\s+(?=[^}]*[a-zA-Z_=])(?:[^%}]|%(?!\}))*%\}(.*?)\{%-?\s*endtrans\s*-?%\}', _re.DOTALL) # _('string') or gettext('string') – single or double quotes, no newlines _pat_underscore = _re.compile(r"(?:_|gettext)\(\s*(['\"])(.+?)\1\s*\)") # site.get('key', 'default value') _pat_site_get = _re.compile( r"site\.get\(\s*['\"][^'\"]+['\"]\s*,\s*(['\"])(.+?)\1\s*\)") # placeholder="text" title="text" label="text" – static HTML attrs _pat_html_attr = _re.compile( r'(?:placeholder|title|alt|label|data-confirm)\s*=\s*"([^"<>{}\n]{4,})"') # flash('msg') from Python _pat_flash = _re.compile(r"flash\(\s*(['\"])(.+?)\1") collected = [] def _scan_file(filepath, file_label, tag): try: with open(filepath, 'r', encoding='utf-8', errors='replace') as fh: lines = fh.readlines() content = "".join(lines) def find_line(pos): c = 0 for idx, ln in enumerate(lines): c += len(ln) if c > pos: return idx + 1 return len(lines) def add(text, pos, kind=''): text = text.strip() if text and len(text) >= 2: collected.append({ "text": text, "file": file_label + (' ' + kind if kind else ''), "line": find_line(pos), }) for m in _pat_trans_simple.finditer(content): add(m.group(1), m.start(), '(trans)') for m in _pat_trans_vars.finditer(content): add(m.group(1), m.start(), '(trans)') for m in _pat_underscore.finditer(content): add(m.group(2), m.start(), '(gettext)') for m in _pat_site_get.finditer(content): add(m.group(2), m.start(), '(site_get)') if tag == 'html': for m in _pat_html_attr.finditer(content): txt = m.group(1).strip() if txt and not txt.startswith('http') and not txt.isdigit(): add(txt, m.start(), '(attr)') if tag == 'py': for m in _pat_flash.finditer(content): add(m.group(2), m.start(), '(flash)') except Exception: pass # Pattern for raw visible text between HTML tags (catches hardcoded strings) # We strip script/style blocks, Jinja blocks, comments, and HTML tags, then # pick up remaining text fragments that look like user-visible phrases. import html as _html_mod def _extract_raw_text_nodes(filepath, file_label): """Extract visible raw text from a Jinja/HTML template.""" try: with open(filepath, 'r', encoding='utf-8', errors='replace') as _fh: src = _fh.read() except Exception: return # Strip and src = _re.sub(r']*>.*?', ' ', src, flags=_re.DOTALL | _re.IGNORECASE) src = _re.sub(r']*>.*?', ' ', src, flags=_re.DOTALL | _re.IGNORECASE) # Strip Jinja comments {# ... #} src = _re.sub(r'\{#.*?#\}', ' ', src, flags=_re.DOTALL) # Strip Jinja blocks {% ... %} and expressions {{ ... }} src = _re.sub(r'\{%-?.*?-?%\}', ' ', src, flags=_re.DOTALL) src = _re.sub(r'\{\{.*?\}\}', ' ', src, flags=_re.DOTALL) # Strip HTML tags src = _re.sub(r'<[^>]+>', ' ', src) # Unescape HTML entities src = _html_mod.unescape(src) # Split into lines for line-number tracking lines = src.splitlines() for lineno, line in enumerate(lines, 1): # Split by common punctuation boundaries to get text fragments for fragment in _re.split(r'[|•·/\\]', line): txt = fragment.strip() # Accept phrases: 5+ chars, has a letter, has a space, not code/jinja leftovers if (len(txt) >= 5 and _re.search(r'[a-zA-Z]', txt) and ' ' in txt and not _re.search(r'[{}<>%=\[\]@#\\]', txt) and not txt.startswith(('http', '//', '--', '/*', '*')) and not txt.isdigit()): collected.append({ "text": txt, "file": file_label + ' (raw)', "line": lineno, }) # Scan all HTML templates tpl_base = os.path.join(dir_path, 'templates') for fp in _glob.glob(os.path.join(tpl_base, '**', '*.html'), recursive=True): label = os.path.relpath(fp, tpl_base) _scan_file(fp, label, 'html') _extract_raw_text_nodes(fp, label) # Scan Python route files for fp in _glob.glob(os.path.join(dir_path, 'routes', '*.py')): label = 'routes/' + os.path.basename(fp) _scan_file(fp, label, 'py') # Scan helpers.py _scan_file(os.path.join(dir_path, 'helpers.py'), 'helpers.py', 'py') # De-duplicate by text content (keep first occurrence) seen = set() unique = [] for item in collected: if item['text'] not in seen: seen.add(item['text']) unique.append(item) _translatable_cache = unique return unique def get_all_plans(): with _db() as c: return [dict(r) for r in c.execute("SELECT * FROM plans ORDER BY price_monthly").fetchall()] def get_plan_by_id(plan_id): with _db() as c: r = c.execute("SELECT * FROM plans WHERE id=?", (plan_id,)).fetchone() return dict(r) if r else None def get_plan_periods(plan_id): with _db() as c: rows = c.execute( "SELECT * FROM plan_billing_periods WHERE plan_id=? ORDER BY months", (plan_id,) ).fetchall() return [dict(r) for r in rows] def get_period_by_id(period_id): with _db() as c: r = c.execute("SELECT * FROM plan_billing_periods WHERE id=?", (period_id,)).fetchone() return dict(r) if r else None def calc_period_price(price_monthly, months, discount_percent): subtotal = price_monthly * months discount_amt = subtotal * (discount_percent / 100.0) total = round(subtotal - discount_amt, 2) per_month = round(total / months, 2) return {'subtotal': round(subtotal, 2), 'discount_amount': round(discount_amt, 2), 'total': total, 'per_month': per_month} def _seed_billing_periods_for_plan(plan_id): with _db() as c: c.execute("DELETE FROM plan_billing_periods WHERE plan_id=? AND months IN (3, 6)", (plan_id,)) for months, label, discount in [(1, 'Monthly', 0), (12, 'Yearly', 20)]: c.execute( "INSERT OR IGNORE INTO plan_billing_periods (plan_id, months, label, discount_percent) VALUES (?,?,?,?)", (plan_id, months, label, discount) ) c.commit() def _downgrade_user_to_free(user_id): with _db() as c: c.execute(""" UPDATE users SET plan_id=1, subscription_status='free', stripe_subscription_id='', paypal_subscription_id='', subscription_expires_at=NULL WHERE id=? """, (user_id,)) c.commit() def _resolve_subscription_status(user: dict) -> dict: """Correct subscription_status in-memory and in the database when a subscription has expired. The scheduler runs every 5 minutes but can lag. This function fires on every user read and immediately downgrades the user to the free plan in the DB if their subscription_expires_at has passed, so the change takes effect at the exact moment of expiry rather than waiting for the next scheduler tick. """ expires_at = (user.get('subscription_expires_at') or '').strip() if not expires_at: return user # Only act on paid-plan users still flagged as 'active' if user.get('subscription_status') != 'active': return user try: # Handle both "YYYY-MM-DD HH:MM:SS" and bare "YYYY-MM-DD" formats s = expires_at[:19] if len(s) == 10: # date only — treat as end-of-day (23:59:59) exp_dt = datetime.strptime(s, '%Y-%m-%d').replace( hour=23, minute=59, second=59, tzinfo=_tz.utc) else: exp_dt = datetime.strptime(s, '%Y-%m-%d %H:%M:%S').replace(tzinfo=_tz.utc) if datetime.now(_tz.utc) <= exp_dt: return user # not yet expired except Exception: return user # Subscription has expired — update the database immediately user = dict(user) # don't mutate the caller's dict user_id = user.get('id') try: with _db() as c: free_plan = c.execute( "SELECT id FROM plans WHERE price_monthly=0 ORDER BY id ASC LIMIT 1" ).fetchone() if free_plan: free_plan_id = free_plan[0] c.execute( """UPDATE users SET plan_id = ?, subscription_status = 'expired', auto_renew = 0 WHERE id = ?""", (free_plan_id, user_id) ) c.commit() user['plan_id'] = free_plan_id user['plan_name'] = 'Free' except Exception: pass user['subscription_status'] = 'expired' return user def get_all_users_admin(page=1, per_page=50, search='', plan_id=None, status=None, sort_by='created_at', sort_dir='desc'): _sort_cols = { 'username': 'u.username', 'email': 'u.email', 'plan': 'p.name', 'created_at': 'u.created_at', 'id': 'u.id', } _sd = 'ASC' if (sort_dir or 'desc').lower() == 'asc' else 'DESC' sort_col = _sort_cols.get(sort_by or 'created_at', 'u.created_at') where, params = [], [] if search: where.append("(u.username LIKE ? OR u.email LIKE ?)") params += [f'%{search}%', f'%{search}%'] if plan_id: where.append("u.plan_id = ?") params.append(int(plan_id)) if status == 'active': where.append("u.subscription_status = 'active'") elif status == 'expired': where.append("u.subscription_status = 'expired'") elif status == 'cancelled': where.append("u.subscription_status = 'cancelled'") elif status == 'free': where.append("(u.subscription_status IS NULL OR u.subscription_status = '' OR u.subscription_status = 'free')") where_sql = ("WHERE " + " AND ".join(where)) if where else "" offset = (page - 1) * per_page with _db() as c: total = c.execute( f"SELECT COUNT(*) FROM users u LEFT JOIN plans p ON p.id=u.plan_id {where_sql}", params ).fetchone()[0] rows = c.execute(f""" SELECT u.*, p.name as plan_name, (SELECT COUNT(*) FROM conversions WHERE user_id=u.id) as conversion_count, (SELECT COUNT(*) FROM file_uploads WHERE user_id=u.id AND deleted_at IS NULL) as file_count FROM users u LEFT JOIN plans p ON p.id=u.plan_id {where_sql} ORDER BY {sort_col} {_sd} LIMIT ? OFFSET ? """, params + [per_page, offset]).fetchall() return [_resolve_subscription_status(dict(r)) for r in rows], total def get_all_file_uploads_admin(page=1, per_page=50, include_deleted=False, search='', sort_by='uploaded_at', sort_dir='desc'): _sort_cols = { 'uploaded_at': 'fu.uploaded_at', 'file_size': 'fu.file_size', 'file_name': 'fu.file_name', 'username': 'u.username', } _sd = 'ASC' if (sort_dir or 'desc').lower() == 'asc' else 'DESC' sort_col = _sort_cols.get(sort_by or 'uploaded_at', 'fu.uploaded_at') where_parts, params = [], [] if not include_deleted: where_parts.append("fu.deleted_at IS NULL") if search: where_parts.append("(fu.file_name LIKE ? OR u.username LIKE ? OR u.email LIKE ?)") params += [f'%{search}%', f'%{search}%', f'%{search}%'] where = ("WHERE " + " AND ".join(where_parts)) if where_parts else "" offset = (page - 1) * per_page with _db() as c: total = c.execute( f"SELECT COUNT(*) FROM file_uploads fu LEFT JOIN users u ON u.id = fu.user_id {where}", params ).fetchone()[0] rows = c.execute(f""" SELECT fu.*, u.username, u.email, c.output_format, c.status AS conv_status, c.output_path AS conv_output_path, c.id AS conv_id FROM file_uploads fu LEFT JOIN users u ON u.id = fu.user_id LEFT JOIN conversions c ON c.upload_id = fu.id {where} ORDER BY {sort_col} {_sd} LIMIT ? OFFSET ? """, params + [per_page, offset]).fetchall() return [dict(r) for r in rows], total def admin_delete_file_upload(upload_id, also_delete_conversions=False): from flask import current_app with _db() as c: row = c.execute("SELECT file_path FROM file_uploads WHERE id=?", (upload_id,)).fetchone() if row: full_folder = os.path.join(current_app.config['UPLOAD_DIR'], os.path.dirname(row['file_path'])) try: import shutil as _sh if os.path.isdir(full_folder): _sh.rmtree(full_folder) elif os.path.exists(os.path.join(current_app.config['UPLOAD_DIR'], row['file_path'])): os.remove(os.path.join(current_app.config['UPLOAD_DIR'], row['file_path'])) except Exception: pass if also_delete_conversions: c.execute("DELETE FROM conversions WHERE upload_id=?", (upload_id,)) c.execute("UPDATE file_uploads SET deleted_at=datetime('now') WHERE id=?", (upload_id,)) c.commit() return row is not None def admin_delete_all_file_uploads(also_delete_conversions=False): from flask import current_app with _db() as c: rows = c.execute("SELECT id, file_path FROM file_uploads WHERE deleted_at IS NULL").fetchall() for row in rows: full_folder = os.path.join(current_app.config['UPLOAD_DIR'], os.path.dirname(row['file_path'])) try: import shutil as _sh if os.path.isdir(full_folder): _sh.rmtree(full_folder) except Exception: pass if also_delete_conversions and rows: ids = [r['id'] for r in rows] c.execute( "DELETE FROM conversions WHERE upload_id IN ({})".format(','.join('?' * len(ids))), ids ) c.execute("UPDATE file_uploads SET deleted_at=datetime('now') WHERE deleted_at IS NULL") c.commit() return len(rows) def get_user_conversions(user_id, page=1, per_page=20): offset = (page - 1) * per_page with _db() as c: total = c.execute( "SELECT COUNT(*) FROM conversions WHERE user_id=? AND status != 'deleted'", (user_id,) ).fetchone()[0] rows = c.execute(""" SELECT * FROM conversions WHERE user_id=? AND status != 'deleted' ORDER BY created_at DESC LIMIT ? OFFSET ? """, (user_id, per_page, offset)).fetchall() return [dict(r) for r in rows], total def get_all_conversions_admin(page=1, per_page=50): offset = (page - 1) * per_page with _db() as c: total = c.execute("SELECT COUNT(*) FROM conversions").fetchone()[0] rows = c.execute(""" SELECT c.*, u.username, u.email FROM conversions c LEFT JOIN users u ON u.id=c.user_id ORDER BY c.created_at DESC LIMIT ? OFFSET ? """, (per_page, offset)).fetchall() return [dict(r) for r in rows], total def get_user_by_id(user_id): with _db() as c: r = c.execute("SELECT * FROM users WHERE id=?", (user_id,)).fetchone() return _resolve_subscription_status(dict(r)) if r else None def get_user_by_email(email): with _db() as c: r = c.execute("SELECT * FROM users WHERE email=?", (email,)).fetchone() return _resolve_subscription_status(dict(r)) if r else None def get_user_by_username(username): with _db() as c: r = c.execute("SELECT * FROM users WHERE username=?", (username,)).fetchone() return _resolve_subscription_status(dict(r)) if r else None def create_user(username, email, password): pw_hash = _hash_password(password) with _db() as c: try: cursor = c.execute( "INSERT INTO users (username, email, password_hash, plan_id) VALUES (?,?,?,1)", (username, email, pw_hash) ) c.commit() return cursor.lastrowid except sqlite3.IntegrityError: return None def get_user_by_google_id(google_id): with _db() as c: r = c.execute("SELECT * FROM users WHERE google_id=? AND google_id != ''", (google_id,)).fetchone() return dict(r) if r else None def _save_social_avatar(source, user_id, *, is_bytes=False, ext='.jpg'): """Download a social login avatar URL (or write raw bytes) to static/avatars/. Returns the local URL path like /static/avatars/avatar_7_abc.jpg, or None on failure. Only call this when the user has no existing profile_picture.""" import uuid as _uuid_mod try: _dp = os.path.dirname(os.path.realpath(__file__)) avatar_dir = os.path.join(_dp, 'static', 'avatars') os.makedirs(avatar_dir, exist_ok=True) filename = f"avatar_{user_id}_{_uuid_mod.uuid4().hex[:8]}{ext}" filepath = os.path.join(avatar_dir, filename) if is_bytes: with open(filepath, 'wb') as _f: _f.write(source) else: import requests as _rq r = _rq.get(source, timeout=8, stream=True) if r.status_code != 200: return None with open(filepath, 'wb') as _f: for chunk in r.iter_content(8192): _f.write(chunk) return f"/static/avatars/{filename}" except Exception: return None def create_or_get_google_user(google_id, email, display_name, picture_url=None): """Find existing user by google_id or email, link google_id, or create new user. If picture_url is provided and the user has no profile picture, download and save it.""" import secrets as _sec with _db() as c: row = c.execute("SELECT * FROM users WHERE google_id=? AND google_id != ''", (google_id,)).fetchone() if row: user = dict(row) # Set profile picture if missing and we have one from Google if picture_url and not user.get('profile_picture'): saved = _save_social_avatar(picture_url, user['id']) if saved: c.execute("UPDATE users SET profile_picture=? WHERE id=?", (saved, user['id'])) c.commit() user['profile_picture'] = saved return user row = c.execute("SELECT * FROM users WHERE email=?", (email,)).fetchone() if row: user = dict(row) c.execute("UPDATE users SET google_id=?, google_email=?, email_verified=1 WHERE id=?", (google_id, email, user['id'])) c.commit() if picture_url and not user.get('profile_picture'): saved = _save_social_avatar(picture_url, user['id']) if saved: c.execute("UPDATE users SET profile_picture=? WHERE id=?", (saved, user['id'])) c.commit() return dict(c.execute("SELECT * FROM users WHERE id=?", (user['id'],)).fetchone()) base = (display_name or email.split('@')[0]).replace(' ', '').lower()[:18] username = base suffix = 0 while c.execute("SELECT id FROM users WHERE username=?", (username,)).fetchone(): suffix += 1 username = f"{base}{suffix}" dummy_pw = _sec.token_hex(24) pw_hash = _hash_password(dummy_pw) cursor = c.execute( "INSERT INTO users (username, email, password_hash, plan_id, google_id, google_email, email_verified, display_name) VALUES (?,?,?,1,?,?,1,?)", (username, email, pw_hash, google_id, email, display_name or username) ) c.commit() new_id = cursor.lastrowid if picture_url: saved = _save_social_avatar(picture_url, new_id) if saved: c.execute("UPDATE users SET profile_picture=? WHERE id=?", (saved, new_id)) c.commit() return dict(c.execute("SELECT * FROM users WHERE id=?", (new_id,)).fetchone()) def get_user_by_microsoft_id(microsoft_id): with _db() as c: r = c.execute("SELECT * FROM users WHERE microsoft_id=? AND microsoft_id != ''", (microsoft_id,)).fetchone() return dict(r) if r else None def create_or_get_microsoft_user(microsoft_id, email, display_name, picture_bytes=None, picture_ext='.jpg'): """Find existing user by microsoft_id or email, link microsoft_id, or create new user. If picture_bytes is provided and the user has no profile picture, save it to static/avatars/.""" import secrets as _sec with _db() as c: row = c.execute("SELECT * FROM users WHERE microsoft_id=? AND microsoft_id != ''", (microsoft_id,)).fetchone() if row: user = dict(row) if picture_bytes and not user.get('profile_picture'): saved = _save_social_avatar(picture_bytes, user['id'], is_bytes=True, ext=picture_ext) if saved: c.execute("UPDATE users SET profile_picture=? WHERE id=?", (saved, user['id'])) c.commit() user['profile_picture'] = saved return user row = c.execute("SELECT * FROM users WHERE email=?", (email,)).fetchone() if row: user = dict(row) c.execute("UPDATE users SET microsoft_id=?, microsoft_email=?, email_verified=1 WHERE id=?", (microsoft_id, email, user['id'])) c.commit() if picture_bytes and not user.get('profile_picture'): saved = _save_social_avatar(picture_bytes, user['id'], is_bytes=True, ext=picture_ext) if saved: c.execute("UPDATE users SET profile_picture=? WHERE id=?", (saved, user['id'])) c.commit() return dict(c.execute("SELECT * FROM users WHERE id=?", (user['id'],)).fetchone()) base = (display_name or email.split('@')[0]).replace(' ', '').lower()[:18] username = base suffix = 0 while c.execute("SELECT id FROM users WHERE username=?", (username,)).fetchone(): suffix += 1 username = f"{base}{suffix}" dummy_pw = _sec.token_hex(24) pw_hash = _hash_password(dummy_pw) cursor = c.execute( "INSERT INTO users (username, email, password_hash, plan_id, microsoft_id, microsoft_email, email_verified, display_name) VALUES (?,?,?,1,?,?,1,?)", (username, email, pw_hash, microsoft_id, email, display_name or username) ) c.commit() new_id = cursor.lastrowid if picture_bytes: saved = _save_social_avatar(picture_bytes, new_id, is_bytes=True, ext=picture_ext) if saved: c.execute("UPDATE users SET profile_picture=? WHERE id=?", (saved, new_id)) c.commit() return dict(c.execute("SELECT * FROM users WHERE id=?", (new_id,)).fetchone()) def get_user_by_apple_id(apple_id): with _db() as c: r = c.execute("SELECT * FROM users WHERE apple_id=? AND apple_id != ''", (apple_id,)).fetchone() return dict(r) if r else None def create_or_get_apple_user(apple_id, email, display_name): """Find existing user by apple_id or email, link apple_id, or create new user.""" import secrets as _sec with _db() as c: row = c.execute("SELECT * FROM users WHERE apple_id=? AND apple_id != ''", (apple_id,)).fetchone() if row: return dict(row) if email: row = c.execute("SELECT * FROM users WHERE email=?", (email,)).fetchone() if row: c.execute("UPDATE users SET apple_id=?, apple_email=?, email_verified=1 WHERE id=?", (apple_id, email, row['id'])) c.commit() return dict(c.execute("SELECT * FROM users WHERE id=?", (row['id'],)).fetchone()) base = (display_name or (email.split('@')[0] if email else 'appleuser')).replace(' ', '').lower()[:18] username = base suffix = 0 while c.execute("SELECT id FROM users WHERE username=?", (username,)).fetchone(): suffix += 1 username = f"{base}{suffix}" dummy_pw = _sec.token_hex(24) pw_hash = _hash_password(dummy_pw) fallback_email = email or f"{apple_id}@privaterelay.appleid.com" cursor = c.execute( "INSERT INTO users (username, email, password_hash, plan_id, apple_id, apple_email, email_verified, display_name) VALUES (?,?,?,1,?,?,1,?)", (username, fallback_email, pw_hash, apple_id, email or '', display_name or username) ) c.commit() return dict(c.execute("SELECT * FROM users WHERE id=?", (cursor.lastrowid,)).fetchone()) def get_current_user(): user_id = session.get('user_id') if not user_id: return None user = get_user_by_id(user_id) if not user: return None # Real-time subscription expiry check — downgrade immediately if expired expires_at = user.get('subscription_expires_at') or '' if expires_at: try: exp_dt = datetime.strptime(expires_at[:19], '%Y-%m-%d %H:%M:%S').replace(tzinfo=_tz.utc) now_dt = datetime.now(_tz.utc) if now_dt > exp_dt: plan = get_plan_by_id(user.get('plan_id', 1)) if plan and plan.get('price_monthly', 0) > 0: # Subscription expired — downgrade to free plan with _db() as _c: free_plan = _c.execute( "SELECT id FROM plans WHERE price_monthly=0 ORDER BY id ASC LIMIT 1" ).fetchone() if free_plan: _c.execute( "UPDATE users SET plan_id=?, subscription_status='expired', auto_renew=0 WHERE id=?", (free_plan[0], user_id) ) _c.commit() # Reload user with updated plan user = get_user_by_id(user_id) except Exception: pass return user def login_required_user(f): @wraps(f) def decorated(*args, **kwargs): if not session.get('user_id'): return redirect(url_for('user_login', next=request.url)) return f(*args, **kwargs) return decorated def user_has_paid_plan(user=None): if user is None: user = get_current_user() if not user: return False if user.get('subscription_status') == 'active': plan = get_plan_by_id(user.get('plan_id', 1)) if plan and plan.get('price_monthly', 0) > 0: return True return False def get_effective_plan_for_user(user_id): """ Return the most applicable plan for a user, considering team-level billing. When the user is an active member of (or owns) teams that have dedicated team plans, the team plan with the highest price_monthly is selected — giving a deterministic, best-available result for multi-team scenarios. Falls back to the user's personal plan when no team plan is available. Used by sharing routes to gate share-link features on the best plan the user has access to. Team-specific resource operations in teams.py use the team's own plan directly via _get_team_effective_plan(). """ with _db() as c: user_row = c.execute("SELECT plan_id FROM users WHERE id=? LIMIT 1", (user_id,)).fetchone() if not user_row: return None personal_plan_id = user_row['plan_id'] team_plan_rows = c.execute( """SELECT DISTINCT t.plan_id FROM teams t WHERE t.plan_id IS NOT NULL AND ( t.owner_user_id = ? OR EXISTS ( SELECT 1 FROM team_members tm WHERE tm.team_id = t.id AND tm.user_id = ? AND tm.status = 'active' ) )""", (user_id, user_id) ).fetchall() best_plan = None best_price = -1 for row in team_plan_rows: plan = get_plan_by_id(row['plan_id']) if plan: price = float(plan.get('price_monthly') or 0) if price > best_price: best_price = price best_plan = plan if best_plan: return best_plan return get_plan_by_id(personal_plan_id) def sign_invite_token(raw_token): """ Return a signed, time-bounded version of raw_token for use in invite URLs. Uses itsdangerous URLSafeTimedSerializer with the Flask app secret key. The raw_token is stored in the DB; the signed token travels in the URL. """ from flask import current_app from itsdangerous import URLSafeTimedSerializer s = URLSafeTimedSerializer(current_app.secret_key, salt='team-invite') return s.dumps(raw_token) def verify_invite_token(signed_token, max_age_days=7): """ Verify and decode a signed invite token. Returns the raw_token on success, or None if the signature is invalid or the token has expired. """ from flask import current_app from itsdangerous import URLSafeTimedSerializer, BadSignature, SignatureExpired s = URLSafeTimedSerializer(current_app.secret_key, salt='team-invite') try: return s.loads(signed_token, max_age=max_age_days * 24 * 3600) except (BadSignature, SignatureExpired): return None def _user_avatar_color(username): colors = ['#3B82F6','#8B5CF6','#10B981','#F59E0B','#EF4444','#06B6D4','#F97316','#EC4899'] return colors[sum(ord(c) for c in (username or 'U')) % len(colors)] def _get_storage_stats(): from flask import current_app upload_dir = current_app.config.get('UPLOAD_DIR', 'storage/uploads') uploads_bytes = 0 conversions_bytes = 0 user_bytes = {} try: with _db() as c: upload_rows = c.execute( "SELECT fu.user_id, u.username, fu.file_path FROM file_uploads fu LEFT JOIN users u ON u.id=fu.user_id WHERE fu.deleted_at IS NULL" ).fetchall() for row in upload_rows: path = os.path.join(upload_dir, row['file_path'] or '') try: sz = os.path.getsize(path) if os.path.exists(path) else 0 except Exception: sz = 0 uploads_bytes += sz uid = row['user_id'] or 0 uname = row['username'] or 'Anonymous' if uid not in user_bytes: user_bytes[uid] = {'username': uname, 'bytes': 0} user_bytes[uid]['bytes'] += sz conv_rows = c.execute( "SELECT output_path FROM conversions WHERE output_path IS NOT NULL AND output_path != ''" ).fetchall() for row in conv_rows: path = os.path.join(upload_dir, row['output_path'] or '') try: sz = os.path.getsize(path) if os.path.exists(path) else 0 except Exception: sz = 0 conversions_bytes += sz except Exception: pass sorted_users = sorted(user_bytes.values(), key=lambda x: x['bytes'], reverse=True) total_bytes = uploads_bytes + conversions_bytes return { 'total_bytes': total_bytes, 'uploads_bytes': uploads_bytes, 'conversions_bytes': conversions_bytes, 'by_user': sorted_users, } def _fmt_bytes(b): for unit in ['B', 'KB', 'MB', 'GB']: if b < 1024: return f"{b:.1f} {unit}" b /= 1024 return f"{b:.1f} TB" def _generate_referral_code(): chars = _string.ascii_letters + _string.digits while True: code = ''.join(random.choices(chars, k=8)) with _db() as c: if not c.execute("SELECT 1 FROM users WHERE referral_code=?", (code,)).fetchone(): return code def ensure_referral_code(user_id): with _db() as c: row = c.execute("SELECT referral_code FROM users WHERE id=?", (user_id,)).fetchone() if row and row['referral_code']: return row['referral_code'] code = _generate_referral_code() c.execute("UPDATE users SET referral_code=? WHERE id=?", (code, user_id)) c.commit() return code def get_user_by_referral_code(code): if not code: return None with _db() as c: r = c.execute("SELECT * FROM users WHERE referral_code=?", (code,)).fetchone() return dict(r) if r else None def get_user_referrals(user_id): with _db() as c: rows = c.execute(""" SELECT u.id, u.username, u.email, u.created_at, p.name as plan_name, p.price_monthly, u.subscription_status, rr.status as reward_status, rr.reward_days, rr.granted_at FROM users u LEFT JOIN plans p ON p.id = u.plan_id LEFT JOIN referral_rewards rr ON rr.referred_user_id = u.id AND rr.referrer_user_id = ? WHERE u.referred_by_user_id = ? ORDER BY u.created_at DESC """, (user_id, user_id)).fetchall() return [dict(r) for r in rows] def get_user_rewards(user_id): with _db() as c: rows = c.execute(""" SELECT rr.*, u.username as referred_username, u.email as referred_email FROM referral_rewards rr JOIN users u ON u.id = rr.referred_user_id WHERE rr.referrer_user_id = ? ORDER BY rr.created_at DESC """, (user_id,)).fetchall() return [dict(r) for r in rows] def grant_referral_reward(referrer_id, referred_id, plan_name=''): with _db() as c: site = {r['key']: r['value'] for r in c.execute( "SELECT key, value FROM site_settings WHERE key IN ('referral_enabled','referral_reward_days')" ).fetchall()} if site.get('referral_enabled', '1') != '1': return False reward_days = int(site.get('referral_reward_days', 30)) try: c.execute(""" INSERT OR IGNORE INTO referral_rewards (referrer_user_id, referred_user_id, reward_days, plan_name, status) VALUES (?,?,?,?,'pending') """, (referrer_id, referred_id, reward_days, plan_name)) c.commit() except Exception: return False reward = c.execute( "SELECT * FROM referral_rewards WHERE referrer_user_id=? AND referred_user_id=? AND status='pending'", (referrer_id, referred_id) ).fetchone() if not reward: return False referrer = c.execute("SELECT * FROM users WHERE id=?", (referrer_id,)).fetchone() if not referrer: return False import datetime as _dt if referrer['subscription_status'] == 'active' and referrer['subscription_expires_at']: try: expires = _dt.datetime.fromisoformat(referrer['subscription_expires_at']) except Exception: expires = _dt.datetime.utcnow() new_expires = expires + _dt.timedelta(days=reward_days) c.execute( "UPDATE users SET subscription_expires_at=? WHERE id=?", (new_expires.isoformat(), referrer_id) ) c.execute(""" UPDATE referral_rewards SET status='granted', granted_at=datetime('now') WHERE referrer_user_id=? AND referred_user_id=? """, (referrer_id, referred_id)) c.commit() return True