app.py

import base64
import csv
import io
import os
import secrets
import sqlite3
from urllib.parse import parse_qsl, urlencode, urlparse, urlunparse

import qrcode
from flask import (
    Flask,
    Response,
    abort,
    g,
    redirect,
    render_template,
    request,
    session,
    url_for,
)

try:
    import psycopg
    from psycopg.rows import dict_row
except ImportError:  # Local SQLite development does not need psycopg installed.
    psycopg = None
    dict_row = None


app = Flask(__name__)
app.config["DATABASE"] = os.environ.get("DATABASE_PATH", "link_tracker.db")
app.config["DATABASE_URL"] = os.environ.get("DATABASE_URL", "")
app.config["DB_BACKEND"] = os.environ.get(
    "DB_BACKEND", "postgres" if os.environ.get("DATABASE_URL") else "sqlite"
)
app.config["SECRET_KEY"] = os.environ.get("SECRET_KEY", "local-dev-secret-key")
app.config["PUBLIC_BASE_URL"] = os.environ.get("PUBLIC_BASE_URL", "").rstrip("/")

if os.environ.get("VERCEL") and not app.config["DATABASE_URL"]:
    app.config["DB_BACKEND"] = "postgres"

DEFAULT_UTM_VARIANTS = "\n".join(
    [
        "qr,offline",
        "poster,offline",
        "flyer,print",
    ]
)
UTM_FIELDS = ("utm_source", "utm_medium", "utm_campaign", "utm_term", "utm_content")


SQLITE_SCHEMA = """
CREATE TABLE IF NOT EXISTS short_links (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    code TEXT NOT NULL UNIQUE,
    destination_url TEXT NOT NULL,
    title TEXT,
    is_active INTEGER NOT NULL DEFAULT 1,
    created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
);

CREATE TABLE IF NOT EXISTS visits (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    short_link_id INTEGER NOT NULL,
    visited_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
    utm_source TEXT,
    utm_medium TEXT,
    utm_campaign TEXT,
    utm_term TEXT,
    utm_content TEXT,
    referrer TEXT,
    user_agent TEXT,
    ip_address TEXT,
    browser TEXT,
    device_type TEXT,
    FOREIGN KEY (short_link_id) REFERENCES short_links (id)
);

CREATE TABLE IF NOT EXISTS generated_links (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    short_link_id INTEGER NOT NULL,
    utm_source TEXT NOT NULL,
    utm_medium TEXT NOT NULL,
    utm_campaign TEXT NOT NULL,
    created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
    FOREIGN KEY (short_link_id) REFERENCES short_links (id)
);

CREATE INDEX IF NOT EXISTS idx_short_links_code ON short_links (code);
CREATE INDEX IF NOT EXISTS idx_generated_links_campaign ON generated_links (utm_campaign);
CREATE INDEX IF NOT EXISTS idx_generated_links_short_link_id ON generated_links (short_link_id);
CREATE INDEX IF NOT EXISTS idx_visits_short_link_id ON visits (short_link_id);
CREATE INDEX IF NOT EXISTS idx_visits_visited_at ON visits (visited_at);
CREATE INDEX IF NOT EXISTS idx_visits_utm_campaign ON visits (utm_campaign);
"""

POSTGRES_SCHEMA = """
CREATE TABLE IF NOT EXISTS short_links (
    id INTEGER GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    code TEXT NOT NULL UNIQUE,
    destination_url TEXT NOT NULL,
    title TEXT,
    is_active INTEGER NOT NULL DEFAULT 1,
    created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
);

CREATE TABLE IF NOT EXISTS visits (
    id INTEGER GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    short_link_id INTEGER NOT NULL REFERENCES short_links (id),
    visited_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
    utm_source TEXT,
    utm_medium TEXT,
    utm_campaign TEXT,
    utm_term TEXT,
    utm_content TEXT,
    referrer TEXT,
    user_agent TEXT,
    ip_address TEXT,
    browser TEXT,
    device_type TEXT
);

CREATE TABLE IF NOT EXISTS generated_links (
    id INTEGER GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    short_link_id INTEGER NOT NULL REFERENCES short_links (id),
    utm_source TEXT NOT NULL,
    utm_medium TEXT NOT NULL,
    utm_campaign TEXT NOT NULL,
    created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
);

CREATE INDEX IF NOT EXISTS idx_short_links_code ON short_links (code);
CREATE INDEX IF NOT EXISTS idx_generated_links_campaign ON generated_links (utm_campaign);
CREATE INDEX IF NOT EXISTS idx_generated_links_short_link_id ON generated_links (short_link_id);
CREATE INDEX IF NOT EXISTS idx_visits_short_link_id ON visits (short_link_id);
CREATE INDEX IF NOT EXISTS idx_visits_visited_at ON visits (visited_at);
CREATE INDEX IF NOT EXISTS idx_visits_utm_campaign ON visits (utm_campaign);
"""


