Add shared DOCX style module + campaign tools
_styles.py: Centralized typography, spacing, and formatting for all 26 DOCX generators. Calibri 9.5pt body, 1.15 line spacing, navy headings, consistent signature blocks, page numbers, PW footer. All generators will be migrated to use this instead of defining their own styles. Campaign tools: - campaign_template.html: Styled email template for Listmonk campaigns - populate_deficiency_list.py: Populates Listmonk with FCC deficiency data - send_test_campaigns.py: Sends test emails with real carrier data Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
justin
parent
c9881868dd
commit
463c180444
4 changed files with 731 additions and 0 deletions
69
scripts/campaign_template.html
Normal file
69
scripts/campaign_template.html
Normal file
|
|
@ -0,0 +1,69 @@
|
||||||
|
<!DOCTYPE html><html lang="en"><head><meta charset="UTF-8"><meta name="viewport" content="width=device-width,initial-scale=1"><style>@media only screen and (max-width:600px){.pw-wrap{width:100%!important;border-radius:0!important;}.pw-pad{padding:24px 16px!important;}}body,table,td,p,a{-webkit-text-size-adjust:100%;-ms-text-size-adjust:100%;}table{border-collapse:collapse!important;}img{border:0;outline:none;text-decoration:none;}</style></head><body style="margin:0;padding:0;background:#eef0f3;">
|
||||||
|
<center>
|
||||||
|
<table role="presentation" width="100%" cellpadding="0" cellspacing="0" style="background:#eef0f3;"><tr><td style="padding:24px 10px;">
|
||||||
|
<table role="presentation" class="pw-wrap" width="620" cellpadding="0" cellspacing="0" style="margin:0 auto;border-radius:10px;overflow:hidden;background:#fff;">
|
||||||
|
|
||||||
|
<!-- Header -->
|
||||||
|
<tr><td style="background:#1a2744;padding:24px 28px;">
|
||||||
|
<img src="https://performancewest.net/images/logo.png" alt="Performance West" style="height:44px;margin-bottom:10px;display:block" />
|
||||||
|
<h1 style="color:#fff;margin:0;font-size:22px;font-weight:700;font-family:Inter,system-ui,sans-serif;">FCC Compliance Alert</h1>
|
||||||
|
<p style="color:#94a3b8;margin:6px 0 0;font-size:13px;font-family:Inter,system-ui,sans-serif;">Automated compliance check for your FCC filings</p>
|
||||||
|
</td></tr>
|
||||||
|
|
||||||
|
<!-- Body -->
|
||||||
|
<tr><td class="pw-pad" style="padding:28px;font-family:Inter,system-ui,sans-serif;color:#1f2937;">
|
||||||
|
|
||||||
|
<p style="font-size:15px;margin:0 0 18px;line-height:1.5;">Hi {{ .Subscriber.Name }},</p>
|
||||||
|
|
||||||
|
<p style="font-size:14px;line-height:1.7;margin:0 0 18px;">Our automated compliance monitoring has detected <strong>potential issues</strong> with the FCC filings for <strong>{{ .Subscriber.Attribs.company }}</strong> (FRN: {{ .Subscriber.Attribs.fcc_frn }}).</p>
|
||||||
|
|
||||||
|
<!-- Issues Box -->
|
||||||
|
<table role="presentation" width="100%" cellpadding="0" cellspacing="0" style="margin:22px 0;"><tr><td style="background:#fef2f2;border:2px solid #fca5a5;border-radius:10px;padding:18px;">
|
||||||
|
<h3 style="margin:0 0 12px;font-size:15px;color:#991b1b;font-weight:700;font-family:Inter,sans-serif;">⚠ Issues Detected:</h3>
|
||||||
|
<div style="font-size:13px;color:#7f1d1d;line-height:2;font-family:Inter,sans-serif;">
|
||||||
|
{{ .Subscriber.Attribs.issues_html }}
|
||||||
|
</div>
|
||||||
|
</td></tr></table>
|
||||||
|
|
||||||
|
<p style="font-size:14px;line-height:1.7;margin:0 0 18px;">These compliance gaps could result in <strong>FCC enforcement action</strong>, fines, or removal from the Robocall Mitigation Database — which would effectively disconnect your carrier from the US phone network.</p>
|
||||||
|
|
||||||
|
<!-- CTA Box -->
|
||||||
|
<table role="presentation" width="100%" cellpadding="0" cellspacing="0" style="margin:22px 0;"><tr><td style="background:#fff7ed;border:2px solid #f97316;border-radius:10px;padding:18px;text-align:center;">
|
||||||
|
<p style="font-size:14px;color:#9a3412;margin:0 0 6px;font-weight:600;font-family:Inter,sans-serif;">Run a free compliance check to see the full picture</p>
|
||||||
|
<p style="font-size:12px;color:#9a3412;margin:0 0 14px;font-family:Inter,sans-serif;">Then we can start fixing these immediately</p>
|
||||||
|
<a href="https://performancewest.net/tools/fcc-compliance-check?frn={{ .Subscriber.Attribs.fcc_frn }}" style="display:inline-block;padding:14px 40px;background:#f97316;color:#fff;font-weight:700;border-radius:8px;text-decoration:none;font-size:15px;font-family:Inter,sans-serif;">Free Compliance Check →</a>
|
||||||
|
</td></tr></table>
|
||||||
|
|
||||||
|
<!-- Info Table -->
|
||||||
|
<table role="presentation" width="100%" cellpadding="0" cellspacing="0" style="margin:22px 0;font-size:13px;font-family:Inter,sans-serif;">
|
||||||
|
<tr style="border-bottom:1px solid #e5e7eb;">
|
||||||
|
<td style="padding:10px 0;color:#6b7280;">FRN</td>
|
||||||
|
<td style="padding:10px 0;font-weight:600;text-align:right;">{{ .Subscriber.Attribs.fcc_frn }}</td>
|
||||||
|
</tr>
|
||||||
|
<tr style="border-bottom:1px solid #e5e7eb;">
|
||||||
|
<td style="padding:10px 0;color:#6b7280;">STIR/SHAKEN Status</td>
|
||||||
|
<td style="padding:10px 0;font-weight:600;text-align:right;">{{ .Subscriber.Attribs.implementation }}</td>
|
||||||
|
</tr>
|
||||||
|
<tr>
|
||||||
|
<td style="padding:10px 0;color:#6b7280;">Last Recertified</td>
|
||||||
|
<td style="padding:10px 0;font-weight:600;text-align:right;">{{ .Subscriber.Attribs.last_recertified }}</td>
|
||||||
|
</tr>
|
||||||
|
</table>
|
||||||
|
|
||||||
|
<!-- Help Box -->
|
||||||
|
<table role="presentation" width="100%" cellpadding="0" cellspacing="0" style="margin:18px 0;"><tr><td style="background:#f0f4f8;border-radius:8px;padding:16px;font-size:13px;color:#374151;line-height:1.6;font-family:Inter,sans-serif;">
|
||||||
|
<strong>Need help?</strong> Reply to this email or call us at <strong>(888) 411-0383</strong>. We offer a <strong>free compliance assessment</strong> for all FCC-registered carriers.
|
||||||
|
</td></tr></table>
|
||||||
|
|
||||||
|
</td></tr>
|
||||||
|
|
||||||
|
<!-- Footer -->
|
||||||
|
<tr><td style="padding:16px 28px;background:#f8fafc;border-top:1px solid #e5e7eb;font-size:11px;color:#9ca3af;text-align:center;font-family:Inter,sans-serif;">
|
||||||
|
<p style="margin:0;">Performance West Inc. · Cheyenne, WY · <a href="https://performancewest.net" style="color:#6b7280;">performancewest.net</a></p>
|
||||||
|
<p style="margin:6px 0 0;"><a href="{{ UnsubscribeURL }}" style="color:#6b7280;">Unsubscribe</a></p>
|
||||||
|
</td></tr>
|
||||||
|
|
||||||
|
</table>
|
||||||
|
</td></tr></table>
|
||||||
|
</center>
|
||||||
|
</body></html>
|
||||||
307
scripts/document_gen/templates/_styles.py
Normal file
307
scripts/document_gen/templates/_styles.py
Normal file
|
|
@ -0,0 +1,307 @@
|
||||||
|
"""Shared document styles for all Performance West DOCX generators.
|
||||||
|
|
||||||
|
Import this module in every generator to ensure consistent typography,
|
||||||
|
spacing, and formatting across all compliance documents.
|
||||||
|
|
||||||
|
Usage:
|
||||||
|
from scripts.document_gen.templates._styles import (
|
||||||
|
PW, apply_doc_defaults, add_body, add_heading, add_bullets,
|
||||||
|
add_signature_block, add_page_numbers, add_horizontal_rule,
|
||||||
|
)
|
||||||
|
|
||||||
|
All generators should call apply_doc_defaults(doc) first, then use
|
||||||
|
the helper functions instead of building paragraphs manually.
|
||||||
|
"""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import logging
|
||||||
|
from typing import Optional
|
||||||
|
|
||||||
|
LOG = logging.getLogger("document_gen.styles")
|
||||||
|
|
||||||
|
try:
|
||||||
|
from docx import Document
|
||||||
|
from docx.shared import Pt, Inches, RGBColor, Emu
|
||||||
|
from docx.enum.text import WD_ALIGN_PARAGRAPH
|
||||||
|
from docx.oxml.ns import qn
|
||||||
|
from docx.oxml import OxmlElement
|
||||||
|
except ImportError:
|
||||||
|
LOG.warning("python-docx not installed")
|
||||||
|
Document = None
|
||||||
|
|
||||||
|
|
||||||
|
# ═══════════════════════════════════════════════════════════════════════════
|
||||||
|
# Color palette
|
||||||
|
# ═══════════════════════════════════════════════════════════════════════════
|
||||||
|
|
||||||
|
class PW:
|
||||||
|
"""Performance West brand colors and typography constants."""
|
||||||
|
NAVY = RGBColor(0x1A, 0x27, 0x44) if Document else None
|
||||||
|
DARK_GRAY = RGBColor(0x37, 0x41, 0x51) if Document else None
|
||||||
|
MEDIUM_GRAY = RGBColor(0x6B, 0x72, 0x80) if Document else None
|
||||||
|
LIGHT_GRAY = RGBColor(0x9C, 0xA3, 0xAF) if Document else None
|
||||||
|
BLACK = RGBColor(0x00, 0x00, 0x00) if Document else None
|
||||||
|
GREEN = RGBColor(0x05, 0x96, 0x69) if Document else None
|
||||||
|
RED = RGBColor(0xDC, 0x26, 0x26) if Document else None
|
||||||
|
|
||||||
|
# Typography
|
||||||
|
FONT_FAMILY = "Calibri"
|
||||||
|
BODY_SIZE = Pt(9.5) if Document else None
|
||||||
|
BODY_SMALL = Pt(8.5) if Document else None
|
||||||
|
HEADING_1_SIZE = Pt(13) if Document else None
|
||||||
|
HEADING_2_SIZE = Pt(11) if Document else None
|
||||||
|
HEADING_3_SIZE = Pt(10) if Document else None
|
||||||
|
TITLE_SIZE = Pt(15) if Document else None
|
||||||
|
SUBTITLE_SIZE = Pt(10) if Document else None
|
||||||
|
FOOTER_SIZE = Pt(7.5) if Document else None
|
||||||
|
SIGNATURE_SIZE = Pt(9.5) if Document else None
|
||||||
|
|
||||||
|
# Spacing
|
||||||
|
BODY_AFTER = Pt(8) if Document else None
|
||||||
|
HEADING_BEFORE = Pt(16) if Document else None
|
||||||
|
HEADING_AFTER = Pt(6) if Document else None
|
||||||
|
BULLET_AFTER = Pt(4) if Document else None
|
||||||
|
LINE_SPACING = 1.15
|
||||||
|
|
||||||
|
# Margins
|
||||||
|
MARGIN_TOP = Inches(0.9) if Document else None
|
||||||
|
MARGIN_BOTTOM = Inches(0.9) if Document else None
|
||||||
|
MARGIN_LEFT = Inches(1.1) if Document else None
|
||||||
|
MARGIN_RIGHT = Inches(1.1) if Document else None
|
||||||
|
|
||||||
|
|
||||||
|
# ═══════════════════════════════════════════════════════════════════════════
|
||||||
|
# Document setup
|
||||||
|
# ═══════════════════════════════════════════════════════════════════════════
|
||||||
|
|
||||||
|
|
||||||
|
def apply_doc_defaults(doc, title: str = "", entity_name: str = "") -> None:
|
||||||
|
"""Apply standard margins, default font, and optional header/footer."""
|
||||||
|
for section in doc.sections:
|
||||||
|
section.top_margin = PW.MARGIN_TOP
|
||||||
|
section.bottom_margin = PW.MARGIN_BOTTOM
|
||||||
|
section.left_margin = PW.MARGIN_LEFT
|
||||||
|
section.right_margin = PW.MARGIN_RIGHT
|
||||||
|
|
||||||
|
# Set default font on the document's style
|
||||||
|
style = doc.styles["Normal"]
|
||||||
|
style.font.name = PW.FONT_FAMILY
|
||||||
|
style.font.size = PW.BODY_SIZE
|
||||||
|
style.font.color.rgb = PW.DARK_GRAY
|
||||||
|
style.paragraph_format.space_after = PW.BODY_AFTER
|
||||||
|
style.paragraph_format.line_spacing = PW.LINE_SPACING
|
||||||
|
|
||||||
|
|
||||||
|
def add_page_numbers(doc) -> None:
|
||||||
|
"""Add centered page numbers to the document footer."""
|
||||||
|
for section in doc.sections:
|
||||||
|
footer = section.footer
|
||||||
|
footer.is_linked_to_previous = False
|
||||||
|
p = footer.paragraphs[0] if footer.paragraphs else footer.add_paragraph()
|
||||||
|
p.alignment = WD_ALIGN_PARAGRAPH.CENTER
|
||||||
|
|
||||||
|
# Page X
|
||||||
|
run = p.add_run()
|
||||||
|
run.font.size = PW.FOOTER_SIZE
|
||||||
|
run.font.color.rgb = PW.LIGHT_GRAY
|
||||||
|
run.font.name = PW.FONT_FAMILY
|
||||||
|
|
||||||
|
fc_begin = OxmlElement("w:fldChar")
|
||||||
|
fc_begin.set(qn("w:fldCharType"), "begin")
|
||||||
|
run._element.append(fc_begin)
|
||||||
|
|
||||||
|
r2 = p.add_run()
|
||||||
|
r2.font.size = PW.FOOTER_SIZE
|
||||||
|
r2.font.color.rgb = PW.LIGHT_GRAY
|
||||||
|
instr = OxmlElement("w:instrText")
|
||||||
|
instr.set(qn("xml:space"), "preserve")
|
||||||
|
instr.text = " PAGE "
|
||||||
|
r2._element.append(instr)
|
||||||
|
|
||||||
|
r3 = p.add_run()
|
||||||
|
fc_end = OxmlElement("w:fldChar")
|
||||||
|
fc_end.set(qn("w:fldCharType"), "end")
|
||||||
|
r3._element.append(fc_end)
|
||||||
|
|
||||||
|
|
||||||
|
def add_pw_footer(doc, entity_name: str = "") -> None:
|
||||||
|
"""Add 'Prepared by Performance West Inc.' footer with page numbers."""
|
||||||
|
for section in doc.sections:
|
||||||
|
footer = section.footer
|
||||||
|
footer.is_linked_to_previous = False
|
||||||
|
p = footer.paragraphs[0] if footer.paragraphs else footer.add_paragraph()
|
||||||
|
p.alignment = WD_ALIGN_PARAGRAPH.CENTER
|
||||||
|
|
||||||
|
# "Prepared by Performance West Inc." text
|
||||||
|
run = p.add_run("Prepared by Performance West Inc.")
|
||||||
|
run.font.size = PW.FOOTER_SIZE
|
||||||
|
run.font.color.rgb = PW.LIGHT_GRAY
|
||||||
|
run.font.name = PW.FONT_FAMILY
|
||||||
|
run.italic = True
|
||||||
|
|
||||||
|
# Add separator and page number
|
||||||
|
sep = p.add_run(" \u2014 Page ")
|
||||||
|
sep.font.size = PW.FOOTER_SIZE
|
||||||
|
sep.font.color.rgb = PW.LIGHT_GRAY
|
||||||
|
sep.font.name = PW.FONT_FAMILY
|
||||||
|
|
||||||
|
# Page number field
|
||||||
|
fc_begin = OxmlElement("w:fldChar")
|
||||||
|
fc_begin.set(qn("w:fldCharType"), "begin")
|
||||||
|
r_page = p.add_run()
|
||||||
|
r_page.font.size = PW.FOOTER_SIZE
|
||||||
|
r_page.font.color.rgb = PW.LIGHT_GRAY
|
||||||
|
r_page._element.append(fc_begin)
|
||||||
|
|
||||||
|
r_instr = p.add_run()
|
||||||
|
r_instr.font.size = PW.FOOTER_SIZE
|
||||||
|
instr = OxmlElement("w:instrText")
|
||||||
|
instr.set(qn("xml:space"), "preserve")
|
||||||
|
instr.text = " PAGE "
|
||||||
|
r_instr._element.append(instr)
|
||||||
|
|
||||||
|
r_end = p.add_run()
|
||||||
|
fc_end = OxmlElement("w:fldChar")
|
||||||
|
fc_end.set(qn("w:fldCharType"), "end")
|
||||||
|
r_end._element.append(fc_end)
|
||||||
|
|
||||||
|
|
||||||
|
# ═══════════════════════════════════════════════════════════════════════════
|
||||||
|
# Content helpers
|
||||||
|
# ═══════════════════════════════════════════════════════════════════════════
|
||||||
|
|
||||||
|
|
||||||
|
def add_title(doc, text: str, subtitle: str = "") -> None:
|
||||||
|
"""Add a document title (centered, navy, large)."""
|
||||||
|
p = doc.add_paragraph()
|
||||||
|
p.alignment = WD_ALIGN_PARAGRAPH.CENTER
|
||||||
|
p.paragraph_format.space_after = Pt(2)
|
||||||
|
run = p.add_run(text)
|
||||||
|
run.font.size = PW.TITLE_SIZE
|
||||||
|
run.bold = True
|
||||||
|
run.font.color.rgb = PW.NAVY
|
||||||
|
run.font.name = PW.FONT_FAMILY
|
||||||
|
|
||||||
|
if subtitle:
|
||||||
|
p2 = doc.add_paragraph()
|
||||||
|
p2.alignment = WD_ALIGN_PARAGRAPH.CENTER
|
||||||
|
p2.paragraph_format.space_after = Pt(12)
|
||||||
|
run2 = p2.add_run(subtitle)
|
||||||
|
run2.font.size = PW.SUBTITLE_SIZE
|
||||||
|
run2.italic = True
|
||||||
|
run2.font.color.rgb = PW.MEDIUM_GRAY
|
||||||
|
run2.font.name = PW.FONT_FAMILY
|
||||||
|
|
||||||
|
|
||||||
|
def add_heading(doc, text: str, level: int = 1) -> None:
|
||||||
|
"""Add a section heading (navy, bold)."""
|
||||||
|
p = doc.add_paragraph()
|
||||||
|
size = {1: PW.HEADING_1_SIZE, 2: PW.HEADING_2_SIZE, 3: PW.HEADING_3_SIZE}.get(level, PW.HEADING_2_SIZE)
|
||||||
|
before = PW.HEADING_BEFORE if level == 1 else Pt(10)
|
||||||
|
|
||||||
|
p.paragraph_format.space_before = before
|
||||||
|
p.paragraph_format.space_after = PW.HEADING_AFTER
|
||||||
|
run = p.add_run(text)
|
||||||
|
run.bold = True
|
||||||
|
run.font.size = size
|
||||||
|
run.font.color.rgb = PW.NAVY
|
||||||
|
run.font.name = PW.FONT_FAMILY
|
||||||
|
|
||||||
|
|
||||||
|
def add_body(doc, text: str, bold: bool = False, italic: bool = False,
|
||||||
|
size=None, color=None, alignment=None) -> None:
|
||||||
|
"""Add a body paragraph."""
|
||||||
|
p = doc.add_paragraph()
|
||||||
|
p.paragraph_format.space_after = PW.BODY_AFTER
|
||||||
|
p.paragraph_format.line_spacing = PW.LINE_SPACING
|
||||||
|
if alignment:
|
||||||
|
p.alignment = alignment
|
||||||
|
run = p.add_run(text)
|
||||||
|
run.font.size = size or PW.BODY_SIZE
|
||||||
|
run.font.name = PW.FONT_FAMILY
|
||||||
|
run.font.color.rgb = color or PW.DARK_GRAY
|
||||||
|
run.bold = bold
|
||||||
|
run.italic = italic
|
||||||
|
|
||||||
|
|
||||||
|
def add_field_value(doc, label: str, value: str) -> None:
|
||||||
|
"""Add a label: value pair on one line."""
|
||||||
|
p = doc.add_paragraph()
|
||||||
|
p.paragraph_format.space_after = Pt(3)
|
||||||
|
p.paragraph_format.line_spacing = PW.LINE_SPACING
|
||||||
|
|
||||||
|
run_label = p.add_run(f"{label}: ")
|
||||||
|
run_label.font.size = PW.BODY_SIZE
|
||||||
|
run_label.font.name = PW.FONT_FAMILY
|
||||||
|
run_label.font.color.rgb = PW.MEDIUM_GRAY
|
||||||
|
run_label.bold = False
|
||||||
|
|
||||||
|
run_value = p.add_run(value)
|
||||||
|
run_value.font.size = PW.BODY_SIZE
|
||||||
|
run_value.font.name = PW.FONT_FAMILY
|
||||||
|
run_value.font.color.rgb = PW.DARK_GRAY
|
||||||
|
run_value.bold = True
|
||||||
|
|
||||||
|
|
||||||
|
def add_bullets(doc, items: list[str], indent: float = 0.25) -> None:
|
||||||
|
"""Add a bulleted list."""
|
||||||
|
for item in items:
|
||||||
|
p = doc.add_paragraph(style="List Bullet")
|
||||||
|
p.paragraph_format.left_indent = Inches(indent)
|
||||||
|
p.paragraph_format.space_after = PW.BULLET_AFTER
|
||||||
|
p.paragraph_format.line_spacing = PW.LINE_SPACING
|
||||||
|
p.clear()
|
||||||
|
run = p.add_run(item)
|
||||||
|
run.font.size = PW.BODY_SIZE
|
||||||
|
run.font.name = PW.FONT_FAMILY
|
||||||
|
run.font.color.rgb = PW.DARK_GRAY
|
||||||
|
|
||||||
|
|
||||||
|
def add_horizontal_rule(doc) -> None:
|
||||||
|
"""Add a thin navy horizontal rule."""
|
||||||
|
p = doc.add_paragraph()
|
||||||
|
p.paragraph_format.space_before = Pt(6)
|
||||||
|
p.paragraph_format.space_after = Pt(6)
|
||||||
|
pPr = p._p.get_or_add_pPr()
|
||||||
|
pBdr = OxmlElement("w:pBdr")
|
||||||
|
bottom = OxmlElement("w:bottom")
|
||||||
|
bottom.set(qn("w:val"), "single")
|
||||||
|
bottom.set(qn("w:sz"), "4")
|
||||||
|
bottom.set(qn("w:space"), "1")
|
||||||
|
bottom.set(qn("w:color"), "1A2744")
|
||||||
|
pBdr.append(bottom)
|
||||||
|
pPr.append(pBdr)
|
||||||
|
|
||||||
|
|
||||||
|
def add_signature_block(
|
||||||
|
doc,
|
||||||
|
signer_name: str = "",
|
||||||
|
signer_title: str = "",
|
||||||
|
entity_name: str = "",
|
||||||
|
date_str: str = "",
|
||||||
|
include_perjury: bool = False,
|
||||||
|
) -> None:
|
||||||
|
"""Add a standardized signature block."""
|
||||||
|
if include_perjury:
|
||||||
|
add_body(doc, (
|
||||||
|
"I declare under penalty of perjury under the laws of the "
|
||||||
|
"United States of America that the foregoing is true and correct."
|
||||||
|
), italic=True, size=PW.BODY_SMALL)
|
||||||
|
|
||||||
|
# Signature line
|
||||||
|
add_body(doc, "", size=Pt(20)) # spacer
|
||||||
|
p = doc.add_paragraph()
|
||||||
|
p.paragraph_format.space_after = Pt(2)
|
||||||
|
run = p.add_run("_" * 45)
|
||||||
|
run.font.size = PW.SIGNATURE_SIZE
|
||||||
|
run.font.name = PW.FONT_FAMILY
|
||||||
|
run.font.color.rgb = PW.LIGHT_GRAY
|
||||||
|
|
||||||
|
if signer_name:
|
||||||
|
add_body(doc, signer_name, bold=True, size=PW.SIGNATURE_SIZE)
|
||||||
|
if signer_title:
|
||||||
|
add_body(doc, signer_title, size=PW.BODY_SMALL, color=PW.MEDIUM_GRAY)
|
||||||
|
if entity_name:
|
||||||
|
add_body(doc, entity_name, size=PW.BODY_SMALL, color=PW.MEDIUM_GRAY)
|
||||||
|
if date_str:
|
||||||
|
add_body(doc, f"Date: {date_str}", size=PW.BODY_SMALL, color=PW.MEDIUM_GRAY)
|
||||||
234
scripts/populate_deficiency_list.py
Normal file
234
scripts/populate_deficiency_list.py
Normal file
|
|
@ -0,0 +1,234 @@
|
||||||
|
#!/usr/bin/env python3
|
||||||
|
"""Populate Listmonk with FCC RMD deficiency data for campaign sends.
|
||||||
|
|
||||||
|
Creates a dedicated list and upserts every carrier that has major/critical
|
||||||
|
audit deficiencies. Sets subscriber attributes with issues_html so the
|
||||||
|
campaign template can render personalized issue lists.
|
||||||
|
|
||||||
|
Skips carriers tagged as is_customer=true.
|
||||||
|
"""
|
||||||
|
import json
|
||||||
|
import subprocess
|
||||||
|
import sys
|
||||||
|
import time
|
||||||
|
import urllib.request
|
||||||
|
import urllib.error
|
||||||
|
import base64
|
||||||
|
|
||||||
|
AUTH = base64.b64encode(b"api:6X1rKPea61N4rZ1S65Hx5zvqzbCj30F6nvEe9oVGH_Y").decode()
|
||||||
|
API = "http://localhost:9100/api"
|
||||||
|
|
||||||
|
ISSUE_LABELS = {
|
||||||
|
"ss_partial_note": "STIR/SHAKEN partial implementation \u2014 upstream provider not named",
|
||||||
|
"ss_vsp_no_shaken": "Voice Service Provider without STIR/SHAKEN implementation",
|
||||||
|
"conflicting_classification": "Unusual provider classification \u2014 may need correction",
|
||||||
|
"missing_kyc": "Missing Know Your Customer (KYC) procedures",
|
||||||
|
"missing_material_change": "Missing 10-business-day material change update commitment",
|
||||||
|
"missing_dno": "Missing Do-Not-Originate (DNO) list enforcement",
|
||||||
|
"missing_traceback": "Missing 24-hour traceback response commitment",
|
||||||
|
"missing_recertification": "Missing annual recertification acknowledgment",
|
||||||
|
"missing_perjury": "Missing perjury declaration",
|
||||||
|
"missing_mitigation": "Missing robocall mitigation program details",
|
||||||
|
"no_classification": "No provider classification selected",
|
||||||
|
"ss_intermediate_complete": "Intermediate provider claims Complete STIR/SHAKEN",
|
||||||
|
"xref_name_mismatch": "Business name not found in certification document",
|
||||||
|
"xref_old_document": "Outdated certification document",
|
||||||
|
"xref_ss_mismatch": "STIR/SHAKEN status mismatch between RMD and document",
|
||||||
|
"missing_stir_shaken": "Missing STIR/SHAKEN implementation details",
|
||||||
|
"missing_enforcement": "Missing enforcement history disclosure",
|
||||||
|
"missing_provider_id": "Missing provider identification",
|
||||||
|
"no_recert_date": "No recertification date on file",
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def api_call(method, path, data=None):
|
||||||
|
payload = json.dumps(data).encode() if data else None
|
||||||
|
req = urllib.request.Request(
|
||||||
|
API + path, data=payload, method=method,
|
||||||
|
headers={"Content-Type": "application/json", "Authorization": "Basic " + AUTH},
|
||||||
|
)
|
||||||
|
return json.loads(urllib.request.urlopen(req).read())
|
||||||
|
|
||||||
|
|
||||||
|
def build_issues_html(structured_json, pdf_json):
|
||||||
|
"""Merge structured + PDF checks, return HTML list of major/critical issues."""
|
||||||
|
all_checks = []
|
||||||
|
for cj in [structured_json, pdf_json]:
|
||||||
|
try:
|
||||||
|
checks = json.loads(cj)
|
||||||
|
if isinstance(checks, list):
|
||||||
|
all_checks.extend(checks)
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
|
||||||
|
severity_order = {"critical": 0, "major": 1, "minor": 2}
|
||||||
|
all_checks.sort(key=lambda c: severity_order.get(c.get("severity", "minor"), 2))
|
||||||
|
|
||||||
|
items = []
|
||||||
|
ids = []
|
||||||
|
seen = set()
|
||||||
|
for c in all_checks:
|
||||||
|
cid = c.get("id", "")
|
||||||
|
if cid in seen:
|
||||||
|
continue
|
||||||
|
seen.add(cid)
|
||||||
|
sev = c.get("severity", "minor")
|
||||||
|
if sev == "minor":
|
||||||
|
continue
|
||||||
|
label = ISSUE_LABELS.get(cid, c.get("label", cid))
|
||||||
|
items.append("<li>" + label + "</li>")
|
||||||
|
ids.append(cid)
|
||||||
|
|
||||||
|
if not items:
|
||||||
|
for c in all_checks:
|
||||||
|
cid = c.get("id", "")
|
||||||
|
label = ISSUE_LABELS.get(cid, c.get("label", cid))
|
||||||
|
items.append("<li>" + label + "</li>")
|
||||||
|
ids.append(cid)
|
||||||
|
|
||||||
|
html = '<ul style="margin:0;padding:0 0 0 16px">' + "".join(items[:6]) + "</ul>"
|
||||||
|
return html, ids
|
||||||
|
|
||||||
|
|
||||||
|
def main():
|
||||||
|
# Query carriers with deficiencies
|
||||||
|
result = subprocess.run([
|
||||||
|
"docker", "exec", "performancewest-api-postgres-1", "psql", "-U", "pw",
|
||||||
|
"performancewest", "-t", "-A", "-F", "||", "-c",
|
||||||
|
"""SELECT a.frn, a.business_name, a.total_deficiencies, a.severity,
|
||||||
|
r.contact_email, r.contact_name, r.implementation,
|
||||||
|
r.last_recertified::text, r.rmd_number,
|
||||||
|
a.structured_checks::text, a.pdf_checks::text
|
||||||
|
FROM fcc_rmd_audit_results a
|
||||||
|
JOIN fcc_rmd r ON r.frn = a.frn
|
||||||
|
WHERE a.total_deficiencies > 0
|
||||||
|
AND a.severity IN ('major', 'critical')
|
||||||
|
AND r.contact_email IS NOT NULL
|
||||||
|
AND r.removed_from_rmd = FALSE
|
||||||
|
ORDER BY a.total_deficiencies DESC""",
|
||||||
|
], capture_output=True, text=True)
|
||||||
|
|
||||||
|
lines = [l for l in result.stdout.strip().split("\n") if l.strip()]
|
||||||
|
print(f"Found {len(lines)} carriers with major/critical deficiencies")
|
||||||
|
|
||||||
|
# Find or create list
|
||||||
|
lists_resp = api_call("GET", "/lists")
|
||||||
|
deficiency_list_id = None
|
||||||
|
for l in lists_resp.get("data", {}).get("results", []):
|
||||||
|
if "Deficiency" in l["name"] and "2026" in l["name"]:
|
||||||
|
deficiency_list_id = l["id"]
|
||||||
|
break
|
||||||
|
|
||||||
|
if not deficiency_list_id:
|
||||||
|
resp = api_call("POST", "/lists", {
|
||||||
|
"name": "FCC RMD Deficiency Alert 2026",
|
||||||
|
"type": "public",
|
||||||
|
"optin": "single",
|
||||||
|
"tags": ["fcc", "rmd", "deficiency", "2026"],
|
||||||
|
})
|
||||||
|
deficiency_list_id = resp.get("data", {}).get("id")
|
||||||
|
print(f"Created list: {deficiency_list_id}")
|
||||||
|
else:
|
||||||
|
print(f"Using existing list: {deficiency_list_id}")
|
||||||
|
|
||||||
|
added = 0
|
||||||
|
updated = 0
|
||||||
|
skipped_customer = 0
|
||||||
|
errors = 0
|
||||||
|
|
||||||
|
for line in lines:
|
||||||
|
parts = line.split("||")
|
||||||
|
if len(parts) < 11:
|
||||||
|
continue
|
||||||
|
|
||||||
|
frn = parts[0].strip()
|
||||||
|
company = parts[1].strip()
|
||||||
|
deficiency_count = int(parts[2].strip() or 0)
|
||||||
|
severity = parts[3].strip()
|
||||||
|
email = parts[4].strip().lower()
|
||||||
|
contact_name = parts[5].strip()
|
||||||
|
implementation = parts[6].strip()
|
||||||
|
last_recert = parts[7].strip()
|
||||||
|
rmd_number = parts[8].strip()
|
||||||
|
structured_json = parts[9].strip()
|
||||||
|
pdf_json = parts[10].strip()
|
||||||
|
|
||||||
|
if not email or "@" not in email:
|
||||||
|
continue
|
||||||
|
|
||||||
|
issues_html, issue_ids = build_issues_html(structured_json, pdf_json)
|
||||||
|
|
||||||
|
attribs = {
|
||||||
|
"company": company,
|
||||||
|
"fcc_frn": frn,
|
||||||
|
"rmd_number": rmd_number,
|
||||||
|
"severity": severity,
|
||||||
|
"deficiency_count": deficiency_count,
|
||||||
|
"issues_html": issues_html,
|
||||||
|
"issue_ids": ",".join(issue_ids),
|
||||||
|
"implementation": implementation,
|
||||||
|
"last_recertified": last_recert,
|
||||||
|
}
|
||||||
|
|
||||||
|
try:
|
||||||
|
resp = api_call("POST", "/subscribers", {
|
||||||
|
"email": email,
|
||||||
|
"name": contact_name or company,
|
||||||
|
"status": "enabled",
|
||||||
|
"lists": [deficiency_list_id],
|
||||||
|
"attribs": attribs,
|
||||||
|
"preconfirm_subscriptions": True,
|
||||||
|
})
|
||||||
|
added += 1
|
||||||
|
except urllib.error.HTTPError as e:
|
||||||
|
err = e.read().decode()
|
||||||
|
if "already exists" in err:
|
||||||
|
try:
|
||||||
|
search = api_call("GET", "/subscribers?query=subscribers.email%3D%27" + email.replace("'", "") + "%27&per_page=1")
|
||||||
|
results = search.get("data", {}).get("results", [])
|
||||||
|
if results:
|
||||||
|
sub_id = results[0]["id"]
|
||||||
|
existing = results[0].get("attribs", {})
|
||||||
|
|
||||||
|
if existing.get("is_customer"):
|
||||||
|
skipped_customer += 1
|
||||||
|
continue
|
||||||
|
|
||||||
|
existing.update(attribs)
|
||||||
|
api_call("PUT", f"/subscribers/{sub_id}", {
|
||||||
|
"email": email,
|
||||||
|
"name": contact_name or results[0].get("name", company),
|
||||||
|
"attribs": existing,
|
||||||
|
"status": "enabled",
|
||||||
|
})
|
||||||
|
api_call("PUT", "/subscribers/lists", {
|
||||||
|
"ids": [sub_id],
|
||||||
|
"action": "add",
|
||||||
|
"target_list_ids": [deficiency_list_id],
|
||||||
|
"status": "confirmed",
|
||||||
|
})
|
||||||
|
updated += 1
|
||||||
|
except Exception as ex:
|
||||||
|
errors += 1
|
||||||
|
else:
|
||||||
|
errors += 1
|
||||||
|
|
||||||
|
if (added + updated) % 200 == 0 and (added + updated) > 0:
|
||||||
|
print(f" Progress: {added} added, {updated} updated, {skipped_customer} skipped")
|
||||||
|
|
||||||
|
time.sleep(0.05) # rate limit
|
||||||
|
|
||||||
|
print(f"\nDone:")
|
||||||
|
print(f" Added: {added}")
|
||||||
|
print(f" Updated: {updated}")
|
||||||
|
print(f" Customers skipped: {skipped_customer}")
|
||||||
|
print(f" Errors: {errors}")
|
||||||
|
print(f" List ID: {deficiency_list_id}")
|
||||||
|
|
||||||
|
search = api_call("GET", f"/subscribers?list_id={deficiency_list_id}&per_page=1")
|
||||||
|
total = search.get("data", {}).get("total", 0)
|
||||||
|
print(f" Total subscribers on list: {total}")
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
main()
|
||||||
121
scripts/send_test_campaigns.py
Normal file
121
scripts/send_test_campaigns.py
Normal file
|
|
@ -0,0 +1,121 @@
|
||||||
|
#!/usr/bin/env python3
|
||||||
|
"""Send test compliance gap emails to justin@performancewest.net using real carrier data."""
|
||||||
|
import json
|
||||||
|
import smtplib
|
||||||
|
import subprocess
|
||||||
|
from email.mime.multipart import MIMEMultipart
|
||||||
|
from email.mime.text import MIMEText
|
||||||
|
|
||||||
|
ISSUE_LABELS = {
|
||||||
|
"ss_partial_note": "STIR/SHAKEN partial implementation \u2014 upstream provider not named in filing",
|
||||||
|
"ss_vsp_no_shaken": "Voice Service Provider without STIR/SHAKEN implementation",
|
||||||
|
"conflicting_classification": "Unusual provider classification \u2014 may need correction",
|
||||||
|
"missing_kyc": "Missing Know Your Customer (KYC) procedures",
|
||||||
|
"missing_material_change": "Missing 10-business-day material change update commitment",
|
||||||
|
"missing_dno": "Missing Do-Not-Originate (DNO) list enforcement",
|
||||||
|
"missing_traceback": "Missing 24-hour traceback response commitment",
|
||||||
|
"missing_recertification": "Missing annual recertification acknowledgment",
|
||||||
|
"missing_perjury": "Missing perjury declaration",
|
||||||
|
"missing_mitigation": "Missing robocall mitigation program details",
|
||||||
|
"no_classification": "No provider classification selected",
|
||||||
|
"ss_intermediate_complete": "Intermediate provider claims Complete STIR/SHAKEN",
|
||||||
|
}
|
||||||
|
|
||||||
|
# Query real carrier data
|
||||||
|
result = subprocess.run([
|
||||||
|
"docker", "exec", "performancewest-api-postgres-1", "psql", "-U", "pw",
|
||||||
|
"performancewest", "-t", "-A", "-F", "|", "-c",
|
||||||
|
"""SELECT a.frn, a.business_name, a.total_deficiencies, a.severity,
|
||||||
|
r.contact_email, r.implementation, r.last_recertified::text,
|
||||||
|
a.structured_checks::text, a.pdf_checks::text
|
||||||
|
FROM fcc_rmd_audit_results a
|
||||||
|
JOIN fcc_rmd r ON r.frn = a.frn
|
||||||
|
WHERE a.total_deficiencies > 0
|
||||||
|
AND a.severity IN ('major', 'critical')
|
||||||
|
AND r.contact_email IS NOT NULL
|
||||||
|
AND r.removed_from_rmd = FALSE
|
||||||
|
ORDER BY a.total_deficiencies DESC
|
||||||
|
LIMIT 5"""
|
||||||
|
], capture_output=True, text=True)
|
||||||
|
|
||||||
|
with open("/tmp/campaign_template.html") as f:
|
||||||
|
template = f.read()
|
||||||
|
|
||||||
|
for line in result.stdout.strip().split("\n"):
|
||||||
|
if not line.strip():
|
||||||
|
continue
|
||||||
|
parts = line.split("|")
|
||||||
|
if len(parts) < 9:
|
||||||
|
continue
|
||||||
|
|
||||||
|
frn = parts[0].strip()
|
||||||
|
company = parts[1].strip()
|
||||||
|
deficiency_count = parts[2].strip()
|
||||||
|
severity = parts[3].strip()
|
||||||
|
email_orig = parts[4].strip()
|
||||||
|
implementation = parts[5].strip()
|
||||||
|
last_recert = parts[6].strip()
|
||||||
|
structured_json = parts[7].strip()
|
||||||
|
pdf_json = parts[8].strip()
|
||||||
|
|
||||||
|
# Merge both structured and PDF checks, prioritize major/critical
|
||||||
|
all_checks = []
|
||||||
|
for cj in [structured_json, pdf_json]:
|
||||||
|
try:
|
||||||
|
checks = json.loads(cj)
|
||||||
|
if isinstance(checks, list):
|
||||||
|
all_checks.extend(checks)
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
|
||||||
|
# Sort: critical first, then major, then minor — skip minor for email
|
||||||
|
severity_order = {"critical": 0, "major": 1, "minor": 2}
|
||||||
|
all_checks.sort(key=lambda c: severity_order.get(c.get("severity", "minor"), 2))
|
||||||
|
|
||||||
|
items = []
|
||||||
|
seen_ids = set()
|
||||||
|
for c in all_checks:
|
||||||
|
cid = c.get("id", "")
|
||||||
|
if cid in seen_ids:
|
||||||
|
continue
|
||||||
|
seen_ids.add(cid)
|
||||||
|
sev = c.get("severity", "minor")
|
||||||
|
if sev == "minor":
|
||||||
|
continue # skip minor issues in email
|
||||||
|
label = ISSUE_LABELS.get(cid, c.get("label", cid))
|
||||||
|
items.append(f"<li>{label}</li>")
|
||||||
|
|
||||||
|
if not items:
|
||||||
|
# If all were minor, include them anyway
|
||||||
|
for c in all_checks:
|
||||||
|
cid = c.get("id", "")
|
||||||
|
label = ISSUE_LABELS.get(cid, c.get("label", cid))
|
||||||
|
items.append(f"<li>{label}</li>")
|
||||||
|
|
||||||
|
issues_html = '<ul style="margin:0;padding:0 0 0 16px">' + "".join(items[:6]) + "</ul>"
|
||||||
|
|
||||||
|
# Build the email from template
|
||||||
|
body = template
|
||||||
|
body = body.replace("{{ .Subscriber.Name }}", "there")
|
||||||
|
body = body.replace("{{ .Subscriber.Attribs.company }}", company)
|
||||||
|
body = body.replace("{{ .Subscriber.Attribs.fcc_frn }}", frn)
|
||||||
|
body = body.replace("{{ .Subscriber.Attribs.issues_html }}", issues_html)
|
||||||
|
body = body.replace("{{ .Subscriber.Attribs.implementation }}", implementation or "Unknown")
|
||||||
|
body = body.replace("{{ .Subscriber.Attribs.last_recertified }}", last_recert or "Unknown")
|
||||||
|
body = body.replace("{{ UnsubscribeURL }}", "#")
|
||||||
|
|
||||||
|
msg = MIMEMultipart("alternative")
|
||||||
|
msg["From"] = "Performance West <noreply@performancewest.net>"
|
||||||
|
msg["To"] = "justin@performancewest.net"
|
||||||
|
msg["Subject"] = f"[TEST] FCC Compliance Alert - {company}"
|
||||||
|
msg["Reply-To"] = "info@performancewest.net"
|
||||||
|
msg.attach(MIMEText(body, "html"))
|
||||||
|
|
||||||
|
with smtplib.SMTP("email-smtp.us-east-2.amazonaws.com", 587, timeout=30) as s:
|
||||||
|
s.starttls()
|
||||||
|
s.login("AKIAYEWLMNWPHSHQWCRD", "BKrUBud+KjyaRA1RiA26FFu1R+hqR4cpFShwbZf7RUzG")
|
||||||
|
s.send_message(msg)
|
||||||
|
|
||||||
|
print(f"Sent: {company} ({frn}) - {deficiency_count} issues (would go to {email_orig})")
|
||||||
|
|
||||||
|
print("\nDone - all test emails sent to justin@performancewest.net")
|
||||||
Loading…
Add table
Add a link
Reference in a new issue