conversation-compliance-auditing

class ComplianceRule:
    def __init__(self, rule_id, name, category, severity):
        self.rule_id = rule_id
        self.name = name
        self.category = category  # regulatory, policy, ethical
        self.severity = severity  # critical, high, medium, low
        self.checks = []
 
    def add_check(self, check_function, description):
        self.checks.append({
            "function": check_function,
            "description": description
        })
 
    def evaluate(self, message, context):
        violations = []
        for check in self.checks:
            result = check["function"](message, context)
            if not result["passed"]:
                violations.append({
                    "rule_id": self.rule_id,
                    "rule_name": self.name,
                    "check": check["description"],
                    "severity": self.severity,
                    "details": result.get("details"),
                    "evidence": result.get("evidence")
                })
        return violations

def create_tcpa_rules():
    rules = []
 
    # Time restrictions
    time_rule = ComplianceRule(
        "TCPA-001", "Calling Hours", "regulatory", "critical"
    )
    time_rule.add_check(
        check_calling_hours,
        "Messages must be sent between 8am-9pm recipient local time"
    )
    rules.append(time_rule)
 
    # Consent verification
    consent_rule = ComplianceRule(
        "TCPA-002", "Prior Express Consent", "regulatory", "critical"
    )
    consent_rule.add_check(
        check_sms_consent,
        "SMS requires prior express written consent"
    )
    consent_rule.add_check(
        check_call_consent,
        "Calls require prior express consent for marketing"
    )
    rules.append(consent_rule)
 
    # Opt-out honoring
    optout_rule = ComplianceRule(
        "TCPA-003", "Opt-Out Compliance", "regulatory", "critical"
    )
    optout_rule.add_check(
        check_optout_honored,
        "Must honor opt-out requests immediately"
    )
    rules.append(optout_rule)
 
    return rules
 
def check_calling_hours(message, context):
    recipient_time = get_recipient_local_time(context.recipient)
    hour = recipient_time.hour
 
    if hour < 8 or hour >= 21:
        return {
            "passed": False,
            "details": f"Message scheduled for {recipient_time.strftime('%I:%M %p')} recipient time",
            "evidence": f"Recipient timezone: {context.recipient.timezone}"
        }
    return {"passed": True}

def create_canspam_rules():
    rules = []
 
    # Accurate headers
    header_rule = ComplianceRule(
        "CAN-SPAM-001", "Accurate Header Information", "regulatory", "critical"
    )
    header_rule.add_check(
        check_from_address,
        "From address must accurately identify sender"
    )
    header_rule.add_check(
        check_subject_line,
        "Subject line must not be deceptive"
    )
    rules.append(header_rule)
 
    # Ad identification
    ad_rule = ComplianceRule(
        "CAN-SPAM-002", "Advertisement Disclosure", "regulatory", "high"
    )
    ad_rule.add_check(
        check_ad_disclosure,
        "Commercial messages must be identifiable as advertisements"
    )
    rules.append(ad_rule)
 
    # Physical address
    address_rule = ComplianceRule(
        "CAN-SPAM-003", "Physical Address Required", "regulatory", "high"
    )
    address_rule.add_check(
        check_physical_address,
        "Must include valid physical postal address"
    )
    rules.append(address_rule)
 
    # Opt-out mechanism
    optout_rule = ComplianceRule(
        "CAN-SPAM-004", "Opt-Out Mechanism", "regulatory", "critical"
    )
    optout_rule.add_check(
        check_unsubscribe_link,
        "Must include clear opt-out mechanism"
    )
    rules.append(optout_rule)
 
    return rules

def create_policy_rules(company_config):
    rules = []
 
    # Pricing claims
    pricing_rule = ComplianceRule(
        "POLICY-001", "Accurate Pricing", "policy", "high"
    )
    pricing_rule.add_check(
        lambda m, c: check_pricing_accuracy(m, c, company_config),
        "Pricing claims must match current price list"
    )
    rules.append(pricing_rule)
 
    # Feature claims
    feature_rule = ComplianceRule(
        "POLICY-002", "Accurate Feature Claims", "policy", "high"
    )
    feature_rule.add_check(
        lambda m, c: check_feature_accuracy(m, c, company_config),
        "Feature claims must match product capabilities"
    )
    rules.append(feature_rule)
 
    # Competitor mentions
    competitor_rule = ComplianceRule(
        "POLICY-003", "Competitor References", "policy", "medium"
    )
    competitor_rule.add_check(
        check_competitor_claims,
        "Competitor claims must be verifiable"
    )
    rules.append(competitor_rule)
 
    # Promise restrictions
    promise_rule = ComplianceRule(
        "POLICY-004", "Unauthorized Promises", "policy", "high"
    )
    promise_rule.add_check(
        check_unauthorized_promises,
        "Cannot make commitments outside approval matrix"
    )
    rules.append(promise_rule)
 
    return rules

class ConversationAuditor:
    def __init__(self):
        self.rules = []
        self.rules.extend(create_tcpa_rules())
        self.rules.extend(create_canspam_rules())
        self.rules.extend(create_gdpr_rules())
        self.rules.extend(create_policy_rules(get_company_config()))
 
    def audit_message(self, message, context):
        """Audit a single message before sending"""
 
        all_violations = []
 
        for rule in self.rules:
            violations = rule.evaluate(message, context)
            all_violations.extend(violations)
 
        # Categorize by severity
        critical = [v for v in all_violations if v["severity"] == "critical"]
        high = [v for v in all_violations if v["severity"] == "high"]
 
        result = {
            "passed": len(critical) == 0,
            "violations": all_violations,
            "critical_count": len(critical),
            "high_count": len(high),
            "can_send": len(critical) == 0 and len(high) == 0,
            "requires_review": len(high) > 0
        }
 
        if not result["passed"]:
            log_compliance_violation(message, result)
 
        return result
 
    def audit_conversation(self, conversation):
        """Audit entire conversation for patterns"""
 
        conversation_violations = []
        for message in conversation.messages:
            if message.sender == "bot":
                result = self.audit_message(message, conversation)
                if result["violations"]:
                    conversation_violations.append({
                        "message_id": message.id,
                        "violations": result["violations"]
                    })
 
        return {
            "conversation_id": conversation.id,
            "total_violations": sum(len(v["violations"]) for v in conversation_violations),
            "messages_with_violations": len(conversation_violations),
            "details": conversation_violations
        }