def db_backend() -> str:
    return app.config["DB_BACKEND"]


def get_db():
    if "db" not in g:
        if db_backend() == "postgres":
            if psycopg is None:
                raise RuntimeError("Install psycopg to use the Postgres backend.")
            if not app.config["DATABASE_URL"]:
                raise RuntimeError(
                    "DATABASE_URL is required for Vercel production. "
                    "Set it to the Neon pooled connection string."
                )
            g.db = psycopg.connect(app.config["DATABASE_URL"], row_factory=dict_row)
        else:
            g.db = sqlite3.connect(app.config["DATABASE"])
            g.db.row_factory = sqlite3.Row
    return g.db


@app.teardown_appcontext
def close_db(_error: Exception | None = None) -> None:
    db = g.pop("db", None)
    if db is not None:
        db.close()


def placeholders(query: str) -> str:
    if db_backend() == "postgres":
        return query.replace("?", "%s")
    return query


def execute(query: str, params: tuple = ()):
    return get_db().execute(placeholders(query), params)


def executescript(script: str) -> None:
    db = get_db()
    if db_backend() == "postgres":
        for statement in script.split(";"):
            if statement.strip():
                db.execute(statement)
    else:
        db.executescript(script)


def integrity_error() -> type[Exception]:
    if db_backend() == "postgres" and psycopg is not None:
        return psycopg.IntegrityError
    return sqlite3.IntegrityError


def init_db() -> None:
    executescript(POSTGRES_SCHEMA if db_backend() == "postgres" else SQLITE_SCHEMA)
    get_db().commit()


@app.before_request
def ensure_database() -> None:
    if not app.config.get("DB_INITIALIZED"):
        init_db()
        app.config["DB_INITIALIZED"] = True


@app.before_request
def require_site_login():
    open_endpoints = {"site_login", "site_logout", "redirect_short_link", "static"}
    if request.endpoint in open_endpoints:
        return None

    if not analytics_password():
        return render_template("login.html", password_missing=True), 503

    if not session.get("site_authenticated"):
        return redirect(url_for("site_login", next=request.full_path))

    return None


def normalize_url(raw_url: str) -> str:
    value = raw_url.strip()
    if not value:
        return ""

    parsed = urlparse(value)
    if not parsed.scheme:
        value = f"https://{value}"

    return value


def is_valid_url(value: str) -> bool:
    parsed = urlparse(value)
    return parsed.scheme in {"http", "https"} and bool(parsed.netloc)


def add_query_params(url: str, params: dict[str, str]) -> str:
    parsed = urlparse(url)
    query_params = dict(parse_qsl(parsed.query, keep_blank_values=True))
    query_params.update(
        {key: value.strip() for key, value in params.items() if value.strip()}
    )
    return urlunparse(parsed._replace(query=urlencode(query_params)))


def parse_utm_variants(raw_variants: str) -> list[dict[str, str]]:
    variants: list[dict[str, str]] = []

    for line in raw_variants.splitlines():
        stripped = line.strip()
        if not stripped:
            continue

        parts = [part.strip() for part in stripped.split(",")]
        if len(parts) != 2 or not all(parts):
            raise ValueError(
                "Each UTM variant must be on its own line as source,medium."
            )

        source, medium = parts
        variants.append(
            {
                "source": source,
                "medium": medium,
                "label": f"{source} / {medium}",
            }
        )

    if not variants:
        raise ValueError("Add at least one UTM variant to generate.")

    return variants


