generated from westfarn/web_django_template
Initial commit
This commit is contained in:
ai_ml_operations
@@ -0,0 +1,10 @@
|
||||
from django.contrib import admin
|
||||
|
||||
from core.models import StoredFile
|
||||
|
||||
|
||||
@admin.register(StoredFile)
|
||||
class StoredFileAdmin(admin.ModelAdmin):
|
||||
list_display = ("filename", "kind", "content_type", "size", "created_at")
|
||||
list_filter = ("kind", "content_type")
|
||||
readonly_fields = ("size", "content_type", "created_at", "updated_at")
|
||||
@@ -0,0 +1,11 @@
|
||||
from django.apps import AppConfig
|
||||
|
||||
|
||||
class CoreConfig(AppConfig):
|
||||
default_auto_field = "django.db.models.BigAutoField"
|
||||
name = "core"
|
||||
|
||||
def ready(self):
|
||||
from core.registry import register_feature
|
||||
|
||||
register_feature("core")
|
||||
@@ -0,0 +1,238 @@
|
||||
"""Campaign tracked homepage links (UTM) plus optional piha.li short URLs.
|
||||
|
||||
utm_source comes from UTM_SOURCE env, else a slug of SITE_NAME — never a
|
||||
hardcoded client name. Shared by email_sms and directmail.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import html
|
||||
import re
|
||||
from urllib.parse import urlencode
|
||||
|
||||
from django.conf import settings
|
||||
from django.urls import NoReverseMatch, reverse
|
||||
|
||||
from contacts.models import Channel
|
||||
from core.shortener import resolve_display_url
|
||||
|
||||
UTM_CAMPAIGN_SLUG_MAX = 80
|
||||
_UTM_ANCHOR_RE = re.compile(r"(<a\b[^>]*>)(.*?)(</a>)", re.IGNORECASE | re.DOTALL)
|
||||
_SHORT_TEXT_URL_RE = re.compile(
|
||||
r"https?://(?:(?:www\.)?(?:beta\.)?piha\.li|(?:127\.0\.0\.1|localhost):\d+)"
|
||||
r"/[a-z0-9]{4,8}",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
|
||||
|
||||
def campaign_utm_slug(name: str) -> str:
|
||||
"""Lowercase hyphenated slug (utm_campaign / utm_source)."""
|
||||
slug = re.sub(r"[^a-z0-9]+", "-", (name or "").strip().lower())
|
||||
slug = re.sub(r"-{2,}", "-", slug).strip("-")
|
||||
return (slug or "campaign")[:UTM_CAMPAIGN_SLUG_MAX]
|
||||
|
||||
|
||||
def campaign_utm_source() -> str:
|
||||
"""Brand slug for utm_source. Override with UTM_SOURCE env."""
|
||||
explicit = (getattr(settings, "UTM_SOURCE", None) or "").strip()
|
||||
if explicit:
|
||||
return campaign_utm_slug(explicit)
|
||||
return campaign_utm_slug(getattr(settings, "SITE_NAME", None) or "campaign")
|
||||
|
||||
|
||||
def public_site_base_url() -> str:
|
||||
"""Public homepage used as the UTM landing URL."""
|
||||
return (getattr(settings, "PUBLIC_SITE_URL", None) or "").rstrip("/")
|
||||
|
||||
|
||||
def public_site_link_label() -> str:
|
||||
"""Visible email-link text (host), or SITE_NAME in local/dev."""
|
||||
host = (
|
||||
public_site_base_url()
|
||||
.replace("https://", "")
|
||||
.replace("http://", "")
|
||||
.split("/")[0]
|
||||
)
|
||||
brand = (getattr(settings, "SITE_NAME", None) or "").strip() or "our website"
|
||||
if (
|
||||
not host
|
||||
or host.startswith("127.")
|
||||
or host.startswith("localhost")
|
||||
or host.startswith("0.0.0.0")
|
||||
):
|
||||
return brand
|
||||
return host
|
||||
|
||||
|
||||
def build_campaign_utm_url(
|
||||
*,
|
||||
name: str,
|
||||
medium: str,
|
||||
base_url: str = "",
|
||||
) -> str:
|
||||
"""Homepage URL with utm_source / utm_medium / utm_campaign."""
|
||||
base = (base_url or public_site_base_url()).rstrip("/")
|
||||
channel = (medium or "").strip().lower()
|
||||
if channel not in Channel.values:
|
||||
channel = Channel.EMAIL
|
||||
query = urlencode(
|
||||
{
|
||||
"utm_source": campaign_utm_source(),
|
||||
"utm_medium": channel,
|
||||
"utm_campaign": campaign_utm_slug(name),
|
||||
}
|
||||
)
|
||||
return f"{base}/?{query}"
|
||||
|
||||
|
||||
def campaign_shortener_ref(
|
||||
*,
|
||||
campaign_id=None,
|
||||
user_id=None,
|
||||
medium: str = "",
|
||||
name: str = "",
|
||||
) -> str:
|
||||
"""external_ref for the shortener (max 64). Prefer campaign UUID."""
|
||||
if campaign_id:
|
||||
return str(campaign_id)[:64]
|
||||
slug = campaign_utm_slug(name)[:40]
|
||||
ch = (medium or "email").strip().lower()[:8]
|
||||
uid = "" if user_id is None else str(user_id)
|
||||
return f"p{uid}-{ch}-{slug}"[:64]
|
||||
|
||||
|
||||
def resolve_campaign_tracked_url(
|
||||
*,
|
||||
name: str,
|
||||
medium: str,
|
||||
campaign_id=None,
|
||||
user_id=None,
|
||||
shorten: bool = True,
|
||||
) -> tuple[str, str]:
|
||||
"""Return (long UTM URL, display URL). Display is short when mint works."""
|
||||
target = build_campaign_utm_url(name=name, medium=medium)
|
||||
if not shorten:
|
||||
return target, target
|
||||
display = resolve_display_url(
|
||||
target,
|
||||
title=f"{(name or 'campaign').strip()} ({medium})"[:200],
|
||||
external_ref=campaign_shortener_ref(
|
||||
campaign_id=campaign_id, user_id=user_id, medium=medium, name=name
|
||||
),
|
||||
)
|
||||
return target, display
|
||||
|
||||
|
||||
def _utm_text_url_re() -> re.Pattern[str]:
|
||||
return re.compile(
|
||||
r"https?://[^\s<>\"]+utm_source="
|
||||
+ re.escape(campaign_utm_source())
|
||||
+ r"[^\s<>\"]*",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
|
||||
|
||||
def _is_our_utm_anchor(attrs: str) -> bool:
|
||||
lower = attrs.lower()
|
||||
source = campaign_utm_source().lower()
|
||||
return "data-campaign-utm" in lower or f"utm_source={source}" in lower
|
||||
|
||||
|
||||
def ensure_campaign_utm_in_html(body: str, url: str, label: str) -> str:
|
||||
"""Insert or refresh the tracked <a> in an HTML (email) body."""
|
||||
href = html.escape(url, quote=True)
|
||||
label_html = html.escape(label)
|
||||
replaced = False
|
||||
|
||||
def _repl(match: re.Match[str]) -> str:
|
||||
nonlocal replaced
|
||||
if replaced or not _is_our_utm_anchor(match.group(1)):
|
||||
return match.group(0)
|
||||
replaced = True
|
||||
inner = match.group(2) if (match.group(2) or "").strip() else label_html
|
||||
return f'<a href="{href}" data-campaign-utm="1">{inner}</a>'
|
||||
|
||||
out = _UTM_ANCHOR_RE.sub(_repl, body or "")
|
||||
if replaced:
|
||||
return out
|
||||
anchor = f'<a href="{href}" data-campaign-utm="1">{label_html}</a>'
|
||||
text = (body or "").rstrip()
|
||||
if text:
|
||||
return f"{text}\n<p>{anchor}</p>"
|
||||
return f"<p>{anchor}</p>"
|
||||
|
||||
|
||||
def ensure_campaign_utm_in_text(body: str, url: str) -> str:
|
||||
"""Insert or refresh the tracked URL in a plain-text (SMS) body."""
|
||||
raw = body or ""
|
||||
utm_re = _utm_text_url_re()
|
||||
if utm_re.search(raw):
|
||||
return utm_re.sub(url, raw, count=1)
|
||||
if _SHORT_TEXT_URL_RE.search(raw):
|
||||
return _SHORT_TEXT_URL_RE.sub(url, raw, count=1)
|
||||
text = raw.rstrip()
|
||||
return f"{text}\n\n{url}" if text else url
|
||||
|
||||
|
||||
def ensure_campaign_utm_link(
|
||||
body: str,
|
||||
*,
|
||||
name: str,
|
||||
medium: str,
|
||||
html: bool,
|
||||
campaign_id=None,
|
||||
shorten: bool | None = None,
|
||||
) -> str:
|
||||
"""Keep a tracked site link in the body, matching campaign name + channel."""
|
||||
do_shorten = bool(campaign_id) if shorten is None else shorten
|
||||
_target, url = resolve_campaign_tracked_url(
|
||||
name=name,
|
||||
medium=medium,
|
||||
campaign_id=campaign_id,
|
||||
shorten=do_shorten,
|
||||
)
|
||||
if html:
|
||||
return ensure_campaign_utm_in_html(body, url, public_site_link_label())
|
||||
return ensure_campaign_utm_in_text(body, url)
|
||||
|
||||
|
||||
def postcard_designer_url() -> str:
|
||||
try:
|
||||
return reverse("directmail:postcard_designer")
|
||||
except NoReverseMatch:
|
||||
return ""
|
||||
|
||||
|
||||
def utm_panel_context(
|
||||
*,
|
||||
campaign=None,
|
||||
live: bool = False,
|
||||
medium: str = "",
|
||||
) -> dict:
|
||||
"""Template context for `_utm_link_panel.html`."""
|
||||
name = getattr(campaign, "name", "") or ""
|
||||
ch = medium or getattr(campaign, "channel", "") or ""
|
||||
display = ""
|
||||
if campaign is not None:
|
||||
_target, display = resolve_campaign_tracked_url(
|
||||
name=name,
|
||||
medium=ch,
|
||||
campaign_id=getattr(campaign, "pk", None),
|
||||
shorten=True,
|
||||
)
|
||||
try:
|
||||
shorten_url = reverse("campaign_short_link")
|
||||
except NoReverseMatch:
|
||||
shorten_url = ""
|
||||
return {
|
||||
"utm_base_url": public_site_base_url(),
|
||||
"utm_link_label": public_site_link_label(),
|
||||
"utm_source": campaign_utm_source(),
|
||||
"utm_shorten_url": shorten_url,
|
||||
"utm_url": display,
|
||||
"utm_live": live,
|
||||
"utm_medium": ch,
|
||||
"utm_campaign_name": name,
|
||||
"utm_campaign_id": str(getattr(campaign, "pk", "") or ""),
|
||||
"utm_postcard_designer_url": postcard_designer_url(),
|
||||
}
|
||||
@@ -0,0 +1,9 @@
|
||||
from core.registry import enabled_features, portal_nav, public_nav
|
||||
|
||||
|
||||
def feature_registry(request):
|
||||
return {
|
||||
"portal_nav": portal_nav(),
|
||||
"public_nav_extra": public_nav(),
|
||||
"enabled_features": enabled_features(),
|
||||
}
|
||||
@@ -0,0 +1,46 @@
|
||||
import time
|
||||
|
||||
from django.core.management.base import BaseCommand
|
||||
|
||||
from core.registry import run_dispatchers
|
||||
|
||||
|
||||
class Command(BaseCommand):
|
||||
help = (
|
||||
"Enqueue due scheduled work registered by installed apps. "
|
||||
"Pass --loop to poll until stopped (worker entrypoint)."
|
||||
)
|
||||
|
||||
def add_arguments(self, parser):
|
||||
parser.add_argument(
|
||||
"--loop",
|
||||
action="store_true",
|
||||
help="Poll forever until SIGINT/SIGTERM.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--interval",
|
||||
type=int,
|
||||
default=15,
|
||||
help="Seconds between polls when --loop is set (default 15).",
|
||||
)
|
||||
|
||||
def handle(self, *args, **options):
|
||||
self.verbosity = int(options.get("verbosity", 1))
|
||||
if options["loop"]:
|
||||
interval = max(1, int(options["interval"] or 15))
|
||||
self.stdout.write(f"Polling due scheduled work every {interval}s.")
|
||||
try:
|
||||
while True:
|
||||
self._dispatch_once()
|
||||
time.sleep(interval)
|
||||
except KeyboardInterrupt:
|
||||
self.stdout.write("dispatch_due loop stopped.")
|
||||
return
|
||||
self._dispatch_once()
|
||||
|
||||
def _dispatch_once(self):
|
||||
enqueued = run_dispatchers()
|
||||
if enqueued:
|
||||
self.stdout.write(self.style.SUCCESS(f"Enqueued {enqueued} due item(s)."))
|
||||
elif self.verbosity >= 2:
|
||||
self.stdout.write("No due items.")
|
||||
@@ -0,0 +1,35 @@
|
||||
# Generated by Django 6.1 on 2026-08-26 11:38
|
||||
|
||||
import django.db.models.deletion
|
||||
import uuid
|
||||
from django.conf import settings
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
initial = True
|
||||
|
||||
dependencies = [
|
||||
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.CreateModel(
|
||||
name='StoredFile',
|
||||
fields=[
|
||||
('created_at', models.DateTimeField(auto_now_add=True)),
|
||||
('updated_at', models.DateTimeField(auto_now=True)),
|
||||
('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
|
||||
('kind', models.CharField(choices=[('campaign_image', 'Campaign image'), ('social_image', 'Social image'), ('social_video', 'Social video'), ('invoice_pdf', 'Invoice PDF')], default='campaign_image', max_length=32)),
|
||||
('filename', models.CharField(blank=True, max_length=255)),
|
||||
('content_type', models.CharField(max_length=128)),
|
||||
('size', models.PositiveIntegerField(default=0)),
|
||||
('data', models.BinaryField()),
|
||||
('uploaded_by', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='uploaded_files', to=settings.AUTH_USER_MODEL)),
|
||||
],
|
||||
options={
|
||||
'ordering': ['-created_at'],
|
||||
},
|
||||
),
|
||||
]
|
||||
@@ -0,0 +1,52 @@
|
||||
import uuid
|
||||
|
||||
from django.conf import settings
|
||||
from django.db import models
|
||||
|
||||
|
||||
class TimeStampedModel(models.Model):
|
||||
"""Abstract base with created/updated timestamps."""
|
||||
|
||||
created_at = models.DateTimeField(auto_now_add=True)
|
||||
updated_at = models.DateTimeField(auto_now=True)
|
||||
|
||||
class Meta:
|
||||
abstract = True
|
||||
|
||||
|
||||
class UUIDPrimaryKeyModel(models.Model):
|
||||
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
|
||||
|
||||
class Meta:
|
||||
abstract = True
|
||||
|
||||
|
||||
class StoredFile(UUIDPrimaryKeyModel, TimeStampedModel):
|
||||
"""Binary file blob in the database (no filesystem media storage)."""
|
||||
|
||||
class Kind(models.TextChoices):
|
||||
CAMPAIGN_IMAGE = "campaign_image", "Campaign image"
|
||||
SOCIAL_IMAGE = "social_image", "Social image"
|
||||
SOCIAL_VIDEO = "social_video", "Social video"
|
||||
INVOICE_PDF = "invoice_pdf", "Invoice PDF"
|
||||
|
||||
kind = models.CharField(
|
||||
max_length=32, choices=Kind.choices, default=Kind.CAMPAIGN_IMAGE
|
||||
)
|
||||
filename = models.CharField(max_length=255, blank=True)
|
||||
content_type = models.CharField(max_length=128)
|
||||
size = models.PositiveIntegerField(default=0)
|
||||
data = models.BinaryField()
|
||||
uploaded_by = models.ForeignKey(
|
||||
settings.AUTH_USER_MODEL,
|
||||
null=True,
|
||||
blank=True,
|
||||
on_delete=models.SET_NULL,
|
||||
related_name="uploaded_files",
|
||||
)
|
||||
|
||||
class Meta:
|
||||
ordering = ["-created_at"]
|
||||
|
||||
def __str__(self) -> str:
|
||||
return self.filename or str(self.pk)
|
||||
@@ -0,0 +1,105 @@
|
||||
"""Optional-app registry. Apps register from AppConfig.ready() — never from always-on imports."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Callable
|
||||
|
||||
from django.urls import reverse
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class NavItem:
|
||||
section: str
|
||||
label: str
|
||||
url_name: str
|
||||
group: str = "Overview"
|
||||
order: int = 100
|
||||
|
||||
|
||||
GROUP_ORDER = {
|
||||
"Overview": 10,
|
||||
"Outreach": 20,
|
||||
"Content": 30,
|
||||
"Social": 40,
|
||||
"Billing": 50,
|
||||
}
|
||||
|
||||
_portal_nav: list[NavItem] = []
|
||||
_public_nav: list[NavItem] = []
|
||||
_dashboard_collectors: list[Callable] = []
|
||||
_dispatchers: list[Callable] = []
|
||||
_features: set[str] = set()
|
||||
|
||||
|
||||
def register_feature(name: str) -> None:
|
||||
_features.add(name)
|
||||
|
||||
|
||||
def enabled_features() -> frozenset[str]:
|
||||
return frozenset(_features)
|
||||
|
||||
|
||||
def register_portal_nav(
|
||||
*,
|
||||
section: str,
|
||||
label: str,
|
||||
url_name: str,
|
||||
group: str = "Overview",
|
||||
order: int = 100,
|
||||
) -> None:
|
||||
item = NavItem(
|
||||
section=section, label=label, url_name=url_name, group=group, order=order
|
||||
)
|
||||
_portal_nav.append(item)
|
||||
|
||||
|
||||
def register_public_nav(
|
||||
*,
|
||||
section: str,
|
||||
label: str,
|
||||
url_name: str,
|
||||
order: int = 100,
|
||||
) -> None:
|
||||
_public_nav.append(
|
||||
NavItem(section=section, label=label, url_name=url_name, order=order)
|
||||
)
|
||||
|
||||
|
||||
def register_dashboard_collector(fn: Callable) -> None:
|
||||
"""fn(request) -> dict merged into dashboard home context."""
|
||||
_dashboard_collectors.append(fn)
|
||||
|
||||
|
||||
def register_dispatcher(fn: Callable) -> None:
|
||||
"""fn() -> int count of items enqueued. Called by dispatch_due."""
|
||||
_dispatchers.append(fn)
|
||||
|
||||
|
||||
def portal_nav() -> list[NavItem]:
|
||||
return sorted(
|
||||
_portal_nav,
|
||||
key=lambda i: (GROUP_ORDER.get(i.group, 99), i.order, i.label),
|
||||
)
|
||||
|
||||
|
||||
def public_nav() -> list[NavItem]:
|
||||
return sorted(_public_nav, key=lambda i: (i.order, i.label))
|
||||
|
||||
|
||||
def collect_dashboard_context(request) -> dict:
|
||||
extra: dict = {}
|
||||
for fn in _dashboard_collectors:
|
||||
extra.update(fn(request) or {})
|
||||
return extra
|
||||
|
||||
|
||||
def run_dispatchers() -> int:
|
||||
total = 0
|
||||
for fn in _dispatchers:
|
||||
total += int(fn() or 0)
|
||||
return total
|
||||
|
||||
|
||||
def reverse_nav(item: NavItem) -> str:
|
||||
return reverse(item.url_name)
|
||||
@@ -0,0 +1,19 @@
|
||||
"""Shared schedule parsing for campaigns and social posts."""
|
||||
|
||||
from datetime import datetime
|
||||
|
||||
from django.utils import timezone
|
||||
|
||||
|
||||
def parse_scheduled_for(raw: str | None):
|
||||
"""Parse optional ``datetime-local`` value into an aware datetime."""
|
||||
value = (raw or "").strip()
|
||||
if not value:
|
||||
return None
|
||||
try:
|
||||
parsed = datetime.fromisoformat(value)
|
||||
except ValueError as exc:
|
||||
raise ValueError("Invalid schedule datetime.") from exc
|
||||
if timezone.is_naive(parsed):
|
||||
return timezone.make_aware(parsed, timezone.get_current_timezone())
|
||||
return parsed
|
||||
@@ -0,0 +1,87 @@
|
||||
"""Server-to-server client for url_shortening_service (piha.li).
|
||||
|
||||
Call the API host (`SHORTENER_BASE_URL`), never the public short domain.
|
||||
Auth: Authorization: Bearer <SHORTENER_API_TOKEN> (name:secret).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
|
||||
import requests
|
||||
from django.conf import settings
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def is_configured() -> bool:
|
||||
base = (getattr(settings, "SHORTENER_BASE_URL", None) or "").strip()
|
||||
token = (getattr(settings, "SHORTENER_API_TOKEN", None) or "").strip()
|
||||
return bool(base and token)
|
||||
|
||||
|
||||
def mint_short_url(
|
||||
target_url: str,
|
||||
*,
|
||||
title: str = "",
|
||||
external_ref: str = "",
|
||||
) -> str | None:
|
||||
"""POST /api/links/. Return short_url, or None if unconfigured / failed.
|
||||
|
||||
Does not raise. Compose must still work when the shortener is down or unset.
|
||||
"""
|
||||
if not is_configured():
|
||||
return None
|
||||
target = (target_url or "").strip()
|
||||
if not target.lower().startswith("https://"):
|
||||
logger.info("shortener skip: target is not https")
|
||||
return None
|
||||
|
||||
base = str(settings.SHORTENER_BASE_URL).rstrip("/")
|
||||
token = str(settings.SHORTENER_API_TOKEN).strip()
|
||||
timeout = int(getattr(settings, "SHORTENER_TIMEOUT_SECONDS", 10) or 10)
|
||||
try:
|
||||
response = requests.post(
|
||||
f"{base}/api/links/",
|
||||
headers={
|
||||
"Authorization": f"Bearer {token}",
|
||||
"Content-Type": "application/json",
|
||||
},
|
||||
json={
|
||||
"target_url": target,
|
||||
"title": (title or "")[:200],
|
||||
"external_ref": (external_ref or "")[:64],
|
||||
},
|
||||
timeout=timeout,
|
||||
)
|
||||
except requests.RequestException:
|
||||
logger.exception("shortener request failed")
|
||||
return None
|
||||
|
||||
if response.status_code not in (200, 201):
|
||||
logger.warning(
|
||||
"shortener mint failed status=%s",
|
||||
response.status_code,
|
||||
)
|
||||
return None
|
||||
try:
|
||||
data = response.json()
|
||||
except ValueError:
|
||||
logger.warning("shortener mint returned non-json")
|
||||
return None
|
||||
if not isinstance(data, dict):
|
||||
return None
|
||||
short = (data.get("short_url") or "").strip()
|
||||
return short or None
|
||||
|
||||
|
||||
def resolve_display_url(
|
||||
target_url: str,
|
||||
*,
|
||||
title: str = "",
|
||||
external_ref: str = "",
|
||||
) -> str:
|
||||
"""Short URL when mint succeeds, otherwise the original target."""
|
||||
return mint_short_url(
|
||||
target_url, title=title, external_ref=external_ref
|
||||
) or target_url
|
||||
@@ -0,0 +1,45 @@
|
||||
<div class="utm-panel" id="utm-link-panel"
|
||||
data-base-url="{{ utm_base_url }}"
|
||||
data-link-label="{{ utm_link_label }}"
|
||||
data-utm-source="{{ utm_source }}"
|
||||
data-medium="{{ utm_medium|default:'' }}"
|
||||
data-campaign-name="{{ utm_campaign_name|default:'' }}"
|
||||
data-shorten-url="{{ utm_shorten_url|default:'' }}"
|
||||
data-campaign-id="{{ utm_campaign_id }}"
|
||||
data-csrf="{{ csrf_token }}"
|
||||
data-live="{% if utm_live %}1{% else %}0{% endif %}">
|
||||
<div class="field" style="margin:0">
|
||||
<label>Tracked site link</label>
|
||||
<p class="hint" data-utm-email-hint hidden>
|
||||
Auto-added to the email as a styled link. Updates as you type the campaign name.
|
||||
<code>utm_medium=email</code> · <code>utm_campaign</code> matches the name.
|
||||
Recipients tap a short <code>piha.li</code> link that redirects to this URL.
|
||||
</p>
|
||||
<p class="hint" data-utm-sms-hint hidden>
|
||||
SMS is plain text — phones cannot show HTML links. A short <code>piha.li</code>
|
||||
URL is in the message and phones make it tappable. Preview below shows it as a link.
|
||||
</p>
|
||||
<p class="hint" data-utm-postcard-hint hidden>
|
||||
Copy the QR image into the postcard designer. Recipients scan it to open the
|
||||
short tracked link (<code>piha.li</code>).
|
||||
</p>
|
||||
<code class="utm-url" data-utm-url>{{ utm_url|default:"" }}</code>
|
||||
<div class="utm-actions">
|
||||
<button type="button" class="btn btn-ghost btn-sm" data-utm-copy-url>Copy link</button>
|
||||
</div>
|
||||
<div class="utm-qr" data-utm-qr-wrap hidden>
|
||||
<canvas data-utm-qr width="192" height="192" aria-label="Campaign QR code"></canvas>
|
||||
<div class="utm-qr-actions">
|
||||
<button type="button" class="btn btn-ghost btn-sm" data-utm-copy-qr>Copy QR image</button>
|
||||
<button type="button" class="btn btn-ghost btn-sm" data-utm-download-qr>Download PNG</button>
|
||||
{% if utm_postcard_designer_url %}
|
||||
<p class="hint" style="margin-top:8px">
|
||||
Paste or upload the PNG in
|
||||
<a href="{{ utm_postcard_designer_url }}">Postcard design</a>.
|
||||
</p>
|
||||
{% endif %}
|
||||
</div>
|
||||
</div>
|
||||
<p class="hint" data-utm-status hidden></p>
|
||||
</div>
|
||||
</div>
|
||||
@@ -0,0 +1,39 @@
|
||||
from io import StringIO
|
||||
|
||||
from django.core.management import call_command
|
||||
from django.test import Client, TestCase, override_settings
|
||||
from django.urls import reverse
|
||||
|
||||
|
||||
class HealthzTests(TestCase):
|
||||
def test_healthz_ok(self):
|
||||
response = Client().get("/healthz/")
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertEqual(response.json()["status"], "ok")
|
||||
|
||||
|
||||
class UnderConstructionTests(TestCase):
|
||||
@override_settings(SITE_UNDER_CONSTRUCTION=True)
|
||||
def test_home_redirects_when_gated(self):
|
||||
response = Client().get("/")
|
||||
self.assertEqual(response.status_code, 302)
|
||||
self.assertIn("/under-construction", response["Location"])
|
||||
|
||||
@override_settings(SITE_UNDER_CONSTRUCTION=False)
|
||||
def test_home_ok_when_open(self):
|
||||
response = Client().get("/")
|
||||
self.assertEqual(response.status_code, 200)
|
||||
|
||||
|
||||
class PublicSmokeTests(TestCase):
|
||||
def test_about_and_contact_get(self):
|
||||
client = Client()
|
||||
self.assertEqual(client.get(reverse("public:about")).status_code, 200)
|
||||
self.assertEqual(client.get(reverse("public:contact")).status_code, 200)
|
||||
|
||||
|
||||
class DispatchDueTests(TestCase):
|
||||
def test_quiet_when_nothing_due(self):
|
||||
out = StringIO()
|
||||
call_command("dispatch_due", stdout=out)
|
||||
self.assertEqual(out.getvalue(), "")
|
||||
@@ -0,0 +1,169 @@
|
||||
"""Campaign UTM + shortener helpers (always-on core)."""
|
||||
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from django.test import SimpleTestCase, TestCase, override_settings
|
||||
from django.urls import reverse
|
||||
|
||||
from core.campaign_utm import (
|
||||
build_campaign_utm_url,
|
||||
campaign_utm_slug,
|
||||
campaign_utm_source,
|
||||
ensure_campaign_utm_in_text,
|
||||
ensure_campaign_utm_link,
|
||||
)
|
||||
from core.shortener import mint_short_url, resolve_display_url
|
||||
|
||||
|
||||
class CampaignUtmHelperTests(SimpleTestCase):
|
||||
def test_slug(self):
|
||||
self.assertEqual(campaign_utm_slug("Spring seller tips"), "spring-seller-tips")
|
||||
|
||||
@override_settings(SITE_NAME="Acme HVAC", UTM_SOURCE="")
|
||||
def test_source_from_site_name(self):
|
||||
self.assertEqual(campaign_utm_source(), "acme-hvac")
|
||||
|
||||
@override_settings(SITE_NAME="Acme HVAC", UTM_SOURCE="acme")
|
||||
def test_source_env_override(self):
|
||||
self.assertEqual(campaign_utm_source(), "acme")
|
||||
|
||||
@override_settings(
|
||||
SITE_NAME="Acme HVAC",
|
||||
UTM_SOURCE="",
|
||||
PUBLIC_SITE_URL="https://acmehvac.com",
|
||||
)
|
||||
def test_url_shape(self):
|
||||
url = build_campaign_utm_url(name="Spring seller tips", medium="email")
|
||||
self.assertEqual(
|
||||
url,
|
||||
"https://acmehvac.com/?utm_source=acme-hvac"
|
||||
"&utm_medium=email&utm_campaign=spring-seller-tips",
|
||||
)
|
||||
|
||||
def test_sms_replaces_short_url(self):
|
||||
updated = ensure_campaign_utm_in_text(
|
||||
"Hi\n\nhttps://piha.li/a3k9xm",
|
||||
"https://piha.li/zzzzzz",
|
||||
)
|
||||
self.assertEqual(updated, "Hi\n\nhttps://piha.li/zzzzzz")
|
||||
|
||||
|
||||
class ShortenerClientTests(SimpleTestCase):
|
||||
@override_settings(SHORTENER_BASE_URL="", SHORTENER_API_TOKEN="")
|
||||
def test_unconfigured_returns_none(self):
|
||||
self.assertIsNone(mint_short_url("https://acmehvac.com/"))
|
||||
self.assertEqual(
|
||||
resolve_display_url("https://acmehvac.com/x"),
|
||||
"https://acmehvac.com/x",
|
||||
)
|
||||
|
||||
@override_settings(
|
||||
SHORTENER_BASE_URL="https://shortener.aimloperations.com",
|
||||
SHORTENER_API_TOKEN="acme:test-token",
|
||||
)
|
||||
def test_mint_posts_bearer(self):
|
||||
mock = MagicMock()
|
||||
mock.status_code = 201
|
||||
mock.json.return_value = {
|
||||
"code": "a3k9xm",
|
||||
"short_url": "https://piha.li/a3k9xm",
|
||||
}
|
||||
with patch("core.shortener.requests.post", return_value=mock) as post:
|
||||
short = mint_short_url(
|
||||
"https://acmehvac.com/?utm_source=acme",
|
||||
title="Spring",
|
||||
external_ref="campaign-uuid",
|
||||
)
|
||||
self.assertEqual(short, "https://piha.li/a3k9xm")
|
||||
post.assert_called_once()
|
||||
headers = post.call_args.kwargs["headers"]
|
||||
self.assertEqual(headers["Authorization"], "Bearer acme:test-token")
|
||||
payload = post.call_args.kwargs["json"]
|
||||
self.assertEqual(payload["target_url"], "https://acmehvac.com/?utm_source=acme")
|
||||
self.assertEqual(payload["external_ref"], "campaign-uuid")
|
||||
|
||||
@override_settings(
|
||||
SHORTENER_BASE_URL="https://shortener.aimloperations.com",
|
||||
SHORTENER_API_TOKEN="acme:test-token",
|
||||
SITE_NAME="Acme HVAC",
|
||||
UTM_SOURCE="",
|
||||
PUBLIC_SITE_URL="https://acmehvac.com",
|
||||
)
|
||||
def test_ensure_link_falls_back_on_401(self):
|
||||
mock = MagicMock()
|
||||
mock.status_code = 401
|
||||
mock.text = '{"detail":"Unauthorized"}'
|
||||
with patch("core.shortener.requests.post", return_value=mock):
|
||||
body = ensure_campaign_utm_link(
|
||||
"See you Saturday",
|
||||
name="Open house",
|
||||
medium="sms",
|
||||
html=False,
|
||||
campaign_id="11111111-1111-1111-1111-111111111111",
|
||||
)
|
||||
self.assertIn("utm_medium=sms", body)
|
||||
self.assertIn("utm_campaign=open-house", body)
|
||||
|
||||
|
||||
class CampaignShortLinkViewTests(TestCase):
|
||||
def setUp(self):
|
||||
from django.contrib.auth import get_user_model
|
||||
|
||||
User = get_user_model()
|
||||
self.user = User.objects.create_user(
|
||||
username="utm-composer", password="test-pass-123"
|
||||
)
|
||||
self.client.login(username="utm-composer", password="test-pass-123")
|
||||
|
||||
def test_requires_login(self):
|
||||
from django.test import Client
|
||||
|
||||
response = Client().post(
|
||||
reverse("campaign_short_link"),
|
||||
data='{"name":"Open house","medium":"sms"}',
|
||||
content_type="application/json",
|
||||
)
|
||||
self.assertEqual(response.status_code, 302)
|
||||
|
||||
@override_settings(
|
||||
PUBLIC_SITE_URL="https://acmehvac.com",
|
||||
SITE_NAME="Acme HVAC",
|
||||
UTM_SOURCE="",
|
||||
SHORTENER_BASE_URL="",
|
||||
SHORTENER_API_TOKEN="",
|
||||
)
|
||||
def test_unconfigured_returns_long_url(self):
|
||||
response = self.client.post(
|
||||
reverse("campaign_short_link"),
|
||||
data='{"name":"Open house","medium":"sms"}',
|
||||
content_type="application/json",
|
||||
)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
data = response.json()
|
||||
self.assertFalse(data["shortened"])
|
||||
self.assertIn("utm_medium=sms", data["display_url"])
|
||||
self.assertIn("utm_source=acme-hvac", data["display_url"])
|
||||
|
||||
@override_settings(
|
||||
PUBLIC_SITE_URL="https://acmehvac.com",
|
||||
SITE_NAME="Acme HVAC",
|
||||
SHORTENER_BASE_URL="https://shortener.aimloperations.com",
|
||||
SHORTENER_API_TOKEN="acme:test-token",
|
||||
)
|
||||
def test_mints_short_url(self):
|
||||
mock = MagicMock()
|
||||
mock.status_code = 201
|
||||
mock.json.return_value = {
|
||||
"code": "a3k9xm",
|
||||
"short_url": "https://piha.li/a3k9xm",
|
||||
}
|
||||
with patch("core.shortener.requests.post", return_value=mock):
|
||||
response = self.client.post(
|
||||
reverse("campaign_short_link"),
|
||||
data='{"name":"Open house","medium":"sms"}',
|
||||
content_type="application/json",
|
||||
)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
data = response.json()
|
||||
self.assertTrue(data["shortened"])
|
||||
self.assertEqual(data["display_url"], "https://piha.li/a3k9xm")
|
||||
@@ -0,0 +1,94 @@
|
||||
"""Feature-flag isolation: flags control INSTALLED_APPS membership and deps."""
|
||||
|
||||
import ast
|
||||
from pathlib import Path
|
||||
|
||||
from django.apps import apps
|
||||
from django.core.exceptions import ImproperlyConfigured
|
||||
from django.test import SimpleTestCase, TestCase
|
||||
from django.urls import reverse
|
||||
|
||||
|
||||
def _read_settings_source() -> str:
|
||||
return (Path(__file__).resolve().parent.parent / "client_site/settings/base.py").read_text()
|
||||
|
||||
|
||||
class FeatureFlagSettingsTests(SimpleTestCase):
|
||||
def test_optional_apps_are_gated_in_settings(self):
|
||||
src = _read_settings_source()
|
||||
self.assertIn("FEATURE_EMAIL_SMS", src)
|
||||
self.assertIn("FEATURE_DIRECT_MAIL", src)
|
||||
self.assertIn("FEATURE_BLOG", src)
|
||||
self.assertIn("FEATURE_PAYMENTS", src)
|
||||
self.assertIn("FEATURE_SOCIAL", src)
|
||||
self.assertIn("FEATURE_SOCIAL_AI", src)
|
||||
self.assertIn("email_sms.apps.EmailSmsConfig", src)
|
||||
self.assertIn("directmail.apps.DirectmailConfig", src)
|
||||
self.assertIn("blog.apps.BlogConfig", src)
|
||||
self.assertIn("payments.apps.PaymentsConfig", src)
|
||||
self.assertIn("social_ai.apps.SocialAiConfig", src)
|
||||
|
||||
def test_payments_requires_email_sms_in_source(self):
|
||||
src = _read_settings_source()
|
||||
self.assertIn("FEATURE_PAYMENTS requires FEATURE_EMAIL_SMS", src)
|
||||
self.assertIn("FEATURE_SOCIAL_AI requires FEATURE_SOCIAL", src)
|
||||
|
||||
|
||||
class AlwaysOnImportIsolationTests(SimpleTestCase):
|
||||
"""Always-on modules must not import optional apps at module level."""
|
||||
|
||||
OPTIONAL = {
|
||||
"email_sms",
|
||||
"directmail",
|
||||
"blog",
|
||||
"payments",
|
||||
"social",
|
||||
"social_ai",
|
||||
}
|
||||
|
||||
ALWAYS_ON = [
|
||||
"dashboard/views.py",
|
||||
"public/views.py",
|
||||
"core/management/commands/dispatch_due.py",
|
||||
"core/shortener.py",
|
||||
"core/campaign_utm.py",
|
||||
"client_site/urls.py",
|
||||
]
|
||||
|
||||
def test_no_module_level_optional_imports(self):
|
||||
root = Path(__file__).resolve().parent.parent
|
||||
for rel in self.ALWAYS_ON:
|
||||
path = root / rel
|
||||
tree = ast.parse(path.read_text())
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, ast.ImportFrom) and node.module:
|
||||
top = node.module.split(".")[0]
|
||||
# Allow inside functions — ast.ImportFrom at module body only
|
||||
if isinstance(node, (ast.Import, ast.ImportFrom)):
|
||||
pass
|
||||
# Check only top-level
|
||||
for node in tree.body:
|
||||
if isinstance(node, ast.Import):
|
||||
for alias in node.names:
|
||||
top = alias.name.split(".")[0]
|
||||
self.assertNotIn(top, self.OPTIONAL, f"{rel} imports {alias.name}")
|
||||
if isinstance(node, ast.ImportFrom) and node.module:
|
||||
top = node.module.split(".")[0]
|
||||
self.assertNotIn(top, self.OPTIONAL, f"{rel} imports {node.module}")
|
||||
|
||||
|
||||
class InstalledOptionalAppsTests(TestCase):
|
||||
"""Dev defaults install optional apps so the template is a working seed."""
|
||||
|
||||
def test_dev_installs_catalog_apps(self):
|
||||
labels = {c.label for c in apps.get_app_configs()}
|
||||
for label in ("email_sms", "directmail", "blog", "payments", "social", "social_ai"):
|
||||
self.assertIn(label, labels)
|
||||
|
||||
def test_optional_urls_resolve_when_installed(self):
|
||||
self.assertTrue(reverse("email_sms:campaign_list").startswith("/portal/messaging/"))
|
||||
self.assertTrue(reverse("directmail:postcard_designer").startswith("/portal/direct-mail/"))
|
||||
self.assertTrue(reverse("blog:list").startswith("/blog/"))
|
||||
self.assertTrue(reverse("payments:invoice_list").startswith("/portal/payments/"))
|
||||
self.assertTrue(reverse("social:composer").startswith("/portal/social/"))
|
||||
self.assertTrue(reverse("social_ai:generate").startswith("/portal/social/api/generate/"))
|
||||
@@ -0,0 +1,9 @@
|
||||
from django.urls import path
|
||||
|
||||
from core import views
|
||||
|
||||
app_name = "core"
|
||||
|
||||
urlpatterns = [
|
||||
path("files/<uuid:pk>/", views.stored_file, name="stored_file"),
|
||||
]
|
||||
@@ -0,0 +1,66 @@
|
||||
import io
|
||||
import json
|
||||
|
||||
from django.contrib.auth.decorators import login_required
|
||||
from django.http import FileResponse, JsonResponse
|
||||
from django.shortcuts import get_object_or_404
|
||||
from django.views.decorators.http import require_GET, require_POST
|
||||
|
||||
from core.campaign_utm import resolve_campaign_tracked_url
|
||||
from core.models import StoredFile
|
||||
|
||||
|
||||
def healthz(_request):
|
||||
"""Liveness probe for deploy / NPM health checks."""
|
||||
return JsonResponse({"status": "ok"})
|
||||
|
||||
|
||||
@require_GET
|
||||
def stored_file(request, pk):
|
||||
"""Public fetch (UUID acts as capability token)."""
|
||||
stored = get_object_or_404(StoredFile, pk=pk)
|
||||
response = FileResponse(
|
||||
io.BytesIO(bytes(stored.data)),
|
||||
content_type=stored.content_type or "application/octet-stream",
|
||||
)
|
||||
if stored.filename:
|
||||
response["Content-Disposition"] = f'inline; filename="{stored.filename}"'
|
||||
response["Cache-Control"] = "public, max-age=86400"
|
||||
return response
|
||||
|
||||
|
||||
@login_required
|
||||
@require_POST
|
||||
def campaign_short_link(request):
|
||||
"""Mint (or reuse) a short URL for the campaign tracked landing link.
|
||||
|
||||
Browser talks to this portal endpoint only. The shortener is server-to-server.
|
||||
Does not import optional campaign apps — campaign_id is only an external_ref.
|
||||
"""
|
||||
try:
|
||||
payload = json.loads(request.body.decode() or "{}")
|
||||
except (json.JSONDecodeError, UnicodeDecodeError):
|
||||
payload = {}
|
||||
if not isinstance(payload, dict):
|
||||
payload = {}
|
||||
name = (payload.get("name") or request.POST.get("name") or "").strip()
|
||||
medium = (payload.get("medium") or request.POST.get("medium") or "").strip()
|
||||
campaign_id = (
|
||||
payload.get("campaign_id") or request.POST.get("campaign_id") or ""
|
||||
)
|
||||
campaign_id = str(campaign_id).strip() or None
|
||||
target, display = resolve_campaign_tracked_url(
|
||||
name=name or "campaign",
|
||||
medium=medium,
|
||||
campaign_id=campaign_id,
|
||||
user_id=getattr(request.user, "pk", None),
|
||||
shorten=True,
|
||||
)
|
||||
return JsonResponse(
|
||||
{
|
||||
"target_url": target,
|
||||
"short_url": display if display != target else "",
|
||||
"display_url": display,
|
||||
"shortened": display != target,
|
||||
}
|
||||
)
|
||||
Reference in New Issue
Block a user