crypto-compliance-agent / src /utils /report_builder.py
arjitmat's picture
Deploy Aegis: multi-agent crypto compliance platform
6c54d57 verified
"""Compliance report builder — markdown and PDF output."""
import io
from datetime import datetime
from fpdf import FPDF
# Brand colours
DARK = (4, 13, 18) # page-bg inspired dark
NAVY = (15, 25, 40) # header bar
TEAL = (0, 201, 167) # primary accent
TEAL_DARK = (0, 140, 116)
WHITE = (255, 255, 255)
LIGHT_GREY = (245, 247, 250)
MID_GREY = (120, 130, 140)
DARK_TEXT = (30, 35, 45)
# Risk colours
RISK_COLOURS = {
"Low": (52, 211, 153),
"Medium": (212, 169, 66),
"High": (207, 123, 46),
"Critical": (217, 69, 69),
}
class ReportBuilder:
"""Generates compliance reports in markdown and PDF formats."""
def build_markdown(self, result: dict) -> str:
"""Build a full compliance report in markdown format."""
scores = result.get("risk_scores", {})
now = datetime.utcnow().strftime("%Y-%m-%d %H:%M UTC")
s = [] # sections
s.append("# Aegis \u2014 Compliance Intelligence Report")
s.append(f"*Generated: {now}*\n")
# 1. Executive Summary
overall = scores.get("overall", 0)
label = self._risk_label(overall)
s.append("## 1. Executive Summary")
s.append(f"**Overall Risk Score: {overall}/100 ({label})**\n")
s.append("| Risk Area | Score | Level |")
s.append("|-----------|-------|-------|")
for area in ["licensing", "aml", "token", "disclosure", "operational"]:
v = scores.get(area, 0)
s.append(f"| {area.title()} | {v}/100 | {self._risk_label(v)} |")
s.append("")
top_risks = result.get("top_risks", [])
if top_risks:
s.append("**Top Risks:**")
for i, r in enumerate(top_risks[:3], 1):
s.append(f"{i}. {r}")
s.append("")
# 2. Business Assessment
s.append("## 2. Business Activity Assessment")
s.append(result.get("business_summary", "*No summary available.*"))
s.append("")
# 3. Jurisdictions
s.append("## 3. Jurisdiction Analysis")
for jx in result.get("jurisdictions", []):
s.append(f"### {jx.get('name', '')}")
s.append(f"**Status:** {jx.get('status', '')}")
if jx.get("detail"):
s.append(jx["detail"])
s.append("")
# 4. Token
s.append("## 4. Token Classification")
tc = result.get("token_classification", {})
if tc:
s.append(f"**Howey Test:** {tc.get('howey_result', 'Not assessed')}")
s.append(f"**MiCA Type:** {tc.get('mica_type', 'Not assessed')}")
s.append(f"**Implications:** {tc.get('implications', 'N/A')}")
s.append("")
# 5. AML
s.append("## 5. AML/KYC Requirements")
aml = result.get("aml_requirements", {})
for item in aml.get("needed", []):
s.append(f"- {item}")
if aml.get("missing"):
s.append("\n**Gaps:**")
for item in aml["missing"]:
s.append(f"- {item}")
s.append("")
# 6. Licensing
s.append("## 6. Licensing Roadmap")
roadmap = result.get("licensing_roadmap", [])
if roadmap:
for i, step in enumerate(roadmap, 1):
s.append(f"{i}. **{step.get('action', '')}** ({step.get('jurisdiction', '')})")
s.append(f" Timeline: {step.get('timeline', 'TBD')} | Cost: {step.get('cost', 'TBD')}")
else:
s.append("*No licensing requirements identified for current activities.*")
s.append("")
# 7. Cases
s.append("## 7. Relevant Enforcement Cases")
for case in result.get("enforcement_cases", [])[:5]:
name = case.get("case_name", case.get("title", ""))
s.append(f"### {name}")
if case.get("outcome"):
s.append(f"**Outcome:** {case['outcome']}")
if case.get("key_lesson"):
s.append(f"**Lesson:** {case['key_lesson']}")
s.append("")
# 8. Actions
s.append("## 8. Priority Action Plan")
actions = result.get("priority_actions", [])
if actions:
for bucket, label_str in [
("[CRITICAL]", "**Immediate (0-30 days):**"),
("[HIGH]", "\n**30-60 days:**"),
("[URGENT]", ""),
("[MEDIUM]", "\n**60-90 days:**"),
("[LOW]", "\n**90+ days:**"),
]:
items = [a for a in actions if bucket in a]
if items and label_str:
s.append(label_str)
for a in items:
clean = a
for tag in ["[CRITICAL] ", "[HIGH] ", "[URGENT] ", "[MEDIUM] ", "[LOW] "]:
clean = clean.replace(tag, "")
s.append(f"- {clean}")
s.append("")
# 9. Disclaimer
s.append("## 9. Disclaimer")
s.append(
"This report is generated by an AI system and provides general regulatory "
"information only. It does not constitute legal advice. Always consult "
"qualified legal counsel before making compliance decisions."
)
return "\n".join(s)
def build_pdf(self, result: dict) -> bytes:
"""Build a professional PDF compliance report."""
pdf = _AegisPDF()
pdf.set_auto_page_break(auto=True, margin=20)
pdf.add_page()
scores = result.get("risk_scores", {})
overall = scores.get("overall", 0)
label = self._risk_label(overall)
now = datetime.utcnow().strftime("%Y-%m-%d %H:%M UTC")
rc = RISK_COLOURS.get(label, TEAL)
# ── TITLE BLOCK ──
pdf.set_font("Helvetica", "B", 22)
pdf.set_text_color(*NAVY)
pdf.cell(0, 14, "Aegis Compliance Report", new_x="LMARGIN", new_y="NEXT")
pdf.set_font("Helvetica", "", 9)
pdf.set_text_color(*MID_GREY)
pdf.cell(0, 5, f"Generated {now} | AI-powered regulatory analysis", new_x="LMARGIN", new_y="NEXT")
pdf.ln(8)
# ── RISK SCORE BANNER ──
# Coloured banner
y = pdf.get_y()
pdf.set_fill_color(*rc)
pdf.rect(10, y, 190, 22, "F")
# Lighter inner area
pdf.set_fill_color(255, 255, 255)
pdf.rect(12, y + 1, 186, 20, "F")
# Score text
pdf.set_xy(16, y + 3)
pdf.set_font("Helvetica", "B", 16)
pdf.set_text_color(*rc)
pdf.cell(50, 7, f"{overall}/100", new_x="END")
pdf.set_font("Helvetica", "B", 12)
pdf.cell(40, 7, label.upper(), new_x="END")
# Sub-scores on same line
pdf.set_font("Helvetica", "", 8)
pdf.set_text_color(*MID_GREY)
sub_parts = []
for area in ["licensing", "aml", "token", "disclosure", "operational"]:
v = scores.get(area, 0)
sub_parts.append(f"{area.title()}: {v}")
pdf.cell(0, 7, " ".join(sub_parts))
# Score bar
bar_y = y + 14
pdf.set_fill_color(230, 235, 240)
pdf.rect(16, bar_y, 180, 4, "F")
bar_w = max(2, 180 * overall / 100)
pdf.set_fill_color(*rc)
pdf.rect(16, bar_y, bar_w, 4, "F")
pdf.set_y(y + 28)
# ── TOP RISKS ──
top_risks = result.get("top_risks", [])
if top_risks and top_risks[0] != "No significant risks identified":
pdf._heading("Key Risks")
for risk in top_risks[:3]:
pdf._bullet(risk, color=RISK_COLOURS.get("High", TEAL_DARK))
pdf.ln(2)
# ── BUSINESS SUMMARY ──
pdf._heading("Business Assessment")
summary = result.get("business_summary", "")
if summary:
pdf._text(summary)
else:
pdf._text("No business summary generated.")
# ── JURISDICTIONS ──
pdf._heading("Jurisdiction Analysis")
jurisdictions = result.get("jurisdictions", [])
if jurisdictions:
for jx in jurisdictions:
name = jx.get("name", "Unknown")
status = jx.get("status", "")
# Status badge
pdf._badge(name, status)
if jx.get("key_requirements") and jx["key_requirements"] != "See detailed analysis":
pdf._text(f"Requirements: {jx['key_requirements'][:300]}")
if jx.get("gaps") and jx["gaps"] != "None identified":
pdf._text(f"Gaps: {jx['gaps'][:300]}")
if jx.get("detail"):
pdf._text(jx["detail"][:600])
pdf.ln(2)
else:
pdf._text("No jurisdiction analysis available.")
# ── TOKEN CLASSIFICATION ──
pdf._heading("Token Classification")
tc = result.get("token_classification", {})
if tc and tc.get("howey_result"):
pdf._kv("Howey Test", tc.get("howey_result", "Not assessed"))
pdf._kv("MiCA Type", tc.get("mica_type", "Not assessed"))
if tc.get("implications") and tc["implications"] != "N/A":
pdf._text(f"Implications: {tc['implications'][:300]}")
else:
pdf._text("No token provided for classification.")
pdf.ln(2)
# ── AML/KYC ──
pdf._heading("AML/KYC Requirements")
aml = result.get("aml_requirements", {})
needed = aml.get("needed", [])
missing = aml.get("missing", [])
if needed:
pdf._sub("Required:")
for item in needed[:8]:
pdf._bullet(item)
if missing:
pdf._sub("Gaps Identified:")
for item in missing:
pdf._bullet(item, color=(207, 123, 46))
if not needed and not missing:
pdf._text("No AML/KYC gaps identified.")
pdf.ln(2)
# ── LICENSING ROADMAP ──
pdf._heading("Licensing Roadmap")
roadmap = result.get("licensing_roadmap", [])
if roadmap:
for i, step in enumerate(roadmap, 1):
action = step.get("action", "")
jx = step.get("jurisdiction", "")
timeline = step.get("timeline", "TBD")
cost = step.get("cost", "TBD")
pdf._step(i, f"{action} - {jx}", f"Timeline: {timeline} | Cost: {cost}")
else:
pdf._text("No additional licences identified for current activities and jurisdictions.")
pdf.ln(2)
# ── ENFORCEMENT CASES ──
pdf._heading("Relevant Enforcement Cases")
cases = result.get("enforcement_cases", [])
if cases:
for case in cases[:4]:
name = case.get("case_name", case.get("title", ""))
if not name:
continue
pdf._sub(name)
if case.get("outcome"):
pdf._text(f"Outcome: {case['outcome'][:300]}")
if case.get("key_lesson"):
pdf._text(f"Lesson: {case['key_lesson'][:300]}")
pdf.ln(1)
else:
pdf._text("No closely analogous enforcement cases identified.")
# ── ACTION PLAN ──
pdf._heading("Priority Action Plan")
actions = result.get("priority_actions", [])
if actions:
for bucket, title in [
("[CRITICAL]", "IMMEDIATE (0-30 DAYS)"),
("[HIGH]", "HIGH PRIORITY (30-60 DAYS)"),
("[URGENT]", None),
("[MEDIUM]", "MEDIUM PRIORITY (60-90 DAYS)"),
("[LOW]", "ONGOING"),
]:
items = [a for a in actions if bucket in a]
if not items:
continue
if title:
pdf._priority_label(title, bucket)
for a in items:
clean = a
for tag in ["[CRITICAL] ", "[HIGH] ", "[URGENT] ", "[MEDIUM] ", "[LOW] "]:
clean = clean.replace(tag, "")
pdf._bullet(clean)
pdf.ln(2)
else:
pdf._text("No action items generated.")
# ── DISCLAIMER ──
pdf.ln(4)
pdf.set_draw_color(*TEAL)
pdf.set_line_width(0.3)
pdf.line(10, pdf.get_y(), 200, pdf.get_y())
pdf.ln(4)
pdf.set_font("Helvetica", "I", 7)
pdf.set_text_color(*MID_GREY)
pdf.set_x(10)
pdf.multi_cell(
0, 3.5,
_AegisPDF._safe(
"DISCLAIMER: This report is generated by an AI system and provides general regulatory "
"information only. It does not constitute legal advice and should not be relied upon as such. "
"The analysis is based on publicly available regulatory texts and may not reflect the most "
"recent amendments. Always consult qualified legal counsel in each relevant jurisdiction "
"before making compliance decisions. Aegis accepts no liability for actions taken based "
"on this output."
),
)
buf = io.BytesIO()
pdf.output(buf)
return buf.getvalue()
@staticmethod
def _risk_label(score: int) -> str:
if score <= 25:
return "Low"
elif score <= 50:
return "Medium"
elif score <= 75:
return "High"
return "Critical"
@staticmethod
def _risk_gauge(score: int) -> str:
filled = score // 5
empty = 20 - filled
return f"`[{'#' * filled}{'.' * empty}]` {score}/100"
class _AegisPDF(FPDF):
"""Professional PDF with Aegis branding."""
def header(self):
# Navy top bar
self.set_fill_color(*NAVY)
self.rect(0, 0, 210, 6, "F")
# Teal accent stripe
self.set_fill_color(*TEAL)
self.rect(0, 6, 210, 1.5, "F")
self.ln(10)
def footer(self):
self.set_y(-12)
self.set_font("Helvetica", "", 7)
self.set_text_color(*MID_GREY)
self.cell(0, 8, f"Aegis Compliance Report | Page {self.page_no()}/{{nb}} | Confidential", align="C")
@staticmethod
def _safe(text: str) -> str:
# Replace common unicode chars with latin-1 equivalents before encoding
text = text.replace("\u2014", " - ") # em-dash
text = text.replace("\u2013", "-") # en-dash
text = text.replace("\u2019", "'") # right single quote
text = text.replace("\u2018", "'") # left single quote
text = text.replace("\u201c", '"') # left double quote
text = text.replace("\u201d", '"') # right double quote
text = text.replace("\u2026", "...") # ellipsis
text = text.replace("\u2022", "-") # bullet
text = text.replace("\u00b7", "-") # middle dot
text = text.replace("\u2192", "->") # right arrow
text = text.replace("\u26a0\ufe0f", "[!]") # warning emoji
text = text.replace("\u2713", "[ok]") # checkmark
return text.encode("latin-1", "replace").decode("latin-1")
def _heading(self, text: str):
"""Section heading with teal left bar."""
self.ln(4)
y = self.get_y()
# Teal left accent bar
self.set_fill_color(*TEAL)
self.rect(10, y, 2.5, 8, "F")
# Heading text
self.set_x(16)
self.set_font("Helvetica", "B", 12)
self.set_text_color(*NAVY)
self.cell(0, 8, self._safe(text), new_x="LMARGIN", new_y="NEXT")
pdf_y = self.get_y()
# Subtle line
self.set_draw_color(220, 225, 230)
self.set_line_width(0.2)
self.line(10, pdf_y, 200, pdf_y)
self.ln(3)
def _sub(self, text: str):
"""Sub-heading."""
self.set_font("Helvetica", "B", 9)
self.set_text_color(*DARK_TEXT)
self.set_x(10)
self.cell(0, 6, self._safe(text), new_x="LMARGIN", new_y="NEXT")
def _text(self, text: str):
self.set_font("Helvetica", "", 9)
self.set_text_color(50, 55, 65)
self.set_x(10)
self.multi_cell(0, 4.5, self._safe(text[:800]))
self.ln(1.5)
def _bullet(self, text: str, color=None):
self.set_x(14)
# Coloured dot
y = self.get_y() + 1.5
c = color or TEAL_DARK
self.set_fill_color(*c)
self.ellipse(14, y, 2, 2, "F")
self.set_x(18)
self.set_font("Helvetica", "", 8.5)
self.set_text_color(50, 55, 65)
safe = self._safe(text[:350])
self.multi_cell(0, 4.5, safe)
def _kv(self, key: str, value: str):
"""Key-value pair on one line."""
self.set_x(10)
self.set_font("Helvetica", "B", 9)
self.set_text_color(*MID_GREY)
self.cell(35, 5, self._safe(key + ":"))
self.set_font("Helvetica", "", 9)
self.set_text_color(*DARK_TEXT)
self.multi_cell(0, 5, self._safe(value[:200]))
self.ln(1)
def _badge(self, name: str, status: str):
"""Jurisdiction name with coloured status badge."""
self.set_x(10)
self.set_font("Helvetica", "B", 10)
self.set_text_color(*NAVY)
self.cell(80, 6, self._safe(name))
# Status badge
badge_colors = {
"Non-compliant": (217, 69, 69),
"Action required": (207, 123, 46),
"Monitoring": (52, 211, 153),
}
bc = badge_colors.get(status, MID_GREY)
self.set_fill_color(*bc)
self.set_text_color(255, 255, 255)
self.set_font("Helvetica", "B", 7)
badge_w = self.get_string_width(self._safe(status.upper())) + 8
self.cell(badge_w, 5, self._safe(status.upper()), fill=True, align="C")
self.ln(7)
def _step(self, num: int, title: str, detail: str):
"""Numbered step for licensing roadmap."""
self.set_x(10)
# Number circle
y = self.get_y()
self.set_fill_color(*TEAL)
self.ellipse(10, y, 6, 6, "F")
self.set_xy(10, y)
self.set_font("Helvetica", "B", 8)
self.set_text_color(255, 255, 255)
self.cell(6, 6, str(num), align="C")
# Title
self.set_xy(18, y)
self.set_font("Helvetica", "B", 9)
self.set_text_color(*DARK_TEXT)
self.cell(0, 6, self._safe(title[:100]), new_x="LMARGIN", new_y="NEXT")
# Detail
self.set_x(18)
self.set_font("Helvetica", "", 8)
self.set_text_color(*MID_GREY)
self.cell(0, 5, self._safe(detail[:150]), new_x="LMARGIN", new_y="NEXT")
self.ln(2)
def _priority_label(self, text: str, bucket: str):
"""Priority category label with colour."""
colors = {
"[CRITICAL]": (217, 69, 69),
"[HIGH]": (207, 123, 46),
"[URGENT]": (207, 123, 46),
"[MEDIUM]": (212, 169, 66),
"[LOW]": (120, 130, 140),
}
c = colors.get(bucket, MID_GREY)
self.set_x(10)
y = self.get_y()
self.set_fill_color(*c)
self.rect(10, y, 60, 5, "F")
self.set_xy(12, y)
self.set_font("Helvetica", "B", 7)
self.set_text_color(255, 255, 255)
self.cell(56, 5, self._safe(text), align="L")
self.ln(7)