def generate_qr_base64(url: str) -> str:
    qr = qrcode.QRCode(box_size=10, border=4)
    qr.add_data(url)
    qr.make(fit=True)

    image = qr.make_image(fill_color="black", back_color="white")
    buffer = io.BytesIO()
    image.save(buffer, format="PNG")

    return base64.b64encode(buffer.getvalue()).decode("ascii")


def unique_code(prefix: str | None = None) -> str:
    if prefix:
        candidate = prefix
        suffix = 1
        while execute(
            "SELECT 1 FROM short_links WHERE code = ?", (candidate,)
        ).fetchone():
            suffix += 1
            candidate = f"{prefix}-{suffix}"
        return candidate

    while True:
        candidate = secrets.token_urlsafe(5).replace("-", "").replace("_", "")[:7]
        if not execute(
            "SELECT 1 FROM short_links WHERE code = ?", (candidate,)
        ).fetchone():
            return candidate


def create_short_link(destination_url: str, title: str = ""):
    for _attempt in range(5):
        code = unique_code()
        try:
            execute(
                """
                INSERT INTO short_links (code, destination_url, title)
                VALUES (?, ?, ?)
                """,
                (code, destination_url, title.strip() or None),
            )
        except integrity_error():
            get_db().rollback()
            continue

        get_db().commit()
        return execute("SELECT * FROM short_links WHERE code = ?", (code,)).fetchone()

    raise RuntimeError("Could not create a unique short code.")


def save_generated_link(
    short_link_id: int, utm_source: str, utm_medium: str, utm_campaign: str
) -> None:
    execute(
        """
        INSERT INTO generated_links (
            short_link_id, utm_source, utm_medium, utm_campaign
        )
        VALUES (?, ?, ?, ?)
        """,
        (short_link_id, utm_source, utm_medium, utm_campaign),
    )


def short_url_for(code: str) -> str:
    path = url_for("redirect_short_link", code=code)
    if app.config["PUBLIC_BASE_URL"]:
        return f"{app.config['PUBLIC_BASE_URL']}{path}"
    return url_for("redirect_short_link", code=code, _external=True)


def classify_browser(user_agent: str) -> str:
    ua = user_agent.lower()
    if "edg/" in ua:
        return "Edge"
    if "chrome/" in ua and "chromium" not in ua:
        return "Chrome"
    if "firefox/" in ua:
        return "Firefox"
    if "safari/" in ua and "chrome/" not in ua:
        return "Safari"
    if user_agent:
        return "Other"
    return "Unknown"


def classify_device(user_agent: str) -> str:
    ua = user_agent.lower()
    if "mobile" in ua or "iphone" in ua or "android" in ua:
        return "Mobile"
    if "ipad" in ua or "tablet" in ua:
        return "Tablet"
    if user_agent:
        return "Desktop"
    return "Unknown"


def client_ip() -> str:
    forwarded = request.headers.get("X-Forwarded-For", "")
    if forwarded:
        return forwarded.split(",", 1)[0].strip()
    return request.remote_addr or ""


def record_visit(link) -> None:
    user_agent = request.headers.get("User-Agent", "")
    values = {field: request.args.get(field, "").strip() or None for field in UTM_FIELDS}
    execute(
        """
        INSERT INTO visits (
            short_link_id, utm_source, utm_medium, utm_campaign,
            utm_term, utm_content, referrer, user_agent, ip_address,
            browser, device_type
        )
        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        """,
        (
            link["id"],
            values["utm_source"],
            values["utm_medium"],
            values["utm_campaign"],
            values["utm_term"],
            values["utm_content"],
            request.referrer,
            user_agent,
            client_ip(),
            classify_browser(user_agent),
            classify_device(user_agent),
        ),
    )
    get_db().commit()


def analytics_password() -> str:
    return os.environ.get("ANALYTICS_PASSWORD", "")