def check_deceptive_claims(message, context):
    """Check for potentially deceptive content"""
 
    deceptive_patterns = [
        r"guarantee[d]?\s+(results|success|roi)",
        r"100%\s+(success|effective|guaranteed)",
        r"risk[- ]free",
        r"no\s+obligation",  # May need context
        r"limited\s+time\s+only",  # If false
        r"act\s+now\s+or\s+miss\s+out"  # Artificial urgency
    ]
 
    violations = []
    for pattern in deceptive_patterns:
        if re.search(pattern, message.text, re.IGNORECASE):
            violations.append({
                "type": "potentially_deceptive",
                "pattern": pattern,
                "match": re.search(pattern, message.text, re.IGNORECASE).group()
            })
 
    return {
        "passed": len(violations) == 0,
        "details": violations,
        "evidence": message.text
    }
 
def check_unauthorized_promises(message, context):
    """Check for promises that require approval"""
 
    promise_patterns = [
        (r"(we('ll| will)|I('ll| will))\s+waive", "fee_waiver"),
        (r"(discount|reduce|lower)\s+(the\s+)?price\s+by", "discount"),
        (r"(extend|lengthen)\s+(the\s+)?(trial|pilot)", "trial_extension"),
        (r"(custom|bespoke|tailored)\s+(feature|development)", "custom_dev"),
        (r"guaranteed?\s+(uptime|availability|sla)", "sla_guarantee")
    ]
 
    violations = []
    for pattern, promise_type in promise_patterns:
        if re.search(pattern, message.text, re.IGNORECASE):
            if not is_authorized(context.sender, promise_type):
                violations.append({
                    "type": "unauthorized_promise",
                    "promise_type": promise_type,
                    "requires_approval_from": get_approval_authority(promise_type)
                })
 
    return {
        "passed": len(violations) == 0,
        "details": violations
    }

def generate_compliance_report(time_period):
    """Generate compliance audit report"""
 
    violations = get_violations(time_period)
 
    report = {
        "period": time_period,
        "summary": {
            "total_messages_audited": count_audited_messages(time_period),
            "total_violations": len(violations),
            "violation_rate": len(violations) / count_audited_messages(time_period),
            "by_severity": count_by_severity(violations),
            "by_category": count_by_category(violations),
            "by_rule": count_by_rule(violations)
        },
        "trends": {
            "vs_previous_period": compare_to_previous(violations, time_period),
            "weekly_trend": calculate_weekly_trend(time_period)
        },
        "top_violations": get_top_violations(violations, n=10),
        "recommendations": generate_recommendations(violations)
    }
 
    return report
 
def generate_recommendations(violations):
    """Generate actionable recommendations from violations"""
 
    recommendations = []
 
    # Group violations by rule
    by_rule = defaultdict(list)
    for v in violations:
        by_rule[v["rule_id"]].append(v)
 
    for rule_id, rule_violations in by_rule.items():
        if len(rule_violations) > 10:  # Significant pattern
            recommendations.append({
                "priority": "high",
                "rule": rule_id,
                "issue": f"{len(rule_violations)} violations of {rule_id}",
                "action": get_remediation_action(rule_id),
                "impact": estimate_risk_impact(rule_violations)
            })
 
    return sorted(recommendations, key=lambda r: r["priority"])

def log_compliance_event(event_type, message, result, context):
    """Maintain audit trail for compliance"""
 
    audit_record = {
        "event_id": generate_uuid(),
        "timestamp": datetime.now(UTC),
        "event_type": event_type,
        "message_id": message.id,
        "conversation_id": context.conversation_id,
        "sender": context.sender,
        "recipient": context.recipient.id,
        "audit_result": {
            "passed": result["passed"],
            "violations": result["violations"],
            "action_taken": result.get("action_taken")
        },
        "message_hash": hash_message(message),  # For verification
        "metadata": {
            "bot_version": get_bot_version(),
            "rules_version": get_rules_version(),
            "channel": context.channel
        }
    }
 
    # Store with immutability guarantees
    compliance_log.insert(audit_record)
 
    return audit_record["event_id"]

def remediate_violation(message, violation):
    """Attempt automatic remediation of violations"""
 
    remediation_handlers = {
        "CAN-SPAM-003": add_physical_address,
        "CAN-SPAM-004": add_unsubscribe_link,
        "POLICY-001": correct_pricing,
        "POLICY-002": remove_inaccurate_claim
    }
 
    if violation["rule_id"] in remediation_handlers:
        handler = remediation_handlers[violation["rule_id"]]
        remediated = handler(message, violation)
        return {
            "remediated": True,
            "original": message.text,
            "corrected": remediated.text,
            "changes": diff_messages(message, remediated)
        }
 
    return {
        "remediated": False,
        "reason": "No auto-remediation available",
        "requires": "manual_review"
    }

Track:
- Violation rate (per 1000 messages)
- Critical violations (should be 0)
- Time to remediation
- Audit coverage (% of messages audited)
- False positive rate

Alert thresholds:
- Any critical violation: Immediate
- >1% violation rate: Same day
- Increasing trend: Weekly review