diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..2d84af7 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,18 @@ +# QRknit .dockerignore +__pycache__/ +*.pyc +*.pyo +*.pyd +*.db +*.sqlite +.env +.env.* +*.log +.git/ +.gitignore +*.md +setup.sh +docker-compose.yml +.DS_Store +venv/ +.venv/ diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..01f42da --- /dev/null +++ b/.gitignore @@ -0,0 +1,5 @@ +__pycache__/ +*.pyc +*.pyo +*.db +.env diff --git a/api-docs.html b/api-docs.html new file mode 100644 index 0000000..908851f --- /dev/null +++ b/api-docs.html @@ -0,0 +1,723 @@ + + + + + + +API Reference — QRknit + + + + + + + + + + + + + + + + + + + + + + + + +
+ +
+
+ + + + +
+ Home + Open App → +
+
+ + + +
+
REST API · Session Auth · JSON
+

API Reference

+

+ Complete reference for the QRknit REST API. Manage links, generate QR codes, pull analytics, and administer users — all from your own code. +

+
Base URL: https://your-instance.com  ·  Format: application/json
+
+ + + +
+ + +
+

Authentication

+

+ All write endpoints require an active session. Log in via the web UI or call POST /api/auth/login with + {"username": "…", "password": "…"} — the server sets an HttpOnly session cookie valid for 30 days. + Include that cookie in all subsequent requests. Endpoints marked Admin additionally require + the authenticated user to have admin privileges. +

+
+ + + + + + +
+
+
+ +
+
Auth
+
+

Session-based login. A successful login sets a 30-day HttpOnly cookie used for all authenticated requests.

+ + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
MethodEndpointAuthDescription
POST/api/auth/login—Login — body: {"username": "…", "password": "…"}. Sets session cookie and returns plan, plan_limits, and org_id.
POST/api/auth/logoutSessionClear the current session and invalidate the cookie.
GET/api/auth/meSessionReturns {"authenticated": true, "username": "…", "is_admin": bool, "org_id": int|null, "plan": "…", "plan_limits": {…}}.
+
+ + + + + + + +
+
+
+ +
+
Utilities
+
+

Stats, plan usage, tags, QR code generation, bulk operations, and CSV import/export.

+ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
MethodEndpointAuthDescription
GET/api/statsSessionDashboard stats: total links, total clicks, clicks in last 7 days, top 5 links, and a 30-day daily click array.
GET/api/planSessionCurrent user's effective plan, limits, and live usage — {plan, limits: {max_links, monthly_clicks, analytics_days}, usage: {active_links, monthly_clicks}}. Org members see pooled usage across all org members.
GET/api/tagsSessionTags scoped to the requesting user — own tags for solo users and admins; all org member tags for org members.
GET/api/fetch-titleSessionServer-side page title fetch. Query: ?url=<url>. Returns {"title": "…"}.
GET/api/qr/:code—QR PNG for a short link. Query params: ?fg=, ?bg=, ?size=, ?style=.
GET/api/qr/custom—QR PNG for any URL. Query: ?url=, ?fg=, ?bg=, ?size=, ?style=. Styles: square, rounded, dots, vertical, horizontal.
POST/api/qr/custom—QR PNG with logo overlay. Body: {url, fg, bg, size, style, logo} where logo is a base64-encoded image string.
POST/api/links/bulkSessionBulk operations. Body: {"action": "delete"|"tag"|"expire", "codes": […]}.
GET/api/links/exportSessionDownload all links as CSV.
POST/api/links/importSessionImport links from CSV. Body: {"csv": "…"}.
GET/api/health—Health check — returns {"status": "ok"}. Used by Docker and uptime monitors.
GET/api/config—Public deployment config — returns non-sensitive values like app_name and base_url.
+
+ + + +
+
+
+ +
+
Admin
+
+

All Admin endpoints require an authenticated admin session. These manage users, organizations, and the message inbox.

+ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
MethodEndpointAuthDescription
GET/api/admin/usersAdminList all users with link counts, plan, org name, and org plan.
POST/api/admin/usersAdminCreate user. Body: {"username": "…", "password": "…", "is_admin": bool, "plan": "…"}.
PATCH/api/admin/users/:idAdminUpdate plan, admin status, and/or org. Body: {plan?, is_admin?, org_id?}. Set org_id: null to remove from org.
DELETE/api/admin/users/:idAdminDelete user. Cannot delete yourself.
PATCH/api/admin/users/:id/passwordAdminChange user password. Body: {"password": "…"}.
GET/api/admin/organizationsAdminList all organizations with member counts and plan.
POST/api/admin/organizationsAdminCreate organization. Body: {"name": "…", "plan": "…"}.
PATCH/api/admin/organizations/:idAdminUpdate organization name and/or plan. Body: {name?, plan?}.
DELETE/api/admin/organizations/:idAdminDelete organization. Members are unassigned but not deleted.
GET/api/admin/messagesAdminList all contact/portal messages, newest first.
DELETE/api/admin/messages/:idAdminDelete a message.
PATCH/api/admin/messages/:id/readAdminMark a message as read.
+
+ + + +
+
+
+ +
+
Contact
+
+

Public endpoint for submitting contact and support messages. No authentication required.

+ + + + + + + + + + + + + + + + + +
MethodEndpointAuthDescription
POST/api/contact—Submit a contact message. Body: {"name": "…", "email": "…", "subject": "…", "body": "…"}.
+
+ + + +
+
+
+ +
+
Public Redirects
+
+

Short link redirection and SEO endpoints. No authentication required.

+ + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
MethodEndpointAuthDescription
GET/:code—Redirect to the destination URL for a short code. Records click analytics (referrer, device, browser, country).
GET/robots.txt—Standard robots.txt for crawler directives.
GET/sitemap.xml—Auto-generated XML sitemap.
+
+ +
+ + + + + + + + + diff --git a/app.py b/app.py new file mode 100644 index 0000000..8adf7c0 --- /dev/null +++ b/app.py @@ -0,0 +1,1666 @@ +# 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 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') + +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 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)", + ]: + 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 _uptime_heartbeat(): + """Daemon thread: records a 'checks_up' ping every 60 s to track daily availability.""" + while True: + try: + today = datetime.now(timezone.utc).date().isoformat() + with get_db() as conn: + conn.execute(""" + INSERT INTO uptime_daily (date, checks_total, checks_up) + VALUES (?, 1, 1) + ON CONFLICT(date) DO UPDATE SET + checks_total = checks_total + 1, + checks_up = checks_up + 1 + """, (today,)) + except Exception: + pass + time.sleep(60) + +threading.Thread(target=_uptime_heartbeat, daemon=True).start() + + +# ───────────────────────────────────────────── +# Auth decorators +# ───────────────────────────────────────────── + +def login_required(f): + @wraps(f) + def decorated(*args, **kwargs): + 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): + 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 + + +# ───────────────────────────────────────────── +# 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 generate_qr_png(data: str, size: int = 300, fg=(0,0,0), bg=(255,255,255), + style: str = 'square', logo_bytes: bytes = None) -> bytes: + """Generate QR PNG. Supports dot styles and logo overlay when qrcode[pil] is installed.""" + try: + import qrcode as qrc + from PIL import Image + + ec = qrc.constants.ERROR_CORRECT_H if logo_bytes else qrc.constants.ERROR_CORRECT_M + qr = qrc.QRCode(error_correction=ec, 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'')) + + +# ───────────────────────────────────────────── +# 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' + +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 get_country_for_request(): + """Return a 2-letter ISO country code for the current request. + + Priority: + 1. CF-IPCountry header (Cloudflare — zero-latency, most reliable) + 2. ip-api.com free JSON API (1-second timeout, fails gracefully) + Returns 'XX' if the IP is private/loopback, 'Unknown' on any failure. + """ + import urllib.request as _ureq, json as _json + + cf = request.headers.get('CF-IPCountry', '').strip().upper() + if cf and len(cf) == 2 and cf.isalpha() and cf != 'XX': + return cf + + ip = get_client_ip() + if not ip or ip in ('127.0.0.1', '::1'): + return 'XX' + # Skip RFC-1918 / loopback ranges — no point querying for private IPs + 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: + pass + try: + req = _ureq.Request( + f'http://ip-api.com/json/{ip}?fields=countryCode', + headers={'User-Agent': 'QRknit/1.0'} + ) + with _ureq.urlopen(req, timeout=1) as resp: + data = _json.loads(resp.read()) + code = data.get('countryCode', '') + return code if code else 'Unknown' + except Exception: + 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'], + '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.""" + today = datetime.now(timezone.utc).date() + start = today - timedelta(days=29) + with get_db() as conn: + rows = conn.execute( + "SELECT date, checks_total, checks_up FROM uptime_daily WHERE date >= ? ORDER BY date", + (start.isoformat(),) + ).fetchall() + row_map = {r['date']: r for r in rows} + days = [] + for i in range(30): + d = (start + timedelta(days=i)).isoformat() + row = row_map.get(d) + if row and row['checks_total'] > 0: + pct = round(row['checks_up'] / row['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', []) + + 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 + + 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) VALUES (?,?,?,?,?,?)', + (code, long_url, title or None, + datetime.now(timezone.utc).replace(tzinfo=None).isoformat(), + expires_at or None, session.get('user_id')) + ) + 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, + '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/', 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/', 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 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/', 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 WHERE code=?', (code,)) + return jsonify({'success': True}) + + +# ───────────────────────────────────────────── +# Analytics +# ───────────────────────────────────────────── + +@app.route('/api/links//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) + + link_id = link['id'] + since = (datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(days=days)).isoformat() + + daily_rows = conn.execute(""" + SELECT substr(clicked_at,1,10) as day, COUNT(*) as count + FROM clicks WHERE link_id=? AND clicked_at>=? + GROUP BY day ORDER BY day + """, (link_id, since)).fetchall() + daily_map = {r['day']: r['count'] 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} + for i in range(days) + ] + for d in daily: d['clicks'] = daily_map.get(d['date'], 0) + + ref_rows = conn.execute( + 'SELECT referrer, COUNT(*) as count FROM clicks WHERE link_id=? AND clicked_at>=? 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( + 'SELECT user_agent, COUNT(*) as count FROM clicks WHERE link_id=? AND clicked_at>=? 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(""" + 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>=? + 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(""" + SELECT COALESCE(NULLIF(country,''),'Unknown') as country, COUNT(*) as count + FROM clicks WHERE link_id=? AND clicked_at>=? + 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, + 'total_clicks': link['clicks'], + 'period_clicks': sum(d['clicks'] for d in daily), + 'daily': daily, 'referrers': referrers, 'devices': devices, 'browsers': browsers, + 'heatmap': heatmap, 'countries': countries, + }) + + +# ───────────────────────────────────────────── +# Click-event CSV export +# ───────────────────────────────────────────── + +@app.route('/api/links//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']+property=["\']og:title["\'][^>]+content=["\']([^"\']*)["\']', html, re.I) + if not m: + m = re.search(r']+content=["\']([^"\']*)["\'][^>]+property=["\']og:title["\']', html, re.I) + if m: + return jsonify({'title': m.group(1).strip()[:200]}) + # Fall back to + m = re.search(r'<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 +# ───────────────────────────────────────────── + +@app.route('/api/qr/') +def qr_code(code): + fg_hex = request.args.get('fg', '000000') + bg_hex = request.args.get('bg', 'ffffff') + size = min(int(request.args.get('size', 300)), 1000) + style = request.args.get('style', 'square') + 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 + png = generate_qr_png(f"{BASE_URL}/{code}", size=size, + fg=hex_to_rgb(fg_hex), bg=hex_to_rgb(bg_hex), style=style) + return Response(png, mimetype='image/png', headers={'Cache-Control': 'public, max-age=3600'}) + + +@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 + fg_hex = request.args.get('fg', '000000') + bg_hex = request.args.get('bg', 'ffffff') + size = min(int(request.args.get('size', 300)), 1000) + style = request.args.get('style', 'square') + png = generate_qr_png(url, size=size, fg=hex_to_rgb(fg_hex), bg=hex_to_rgb(bg_hex), style=style) + return Response(png, mimetype='image/png') + + +@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 + fg_hex = (data.get('fg') or '000000').lstrip('#') + bg_hex = (data.get('bg') or 'ffffff').lstrip('#') + size = min(int(data.get('size', 300)), 1000) + style = data.get('style', 'square') + 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 + png = generate_qr_png(url, size=size, fg=hex_to_rgb(fg_hex), bg=hex_to_rgb(bg_hex), + style=style, logo_bytes=logo_bytes) + return Response(png, mimetype='image/png') + + +# ───────────────────────────────────────────── +# 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 WHERE code IN ({placeholders})', codes) + else: + conn.execute( + f'UPDATE links SET is_active=0 WHERE code IN ({placeholders}) AND user_id=?', + 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}) + + +# ───────────────────────────────────────────── +# 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/', 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/', 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//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/', 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//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/', 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/', 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""" + + + {BASE_URL}/ + {today} + weekly + 1.0 + + + {BASE_URL}/landing-os + {today} + monthly + 0.7 + + + {BASE_URL}/api-docs + {today} + monthly + 0.6 + +""" + return Response(body, mimetype='application/xml') + + +# ───────────────────────────────────────────── +# Redirect +# ───────────────────────────────────────────── + +@app.route('/') +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 redirect('/?error=not_found') + if link['expires_at'] and link['expires_at'] < datetime.now(timezone.utc).replace(tzinfo=None).isoformat(): + return redirect('/?error=expired') + + # Monthly click quota check (skip for links with no owner) + 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 redirect('/?error=quota_exceeded') + + country = get_country_for_request() + client_ip = get_client_ip() + conn.execute( + 'INSERT INTO clicks (link_id,clicked_at,referrer,user_agent,ip_address,country) VALUES (?,?,?,?,?,?)', + (link['id'], datetime.now(timezone.utc).replace(tzinfo=None).isoformat(), + request.referrer, request.headers.get('User-Agent','')[:500], + client_ip[:45], country) + ) + conn.execute('UPDATE links SET clicks=clicks+1 WHERE id=?', (link['id'],)) + return redirect(link['long_url'], code=301) + + +# ───────────────────────────────────────────── +# 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/') +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') diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..3056b1f --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,48 @@ +# ───────────────────────────────────────────── +# QRknit — docker-compose.yml +# For local dev or non-Unraid deployments +# On Unraid: use the Docker tab UI instead +# ───────────────────────────────────────────── + +services: + qrknit: + build: . + # Or use a pre-built image from Docker Hub: + # image: yourdockerhubuser/qrknit:latest + container_name: qrknit + restart: unless-stopped + + ports: + - "5000:5000" # Change left side to use a different host port + # e.g. "80:5000" to serve on port 80 + + volumes: + - qrknit-data:/app/data # Persistent SQLite database + + environment: + - BASE_URL=https://yourdomain.com # ← Your public URL (domain or subdomain) + - APP_NAME=My.Links # ← Displayed in the UI header/logo/title + - ADMIN_USERNAME=admin # ← Admin account username (default: admin) + - PORT=5000 + - DEBUG=false + - SECRET_KEY=replace-with-a-long-random-string # ← Change this! + - ADMIN_PASSWORD=replace-with-a-strong-password # ← Change this! + - COOKIE_SECURE=false # Set to true only if Flask receives HTTPS directly (not behind a proxy) + + # Optional: resource limits + deploy: + resources: + limits: + cpus: '0.5' + memory: 256M + + healthcheck: + test: ["CMD", "python3", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:5000/api/health')"] + interval: 30s + timeout: 10s + retries: 3 + start_period: 15s + +volumes: + qrknit-data: + driver: local