def first_value(row) -> int:
    if row is None:
        return 0
    if isinstance(row, dict):
        return next(iter(row.values()))
    return row[0]


def row_dict(row) -> dict:
    return dict(row)


def scalar(query: str, params: tuple = ()) -> int:
    row = execute(query, params).fetchone()
    value = first_value(row)
    return value if value is not None else 0


def analytics_metric() -> str:
    return request.args.get("metric", "visits") if request.args.get("metric") in {"visits", "uniques"} else "visits"


def unique_visitor_expr(table_prefix: str = "visits") -> str:
    return (
        f"CAST(DATE({table_prefix}.visited_at) AS TEXT) || '|' || "
        f"COALESCE({table_prefix}.ip_address, '') || '|' || "
        f"COALESCE({table_prefix}.user_agent, '')"
    )


def count_expr(metric: str, table_prefix: str = "visits") -> str:
    if metric == "uniques":
        return f"COUNT(DISTINCT {unique_visitor_expr(table_prefix)})"
    return f"COUNT({table_prefix}.id)"


def metric_label(metric: str) -> str:
    return "Unique Visitors" if metric == "uniques" else "Visits"


@app.route("/", methods=["GET", "POST"])
def index():
    url = ""
    title = ""
    error = None
    result = None
    mode = "short_link"
    enable_utm_set = False
    generate_qrs = False
    utm_campaign = ""
    utm_variants = DEFAULT_UTM_VARIANTS

    if request.method == "POST":
        url = normalize_url(request.form.get("url", ""))
        title = request.form.get("title", "").strip()
        mode = request.form.get("mode", "short_link")
        if mode not in {"short_link", "qr_only"}:
            mode = "short_link"
        enable_utm_set = request.form.get("enable_utm_set") == "on"
        generate_qrs = request.form.get("generate_qrs") == "on"
        utm_campaign = request.form.get("utm_campaign", "").strip()
        utm_variants = request.form.get("utm_variants", DEFAULT_UTM_VARIANTS).strip()

        if not url:
            error = "Enter a destination link."
        elif not is_valid_url(url):
            error = "Enter a valid http or https link."
        elif mode == "qr_only":
            result = {
                "mode": "qr_only",
                "destination_url": url,
                "qr_image": generate_qr_base64(url),
            }
        elif enable_utm_set and not utm_campaign:
            error = "Enter a campaign name to generate UTM links."
        else:
            try:
                variants = parse_utm_variants(utm_variants) if enable_utm_set else []
            except ValueError as exc:
                error = str(exc)
            else:
                link = create_short_link(url, title)
                base_short_url = short_url_for(link["code"])
                generated_variants = []

                for variant in variants:
                    save_generated_link(
                        link["id"],
                        variant["source"],
                        variant["medium"],
                        utm_campaign,
                    )
                    variant_url = add_query_params(
                        base_short_url,
                        {
                            "utm_source": variant["source"],
                            "utm_medium": variant["medium"],
                            "utm_campaign": utm_campaign,
                        },
                    )
                    generated_variants.append(
                        {
                            **variant,
                            "url": variant_url,
                            "qr_image": generate_qr_base64(variant_url)
                            if generate_qrs
                            else "",
                        }
                    )
                if variants:
                    get_db().commit()

                result = {
                    "mode": "short_link",
                    "code": link["code"],
                    "destination_url": link["destination_url"],
                    "short_url": base_short_url,
                    "qr_image": generate_qr_base64(base_short_url)
                    if generate_qrs and not generated_variants
                    else "",
                    "variants": generated_variants,
                }

    return render_template(
        "index.html",
        url=url,
        title=title,
        error=error,
        result=result,
        mode=mode,
        enable_utm_set=enable_utm_set,
        generate_qrs=generate_qrs,
        utm_campaign=utm_campaign,
        utm_variants=utm_variants,
    )


@app.route("/s/<code>")
def redirect_short_link(code: str):
    link = execute(
        "SELECT * FROM short_links WHERE code = ? AND is_active = 1", (code,)
    ).fetchone()
    if not link:
        abort(404)

    record_visit(link)
    utm_values = {
        field: request.args.get(field, "").strip()
        for field in UTM_FIELDS
        if request.args.get(field, "").strip()
    }
    return redirect(add_query_params(link["destination_url"], utm_values), code=302)


