Jason Stedwell
14d6e8c50e
- Redirects default to 302 so browsers stop permanently caching the hop — click analytics stay accurate and destination edits take effect for returning visitors. 301 is a per-link opt-in (create/edit/API). - Links can carry a password: visitors get a branded unlock page at /:code and POST to /:code/unlock. Unlocks always redirect with 302. - Missing, expired, and quota-limited links now show branded standalone visitor pages (404/410/429) instead of bouncing to the marketing site with ?error= query params. - Clicks are recorded through the async pipeline with bot/crawler tagging (UA heuristics), GeoLite2 country lookup (ip-api fallback gated behind GEO_API_FALLBACK), and the IP_ANONYMIZE policy applied at write time. - Deletes stamp deleted_at for later hard-purge. - Dashboard: redirect-type selector and password set/replace/remove in the edit form, lock badge on protected links. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2036 lines
85 KiB
Python
2036 lines
85 KiB
Python
# 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 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'
|
||
|
||
_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)
|
||
|
||
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/<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
|
||
# ─────────────────────────────────────────────
|
||
|
||
@app.route('/api/qr/<code>')
|
||
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, 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})
|
||
|
||
|
||
# ─────────────────────────────────────────────
|
||
# 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')
|