new-site/scripts/workers/services/ccpa_audit.py

"""CCPA/CPRA Compliance Audit handler (LLM-based).

Generates a California Consumer Privacy Act / California Privacy Rights Act
compliance audit report. Also used for the "data-mapping" service which
shares similar analysis.
"""

from __future__ import annotations

import os

from .base_handler import BaseServiceHandler

SERVICE_SYSTEM_PROMPT = """You are a compliance analyst at Performance West Inc.
generating a CCPA/CPRA Privacy Compliance Audit report.

RULES:
- Write in professional, clear business English
- Cite specific CCPA/CPRA sections (Cal. Civ. Code §§ 1798.100-1798.199.100)
- Reference CPRA amendments effective January 1, 2023
- Never provide legal advice — use "we recommend" not "you must"
- For each finding: what was found, regulation, risk level (Low/Medium/High/Critical), remediation
- Consider CCPA applicability thresholds ($25M revenue, 100k consumers, 50% revenue from data)
- Address all consumer rights: know, delete, opt-out, correct, limit use of sensitive PI
- Reference CPPA enforcement guidance where applicable
"""

SECTIONS = [
    {
        "name": "executive_summary",
        "prompt": (
            "Write a 200-word executive summary of the CCPA/CPRA audit findings. "
            "Include applicability determination, scope, overall compliance posture, "
            "and critical findings."
        ),
    },
    {
        "name": "data_inventory",
        "prompt": (
            "Analyze the organization's personal information inventory. Categories "
            "of PI collected, sources, purposes, third-party sharing, sensitive PI "
            "processing, retention periods, and cross-border transfers. Identify "
            "gaps in data mapping documentation."
        ),
    },
    {
        "name": "consumer_rights",
        "prompt": (
            "Assess compliance with each CCPA/CPRA consumer right: right to know, "
            "right to delete, right to opt-out of sale/sharing, right to correct, "
            "right to limit use of sensitive PI, right to non-discrimination. "
            "For each: process exists (Y/N), response timeline, verification method, gaps."
        ),
    },
    {
        "name": "privacy_notices",
        "prompt": (
            "Review privacy notices and disclosures: at-or-before-collection notice, "
            "privacy policy, financial incentive notices, opt-out preference signals "
            "compliance (GPC), 'Do Not Sell or Share' link. Assess content completeness "
            "and accessibility."
        ),
    },
    {
        "name": "vendor_management",
        "prompt": (
            "Review service provider and contractor agreements for CCPA/CPRA "
            "compliance. Assess: data processing agreements, flow-down obligations, "
            "audit rights, breach notification, and distinction between service "
            "providers, contractors, and third parties under CPRA."
        ),
    },
    {
        "name": "security_measures",
        "prompt": (
            "Evaluate reasonable security measures under CCPA § 1798.150 (private "
            "right of action for data breaches). Review technical and organizational "
            "safeguards, breach detection capabilities, and incident response readiness."
        ),
    },
    {
        "name": "remediation_plan",
        "prompt": (
            "Provide a prioritized remediation plan. For each finding: reference, "
            "risk level, recommended action, responsible party, timeline. Consider "
            "CPPA enforcement priorities and the cure period provisions."
        ),
    },
]


class CCPAAuditHandler(BaseServiceHandler):
    SERVICE_SLUG = "ccpa-audit"
    SERVICE_NAME = "CCPA/CPRA Compliance Audit"
    TEMPLATE_NAME = "ccpa_audit_template.docx"
    REQUIRES_LLM = True

    async def process(self, order_data: dict) -> list[str]:
        work_dir = self._make_work_dir()
        order_number = order_data["name"]
        context = self._extract_order_context(order_data)

        # data-mapping uses same handler with a different template
        items = order_data.get("items", [])
        item_code = items[0].get("item_code", "") if items else ""
        if item_code == "data-mapping":
            template_name = "data_mapping_template.docx"
        else:
            template_name = self.TEMPLATE_NAME

        template_path = self._get_template_path(template_name)
        docx_filename = self._output_filename(order_number, "docx")
        docx_path = os.path.join(work_dir, docx_filename)

        variables = {
            "order_number": order_number,
            "customer_name": order_data.get("customer_name", ""),
            "date": __import__("datetime").datetime.now().strftime("%B %d, %Y"),
            "service_name": self.SERVICE_NAME,
            "company_size": order_data.get("custom_company_size", "N/A"),
            "industry": order_data.get("custom_industry", "N/A"),
            "state": order_data.get("custom_state", "N/A"),
        }
        self._fill_template(template_path, variables, docx_path)

        sections = await self._generate_sections(
            SERVICE_SYSTEM_PROMPT, SECTIONS, context
        )
        self._add_sections_to_doc(docx_path, sections)

        pdf_path = self._convert_to_pdf(docx_path)
        return [docx_path, pdf_path]