@app.route("/login", methods=["GET", "POST"])
def site_login():
    if not analytics_password():
        return render_template("login.html", password_missing=True), 503

    error = None
    if request.method == "POST":
        if request.form.get("password", "") == analytics_password():
            session["site_authenticated"] = True
            return redirect(request.args.get("next") or url_for("index"))
        error = "That password did not match."

    return render_template("login.html", error=error, password_missing=False)


@app.route("/logout")
def site_logout():
    session.pop("site_authenticated", None)
    return redirect(url_for("site_login"))


@app.route("/analytics/login", methods=["GET", "POST"])
def analytics_login():
    return redirect(url_for("site_login", next=url_for("analytics_dashboard")))


@app.route("/analytics/logout")
def analytics_logout():
    return redirect(url_for("site_logout"))


@app.route("/analytics")
def analytics_dashboard():
    metric = analytics_metric()
    selected_count = count_expr(metric)
    top_links = execute(
        f"""
        SELECT
            short_links.code,
            short_links.title,
            short_links.destination_url,
            {selected_count} AS visit_count,
            MAX(visits.visited_at) AS last_visit
        FROM short_links
        LEFT JOIN visits ON visits.short_link_id = short_links.id
        GROUP BY short_links.id
        ORDER BY visit_count DESC, short_links.created_at DESC
        LIMIT 10
        """
    ).fetchall()
    daily_visits = execute(
        f"""
        SELECT DATE(visited_at) AS visit_date, {selected_count} AS visit_count
        FROM visits
        GROUP BY DATE(visited_at)
        ORDER BY visit_date DESC
        LIMIT 14
        """
    ).fetchall()
    recent_visits = execute(
        """
        SELECT visits.*, short_links.code, short_links.title, short_links.destination_url
        FROM visits
        JOIN short_links ON short_links.id = visits.short_link_id
        ORDER BY visits.visited_at DESC
        LIMIT 25
        """
    ).fetchall()
    generated_campaigns = execute(
        """
        SELECT
            utm_campaign,
            COUNT(*) AS link_count,
            MAX(created_at) AS last_created
        FROM generated_links
        GROUP BY utm_campaign
        ORDER BY last_created DESC
        LIMIT 20
        """
    ).fetchall()

    return render_template(
        "analytics_dashboard.html",
        metric=metric,
        metric_label=metric_label(metric),
        total_visits=scalar("SELECT COUNT(*) FROM visits"),
        total_uniques=unique_count(),
        visits_today=visits_today_count(),
        unique_visitors_today=unique_count(today=True),
        total_links=scalar("SELECT COUNT(*) FROM short_links"),
        top_links=top_links,
        daily_visits=list(reversed(daily_visits)),
        top_campaigns=top_dimension("utm_campaign", metric=metric),
        top_sources=top_dimension("utm_source", metric=metric),
        top_mediums=top_dimension("utm_medium", metric=metric),
        generated_campaigns=generated_campaigns,
        recent_visits=recent_visits,
    )


def unique_count(short_link_id: int | None = None, today: bool = False) -> int:
    filters = []
    params = []
    if short_link_id is not None:
        filters.append("short_link_id = ?")
        params.append(short_link_id)
    if today:
        if db_backend() == "postgres":
            filters.append("DATE(visited_at) = CURRENT_DATE")
        else:
            filters.append("DATE(visited_at) = DATE('now')")

    where = f" WHERE {' AND '.join(filters)}" if filters else ""
    return scalar(
        f"SELECT COUNT(DISTINCT {unique_visitor_expr()}) FROM visits{where}",
        tuple(params),
    )


def visits_today_count() -> int:
    if db_backend() == "postgres":
        return scalar("SELECT COUNT(*) FROM visits WHERE DATE(visited_at) = CURRENT_DATE")
    return scalar("SELECT COUNT(*) FROM visits WHERE DATE(visited_at) = DATE('now')")


