1667 lines
68 KiB
Python
1667 lines
68 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 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/<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 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 WHERE code=?', (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 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/<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
|
||
# ─────────────────────────────────────────────
|
||
|
||
@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 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/<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')
|