jason/qrknit
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
a321ff61714ff895b0d75ccb43bcf1da9ad8e677
qrknit/app.py
T
Jason Stedwell a321ff6171 feat: unique visitors, bot filtering, and account-wide audience analytics 
- Per-link analytics gain unique-visitor counts (period total and per-day),
  a bot_clicks figure, and an exclude_bots=1 filter applied across every
  breakdown, the daily series, and the heatmap. Bot classification comes
  from the is_bot flag stamped at click time.
- New GET /api/analytics/aggregate: one chart across the whole account (or
  org pool), optionally scoped to a single tag — daily clicks + uniques,
  referrers, devices, browsers, countries, and top links, capped to the
  plan's analytics window.
- Dashboard: Audience panel with tag/window/bot controls wired into
  loadDashboard, plus a NO BOTS toggle and PERIOD/UNIQUE/BOTS stat line on
  each link's analytics.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-03 01:40:14 -05:00

2317 lines
98 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# Copyright © 2025 QRknit. All rights reserved.
# Unauthorised copying, modification, or distribution of this software is prohibited.
#
# Third-party libraries used under their respective licences — see NOTICES.md.
# Flask BSD-3-Clause https://flask.palletsprojects.com
# Werkzeug BSD-3-Clause https://werkzeug.palletsprojects.com
# qrcode MIT https://github.com/lincolnloop/python-qrcode
# Pillow HPND https://python-pillow.org
# Gunicorn MIT https://gunicorn.org
import os
import re
import csv
import base64
import sqlite3
import hashlib
import hmac
import time
import io
import json
import queue
import secrets
import struct
import zlib
import threading
from datetime import datetime, timedelta, timezone
from functools import wraps
from flask import Flask, request, jsonify, redirect, Response, session
from werkzeug.security import generate_password_hash, check_password_hash
app = Flask(__name__, static_folder='static', template_folder='templates')
app.config['SECRET_KEY'] = os.environ.get('SECRET_KEY')
if not app.config['SECRET_KEY']:
raise RuntimeError("SECRET_KEY environment variable must be set")
ADMIN_PASSWORD = os.environ.get('ADMIN_PASSWORD', '')
if not ADMIN_PASSWORD:
raise RuntimeError("ADMIN_PASSWORD environment variable must be set")
ADMIN_USERNAME = os.environ.get('ADMIN_USERNAME', 'admin')
DB_PATH = os.environ.get('DB_PATH', '/app/data/qrknit.db')
BASE_URL = os.environ.get('BASE_URL', 'http://localhost:5000').rstrip('/')
APP_NAME = os.environ.get('APP_NAME', 'to.ALWISP')
# Operational settings — see INSTALL.md
WEBHOOK_URL = os.environ.get('WEBHOOK_URL', '') # Slack/Discord-compatible webhook for admin notifications
GEOIP_DB_PATH = os.environ.get('GEOIP_DB_PATH', '/app/data/GeoLite2-Country.mmdb')
GEO_API_FALLBACK = os.environ.get('GEO_API_FALLBACK', 'false').lower() == 'true' # ip-api.com is non-commercial-use only
IP_ANONYMIZE = os.environ.get('IP_ANONYMIZE', 'none').lower() # none | truncate | hash
CLICK_RETENTION_DAYS = int(os.environ.get('CLICK_RETENTION_DAYS', '0')) # 0 = keep click events forever
BACKUPS_ENABLED = os.environ.get('BACKUPS', 'true').lower() == 'true'
BACKUP_DIR = os.environ.get('BACKUP_DIR', '') # defaults to <db dir>/backups
BACKUP_KEEP = int(os.environ.get('BACKUP_KEEP', '7'))
COOKIE_SECURE = os.environ.get('COOKIE_SECURE', 'false').lower() == 'true'
app.config['SESSION_COOKIE_HTTPONLY'] = True
app.config['SESSION_COOKIE_SAMESITE'] = 'Lax'
app.config['SESSION_COOKIE_SECURE'] = COOKIE_SECURE
app.config['PERMANENT_SESSION_LIFETIME'] = timedelta(days=30)
# ─────────────────────────────────────────────
# Plan / tier definitions
# None means unlimited
# ─────────────────────────────────────────────
VALID_PLANS = ('free', 'starter', 'pro', 'team', 'admin')
PLAN_LIMITS = {
'free': {'max_links': 10, 'monthly_clicks': 1_000, 'analytics_days': 7},
'starter': {'max_links': 500, 'monthly_clicks': 50_000, 'analytics_days': 30},
'pro': {'max_links': None, 'monthly_clicks': 500_000, 'analytics_days': 90},
'team': {'max_links': None, 'monthly_clicks': None, 'analytics_days': 90},
'admin': {'max_links': None, 'monthly_clicks': None, 'analytics_days': 90},
}
# ─────────────────────────────────────────────
# Database
# ─────────────────────────────────────────────
def get_db():
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA foreign_keys=ON")
return conn
def init_db():
os.makedirs(os.path.dirname(DB_PATH), exist_ok=True)
with get_db() as conn:
conn.executescript("""
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT UNIQUE NOT NULL,
password_hash TEXT NOT NULL,
is_admin INTEGER DEFAULT 0,
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS links (
id INTEGER PRIMARY KEY AUTOINCREMENT,
code TEXT UNIQUE NOT NULL,
long_url TEXT NOT NULL,
title TEXT,
created_at TEXT NOT NULL,
expires_at TEXT,
clicks INTEGER DEFAULT 0,
is_active INTEGER DEFAULT 1,
is_pinned INTEGER DEFAULT 0
);
CREATE TABLE IF NOT EXISTS clicks (
id INTEGER PRIMARY KEY AUTOINCREMENT,
link_id INTEGER NOT NULL,
clicked_at TEXT NOT NULL,
referrer TEXT,
user_agent TEXT,
FOREIGN KEY (link_id) REFERENCES links(id)
);
CREATE TABLE IF NOT EXISTS tags (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT UNIQUE NOT NULL
);
CREATE TABLE IF NOT EXISTS link_tags (
link_id INTEGER NOT NULL,
tag_id INTEGER NOT NULL,
PRIMARY KEY (link_id, tag_id),
FOREIGN KEY (link_id) REFERENCES links(id),
FOREIGN KEY (tag_id) REFERENCES tags(id)
);
CREATE TABLE IF NOT EXISTS messages (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
email TEXT NOT NULL,
subject TEXT NOT NULL,
body TEXT,
created_at TEXT NOT NULL,
is_read INTEGER DEFAULT 0
);
CREATE TABLE IF NOT EXISTS organizations (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT UNIQUE NOT NULL,
plan TEXT NOT NULL DEFAULT 'team',
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS uptime_daily (
date TEXT PRIMARY KEY,
checks_total INTEGER DEFAULT 0,
checks_up INTEGER DEFAULT 0
);
CREATE TABLE IF NOT EXISTS uptime_minutes (
minute TEXT PRIMARY KEY
);
CREATE TABLE IF NOT EXISTS api_keys (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL REFERENCES users(id),
name TEXT NOT NULL,
key_hash TEXT UNIQUE NOT NULL,
prefix TEXT NOT NULL,
scope TEXT NOT NULL DEFAULT 'read',
created_at TEXT NOT NULL,
last_used_at TEXT,
revoked INTEGER DEFAULT 0
);
CREATE TABLE IF NOT EXISTS audit_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
actor TEXT,
action TEXT NOT NULL,
target TEXT,
details TEXT,
created_at TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_links_code ON links(code);
CREATE INDEX IF NOT EXISTS idx_clicks_link ON clicks(link_id);
CREATE INDEX IF NOT EXISTS idx_clicks_at ON clicks(clicked_at);
""")
# Idempotent migrations — safe to run on existing databases
for migration in [
"ALTER TABLE links ADD COLUMN is_pinned INTEGER DEFAULT 0",
"ALTER TABLE links ADD COLUMN user_id INTEGER REFERENCES users(id)",
"ALTER TABLE clicks ADD COLUMN ip_address TEXT",
"ALTER TABLE clicks ADD COLUMN country TEXT",
"ALTER TABLE users ADD COLUMN plan TEXT NOT NULL DEFAULT 'free'",
"ALTER TABLE users ADD COLUMN org_id INTEGER REFERENCES organizations(id)",
"ALTER TABLE links ADD COLUMN redirect_type INTEGER DEFAULT 302",
"ALTER TABLE links ADD COLUMN password_hash TEXT",
"ALTER TABLE links ADD COLUMN deleted_at TEXT",
"ALTER TABLE clicks ADD COLUMN is_bot INTEGER DEFAULT 0",
]:
try:
conn.execute(migration)
except Exception:
pass
# Migrate tags table to add user_id scoping (recreate if needed)
tags_cols = [col[1] for col in conn.execute('PRAGMA table_info(tags)').fetchall()]
if 'user_id' not in tags_cols:
conn.executescript("""
CREATE TABLE tags_new (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
user_id INTEGER REFERENCES users(id),
UNIQUE(name, user_id)
);
INSERT OR IGNORE INTO tags_new (id, name, user_id)
SELECT id, name, NULL FROM tags;
DROP TABLE tags;
ALTER TABLE tags_new RENAME TO tags;
""")
def seed_admin():
"""Upsert the admin account from env vars on every startup."""
pw_hash = generate_password_hash(ADMIN_PASSWORD)
now = datetime.now(timezone.utc).replace(tzinfo=None).isoformat()
with get_db() as conn:
existing = conn.execute('SELECT id FROM users WHERE username=?', (ADMIN_USERNAME,)).fetchone()
if existing:
conn.execute('UPDATE users SET password_hash=?, is_admin=1, plan=? WHERE username=?',
(pw_hash, 'admin', ADMIN_USERNAME))
else:
conn.execute(
'INSERT INTO users (username, password_hash, is_admin, plan, created_at) VALUES (?,?,1,?,?)',
(ADMIN_USERNAME, pw_hash, 'admin', now)
)
init_db()
seed_admin()
def now_iso():
return datetime.now(timezone.utc).replace(tzinfo=None).isoformat()
def _uptime_heartbeat():
"""Daemon thread: marks the current UTC minute as 'up' once per minute.
INSERT OR IGNORE on the minute primary key makes this safe to run in every
Gunicorn worker — multiple workers collapse to a single mark per minute,
so uptime is no longer over-counted in multi-worker deployments.
"""
while True:
try:
minute = datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M')
with get_db() as conn:
conn.execute('INSERT OR IGNORE INTO uptime_minutes (minute) VALUES (?)', (minute,))
except Exception:
pass
time.sleep(60)
threading.Thread(target=_uptime_heartbeat, daemon=True).start()
# ─────────────────────────────────────────────
# Async click recording
# Clicks are enqueued from the redirect path and written by a background
# thread, so redirects never block on DB writes or geo lookups.
# ─────────────────────────────────────────────
_click_queue: "queue.Queue" = queue.Queue(maxsize=10000)
def _click_writer():
while True:
item = _click_queue.get()
try:
country = item['cf_country'] or get_country_for_ip(item['ip'])
with get_db() as conn:
conn.execute(
'INSERT INTO clicks (link_id,clicked_at,referrer,user_agent,ip_address,country,is_bot) '
'VALUES (?,?,?,?,?,?,?)',
(item['link_id'], item['ts'], item['referrer'], item['ua'],
anonymize_ip(item['ip']), country, item['is_bot'])
)
conn.execute('UPDATE links SET clicks=clicks+1 WHERE id=?', (item['link_id'],))
except Exception:
pass
threading.Thread(target=_click_writer, daemon=True).start()
# ─────────────────────────────────────────────
# Daily maintenance — backups, retention, pruning
# ─────────────────────────────────────────────
def _run_maintenance():
today = datetime.now(timezone.utc).date().isoformat()
cutoff = (datetime.now(timezone.utc) - timedelta(days=31)).strftime('%Y-%m-%dT%H:%M')
with get_db() as conn:
conn.execute('DELETE FROM uptime_minutes WHERE minute < ?', (cutoff,))
if CLICK_RETENTION_DAYS > 0:
click_cutoff = (datetime.now(timezone.utc).replace(tzinfo=None)
- timedelta(days=CLICK_RETENTION_DAYS)).isoformat()
conn.execute('DELETE FROM clicks WHERE clicked_at < ?', (click_cutoff,))
if BACKUPS_ENABLED:
backup_dir = BACKUP_DIR or os.path.join(os.path.dirname(DB_PATH), 'backups')
os.makedirs(backup_dir, exist_ok=True)
path = os.path.join(backup_dir, f'qrknit-{today}.db')
if not os.path.exists(path):
# Vacuum into a per-process temp file, then rename atomically —
# concurrent workers can't trip over a half-written target.
tmp = f'{path}.tmp{os.getpid()}'
try:
conn = sqlite3.connect(DB_PATH)
try:
conn.execute('VACUUM INTO ?', (tmp,))
finally:
conn.close()
os.replace(tmp, path)
except Exception:
try:
os.remove(tmp)
except OSError:
pass
backups = sorted(f for f in os.listdir(backup_dir)
if f.startswith('qrknit-') and f.endswith('.db'))
for old in backups[:-BACKUP_KEEP] if BACKUP_KEEP > 0 else []:
try:
os.remove(os.path.join(backup_dir, old))
except OSError:
pass
def _maintenance_loop():
while True:
try:
_run_maintenance()
except Exception:
pass
time.sleep(3600)
threading.Thread(target=_maintenance_loop, daemon=True).start()
# ─────────────────────────────────────────────
# Webhook notifications (Slack / Discord compatible)
# ─────────────────────────────────────────────
def notify_webhook(text):
"""POST a notification to WEBHOOK_URL. Fire-and-forget — never blocks a request."""
if not WEBHOOK_URL:
return
def _send():
try:
import urllib.request as _ureq
payload = json.dumps({'text': text, 'content': text}).encode()
req = _ureq.Request(WEBHOOK_URL, data=payload,
headers={'Content-Type': 'application/json', 'User-Agent': 'QRknit/1.0'})
_ureq.urlopen(req, timeout=5)
except Exception:
pass
threading.Thread(target=_send, daemon=True).start()
# ─────────────────────────────────────────────
# Auth decorators
# Requests authenticate with either a session cookie or an API key
# (`Authorization: Bearer qrk_…`). Keys carry a read or write scope.
# ─────────────────────────────────────────────
def _try_api_key_auth():
"""If a Bearer API key is present, validate it and load the key's user into
the request's session (without emitting a Set-Cookie). Returns an error
response tuple on a bad key / insufficient scope, else None."""
auth = request.headers.get('Authorization', '')
if not auth.startswith('Bearer '):
return None
token = auth[7:].strip()
if not token.startswith('qrk_'):
return jsonify({'error': 'Invalid API key'}), 401
key_hash = hashlib.sha256(token.encode()).hexdigest()
with get_db() as conn:
row = conn.execute(
'SELECT k.id, k.scope, u.id AS user_id, u.username, u.is_admin, u.org_id '
'FROM api_keys k JOIN users u ON k.user_id=u.id '
'WHERE k.key_hash=? AND k.revoked=0', (key_hash,)
).fetchone()
if not row:
return jsonify({'error': 'Invalid API key'}), 401
if row['scope'] != 'write' and request.method not in ('GET', 'HEAD', 'OPTIONS'):
return jsonify({'error': 'This API key is read-only'}), 403
conn.execute('UPDATE api_keys SET last_used_at=? WHERE id=?', (now_iso(), row['id']))
session['authenticated'] = True
session['user_id'] = row['user_id']
session['username'] = row['username']
session['is_admin'] = bool(row['is_admin'])
session['org_id'] = row['org_id']
session.modified = False # request-scoped only — no session cookie for API clients
return None
def login_required(f):
@wraps(f)
def decorated(*args, **kwargs):
err = _try_api_key_auth()
if err:
return err
if not session.get('authenticated') or not session.get('user_id'):
return jsonify({'error': 'Authentication required'}), 401
return f(*args, **kwargs)
return decorated
def admin_required(f):
@wraps(f)
def decorated(*args, **kwargs):
err = _try_api_key_auth()
if err:
return err
if not session.get('authenticated') or not session.get('user_id'):
return jsonify({'error': 'Authentication required'}), 401
if not session.get('is_admin'):
return jsonify({'error': 'Admin access required'}), 403
return f(*args, **kwargs)
return decorated
def log_audit(conn, action, target=None, details=None):
"""Record an administrative action in the audit log."""
conn.execute(
'INSERT INTO audit_log (actor, action, target, details, created_at) VALUES (?,?,?,?,?)',
(session.get('username'), action, target, details, now_iso())
)
# ─────────────────────────────────────────────
# Plan / usage helpers
# ─────────────────────────────────────────────
def get_user_plan(conn, user_id):
"""Return the effective plan for a user.
Admin users always get the 'admin' tier.
Org members inherit their org's plan.
Solo users use their own stored plan.
"""
row = conn.execute('SELECT plan, is_admin, org_id FROM users WHERE id=?', (user_id,)).fetchone()
if not row:
return 'free'
if row['is_admin']:
return 'admin'
if row['org_id']:
org = conn.execute('SELECT plan FROM organizations WHERE id=?', (row['org_id'],)).fetchone()
if org and org['plan'] in VALID_PLANS:
return org['plan']
return row['plan'] if row['plan'] in VALID_PLANS else 'free'
def get_user_org_id(conn, user_id):
"""Return the org_id for a user, or None if not in an org."""
row = conn.execute('SELECT org_id FROM users WHERE id=?', (user_id,)).fetchone()
return row['org_id'] if row else None
def get_user_active_link_count(conn, user_id):
"""Count active links for a user, pooled across their org if they have one."""
org_id = get_user_org_id(conn, user_id)
if org_id:
return conn.execute(
'SELECT COUNT(*) FROM links WHERE user_id IN '
'(SELECT id FROM users WHERE org_id=?) AND is_active=1',
(org_id,)
).fetchone()[0]
return conn.execute(
'SELECT COUNT(*) FROM links WHERE user_id=? AND is_active=1', (user_id,)
).fetchone()[0]
def get_user_monthly_clicks(conn, user_id):
"""Count clicks in the current calendar month (UTC), pooled across org if applicable."""
now = datetime.now(timezone.utc).replace(tzinfo=None)
since = now.replace(day=1, hour=0, minute=0, second=0, microsecond=0).isoformat()
org_id = get_user_org_id(conn, user_id)
if org_id:
return conn.execute(
'SELECT COUNT(*) FROM clicks c JOIN links l ON c.link_id=l.id '
'WHERE l.user_id IN (SELECT id FROM users WHERE org_id=?) AND c.clicked_at>=?',
(org_id, since)
).fetchone()[0]
return conn.execute(
'SELECT COUNT(*) FROM clicks c JOIN links l ON c.link_id=l.id '
'WHERE l.user_id=? AND c.clicked_at>=?',
(user_id, since)
).fetchone()[0]
# ─────────────────────────────────────────────
# QR Generator
# ─────────────────────────────────────────────
def _qr_ec_constant(qrc, ec: str, logo_bytes=None):
"""Map an L/M/Q/H letter to a qrcode error-correction constant.
A logo overlay always forces H so the covered modules stay recoverable."""
if logo_bytes:
return qrc.constants.ERROR_CORRECT_H
return {
'L': qrc.constants.ERROR_CORRECT_L,
'M': qrc.constants.ERROR_CORRECT_M,
'Q': qrc.constants.ERROR_CORRECT_Q,
'H': qrc.constants.ERROR_CORRECT_H,
}.get((ec or 'M').upper(), qrc.constants.ERROR_CORRECT_M)
def generate_qr_png(data: str, size: int = 300, fg=(0,0,0), bg=(255,255,255),
style: str = 'square', logo_bytes: bytes = None, ec: str = 'M') -> bytes:
"""Generate QR PNG. Supports dot styles and logo overlay when qrcode[pil] is installed."""
try:
import qrcode as qrc
from PIL import Image
qr = qrc.QRCode(error_correction=_qr_ec_constant(qrc, ec, logo_bytes), border=2)
qr.add_data(data)
qr.make(fit=True)
pil_img = None
if style and style != 'square':
try:
from qrcode.image.styledpil import StyledPilImage
from qrcode.image.styles.moduledrawers.pil import (
RoundedModuleDrawer, CircleModuleDrawer,
VerticalBarsDrawer, HorizontalBarsDrawer,
)
_drawers = {
'rounded': RoundedModuleDrawer,
'dots': CircleModuleDrawer,
'vertical': VerticalBarsDrawer,
'horizontal': HorizontalBarsDrawer,
}
drawer_cls = _drawers.get(style)
if drawer_cls:
qr_obj = qr.make_image(
image_factory=StyledPilImage,
module_drawer=drawer_cls(),
fill_color=fg,
back_color=bg,
)
tmp = io.BytesIO()
qr_obj.save(tmp, 'PNG')
tmp.seek(0)
pil_img = Image.open(tmp).convert('RGB')
except (ImportError, Exception):
pass
if pil_img is None:
qr_obj = qr.make_image(fill_color=fg, back_color=bg)
tmp = io.BytesIO()
qr_obj.save(tmp, 'PNG')
tmp.seek(0)
pil_img = Image.open(tmp).convert('RGB')
pil_img = pil_img.resize((size, size), Image.LANCZOS)
if logo_bytes:
from PIL import ImageDraw
logo = Image.open(io.BytesIO(logo_bytes)).convert('RGBA')
logo_size = size // 4
logo = logo.resize((logo_size, logo_size), Image.LANCZOS)
# Erase a square tile at the centre to the background colour so
# QR modules appear to wrap around the logo rather than being
# covered by a floating patch. A 2–3 px gutter keeps the nearest
# module from butting right up against the logo edge.
pad = max(2, size // 100)
zone = logo_size + pad * 2
cx = (size - zone) // 2
cy = (size - zone) // 2
pil_img = pil_img.convert('RGBA')
draw = ImageDraw.Draw(pil_img)
draw.rectangle([cx, cy, cx + zone - 1, cy + zone - 1], fill=(*bg, 255))
# Paste logo centred inside the cleared tile
pil_img.paste(logo, (cx + pad, cy + pad), logo)
pil_img = pil_img.convert('RGB')
buf = io.BytesIO()
pil_img.save(buf, 'PNG')
return buf.getvalue()
except ImportError:
pass
# Fallback pure-python PNG renderer (no styles/logo support)
module_count = 21
cell = max(4, size // module_count)
img_size = cell * module_count
pixels = []
for row in range(img_size):
row_pixels = b''
for col in range(img_size):
r, c = row // cell, col // cell
in_finder = (
(r < 7 and c < 7) or (r < 7 and c >= module_count - 7) or
(r >= module_count - 7 and c < 7)
)
is_dark = False
if in_finder:
lr, lc = r % 7, c % 7
is_dark = (lr == 0 or lr == 6 or lc == 0 or lc == 6 or
(2 <= lr <= 4 and 2 <= lc <= 4))
elif (row // cell + col // cell) % 2 == 0:
is_dark = (r == 6 or c == 6) and (r % 2 == 0 or c % 2 == 0)
row_pixels += bytes(fg if is_dark else bg)
pixels.append(row_pixels)
def png_chunk(t, d):
c = t + d
return struct.pack('>I', len(d)) + c + struct.pack('>I', zlib.crc32(c) & 0xffffffff)
raw = b''.join(b'\x00' + row for row in pixels)
return (b'\x89PNG\r\n\x1a\n'
+ png_chunk(b'IHDR', struct.pack('>IIBBBBB', img_size, img_size, 8, 2, 0, 0, 0))
+ png_chunk(b'IDAT', zlib.compress(raw))
+ png_chunk(b'IEND', b''))
def _sniff_image_mime(data: bytes) -> str:
if data[:8] == b'\x89PNG\r\n\x1a\n': return 'image/png'
if data[:2] == b'\xff\xd8': return 'image/jpeg'
if data[:6] in (b'GIF87a', b'GIF89a'): return 'image/gif'
if data[:4] == b'RIFF' and data[8:12] == b'WEBP': return 'image/webp'
return 'image/png'
def generate_qr_svg(data: str, size: int = 300, fg: str = '#000000', bg: str = '#ffffff',
ec: str = 'M', logo_bytes: bytes = None) -> bytes:
"""Generate a vector QR as SVG (print-ready, scales losslessly).
A logo, when given, is embedded as a data URI over a cleared centre tile.
Dot styles are raster-only; SVG always renders square modules."""
import qrcode as qrc
qr = qrc.QRCode(error_correction=_qr_ec_constant(qrc, ec, logo_bytes), border=2)
qr.add_data(data)
qr.make(fit=True)
matrix = qr.get_matrix()
n = len(matrix)
path = ''.join(
f'M{x} {y}h1v1h-1z'
for y, row in enumerate(matrix)
for x, dark in enumerate(row) if dark
)
logo_svg = ''
if logo_bytes:
# Mirror the PNG renderer: clear a centre tile (logo = 1/4 of the code,
# plus a small gutter) and place the logo inside it.
logo_units = n / 4
pad = max(0.5, n / 100)
zone = logo_units + pad * 2
cx = (n - zone) / 2
b64 = base64.b64encode(logo_bytes).decode()
mime = _sniff_image_mime(logo_bytes)
logo_svg = (
f'<rect x="{cx:.2f}" y="{cx:.2f}" width="{zone:.2f}" height="{zone:.2f}" fill="{bg}"/>'
f'<image href="data:{mime};base64,{b64}" x="{cx + pad:.2f}" y="{cx + pad:.2f}" '
f'width="{logo_units:.2f}" height="{logo_units:.2f}" preserveAspectRatio="xMidYMid meet"/>'
)
svg = (
f'<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 {n} {n}" '
f'width="{size}" height="{size}" shape-rendering="crispEdges">'
f'<rect width="{n}" height="{n}" fill="{bg}"/>'
f'<path d="{path}" fill="{fg}"/>{logo_svg}</svg>'
)
return svg.encode('utf-8')
def png_to_pdf(png_bytes: bytes) -> bytes:
"""Wrap a rendered QR PNG in a single-page PDF (requires Pillow)."""
from PIL import Image
img = Image.open(io.BytesIO(png_bytes)).convert('RGB')
buf = io.BytesIO()
img.save(buf, 'PDF')
return buf.getvalue()
def qr_response(data_str: str, params, logo_bytes: bytes = None,
filename: str = 'qr', cache: bool = False):
"""Shared QR endpoint renderer — handles format (png/svg/pdf), colours,
size, style, error-correction, and optional attachment download."""
fmt = (params.get('format') or 'png').lower()
ec = (params.get('ec') or 'M').upper()
fg_hex = (params.get('fg') or '000000').lstrip('#')
bg_hex = (params.get('bg') or 'ffffff').lstrip('#')
size = min(int(params.get('size', 300) or 300), 1000)
style = params.get('style') or 'square'
download = str(params.get('download') or '').lower() in ('1', 'true')
if fmt == 'svg':
try:
body = generate_qr_svg(data_str, size=size, fg='#'+fg_hex, bg='#'+bg_hex,
ec=ec, logo_bytes=logo_bytes)
except ImportError:
return jsonify({'error': 'SVG output requires the qrcode library'}), 501
mime, ext = 'image/svg+xml', 'svg'
elif fmt == 'pdf':
png = generate_qr_png(data_str, size=size, fg=hex_to_rgb(fg_hex), bg=hex_to_rgb(bg_hex),
style=style, logo_bytes=logo_bytes, ec=ec)
try:
body = png_to_pdf(png)
except ImportError:
return jsonify({'error': 'PDF output requires Pillow'}), 501
mime, ext = 'application/pdf', 'pdf'
elif fmt == 'png':
body = generate_qr_png(data_str, size=size, fg=hex_to_rgb(fg_hex), bg=hex_to_rgb(bg_hex),
style=style, logo_bytes=logo_bytes, ec=ec)
mime, ext = 'image/png', 'png'
else:
return jsonify({'error': 'format must be png, svg, or pdf'}), 400
headers = {}
if cache:
headers['Cache-Control'] = 'public, max-age=3600'
if download:
headers['Content-Disposition'] = f'attachment; filename="{filename}.{ext}"'
return Response(body, mimetype=mime, headers=headers)
# ─────────────────────────────────────────────
# Shared helpers
# ─────────────────────────────────────────────
def generate_code(url: str, length: int = 6) -> str:
return hashlib.sha256(f"{url}{time.time()}".encode()).hexdigest()[:length]
def validate_url(url: str) -> bool:
return url.startswith(('http://', 'https://'))
def hex_to_rgb(h):
h = h.lstrip('#')
return tuple(int(h[i:i+2], 16) for i in (0, 2, 4))
def parse_device(ua):
if not ua: return 'Unknown'
u = ua.lower()
if any(x in u for x in ('mobile','android','iphone')): return 'Mobile'
if any(x in u for x in ('tablet','ipad')): return 'Tablet'
return 'Desktop'
def parse_browser(ua):
if not ua: return 'Unknown'
u = ua.lower()
if 'edg/' in u: return 'Edge'
if 'opr/' in u: return 'Opera'
if 'chrome/' in u: return 'Chrome'
if 'firefox/' in u: return 'Firefox'
if 'safari/' in u: return 'Safari'
if 'curl' in u: return 'curl'
if 'python' in u: return 'Python'
return 'Other'
def parse_referrer(ref):
if not ref: return 'Direct'
r = ref.lower()
if 'google' in r: return 'Google'
if 'bing' in r: return 'Bing'
if 'facebook' in r or 'fb.com' in r: return 'Facebook'
if 'twitter' in r or 't.co' in r or 'x.com' in r: return 'Twitter/X'
if 'linkedin' in r: return 'LinkedIn'
if 'reddit' in r: return 'Reddit'
if 'youtube' in r: return 'YouTube'
if 'instagram' in r: return 'Instagram'
return 'Other'
_BOT_UA_MARKERS = (
'bot', 'crawler', 'spider', 'slurp', 'curl', 'wget', 'python-requests',
'python-urllib', 'httpclient', 'go-http-client', 'okhttp', 'scrapy',
'headless', 'phantomjs', 'facebookexternalhit', 'whatsapp', 'telegrambot',
'discordbot', 'slackbot', 'linkedinbot', 'twitterbot', 'pinterest',
'preview', 'monitor', 'pingdom', 'uptime', 'postmanruntime', 'java/',
)
def is_bot_ua(ua):
"""Heuristic bot/crawler detection from the User-Agent string."""
if not ua:
return True
u = ua.lower()
return any(m in u for m in _BOT_UA_MARKERS)
def get_client_ip():
"""Return the real client IP, honouring X-Forwarded-For from trusted proxies."""
xff = request.headers.get('X-Forwarded-For', '')
if xff:
return xff.split(',')[0].strip()
return request.remote_addr or ''
def anonymize_ip(ip):
"""Apply the configured IP_ANONYMIZE policy before an IP is stored.
'truncate' zeroes the host part; 'hash' keeps unique-visitor counting
possible without retaining the address itself."""
if not ip or IP_ANONYMIZE == 'none':
return ip
if IP_ANONYMIZE == 'hash':
return hashlib.sha256(ip.encode()).hexdigest()[:16]
if IP_ANONYMIZE == 'truncate':
if ':' in ip:
parts = ip.split(':')
return ':'.join(parts[:3]) + '::'
parts = ip.split('.')
if len(parts) == 4:
return '.'.join(parts[:3]) + '.0'
return ip
# Local GeoLite2 country database (preferred over any network lookup).
# Download GeoLite2-Country.mmdb from MaxMind and mount it at GEOIP_DB_PATH.
_geoip_reader = None
try:
import geoip2.database as _geoip2_db
if os.path.exists(GEOIP_DB_PATH):
_geoip_reader = _geoip2_db.Reader(GEOIP_DB_PATH)
except Exception:
_geoip_reader = None
def get_country_for_ip(ip):
"""Return a 2-letter ISO country code for an IP address.
Priority:
1. Local MaxMind GeoLite2 database (offline, no rate limits, commercial-safe)
2. ip-api.com free JSON API — only if GEO_API_FALLBACK=true (their free
tier is rate-limited and licensed for non-commercial use only)
Returns 'XX' for private/loopback IPs, 'Unknown' on any failure.
Called from the background click writer, never from the request path.
"""
if not ip:
return 'Unknown'
try:
import ipaddress
parsed = ipaddress.ip_address(ip)
if parsed.is_private or parsed.is_loopback or parsed.is_link_local:
return 'XX'
except ValueError:
return 'Unknown'
if _geoip_reader:
try:
return _geoip_reader.country(ip).country.iso_code or 'Unknown'
except Exception:
return 'Unknown'
if GEO_API_FALLBACK:
try:
import urllib.request as _ureq
req = _ureq.Request(
f'http://ip-api.com/json/{ip}?fields=countryCode',
headers={'User-Agent': 'QRknit/1.0'}
)
with _ureq.urlopen(req, timeout=2) as resp:
data = json.loads(resp.read())
code = data.get('countryCode', '')
return code if code else 'Unknown'
except Exception:
return 'Unknown'
return 'Unknown'
def get_link_tags(conn, link_id):
rows = conn.execute(
'SELECT t.id, t.name FROM tags t JOIN link_tags lt ON t.id=lt.tag_id WHERE lt.link_id=?',
(link_id,)
).fetchall()
return [{'id': r['id'], 'name': r['name']} for r in rows]
def set_link_tags(conn, link_id, tag_names, user_id=None):
"""Replace all tags on a link. Tags are scoped to the owning user."""
conn.execute('DELETE FROM link_tags WHERE link_id=?', (link_id,))
for name in tag_names:
name = name.strip().lower()
if not name: continue
conn.execute('INSERT OR IGNORE INTO tags (name, user_id) VALUES (?,?)', (name, user_id))
if user_id is not None:
row = conn.execute(
'SELECT id FROM tags WHERE name=? AND user_id=?', (name, user_id)
).fetchone()
else:
row = conn.execute(
'SELECT id FROM tags WHERE name=? AND user_id IS NULL', (name,)
).fetchone()
if row:
conn.execute('INSERT OR IGNORE INTO link_tags (link_id,tag_id) VALUES (?,?)', (link_id, row['id']))
def format_link(row, conn):
owner = conn.execute('SELECT username FROM users WHERE id=?', (row['user_id'],)).fetchone()
return {
'id': row['id'],
'code': row['code'],
'long_url': row['long_url'],
'title': row['title'],
'created_at': row['created_at'],
'expires_at': row['expires_at'],
'clicks': row['clicks'],
'is_active': row['is_active'],
'is_pinned': row['is_pinned'],
'redirect_type': row['redirect_type'] or 302,
'has_password': bool(row['password_hash']),
'short_url': f"{BASE_URL}/{row['code']}",
'qr_url': f"{BASE_URL}/api/qr/{row['code']}",
'tags': get_link_tags(conn, row['id']),
'created_by': owner['username'] if owner else None,
}
# ─────────────────────────────────────────────
# Auth Routes
# ─────────────────────────────────────────────
# Health check (public — used by Docker healthcheck)
# ─────────────────────────────────────────────
@app.route('/api/health')
def health():
return jsonify({'status': 'ok'})
@app.route('/api/uptime')
def get_uptime():
"""Public endpoint — 30-day daily uptime history for the header SLA graph.
Computed from per-minute 'up' marks: pct = minutes marked / minutes elapsed
that day, so downtime actually shows up (the old per-process counter could
only ever report 100%, multiplied per worker). Falls back to legacy
uptime_daily rows for days that predate the minute marks.
"""
now = datetime.now(timezone.utc)
today = now.date()
start = today - timedelta(days=29)
with get_db() as conn:
rows = conn.execute(
"SELECT substr(minute,1,10) as day, COUNT(*) as mins, MIN(minute) as first "
"FROM uptime_minutes WHERE minute >= ? GROUP BY day",
(start.isoformat(),)
).fetchall()
legacy = conn.execute(
"SELECT date, checks_total, checks_up FROM uptime_daily WHERE date >= ?",
(start.isoformat(),)
).fetchall()
min_map = {r['day']: r for r in rows}
legacy_map = {r['date']: r for r in legacy}
days = []
for i in range(30):
d_date = start + timedelta(days=i)
d = d_date.isoformat()
row = min_map.get(d)
if row:
# Window runs from the day's first heartbeat (so fresh installs and
# mid-day deployments aren't penalised) to end-of-day / now.
first_min = int(row['first'][11:13]) * 60 + int(row['first'][14:16])
end_min = 1439 if d_date < today else now.hour * 60 + now.minute
expected = max(1, end_min - first_min + 1)
pct = round(min(row['mins'] / expected, 1.0) * 100, 2)
elif d in legacy_map and legacy_map[d]['checks_total'] > 0:
r = legacy_map[d]
pct = round(r['checks_up'] / r['checks_total'] * 100, 2)
else:
pct = None
days.append({'date': d, 'pct': pct})
valid = [x for x in days if x['pct'] is not None]
overall = round(sum(x['pct'] for x in valid) / len(valid), 2) if valid else None
return jsonify({'days': days, 'overall': overall})
@app.route('/api/config')
def get_config():
"""Public endpoint — exposes non-sensitive deployment config to the frontend."""
return jsonify({'app_name': APP_NAME, 'base_url': BASE_URL})
@app.route('/api/auth/login', methods=['POST'])
def login():
data = request.get_json(silent=True) or {}
username = (data.get('username') or '').strip()
password = data.get('password') or ''
if not username or not password:
return jsonify({'error': 'Username and password required'}), 400
with get_db() as conn:
user = conn.execute('SELECT * FROM users WHERE username=?', (username,)).fetchone()
if not user or not check_password_hash(user['password_hash'], password):
return jsonify({'error': 'Invalid username or password'}), 401
session.permanent = True
session['authenticated'] = True
session['user_id'] = user['id']
session['username'] = user['username']
session['is_admin'] = bool(user['is_admin'])
session['org_id'] = user['org_id'] if user['org_id'] else None
with get_db() as conn:
plan = get_user_plan(conn, user['id'])
return jsonify({
'authenticated': True,
'id': user['id'],
'username': user['username'],
'is_admin': bool(user['is_admin']),
'org_id': user['org_id'],
'plan': plan,
'plan_limits': PLAN_LIMITS[plan],
})
@app.route('/api/auth/logout', methods=['POST'])
def logout():
session.clear()
return jsonify({'success': True})
@app.route('/api/auth/me', methods=['GET'])
def me():
if session.get('authenticated') and session.get('user_id'):
with get_db() as conn:
plan = get_user_plan(conn, session.get('user_id'))
return jsonify({
'authenticated': True,
'id': session.get('user_id'),
'username': session.get('username'),
'is_admin': session.get('is_admin', False),
'org_id': session.get('org_id'),
'plan': plan,
'plan_limits': PLAN_LIMITS[plan],
})
return jsonify({'authenticated': False}), 401
# ─────────────────────────────────────────────
# Links
# ─────────────────────────────────────────────
@app.route('/api/shorten', methods=['POST'])
@login_required
def shorten():
data = request.get_json(silent=True) or {}
long_url = (data.get('url') or '').strip()
custom_code = (data.get('custom_code') or '').strip()
title = (data.get('title') or '').strip()
expires_at = data.get('expires_at')
tags = data.get('tags', [])
redirect_type = data.get('redirect_type', 302)
password = data.get('password') or ''
if not long_url:
return jsonify({'error': 'URL is required'}), 400
if not validate_url(long_url):
return jsonify({'error': 'URL must start with http:// or https://'}), 400
if custom_code and not re.match(r'^[a-zA-Z0-9]{1,20}$', custom_code):
return jsonify({'error': 'Custom code must be 1–20 alphanumeric characters'}), 400
if redirect_type not in (301, 302, '301', '302'):
return jsonify({'error': 'redirect_type must be 301 or 302'}), 400
redirect_type = int(redirect_type)
code = custom_code or generate_code(long_url)
with get_db() as conn:
uid = session.get('user_id')
plan = get_user_plan(conn, uid)
limit = PLAN_LIMITS[plan]['max_links']
if limit is not None and get_user_active_link_count(conn, uid) >= limit:
return jsonify({
'error': f'Active link limit reached for your {plan.capitalize()} plan ({limit} links). '
'Please upgrade to create more links.',
'limit_reached': True,
'plan': plan,
}), 403
existing = conn.execute('SELECT code FROM links WHERE code=?', (code,)).fetchone()
if existing:
if custom_code:
return jsonify({'error': 'Custom code already taken'}), 409
code = generate_code(long_url + str(time.time()))
conn.execute(
'INSERT INTO links (code,long_url,title,created_at,expires_at,user_id,redirect_type,password_hash) '
'VALUES (?,?,?,?,?,?,?,?)',
(code, long_url, title or None,
datetime.now(timezone.utc).replace(tzinfo=None).isoformat(),
expires_at or None, session.get('user_id'), redirect_type,
generate_password_hash(password) if password else None)
)
link_id = conn.execute('SELECT id FROM links WHERE code=?', (code,)).fetchone()['id']
if tags:
set_link_tags(conn, link_id, tags, session.get('user_id'))
return jsonify({
'code': code,
'short_url': f"{BASE_URL}/{code}",
'long_url': long_url,
'title': title,
'tags': tags,
'redirect_type': redirect_type,
'has_password': bool(password),
'qr_url': f"{BASE_URL}/api/qr/{code}",
}), 201
@app.route('/api/links', methods=['GET'])
@login_required
def list_links():
page = int(request.args.get('page', 1))
per_page = min(int(request.args.get('per_page', 20)), 100)
offset = (page - 1) * per_page
search = (request.args.get('q') or '').strip()
tag_filter = (request.args.get('tag') or '').strip().lower()
user_filter = (request.args.get('user') or '').strip()
is_admin = session.get('is_admin', False)
user_id = session.get('user_id')
with get_db() as conn:
org_id = get_user_org_id(conn, user_id) if not is_admin else None
where_clauses = ['l.is_active=1']
params = []
# Admins see only their own links (not cluttered by all users)
if is_admin:
if user_filter:
where_clauses.append(
'l.user_id=(SELECT id FROM users WHERE username=?)'
)
params.append(user_filter)
else:
where_clauses.append('l.user_id=?')
params.append(user_id)
elif org_id:
# Org members see all links within their org
where_clauses.append(
'l.user_id IN (SELECT id FROM users WHERE org_id=?)'
)
params.append(org_id)
else:
# Solo users see only their own links
where_clauses.append('l.user_id=?')
params.append(user_id)
if search:
where_clauses.append('(l.code LIKE ? OR l.long_url LIKE ? OR l.title LIKE ?)')
s = f'%{search}%'
params += [s, s, s]
if tag_filter:
where_clauses.append(
'l.id IN (SELECT lt.link_id FROM link_tags lt '
'JOIN tags t ON lt.tag_id=t.id WHERE t.name=?)'
)
params.append(tag_filter)
where_sql = ' AND '.join(where_clauses)
with get_db() as conn:
total = conn.execute(f'SELECT COUNT(*) FROM links l WHERE {where_sql}', params).fetchone()[0]
rows = conn.execute(
f'SELECT * FROM links l WHERE {where_sql} ORDER BY l.is_pinned DESC, l.created_at DESC LIMIT ? OFFSET ?',
params + [per_page, offset]
).fetchall()
links = [format_link(r, conn) for r in rows]
return jsonify({'links': links, 'total': total, 'page': page, 'per_page': per_page})
def _can_access_link(link, conn):
"""Return True if the current session user may read/write this link."""
if session.get('is_admin'):
return True
if link['user_id'] == session.get('user_id'):
return True
# Org members can access each other's links
user_org = session.get('org_id')
if user_org and link['user_id']:
owner = conn.execute('SELECT org_id FROM users WHERE id=?', (link['user_id'],)).fetchone()
if owner and owner['org_id'] == user_org:
return True
return False
@app.route('/api/links/<code>', methods=['GET'])
@login_required
def link_detail(code):
with get_db() as conn:
link = conn.execute('SELECT * FROM links WHERE code=?', (code,)).fetchone()
if not link or not _can_access_link(link, conn):
return jsonify({'error': 'Not found'}), 404
return jsonify(format_link(link, conn))
@app.route('/api/links/<code>', methods=['PATCH'])
@login_required
def edit_link(code):
with get_db() as conn:
link = conn.execute('SELECT * FROM links WHERE code=? AND is_active=1', (code,)).fetchone()
if not link or not _can_access_link(link, conn):
return jsonify({'error': 'Not found'}), 404
data = request.get_json(silent=True) or {}
updates = {}
if 'url' in data:
url = data['url'].strip()
if not validate_url(url): return jsonify({'error': 'Invalid URL'}), 400
updates['long_url'] = url
if 'title' in data: updates['title'] = data['title'].strip() or None
if 'expires_at' in data: updates['expires_at'] = data['expires_at'] or None
if 'is_pinned' in data: updates['is_pinned'] = 1 if data['is_pinned'] else 0
if 'redirect_type' in data:
if data['redirect_type'] not in (301, 302, '301', '302'):
return jsonify({'error': 'redirect_type must be 301 or 302'}), 400
updates['redirect_type'] = int(data['redirect_type'])
if 'password' in data:
# Non-empty string sets/updates the password; empty/null removes it
pw = data['password'] or ''
updates['password_hash'] = generate_password_hash(pw) if pw else None
if updates:
set_clause = ', '.join(f'{k}=?' for k in updates)
conn.execute(f'UPDATE links SET {set_clause} WHERE code=?',
list(updates.values()) + [code])
if 'tags' in data:
set_link_tags(conn, link['id'], data['tags'], link['user_id'])
updated = conn.execute('SELECT * FROM links WHERE code=?', (code,)).fetchone()
return jsonify(format_link(updated, conn))
@app.route('/api/links/<code>', methods=['DELETE'])
@login_required
def delete_link(code):
with get_db() as conn:
link = conn.execute('SELECT * FROM links WHERE code=?', (code,)).fetchone()
if not link or not _can_access_link(link, conn):
return jsonify({'error': 'Not found'}), 404
conn.execute('UPDATE links SET is_active=0, deleted_at=? WHERE code=?', (now_iso(), code))
return jsonify({'success': True})
# ─────────────────────────────────────────────
# Analytics
# ─────────────────────────────────────────────
@app.route('/api/links/<code>/analytics')
@login_required
def link_analytics(code):
with get_db() as conn:
link = conn.execute('SELECT * FROM links WHERE code=?', (code,)).fetchone()
if not link or not _can_access_link(link, conn):
return jsonify({'error': 'Not found'}), 404
plan = get_user_plan(conn, session.get('user_id'))
max_days = PLAN_LIMITS[plan]['analytics_days']
days = min(int(request.args.get('days', 30)), max_days)
exclude_bots = str(request.args.get('exclude_bots') or '').lower() in ('1', 'true')
bot_clause = ' AND COALESCE(is_bot,0)=0' if exclude_bots else ''
link_id = link['id']
since = (datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(days=days)).isoformat()
daily_rows = conn.execute(f"""
SELECT substr(clicked_at,1,10) as day, COUNT(*) as count,
COUNT(DISTINCT NULLIF(ip_address,'')) as uniq
FROM clicks WHERE link_id=? AND clicked_at>=?{bot_clause}
GROUP BY day ORDER BY day
""", (link_id, since)).fetchall()
daily_map = {r['day']: r['count'] for r in daily_rows}
uniq_map = {r['day']: r['uniq'] for r in daily_rows}
daily = [
{'date': (datetime.now(timezone.utc).replace(tzinfo=None)-timedelta(days=days-1-i)).strftime('%Y-%m-%d'), 'clicks': 0, 'unique': 0}
for i in range(days)
]
for d in daily:
d['clicks'] = daily_map.get(d['date'], 0)
d['unique'] = uniq_map.get(d['date'], 0)
unique_visitors = conn.execute(
f"SELECT COUNT(DISTINCT NULLIF(ip_address,'')) FROM clicks "
f"WHERE link_id=? AND clicked_at>=?{bot_clause}",
(link_id, since)
).fetchone()[0]
bot_clicks = conn.execute(
'SELECT COUNT(*) FROM clicks WHERE link_id=? AND clicked_at>=? AND COALESCE(is_bot,0)=1',
(link_id, since)
).fetchone()[0]
ref_rows = conn.execute(
f'SELECT referrer, COUNT(*) as count FROM clicks WHERE link_id=? AND clicked_at>=?{bot_clause} GROUP BY referrer',
(link_id, since)
).fetchall()
referrers = {}
for r in ref_rows:
b = parse_referrer(r['referrer']); referrers[b] = referrers.get(b,0) + r['count']
referrers = [{'source':k,'count':v} for k,v in sorted(referrers.items(), key=lambda x:-x[1])]
ua_rows = conn.execute(
f'SELECT user_agent, COUNT(*) as count FROM clicks WHERE link_id=? AND clicked_at>=?{bot_clause} GROUP BY user_agent',
(link_id, since)
).fetchall()
devices = {}; browsers = {}
for r in ua_rows:
ua = r['user_agent'] or ''
devices[parse_device(ua)] = devices.get(parse_device(ua),0) + r['count']
browsers[parse_browser(ua)] = browsers.get(parse_browser(ua),0) + r['count']
devices = [{'device':k,'count':v} for k,v in sorted(devices.items(), key=lambda x:-x[1])]
browsers = [{'browser':k,'count':v} for k,v in sorted(browsers.items(), key=lambda x:-x[1])]
# Hourly heatmap: 7 days-of-week × 24 hours
# SQLite strftime('%w') returns 0=Sunday … 6=Saturday; we map to 0=Monday … 6=Sunday
hourly_rows = conn.execute(f"""
SELECT CAST(strftime('%w', clicked_at) AS INTEGER) as dow,
CAST(strftime('%H', clicked_at) AS INTEGER) as hr,
COUNT(*) as count
FROM clicks WHERE link_id=? AND clicked_at>=?{bot_clause}
GROUP BY dow, hr
""", (link_id, since)).fetchall()
# heatmap[day_of_week 0=Mon][hour 0-23]
heatmap = [[0] * 24 for _ in range(7)]
for r in hourly_rows:
mon_dow = (r['dow'] - 1) % 7 # 0=Sun→6, 1=Mon→0, …
heatmap[mon_dow][r['hr']] = r['count']
# Geographic breakdown
country_rows = conn.execute(f"""
SELECT COALESCE(NULLIF(country,''),'Unknown') as country, COUNT(*) as count
FROM clicks WHERE link_id=? AND clicked_at>=?{bot_clause}
GROUP BY country ORDER BY count DESC LIMIT 20
""", (link_id, since)).fetchall()
countries = [{'country': r['country'], 'count': r['count']} for r in country_rows]
return jsonify({
'code': code, 'days': days, 'max_days': max_days,
'exclude_bots': exclude_bots,
'total_clicks': link['clicks'],
'period_clicks': sum(d['clicks'] for d in daily),
'unique_visitors': unique_visitors,
'bot_clicks': bot_clicks,
'daily': daily, 'referrers': referrers, 'devices': devices, 'browsers': browsers,
'heatmap': heatmap, 'countries': countries,
})
# ─────────────────────────────────────────────
# Aggregate analytics — across the whole account (or org), optionally
# filtered to a tag. Powers the dashboard "audience" panel and lets
# campaign runners see one chart for N tagged links.
# ─────────────────────────────────────────────
@app.route('/api/analytics/aggregate')
@login_required
def aggregate_analytics():
user_id = session.get('user_id')
is_admin = session.get('is_admin', False)
tag = (request.args.get('tag') or '').strip().lower()
exclude_bots = str(request.args.get('exclude_bots') or '').lower() in ('1', 'true')
bot_clause = ' AND COALESCE(c.is_bot,0)=0' if exclude_bots else ''
with get_db() as conn:
plan = get_user_plan(conn, user_id)
max_days = PLAN_LIMITS[plan]['analytics_days']
days = min(int(request.args.get('days', 30)), max_days)
since = (datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(days=days)).isoformat()
org_id = get_user_org_id(conn, user_id)
if org_id and not is_admin:
scope_clause = 'l.user_id IN (SELECT id FROM users WHERE org_id=?)'
scope_params = [org_id]
else:
scope_clause = 'l.user_id=?'
scope_params = [user_id]
where = f'l.is_active=1 AND {scope_clause} AND c.clicked_at>=?{bot_clause}'
params = list(scope_params) + [since]
if tag:
where += (' AND l.id IN (SELECT lt.link_id FROM link_tags lt '
'JOIN tags t ON lt.tag_id=t.id WHERE t.name=?)')
params.append(tag)
base = f'FROM clicks c JOIN links l ON c.link_id=l.id WHERE {where}'
daily_rows = conn.execute(f"""
SELECT substr(c.clicked_at,1,10) as day, COUNT(*) as count,
COUNT(DISTINCT NULLIF(c.ip_address,'')) as uniq
{base} GROUP BY day ORDER BY day
""", params).fetchall()
daily_map = {r['day']: r['count'] for r in daily_rows}
uniq_map = {r['day']: r['uniq'] for r in daily_rows}
daily = [
{'date': (datetime.now(timezone.utc).replace(tzinfo=None)-timedelta(days=days-1-i)).strftime('%Y-%m-%d'), 'clicks': 0, 'unique': 0}
for i in range(days)
]
for d in daily:
d['clicks'] = daily_map.get(d['date'], 0)
d['unique'] = uniq_map.get(d['date'], 0)
unique_visitors = conn.execute(
f"SELECT COUNT(DISTINCT NULLIF(c.ip_address,'')) {base}", params
).fetchone()[0]
ref_rows = conn.execute(
f'SELECT c.referrer, COUNT(*) as count {base} GROUP BY c.referrer', params
).fetchall()
referrers = {}
for r in ref_rows:
b = parse_referrer(r['referrer']); referrers[b] = referrers.get(b,0) + r['count']
referrers = [{'source':k,'count':v} for k,v in sorted(referrers.items(), key=lambda x:-x[1])]
ua_rows = conn.execute(
f'SELECT c.user_agent, COUNT(*) as count {base} GROUP BY c.user_agent', params
).fetchall()
devices = {}; browsers = {}
for r in ua_rows:
ua = r['user_agent'] or ''
devices[parse_device(ua)] = devices.get(parse_device(ua),0) + r['count']
browsers[parse_browser(ua)] = browsers.get(parse_browser(ua),0) + r['count']
devices = [{'device':k,'count':v} for k,v in sorted(devices.items(), key=lambda x:-x[1])]
browsers = [{'browser':k,'count':v} for k,v in sorted(browsers.items(), key=lambda x:-x[1])]
country_rows = conn.execute(f"""
SELECT COALESCE(NULLIF(c.country,''),'Unknown') as country, COUNT(*) as count
{base} GROUP BY country ORDER BY count DESC LIMIT 20
""", params).fetchall()
countries = [{'country': r['country'], 'count': r['count']} for r in country_rows]
top_rows = conn.execute(f"""
SELECT l.code, l.title, l.long_url, COUNT(*) as count
{base} GROUP BY l.id ORDER BY count DESC LIMIT 10
""", params).fetchall()
top_links = [dict(r) for r in top_rows]
return jsonify({
'days': days, 'max_days': max_days, 'tag': tag or None,
'exclude_bots': exclude_bots,
'period_clicks': sum(d['clicks'] for d in daily),
'unique_visitors': unique_visitors,
'daily': daily, 'referrers': referrers, 'devices': devices,
'browsers': browsers, 'countries': countries, 'top_links': top_links,
})
# ─────────────────────────────────────────────
# Click-event CSV export
# ─────────────────────────────────────────────
@app.route('/api/links/<code>/clicks/export')
@login_required
def export_clicks(code):
with get_db() as conn:
link = conn.execute('SELECT * FROM links WHERE code=?', (code,)).fetchone()
if not link or not _can_access_link(link, conn):
return jsonify({'error': 'Not found'}), 404
rows = conn.execute(
'SELECT clicked_at, referrer, user_agent, country FROM clicks '
'WHERE link_id=? ORDER BY clicked_at DESC',
(link['id'],)
).fetchall()
buf = io.StringIO()
writer = csv.writer(buf)
writer.writerow(['timestamp', 'referrer', 'device', 'browser', 'country'])
for row in rows:
ua = row['user_agent'] or ''
writer.writerow([
row['clicked_at'],
row['referrer'] or '',
parse_device(ua),
parse_browser(ua),
row['country'] or '',
])
return Response(
buf.getvalue().encode('utf-8'),
mimetype='text/csv',
headers={'Content-Disposition': f'attachment; filename="clicks-{code}.csv"'}
)
# ─────────────────────────────────────────────
# Fetch Title
# ─────────────────────────────────────────────
@app.route('/api/fetch-title')
@login_required
def fetch_title():
"""Fetch the page title for a URL server-side (avoids CORS)."""
import urllib.request as urllib_req
url = (request.args.get('url') or '').strip()
if not url or not validate_url(url):
return jsonify({'title': ''})
try:
req = urllib_req.Request(url, headers={
'User-Agent': 'Mozilla/5.0 (compatible; QRknit-title-fetcher/1.0)',
'Accept': 'text/html',
})
with urllib_req.urlopen(req, timeout=5) as resp:
ct = resp.headers.get('Content-Type', '')
if 'html' not in ct.lower():
return jsonify({'title': ''})
html = resp.read(65536).decode('utf-8', errors='replace')
# og:title (two attribute orderings)
m = re.search(r'<meta[^>]+property=["\']og:title["\'][^>]+content=["\']([^"\']*)["\']', html, re.I)
if not m:
m = re.search(r'<meta[^>]+content=["\']([^"\']*)["\'][^>]+property=["\']og:title["\']', html, re.I)
if m:
return jsonify({'title': m.group(1).strip()[:200]})
# Fall back to <title>
m = re.search(r'<title[^>]*>([^<]+)</title>', html, re.I)
if m:
return jsonify({'title': m.group(1).strip()[:200]})
return jsonify({'title': ''})
except Exception:
return jsonify({'title': ''})
# ─────────────────────────────────────────────
# Plan usage
# ─────────────────────────────────────────────
@app.route('/api/plan')
@login_required
def plan_usage():
uid = session.get('user_id')
with get_db() as conn:
plan = get_user_plan(conn, uid)
link_count = get_user_active_link_count(conn, uid)
monthly_clicks = get_user_monthly_clicks(conn, uid)
limits = PLAN_LIMITS[plan]
return jsonify({
'plan': plan,
'limits': limits,
'usage': {
'active_links': link_count,
'monthly_clicks': monthly_clicks,
},
})
# ─────────────────────────────────────────────
# Tags & Stats
# ─────────────────────────────────────────────
@app.route('/api/tags')
@login_required
def list_tags():
user_id = session.get('user_id')
is_admin = session.get('is_admin', False)
with get_db() as conn:
if is_admin:
# Admins see only their own tags
rows = conn.execute("""
SELECT t.id, t.name, COUNT(lt.link_id) as link_count
FROM tags t LEFT JOIN link_tags lt ON t.id=lt.tag_id
WHERE t.user_id=?
GROUP BY t.id ORDER BY t.name
""", (user_id,)).fetchall()
else:
org_id = get_user_org_id(conn, user_id)
if org_id:
# Org members see tags from all members of their org
rows = conn.execute("""
SELECT t.id, t.name, COUNT(lt.link_id) as link_count
FROM tags t LEFT JOIN link_tags lt ON t.id=lt.tag_id
WHERE t.user_id IN (SELECT id FROM users WHERE org_id=?)
GROUP BY t.id ORDER BY t.name
""", (org_id,)).fetchall()
else:
# Solo users see only their own tags
rows = conn.execute("""
SELECT t.id, t.name, COUNT(lt.link_id) as link_count
FROM tags t LEFT JOIN link_tags lt ON t.id=lt.tag_id
WHERE t.user_id=?
GROUP BY t.id ORDER BY t.name
""", (user_id,)).fetchall()
return jsonify({'tags': [dict(r) for r in rows]})
@app.route('/api/stats')
@login_required
def stats():
is_admin = session.get('is_admin', False)
user_id = session.get('user_id')
with get_db() as conn:
org_id = get_user_org_id(conn, user_id)
# Determine the user_id filter clause and params
# Admins see only their own stats; org members pool across org members
if org_id and not is_admin:
scope_clause = 'l.user_id IN (SELECT id FROM users WHERE org_id=?)'
scope_param = org_id
else:
scope_clause = 'l.user_id=?'
scope_param = user_id
total_links = conn.execute(
f'SELECT COUNT(*) FROM links l WHERE is_active=1 AND {scope_clause}', (scope_param,)
).fetchone()[0]
total_clicks = conn.execute(
f'SELECT COALESCE(SUM(clicks),0) FROM links l WHERE is_active=1 AND {scope_clause}', (scope_param,)
).fetchone()[0]
since_7d = (datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(days=7)).isoformat()
since_30d = (datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(days=30)).isoformat()
clicks_7d = conn.execute(
f'SELECT COUNT(*) FROM clicks c JOIN links l ON c.link_id=l.id '
f'WHERE c.clicked_at>=? AND l.is_active=1 AND {scope_clause}', (since_7d, scope_param)
).fetchone()[0]
top_links = conn.execute(
f'SELECT code, long_url, title, clicks FROM links l WHERE is_active=1 AND {scope_clause} '
f'ORDER BY clicks DESC LIMIT 5', (scope_param,)
).fetchall()
daily_rows = conn.execute(f"""
SELECT substr(c.clicked_at,1,10) as day, COUNT(*) as count
FROM clicks c JOIN links l ON c.link_id=l.id
WHERE c.clicked_at>=? AND l.is_active=1 AND {scope_clause}
GROUP BY day ORDER BY day
""", (since_30d, scope_param)).fetchall()
daily_map = {r['day']: r['count'] for r in daily_rows}
daily = [
{'date': (datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(days=29-i)).strftime('%Y-%m-%d'), 'clicks': 0}
for i in range(30)
]
for d in daily:
d['clicks'] = daily_map.get(d['date'], 0)
return jsonify({
'total_links': total_links,
'total_clicks': total_clicks,
'clicks_7d': clicks_7d,
'top_links': [dict(r) for r in top_links],
'daily': daily,
})
# ─────────────────────────────────────────────
# QR Routes
# ─────────────────────────────────────────────
# All QR endpoints accept: format=png|svg|pdf, ec=L|M|Q|H, fg, bg, size,
# style (raster formats only), download=1 for an attachment response.
@app.route('/api/qr/<code>')
def qr_code(code):
with get_db() as conn:
link = conn.execute('SELECT 1 FROM links WHERE code=? AND is_active=1', (code,)).fetchone()
if not link:
return jsonify({'error': 'Not found'}), 404
return qr_response(f"{BASE_URL}/{code}", request.args,
filename=f'qr-{code}', cache=True)
@app.route('/api/qr/custom', methods=['GET'])
def qr_custom():
url = request.args.get('url', '').strip()
if not url or not validate_url(url):
return jsonify({'error': 'Valid URL required'}), 400
return qr_response(url, request.args, filename='qrcode')
@app.route('/api/qr/custom', methods=['POST'])
def qr_custom_post():
data = request.get_json(silent=True) or {}
url = (data.get('url') or '').strip()
if not url or not validate_url(url):
return jsonify({'error': 'Valid URL required'}), 400
logo_bytes = None
logo_b64 = data.get('logo', '')
if logo_b64:
try:
logo_bytes = base64.b64decode(logo_b64)
except Exception:
return jsonify({'error': 'Invalid logo data'}), 400
return qr_response(url, data, logo_bytes=logo_bytes, filename='qrcode')
# ─────────────────────────────────────────────
# Bulk Operations
# ─────────────────────────────────────────────
@app.route('/api/links/bulk', methods=['POST'])
@login_required
def bulk_links():
data = request.get_json(silent=True) or {}
action = data.get('action')
codes = data.get('codes', [])
if not codes:
return jsonify({'error': 'No codes provided'}), 400
if action not in ('delete', 'tag', 'expire'):
return jsonify({'error': 'Invalid action'}), 400
is_admin = session.get('is_admin', False)
user_id = session.get('user_id')
placeholders = ','.join('?' * len(codes))
with get_db() as conn:
if action == 'delete':
if is_admin:
conn.execute(f'UPDATE links SET is_active=0, deleted_at=? WHERE code IN ({placeholders})',
[now_iso()] + list(codes))
else:
conn.execute(
f'UPDATE links SET is_active=0, deleted_at=? WHERE code IN ({placeholders}) AND user_id=?',
[now_iso()] + list(codes) + [user_id]
)
return jsonify({'deleted': len(codes)})
elif action == 'tag':
tags = data.get('tags', [])
user_org = session.get('org_id')
for code in codes:
q = 'SELECT id,user_id FROM links WHERE code=? AND is_active=1'
link = conn.execute(q, (code,)).fetchone()
if not link:
continue
same_org = False
if user_org and link['user_id']:
owner = conn.execute('SELECT org_id FROM users WHERE id=?', (link['user_id'],)).fetchone()
same_org = owner and owner['org_id'] == user_org
if is_admin or link['user_id'] == user_id or same_org:
set_link_tags(conn, link['id'], tags, link['user_id'])
return jsonify({'tagged': len(codes)})
elif action == 'expire':
expires_at = data.get('expires_at') or None
if is_admin:
conn.execute(
f'UPDATE links SET expires_at=? WHERE code IN ({placeholders})',
[expires_at] + list(codes)
)
else:
conn.execute(
f'UPDATE links SET expires_at=? WHERE code IN ({placeholders}) AND user_id=?',
[expires_at] + list(codes) + [user_id]
)
return jsonify({'updated': len(codes)})
# ─────────────────────────────────────────────
# CSV Export / Import
# ─────────────────────────────────────────────
@app.route('/api/links/export')
@login_required
def export_links():
is_admin = session.get('is_admin', False)
user_id = session.get('user_id')
with get_db() as conn:
if is_admin:
rows = conn.execute(
'SELECT l.*, GROUP_CONCAT(t.name) as tag_names '
'FROM links l '
'LEFT JOIN link_tags lt ON l.id=lt.link_id '
'LEFT JOIN tags t ON lt.tag_id=t.id '
'WHERE l.is_active=1 '
'GROUP BY l.id ORDER BY l.created_at DESC'
).fetchall()
else:
rows = conn.execute(
'SELECT l.*, GROUP_CONCAT(t.name) as tag_names '
'FROM links l '
'LEFT JOIN link_tags lt ON l.id=lt.link_id '
'LEFT JOIN tags t ON lt.tag_id=t.id '
'WHERE l.is_active=1 AND l.user_id=? '
'GROUP BY l.id ORDER BY l.created_at DESC',
(user_id,)
).fetchall()
buf = io.StringIO()
writer = csv.writer(buf)
writer.writerow(['code', 'short_url', 'long_url', 'title', 'tags', 'created_at', 'expires_at', 'clicks'])
for row in rows:
writer.writerow([
row['code'],
f"{BASE_URL}/{row['code']}",
row['long_url'],
row['title'] or '',
row['tag_names'] or '',
row['created_at'],
row['expires_at'] or '',
row['clicks'],
])
return Response(
buf.getvalue().encode('utf-8'),
mimetype='text/csv',
headers={'Content-Disposition': f'attachment; filename="{re.sub(r"[^a-z0-9]+", "-", APP_NAME.lower()).strip("-")}-export.csv"'}
)
@app.route('/api/links/import', methods=['POST'])
@login_required
def import_links():
data = request.get_json(silent=True) or {}
csv_text = (data.get('csv') or '').strip()
if not csv_text:
return jsonify({'error': 'No CSV data provided'}), 400
reader = csv.DictReader(io.StringIO(csv_text))
created = 0
errors = []
with get_db() as conn:
for i, row in enumerate(reader, start=2):
url = (row.get('url') or row.get('long_url') or '').strip()
if not url:
errors.append(f'Row {i}: missing URL'); continue
if not validate_url(url):
errors.append(f'Row {i}: invalid URL "{url[:50]}"'); continue
custom_code = (row.get('custom_code') or row.get('code') or '').strip()
title = (row.get('title') or '').strip()
expires_at = (row.get('expires_at') or '').strip() or None
tags_str = (row.get('tags') or '').strip()
tag_names = [t.strip() for t in tags_str.split(',') if t.strip()] if tags_str else []
if custom_code and not re.match(r'^[a-zA-Z0-9]{1,20}$', custom_code):
errors.append(f'Row {i}: invalid code "{custom_code}"'); continue
code = custom_code or generate_code(url)
if conn.execute('SELECT 1 FROM links WHERE code=?', (code,)).fetchone():
if custom_code:
errors.append(f'Row {i}: code "{custom_code}" already taken'); continue
code = generate_code(url + str(time.time()))
conn.execute(
'INSERT INTO links (code, long_url, title, created_at, expires_at, user_id) VALUES (?,?,?,?,?,?)',
(code, url, title or None,
datetime.now(timezone.utc).replace(tzinfo=None).isoformat(), expires_at,
session.get('user_id'))
)
link_id = conn.execute('SELECT id FROM links WHERE code=?', (code,)).fetchone()['id']
if tag_names:
set_link_tags(conn, link_id, tag_names, session.get('user_id'))
created += 1
return jsonify({'created': created, 'errors': errors})
# ─────────────────────────────────────────────
# API keys — self-serve programmatic access
# Tokens look like qrk_… and are shown exactly once at creation.
# Only a SHA-256 hash is stored.
# ─────────────────────────────────────────────
@app.route('/api/keys', methods=['GET'])
@login_required
def list_api_keys():
with get_db() as conn:
rows = conn.execute(
'SELECT id, name, prefix, scope, created_at, last_used_at, revoked '
'FROM api_keys WHERE user_id=? ORDER BY created_at DESC',
(session.get('user_id'),)
).fetchall()
return jsonify({'keys': [dict(r) for r in rows]})
@app.route('/api/keys', methods=['POST'])
@login_required
def create_api_key():
data = request.get_json(silent=True) or {}
name = (data.get('name') or '').strip()[:64] or 'API key'
scope = (data.get('scope') or 'read').lower()
if scope not in ('read', 'write'):
return jsonify({'error': "scope must be 'read' or 'write'"}), 400
token = 'qrk_' + secrets.token_urlsafe(32)
key_hash = hashlib.sha256(token.encode()).hexdigest()
prefix = token[:12]
with get_db() as conn:
cur = conn.execute(
'INSERT INTO api_keys (user_id, name, key_hash, prefix, scope, created_at) VALUES (?,?,?,?,?,?)',
(session.get('user_id'), name, key_hash, prefix, scope, now_iso())
)
key_id = cur.lastrowid
log_audit(conn, 'api_key.create', target=prefix, details=f'name={name} scope={scope}')
return jsonify({
'id': key_id, 'name': name, 'prefix': prefix, 'scope': scope,
'token': token, # shown once — not retrievable later
}), 201
@app.route('/api/keys/<int:key_id>', methods=['DELETE'])
@login_required
def revoke_api_key(key_id):
with get_db() as conn:
key = conn.execute('SELECT * FROM api_keys WHERE id=?', (key_id,)).fetchone()
if not key or (key['user_id'] != session.get('user_id') and not session.get('is_admin')):
return jsonify({'error': 'Not found'}), 404
conn.execute('UPDATE api_keys SET revoked=1 WHERE id=?', (key_id,))
log_audit(conn, 'api_key.revoke', target=key['prefix'])
return jsonify({'success': True})
@app.route('/api/admin/keys', methods=['GET'])
@admin_required
def admin_list_api_keys():
with get_db() as conn:
rows = conn.execute(
'SELECT k.id, k.name, k.prefix, k.scope, k.created_at, k.last_used_at, k.revoked, '
'u.username FROM api_keys k JOIN users u ON k.user_id=u.id '
'ORDER BY k.created_at DESC'
).fetchall()
return jsonify({'keys': [dict(r) for r in rows]})
# ─────────────────────────────────────────────
# Admin — User Management
# ─────────────────────────────────────────────
@app.route('/api/admin/users', methods=['GET'])
@admin_required
def admin_list_users():
with get_db() as conn:
rows = conn.execute(
'SELECT u.id, u.username, u.is_admin, u.plan, u.created_at, u.org_id, '
'o.name as org_name, o.plan as org_plan, '
'(SELECT COUNT(*) FROM links WHERE user_id=u.id AND is_active=1) as link_count '
'FROM users u LEFT JOIN organizations o ON u.org_id=o.id '
'ORDER BY u.created_at ASC'
).fetchall()
return jsonify({'users': [dict(r) for r in rows]})
@app.route('/api/admin/users', methods=['POST'])
@admin_required
def admin_create_user():
data = request.get_json(silent=True) or {}
username = (data.get('username') or '').strip()
password = (data.get('password') or '').strip()
is_admin = bool(data.get('is_admin', False))
plan = (data.get('plan') or 'free').lower()
if plan not in VALID_PLANS or plan == 'admin':
plan = 'free'
if not username or not password:
return jsonify({'error': 'Username and password required'}), 400
if not re.match(r'^[a-zA-Z0-9_.-]{2,32}$', username):
return jsonify({'error': 'Username must be 2–32 alphanumeric/._- characters'}), 400
with get_db() as conn:
if conn.execute('SELECT 1 FROM users WHERE username=?', (username,)).fetchone():
return jsonify({'error': 'Username already taken'}), 409
conn.execute(
'INSERT INTO users (username, password_hash, is_admin, plan, created_at) VALUES (?,?,?,?,?)',
(username, generate_password_hash(password), 1 if is_admin else 0, plan,
datetime.now(timezone.utc).replace(tzinfo=None).isoformat())
)
user = conn.execute('SELECT id, username, is_admin, plan, created_at FROM users WHERE username=?',
(username,)).fetchone()
return jsonify(dict(user)), 201
@app.route('/api/admin/users/<int:user_id>', methods=['PATCH'])
@admin_required
def admin_update_user(user_id):
data = request.get_json(silent=True) or {}
updates = {}
if 'plan' in data:
plan = (data['plan'] or 'free').lower()
if plan not in VALID_PLANS or plan == 'admin':
return jsonify({'error': f'Invalid plan. Valid plans: {", ".join(p for p in VALID_PLANS if p != "admin")}'}), 400
updates['plan'] = plan
if 'is_admin' in data:
if user_id == session.get('user_id'):
return jsonify({'error': 'Cannot change your own admin status'}), 400
updates['is_admin'] = 1 if data['is_admin'] else 0
if 'org_id' in data:
org_id = data['org_id'] # None to remove from org
updates['org_id'] = org_id
if not updates:
return jsonify({'error': 'No valid fields to update'}), 400
with get_db() as conn:
if not conn.execute('SELECT 1 FROM users WHERE id=?', (user_id,)).fetchone():
return jsonify({'error': 'Not found'}), 404
if 'org_id' in updates and updates['org_id'] is not None:
if not conn.execute('SELECT 1 FROM organizations WHERE id=?', (updates['org_id'],)).fetchone():
return jsonify({'error': 'Organization not found'}), 404
set_clause = ', '.join(f'{k}=?' for k in updates)
conn.execute(f'UPDATE users SET {set_clause} WHERE id=?',
list(updates.values()) + [user_id])
user = conn.execute(
'SELECT u.id, u.username, u.is_admin, u.plan, u.created_at, u.org_id, '
'o.name as org_name FROM users u LEFT JOIN organizations o ON u.org_id=o.id '
'WHERE u.id=?', (user_id,)
).fetchone()
return jsonify(dict(user))
@app.route('/api/admin/users/<int:user_id>', methods=['DELETE'])
@admin_required
def admin_delete_user(user_id):
if user_id == session.get('user_id'):
return jsonify({'error': 'Cannot delete your own account'}), 400
with get_db() as conn:
user = conn.execute('SELECT * FROM users WHERE id=?', (user_id,)).fetchone()
if not user:
return jsonify({'error': 'Not found'}), 404
conn.execute('DELETE FROM users WHERE id=?', (user_id,))
return jsonify({'success': True})
@app.route('/api/admin/users/<int:user_id>/password', methods=['PATCH'])
@admin_required
def admin_change_password(user_id):
data = request.get_json(silent=True) or {}
password = (data.get('password') or '').strip()
if not password:
return jsonify({'error': 'Password required'}), 400
with get_db() as conn:
if not conn.execute('SELECT 1 FROM users WHERE id=?', (user_id,)).fetchone():
return jsonify({'error': 'Not found'}), 404
conn.execute('UPDATE users SET password_hash=? WHERE id=?',
(generate_password_hash(password), user_id))
return jsonify({'success': True})
# ─────────────────────────────────────────────
# Contact / Portal messages
# ─────────────────────────────────────────────
@app.route('/api/contact', methods=['POST'])
def submit_contact():
data = request.get_json(silent=True) or {}
name = (data.get('name') or '').strip()
email = (data.get('email') or '').strip()
subject = (data.get('subject') or '').strip()
body = (data.get('body') or '').strip()
if not name or not email or not subject:
return jsonify({'error': 'name, email and subject are required'}), 400
now = datetime.now(timezone.utc).replace(tzinfo=None).isoformat()
with get_db() as conn:
conn.execute(
'INSERT INTO messages (name, email, subject, body, created_at) VALUES (?,?,?,?,?)',
(name, email, subject, body, now)
)
return jsonify({'success': True})
@app.route('/api/admin/messages', methods=['GET'])
@admin_required
def admin_list_messages():
with get_db() as conn:
rows = conn.execute(
'SELECT * FROM messages ORDER BY created_at DESC'
).fetchall()
return jsonify([dict(r) for r in rows])
@app.route('/api/admin/messages/<int:msg_id>', methods=['DELETE'])
@admin_required
def admin_delete_message(msg_id):
with get_db() as conn:
if not conn.execute('SELECT 1 FROM messages WHERE id=?', (msg_id,)).fetchone():
return jsonify({'error': 'Not found'}), 404
conn.execute('DELETE FROM messages WHERE id=?', (msg_id,))
return jsonify({'success': True})
@app.route('/api/admin/messages/<int:msg_id>/read', methods=['PATCH'])
@admin_required
def admin_mark_message_read(msg_id):
with get_db() as conn:
conn.execute('UPDATE messages SET is_read=1 WHERE id=?', (msg_id,))
return jsonify({'success': True})
# ─────────────────────────────────────────────
# Admin — Organization Management
# ─────────────────────────────────────────────
@app.route('/api/admin/organizations', methods=['GET'])
@admin_required
def admin_list_organizations():
with get_db() as conn:
rows = conn.execute(
'SELECT o.id, o.name, o.plan, o.created_at, '
'COUNT(u.id) as member_count '
'FROM organizations o LEFT JOIN users u ON u.org_id=o.id '
'GROUP BY o.id ORDER BY o.name ASC'
).fetchall()
return jsonify({'organizations': [dict(r) for r in rows]})
@app.route('/api/admin/organizations', methods=['POST'])
@admin_required
def admin_create_organization():
data = request.get_json(silent=True) or {}
name = (data.get('name') or '').strip()
plan = (data.get('plan') or 'team').lower()
if not name:
return jsonify({'error': 'Organization name required'}), 400
if plan not in VALID_PLANS or plan == 'admin':
plan = 'team'
now = datetime.now(timezone.utc).replace(tzinfo=None).isoformat()
with get_db() as conn:
if conn.execute('SELECT 1 FROM organizations WHERE name=?', (name,)).fetchone():
return jsonify({'error': 'Organization name already taken'}), 409
conn.execute('INSERT INTO organizations (name, plan, created_at) VALUES (?,?,?)',
(name, plan, now))
org = conn.execute(
'SELECT id, name, plan, created_at FROM organizations WHERE name=?', (name,)
).fetchone()
return jsonify(dict(org)), 201
@app.route('/api/admin/organizations/<int:org_id>', methods=['PATCH'])
@admin_required
def admin_update_organization(org_id):
data = request.get_json(silent=True) or {}
updates = {}
if 'name' in data:
updates['name'] = (data['name'] or '').strip()
if not updates['name']:
return jsonify({'error': 'Name cannot be empty'}), 400
if 'plan' in data:
plan = (data['plan'] or 'team').lower()
if plan not in VALID_PLANS or plan == 'admin':
return jsonify({'error': 'Invalid plan'}), 400
updates['plan'] = plan
if not updates:
return jsonify({'error': 'No valid fields to update'}), 400
with get_db() as conn:
if not conn.execute('SELECT 1 FROM organizations WHERE id=?', (org_id,)).fetchone():
return jsonify({'error': 'Not found'}), 404
set_clause = ', '.join(f'{k}=?' for k in updates)
conn.execute(f'UPDATE organizations SET {set_clause} WHERE id=?',
list(updates.values()) + [org_id])
org = conn.execute(
'SELECT o.id, o.name, o.plan, o.created_at, COUNT(u.id) as member_count '
'FROM organizations o LEFT JOIN users u ON u.org_id=o.id WHERE o.id=?',
(org_id,)
).fetchone()
return jsonify(dict(org))
@app.route('/api/admin/organizations/<int:org_id>', methods=['DELETE'])
@admin_required
def admin_delete_organization(org_id):
with get_db() as conn:
if not conn.execute('SELECT 1 FROM organizations WHERE id=?', (org_id,)).fetchone():
return jsonify({'error': 'Not found'}), 404
# Remove org_id from all members before deleting
conn.execute('UPDATE users SET org_id=NULL WHERE org_id=?', (org_id,))
conn.execute('DELETE FROM organizations WHERE id=?', (org_id,))
return jsonify({'success': True})
# ─────────────────────────────────────────────
# SEO — robots.txt & sitemap.xml
# ─────────────────────────────────────────────
@app.route('/robots.txt')
def robots_txt():
body = f"""User-agent: *
Allow: /
Allow: /landing-os
Disallow: /app
Disallow: /api/
Disallow: /static/
Sitemap: {BASE_URL}/sitemap.xml
"""
return Response(body, mimetype='text/plain')
@app.route('/sitemap.xml')
def sitemap_xml():
from datetime import date
today = date.today().isoformat()
body = f"""<?xml version="1.0" encoding="UTF-8"?>
<urlset xmlns="http://www.sitemaps.org/schemas/sitemap/0.9"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.sitemaps.org/schemas/sitemap/0.9
http://www.sitemaps.org/schemas/sitemap/0.9/sitemap.xsd">
<url>
<loc>{BASE_URL}/</loc>
<lastmod>{today}</lastmod>
<changefreq>weekly</changefreq>
<priority>1.0</priority>
</url>
<url>
<loc>{BASE_URL}/landing-os</loc>
<lastmod>{today}</lastmod>
<changefreq>monthly</changefreq>
<priority>0.7</priority>
</url>
<url>
<loc>{BASE_URL}/api-docs</loc>
<lastmod>{today}</lastmod>
<changefreq>monthly</changefreq>
<priority>0.6</priority>
</url>
</urlset>"""
return Response(body, mimetype='application/xml')
# ─────────────────────────────────────────────
# Redirect — branded visitor pages, password gate, async click logging
# ─────────────────────────────────────────────
def _visitor_page(heading, message, body_html='', status=200):
"""Branded standalone page shown to link visitors (expired, missing,
password-protected, …) — they never see the app or marketing site."""
html = f"""<!DOCTYPE html>
<html lang="en"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1">
<meta name="robots" content="noindex"><title>{heading} — {APP_NAME}</title>
<style>
:root {{ --bg:#0a0a12; --card:#12121e; --border:#23233a; --text:#e8e8f0; --muted:#6b6b80; --accent:#00b8d4; }}
* {{ box-sizing:border-box; margin:0; padding:0; }}
body {{ background:var(--bg); color:var(--text); font-family:'Segoe UI',system-ui,sans-serif;
min-height:100vh; display:flex; align-items:center; justify-content:center; padding:20px; }}
.card {{ background:var(--card); border:1px solid var(--border); border-radius:12px;
padding:40px 36px; max-width:420px; width:100%; text-align:center; }}
.brand {{ font-family:monospace; font-size:13px; letter-spacing:.14em; color:var(--muted);
text-transform:uppercase; margin-bottom:22px; }}
.brand b {{ color:var(--accent); }}
h1 {{ font-size:21px; margin-bottom:10px; }}
p {{ color:var(--muted); font-size:14px; line-height:1.6; }}
form {{ margin-top:22px; display:flex; flex-direction:column; gap:10px; }}
input[type=password] {{ background:var(--bg); border:1px solid var(--border); border-radius:8px;
color:var(--text); padding:11px 14px; font-size:14px; outline:none; }}
input[type=password]:focus {{ border-color:var(--accent); }}
button {{ background:var(--accent); border:none; border-radius:8px; color:#001318;
font-weight:600; padding:11px; font-size:14px; cursor:pointer; }}
.err {{ color:#ff5470; font-size:13px; margin-top:10px; }}
</style></head><body>
<div class="card">
<div class="brand"><b>▮</b> {APP_NAME}</div>
<h1>{heading}</h1>
<p>{message}</p>
{body_html}
</div>
</body></html>"""
return Response(html, status=status, mimetype='text/html',
headers={'Cache-Control': 'no-store'})
def _password_form(code, error=''):
err_html = f'<div class="err">{error}</div>' if error else ''
return _visitor_page(
'This link is password protected',
'Enter the password to continue to the destination.',
f'''<form method="POST" action="/{code}/unlock">
<input type="password" name="password" placeholder="Password" autofocus autocomplete="off">
<button type="submit">Unlock →</button>
</form>{err_html}''',
status=200
)
def _record_click(link_row):
"""Capture request details and enqueue the click for the background writer."""
ua = request.headers.get('User-Agent', '')[:500]
cf = request.headers.get('CF-IPCountry', '').strip().upper()
item = {
'link_id': link_row['id'],
'ts': now_iso(),
'referrer': request.referrer,
'ua': ua,
'ip': get_client_ip()[:45],
'cf_country': cf if (len(cf) == 2 and cf.isalpha() and cf != 'XX') else '',
'is_bot': 1 if is_bot_ua(ua) else 0,
}
try:
_click_queue.put_nowait(item)
except queue.Full:
pass
def _check_link_gates(conn, link):
"""Expiry and quota checks shared by the redirect and unlock routes.
Returns an error Response, or None if the link may be followed."""
if link['expires_at'] and link['expires_at'] < now_iso():
return _visitor_page('Link expired',
'This short link has passed its expiry date and is no longer available.',
status=410)
if link['user_id']:
plan = get_user_plan(conn, link['user_id'])
click_limit = PLAN_LIMITS[plan]['monthly_clicks']
if click_limit is not None and get_user_monthly_clicks(conn, link['user_id']) >= click_limit:
return _visitor_page('Link temporarily unavailable',
'This link has reached its monthly click limit. Please try again later.',
status=429)
return None
@app.route('/<code>')
def redirect_link(code):
if code in ('static', 'api', 'favicon.ico', 'robots.txt', 'sitemap.xml'):
return 'Not found', 404
with get_db() as conn:
link = conn.execute('SELECT * FROM links WHERE code=? AND is_active=1', (code,)).fetchone()
if not link:
return _visitor_page('Link not found',
"This short link doesn't exist or has been removed.",
status=404)
gate = _check_link_gates(conn, link)
if gate:
return gate
if link['password_hash']:
return _password_form(code)
_record_click(link)
# 302 by default so browsers don't permanently cache the hop —
# keeps analytics accurate and destination edits effective. 301 is per-link opt-in.
return redirect(link['long_url'], code=link['redirect_type'] or 302)
@app.route('/<code>/unlock', methods=['POST'])
def unlock_link(code):
with get_db() as conn:
link = conn.execute('SELECT * FROM links WHERE code=? AND is_active=1', (code,)).fetchone()
if not link or not link['password_hash']:
return _visitor_page('Link not found',
"This short link doesn't exist or has been removed.",
status=404)
gate = _check_link_gates(conn, link)
if gate:
return gate
password = request.form.get('password', '')
if not check_password_hash(link['password_hash'], password):
return _password_form(code, error='Incorrect password — try again.')
_record_click(link)
# Always 302 for password links — the unlock must never be cached away
return redirect(link['long_url'], code=302)
# ─────────────────────────────────────────────
# Frontend routes
# ─────────────────────────────────────────────
@app.route('/')
def landing():
with open(os.path.join(os.path.dirname(__file__), 'landing.html')) as f:
return f.read()
@app.route('/landing-os')
def landing_os():
with open(os.path.join(os.path.dirname(__file__), 'landing-os.html')) as f:
return f.read()
@app.route('/api-docs')
def api_docs():
with open(os.path.join(os.path.dirname(__file__), 'api-docs.html')) as f:
return f.read()
@app.route('/app', defaults={'subpath': ''})
@app.route('/app/<path:subpath>')
def app_frontend(subpath):
with open(os.path.join(os.path.dirname(__file__), 'index.html')) as f:
return f.read()
if __name__ == '__main__':
port = int(os.environ.get('PORT', 5000))
app.run(host='0.0.0.0', port=port,
debug=os.environ.get('DEBUG', 'false').lower() == 'true')
Reference in New Issue View Git Blame Copy Permalink