def top_dimension(
    column: str, short_link_id: int | None = None, metric: str = "visits"
) -> list:
    if column not in UTM_FIELDS:
        raise ValueError("Unsupported analytics column.")

    params: tuple = ()
    link_filter = ""
    if short_link_id is not None:
        link_filter = "AND short_link_id = ?"
        params = (short_link_id,)

    return execute(
        f"""
        SELECT {column} AS label, {count_expr(metric)} AS visit_count
        FROM visits
        WHERE {column} IS NOT NULL AND {column} != '' {link_filter}
        GROUP BY {column}
        ORDER BY visit_count DESC, label ASC
        LIMIT 10
        """,
        params,
    ).fetchall()


def generated_links_for_campaign(campaign: str) -> list[dict]:
    rows = execute(
        f"""
        SELECT
            generated_links.id,
            generated_links.utm_source,
            generated_links.utm_medium,
            generated_links.utm_campaign,
            generated_links.created_at,
            short_links.code,
            short_links.title,
            short_links.destination_url,
            {count_expr("visits", "visits")} AS visit_count,
            {count_expr("uniques", "visits")} AS unique_count,
            MAX(visits.visited_at) AS last_visit
        FROM generated_links
        JOIN short_links ON short_links.id = generated_links.short_link_id
        LEFT JOIN visits ON visits.short_link_id = generated_links.short_link_id
            AND visits.utm_campaign = generated_links.utm_campaign
            AND visits.utm_source = generated_links.utm_source
            AND visits.utm_medium = generated_links.utm_medium
        WHERE generated_links.utm_campaign = ?
        GROUP BY
            generated_links.id,
            generated_links.utm_source,
            generated_links.utm_medium,
            generated_links.utm_campaign,
            generated_links.created_at,
            short_links.code,
            short_links.title,
            short_links.destination_url
        ORDER BY generated_links.created_at DESC, generated_links.id DESC
        """,
        (campaign,),
    ).fetchall()

    links = []
    for row in rows:
        url = add_query_params(
            short_url_for(row["code"]),
            {
                "utm_source": row["utm_source"],
                "utm_medium": row["utm_medium"],
                "utm_campaign": row["utm_campaign"],
            },
        )
        links.append({**dict(row), "url": url})
    return links


def campaign_link_count(campaign: str) -> int:
    return scalar(
        """
        SELECT COUNT(DISTINCT short_link_id)
        FROM generated_links
        WHERE utm_campaign = ?
        """,
        (campaign,),
    )


def campaign_active_link_count(campaign: str) -> int:
    return scalar(
        """
        SELECT COUNT(DISTINCT short_links.id)
        FROM generated_links
        JOIN short_links ON short_links.id = generated_links.short_link_id
        WHERE generated_links.utm_campaign = ? AND short_links.is_active = 1
        """,
        (campaign,),
    )


def campaign_export_rows(campaign: str) -> list[dict]:
    generated_rows = execute(
        """
        SELECT
            generated_links.created_at AS generated_at,
            generated_links.utm_campaign,
            generated_links.utm_source,
            generated_links.utm_medium,
            short_links.code,
            short_links.title,
            short_links.destination_url
        FROM generated_links
        JOIN short_links ON short_links.id = generated_links.short_link_id
        WHERE generated_links.utm_campaign = ?
        ORDER BY generated_links.created_at ASC, generated_links.id ASC
        """,
        (campaign,),
    ).fetchall()
    visit_rows = execute(
        """
        SELECT
            visits.visited_at,
            visits.utm_campaign,
            visits.utm_source,
            visits.utm_medium,
            visits.utm_term,
            visits.utm_content,
            visits.referrer,
            visits.browser,
            visits.device_type,
            visits.user_agent,
            visits.ip_address,
            short_links.code,
            short_links.title,
            short_links.destination_url
        FROM visits
        JOIN short_links ON short_links.id = visits.short_link_id
        WHERE visits.utm_campaign = ?
        ORDER BY visits.visited_at ASC, visits.id ASC
        """,
        (campaign,),
    ).fetchall()

    rows = []
    for row in generated_rows:
        data = row_dict(row)
        short_url = add_query_params(
            short_url_for(data["code"]),
            {
                "utm_source": data["utm_source"],
                "utm_medium": data["utm_medium"],
                "utm_campaign": data["utm_campaign"],
            },
        )
        rows.append(
            {
                "row_type": "generated_link",
                "generated_at": data["generated_at"],
                "visited_at": "",
                "campaign": data["utm_campaign"],
                "source": data["utm_source"],
                "medium": data["utm_medium"],
                "term": "",
                "content": "",
                "short_url": short_url,
                "short_code": data["code"],
                "destination_url": data["destination_url"],
                "title": data["title"] or "",
                "referrer": "",
                "browser": "",
                "device_type": "",
                "user_agent": "",
                "ip_address": "",
            }
        )

    for row in visit_rows:
        data = row_dict(row)
        short_url = add_query_params(
            short_url_for(data["code"]),
            {
                "utm_source": data["utm_source"] or "",
                "utm_medium": data["utm_medium"] or "",
                "utm_campaign": data["utm_campaign"] or "",
                "utm_term": data["utm_term"] or "",
                "utm_content": data["utm_content"] or "",
            },
        )
        rows.append(
            {
                "row_type": "visit",
                "generated_at": "",
                "visited_at": data["visited_at"],
                "campaign": data["utm_campaign"] or "",
                "source": data["utm_source"] or "",
                "medium": data["utm_medium"] or "",
                "term": data["utm_term"] or "",
                "content": data["utm_content"] or "",
                "short_url": short_url,
                "short_code": data["code"],
                "destination_url": data["destination_url"],
                "title": data["title"] or "",
                "referrer": data["referrer"] or "",
                "browser": data["browser"] or "",
                "device_type": data["device_type"] or "",
                "user_agent": data["user_agent"] or "",
                "ip_address": data["ip_address"] or "",
            }
        )
    return rows


@app.route("/analytics/campaign/<campaign>/export.csv")
def export_campaign_csv(campaign: str):
    headers = [
        "row_type",
        "generated_at",
        "visited_at",
        "campaign",
        "source",
        "medium",
        "term",
        "content",
        "short_url",
        "short_code",
        "destination_url",
        "title",
        "referrer",
        "browser",
        "device_type",
        "user_agent",
        "ip_address",
    ]
    buffer = io.StringIO()
    writer = csv.DictWriter(buffer, fieldnames=headers)
    writer.writeheader()
    writer.writerows(campaign_export_rows(campaign))

    filename = "".join(char if char.isalnum() or char in "-_" else "_" for char in campaign)
    return Response(
        buffer.getvalue(),
        mimetype="text/csv",
        headers={
            "Content-Disposition": f"attachment; filename={filename or 'campaign'}-export.csv"
        },
    )


@app.route("/analytics/campaign/<campaign>/cleanup", methods=["POST"])
def cleanup_campaign(campaign: str):
    if request.form.get("confirm_campaign", "") != campaign:
        return redirect(
            url_for(
                "analytics_campaign_detail",
                campaign=campaign,
                cleanup_error="Type the exact campaign name before cleanup.",
            )
        )
    if request.form.get("export_confirmed") != "on":
        return redirect(
            url_for(
                "analytics_campaign_detail",
                campaign=campaign,
                cleanup_error="Confirm that you exported the campaign before cleanup.",
            )
        )

    delete_generated = request.form.get("delete_generated") == "on"
    deactivate_links = request.form.get("deactivate_links") == "on"
    visits_deleted = scalar("SELECT COUNT(*) FROM visits WHERE utm_campaign = ?", (campaign,))
    generated_deleted = (
        scalar("SELECT COUNT(*) FROM generated_links WHERE utm_campaign = ?", (campaign,))
        if delete_generated
        else 0
    )
    links_deactivated = campaign_active_link_count(campaign) if deactivate_links else 0

    if deactivate_links:
        execute(
            """
            UPDATE short_links
            SET is_active = 0
            WHERE id IN (
                SELECT short_link_id
                FROM generated_links
                WHERE utm_campaign = ?
            )
            """,
            (campaign,),
        )

    execute("DELETE FROM visits WHERE utm_campaign = ?", (campaign,))

    if delete_generated:
        execute("DELETE FROM generated_links WHERE utm_campaign = ?", (campaign,))

    get_db().commit()
    return redirect(
        url_for(
            "analytics_campaign_detail",
            campaign=campaign,
            visits_deleted=visits_deleted,
            generated_deleted=generated_deleted,
            links_deactivated=links_deactivated,
        )
    )


@app.route("/analytics/campaign/<campaign>")
def analytics_campaign_detail(campaign: str):
    metric = analytics_metric()
    selected_count = count_expr(metric)
    daily_visits = execute(
        f"""
        SELECT DATE(visited_at) AS visit_date, {selected_count} AS visit_count
        FROM visits
        WHERE utm_campaign = ?
        GROUP BY DATE(visited_at)
        ORDER BY visit_date DESC
        LIMIT 14
        """,
        (campaign,),
    ).fetchall()
    recent_visits = execute(
        """
        SELECT visits.*, short_links.code, short_links.title, short_links.destination_url
        FROM visits
        JOIN short_links ON short_links.id = visits.short_link_id
        WHERE visits.utm_campaign = ?
        ORDER BY visits.visited_at DESC
        LIMIT 50
        """,
        (campaign,),
    ).fetchall()

    return render_template(
        "analytics_campaign.html",
        campaign=campaign,
        metric=metric,
        metric_label=metric_label(metric),
        generated_links=generated_links_for_campaign(campaign),
        active_link_count=campaign_active_link_count(campaign),
        link_count=campaign_link_count(campaign),
        cleanup_error=request.args.get("cleanup_error", ""),
        cleanup_result={
            "visits_deleted": request.args.get("visits_deleted"),
            "generated_deleted": request.args.get("generated_deleted"),
            "links_deactivated": request.args.get("links_deactivated"),
        },
        total_visits=scalar("SELECT COUNT(*) FROM visits WHERE utm_campaign = ?", (campaign,)),
        total_uniques=scalar(
            f"""
            SELECT COUNT(DISTINCT {unique_visitor_expr()})
            FROM visits
            WHERE utm_campaign = ?
            """,
            (campaign,),
        ),
        daily_visits=list(reversed(daily_visits)),
        recent_visits=recent_visits,
    )


@app.route("/analytics/link/<code>")
def analytics_link_detail(code: str):
    metric = analytics_metric()
    selected_count = count_expr(metric)
    link = execute("SELECT * FROM short_links WHERE code = ?", (code,)).fetchone()
    if not link:
        abort(404)

    daily_visits = execute(
        f"""
        SELECT DATE(visited_at) AS visit_date, {selected_count} AS visit_count
        FROM visits
        WHERE short_link_id = ?
        GROUP BY DATE(visited_at)
        ORDER BY visit_date DESC
        LIMIT 14
        """,
        (link["id"],),
    ).fetchall()
    recent_visits = execute(
        """
        SELECT *
        FROM visits
        WHERE short_link_id = ?
        ORDER BY visited_at DESC
        LIMIT 50
        """,
        (link["id"],),
    ).fetchall()

    return render_template(
        "analytics_link.html",
        metric=metric,
        metric_label=metric_label(metric),
        link=link,
        short_url=short_url_for(code),
        total_visits=scalar(
            "SELECT COUNT(*) FROM visits WHERE short_link_id = ?", (link["id"],)
        ),
        total_uniques=unique_count(short_link_id=link["id"]),
        daily_visits=list(reversed(daily_visits)),
        top_campaigns=top_dimension("utm_campaign", link["id"], metric),
        top_sources=top_dimension("utm_source", link["id"], metric),
        top_mediums=top_dimension("utm_medium", link["id"], metric),
        recent_visits=recent_visits,
    )


if __name__ == "__main__":
    with app.app_context():
        init_db()
        app.config["DB_INITIALIZED"] = True
    app.run(debug=True)