고객 서비스 데이터 템플릿

This extraction is configured through the Zendesk Explore graphical user interface. There is no single script to execute. The configuration below outlines the components to add in the query builder.

**Dataset**: `Support: Tickets`

**Metrics**:
1.  `COUNT(Tickets)`

**Rows (Attributes)**:
1.  `Ticket ID`
2.  `Assignee name`
3.  `Ticket type`
4.  `Ticket priority`
5.  `Ticket channel`
6.  `SLA resolution time`
7.  `SLA breach status`
8.  `Full resolution time (min)`
9.  `MIN(Ticket created - Timestamp)`
10. `MIN(Ticket assigned - Timestamp)`
11. `MIN(First reply time - Timestamp)`
12. `MIN(Ticket solved - Timestamp)`
13. `MIN(Ticket reopened - Timestamp)`
14. `MIN(Satisfaction rating received - Timestamp)`
15. `MIN(Ticket closed - Timestamp)`
16. `MIN(Timestamp - Categorized and Prioritized)` (Custom calculated attribute)
17. `MIN(Timestamp - Info Requested)` (Custom calculated attribute)
18. `MIN(Timestamp - Customer Responded)` (Custom calculated attribute)
19. `MIN(Timestamp - Escalated)` (Custom calculated attribute)
20. `MIN(Timestamp - SLA Breach)` (Custom calculated attribute)

**Filters**:
1.  `Ticket created - Date`: Select a date range (e.g., last 180 days)

**Standard Calculated Attribute Formulas**:
*   `Timestamp - Categorized and Prioritized`: `IF ([Changes - Field name] = "priority" OR [Changes - Field name] = "type") THEN [Update - Timestamp] ENDIF`
*   `Timestamp - Info Requested`: `IF ([Changes - Field name] = "status" AND [Changes - New value] = "pending") THEN [Update - Timestamp] ENDIF`
*   `Timestamp - Customer Responded`: `IF ([Changes - Field name] = "status" AND [Changes - Previous value] = "pending" AND [Changes - New value] = "open") THEN [Update - Timestamp] ENDIF`
*   `Timestamp - Escalated`: `IF ([Changes - Field name] = "group") THEN [Update - Timestamp] ENDIF`
*   `Timestamp - SLA Breach`: `IF ([SLA breach status]=TRUE) THEN [SLA update - Timestamp] ENDIF`

import requests
import json
import time
import datetime
import csv

class=class="hl-string">"hl-comment"># --- Configuration ---
class=class="hl-string">"hl-comment"># Replace with your Zendesk details
ZENDESK_SUBDOMAIN = class="hl-string">'[your_zendesk_subdomain]'
ZENDESK_EMAIL = class="hl-string">'[your_zendesk_login_email]'
API_TOKEN = class="hl-string">'[your_zendesk_api_token]'

class=class="hl-string">"hl-comment"># Set the start time for the export (Unix epoch timestamp)
class=class="hl-string">"hl-comment"># Example: 1672531200 is Jan 1, 2023 00:00:00 GMT
START_TIMESTAMP = 1672531200

class=class="hl-string">"hl-comment"># --- Script --- 

BASE_URL = fclass="hl-string">'https://{ZENDESK_SUBDOMAIN}.zendesk.com/api/v2'
AUTH = (fclass="hl-string">'{ZENDESK_EMAIL}/token', API_TOKEN)
SOURCE_SYSTEM = class="hl-string">'Zendesk Support'
LAST_DATA_UPDATE = datetime.datetime.utcnow().isoformat() + class="hl-string">'Z'

CSV_FILENAME = class="hl-string">'zendesk_event_log.csv'
CSV_HEADERS = [
    class="hl-string">'ServiceRequest', class="hl-string">'ActivityName', class="hl-string">'StartTime', class="hl-string">'SourceSystem', class="hl-string">'LastDataUpdate',
    class="hl-string">'AssignedAgent', class="hl-string">'ServiceRequestType', class="hl-string">'Priority', class="hl-string">'CommunicationChannel',
    class="hl-string">'SlaTargetResolutionTime', class="hl-string">'IsSlaBreached', class="hl-string">'ServiceRequestResolutionTime'
]

def get_user_name(user_id, users_cache, session):
    if user_id in users_cache:
        return users_cache[user_id]
    if user_id is None:
        return None
    try:
        response = session.get(fclass="hl-string">'{BASE_URL}/users/{user_id}.json')
        response.raise_for_status()
        user_name = response.json().get(class="hl-string">'user', {}).get(class="hl-string">'name')
        users_cache[user_id] = user_name
        return user_name
    except requests.exceptions.RequestException as e:
        print(fclass="hl-string">'Warning: Could not fetch user {user_id}. Error: {e}')
        users_cache[user_id] = str(user_id)
        return str(user_id)

def format_duration(seconds):
    if seconds is None:
        return None
    return str(datetime.timedelta(seconds=seconds))

def main():
    session = requests.Session()
    session.auth = AUTH
    users_cache = {}
    event_log = []

    class=class="hl-string">"hl-comment"># Fetch tickets using the incremental export API
    url = fclass="hl-string">'{BASE_URL}/incremental/tickets/cursor.json?start_time={START_TIMESTAMP}'
    ticket_count = 0

    while url:
        print(fclass="hl-string">'Fetching tickets from: {url}')
        try:
            response = session.get(url)

            if response.status_code == 429:
                retry_after = int(response.headers.get(class="hl-string">'Retry-After', 60))
                print(fclass="hl-string">'Rate limit hit. Retrying after {retry_after} seconds.')
                time.sleep(retry_after)
                continue

            response.raise_for_status()
            data = response.json()
            tickets = data.get(class="hl-string">'tickets', [])

            for ticket in tickets:
                ticket_count += 1
                ticket_id = ticket[class="hl-string">'id']
                print(fclass="hl-string">'Processing ticket {ticket_id} ({ticket_count} total)...')

                class=class="hl-string">"hl-comment"># Fetch ticket metrics for SLA and resolution time
                metrics_response = session.get(fclass="hl-string">'{BASE_URL}/tickets/{ticket_id}/metrics.json')
                ticket_metrics = metrics_response.json().get(class="hl-string">'ticket_metric', {}) if metrics_response.ok else {}

                resolution_time_seconds = ticket_metrics.get(class="hl-string">'full_resolution_time_in_minutes', {}).get(class="hl-string">'business')
                if resolution_time_seconds:
                    resolution_time_seconds *= 60

                class=class="hl-string">"hl-comment"># Fetch ticket audits for events
                audits_url = fclass="hl-string">'{BASE_URL}/tickets/{ticket_id}/audits.json'
                audits_response = session.get(audits_url)
                audits = audits_response.json().get(class="hl-string">'audits', []) if audits_response.ok else []

                first_agent_reply_sent = False

                for audit in audits:
                    audit_created_at = audit[class="hl-string">'created_at']
                    base_event = {
                        class="hl-string">'ServiceRequest': ticket_id,
                        class="hl-string">'StartTime': audit_created_at,
                        class="hl-string">'SourceSystem': SOURCE_SYSTEM,
                        class="hl-string">'LastDataUpdate': LAST_DATA_UPDATE,
                        class="hl-string">'AssignedAgent': get_user_name(ticket.get(class="hl-string">'assignee_id'), users_cache, session),
                        class="hl-string">'ServiceRequestType': ticket.get(class="hl-string">'type'),
                        class="hl-string">'Priority': ticket.get(class="hl-string">'priority'),
                        class="hl-string">'CommunicationChannel': ticket.get(class="hl-string">'via', {}).get(class="hl-string">'channel'),
                        class="hl-string">'SlaTargetResolutionTime': None, class=class="hl-string">"hl-comment"># Requires parsing SLA policies, complex
                        class="hl-string">'IsSlaBreached': ticket_metrics.get(class="hl-string">'sla_breach_at') is not None,
                        class="hl-string">'ServiceRequestResolutionTime': format_duration(resolution_time_seconds)
                    }

                    for event in audit.get(class="hl-string">'events', []):
                        event_type = event.get(class="hl-string">'type')
                        class=class="hl-string">"hl-comment"># Service Request Created
                        if event_type == class="hl-string">'Create' and event.get(class="hl-string">'field_name') == class="hl-string">'id':
                            new_event = base_event.copy()
                            new_event[class="hl-string">'ActivityName'] = class="hl-string">'Service Request Created'
                            event_log.append(new_event)

                        class=class="hl-string">"hl-comment"># Initial Acknowledgment Sent (Proxy by Notification event)
                        if event_type == class="hl-string">'Notification':
                            new_event = base_event.copy()
                            new_event[class="hl-string">'ActivityName'] = class="hl-string">'Initial Acknowledgment Sent'
                            event_log.append(new_event)

                        class=class="hl-string">"hl-comment"># Request Categorized And Prioritized, Assigned, Status Changes
                        if event_type == class="hl-string">'Change':
                            field = event.get(class="hl-string">'field_name')
                            if field in [class="hl-string">'type', class="hl-string">'priority', class="hl-string">'custom_field_[your_category_field_id]']:
                                new_event = base_event.copy()
                                new_event[class="hl-string">'ActivityName'] = class="hl-string">'Request Categorized And Prioritized'
                                event_log.append(new_event)
                            
                            if field == class="hl-string">'assignee_id':
                                new_event = base_event.copy()
                                new_event[class="hl-string">'ActivityName'] = class="hl-string">'Request Assigned To Agent'
                                new_event[class="hl-string">'AssignedAgent'] = get_user_name(event.get(class="hl-string">'value'), users_cache, session)
                                event_log.append(new_event)

                            if field == class="hl-string">'group_id':
                                new_event = base_event.copy()
                                new_event[class="hl-string">'ActivityName'] = class="hl-string">'Internal Escalation Triggered'
                                event_log.append(new_event)

                            if field == class="hl-string">'status':
                                new_status = event.get(class="hl-string">'value')
                                if new_status == class="hl-string">'pending':
                                    new_event = base_event.copy()
                                    new_event[class="hl-string">'ActivityName'] = class="hl-string">'Information Requested From Customer'
                                    event_log.append(new_event)
                                elif new_status == class="hl-string">'solved':
                                    new_event = base_event.copy()
                                    new_event[class="hl-string">'ActivityName'] = class="hl-string">'Service Request Resolved'
                                    event_log.append(new_event)
                                    class=class="hl-string">"hl-comment"># Satisfaction Survey Sent (Proxy: occurs after solve)
                                    solve_time = datetime.datetime.fromisoformat(audit_created_at.replace(class="hl-string">'Z', class="hl-string">'+00:00'))
                                    survey_time = solve_time + datetime.timedelta(hours=1)
                                    survey_event = new_event.copy()
                                    survey_event[class="hl-string">'ActivityName'] = class="hl-string">'Satisfaction Survey Sent'
                                    survey_event[class="hl-string">'StartTime'] = survey_time.isoformat().replace(class="hl-string">'+00:00', class="hl-string">'Z')
                                    event_log.append(survey_event)
                                elif new_status == class="hl-string">'closed':
                                    new_event = base_event.copy()
                                    new_event[class="hl-string">'ActivityName'] = class="hl-string">'Service Request Closed'
                                    event_log.append(new_event)
                                elif new_status == class="hl-string">'open' and event.get(class="hl-string">'previous_value') in [class="hl-string">'solved', class="hl-string">'pending']:
                                    activity = class="hl-string">'Service Request Reopened' if event.get(class="hl-string">'previous_value') == class="hl-string">'solved' else class="hl-string">'Customer Responded'
                                    new_event = base_event.copy()
                                    new_event[class="hl-string">'ActivityName'] = activity
                                    event_log.append(new_event)
                            
                            if field == class="hl-string">'satisfaction_score':
                                new_event = base_event.copy()
                                new_event[class="hl-string">'ActivityName'] = class="hl-string">'Satisfaction Rating Received'
                                event_log.append(new_event)

                        class=class="hl-string">"hl-comment"># First Agent Public Reply Sent & Customer Responded
                        if event_type == class="hl-string">'Comment':
                            author_id = event.get(class="hl-string">'author_id')
                            author_details_response = session.get(fclass="hl-string">'{BASE_URL}/users/{author_id}.json')
                            author_role = author_details_response.json().get(class="hl-string">'user',{}).get(class="hl-string">'role', class="hl-string">'end-user') if author_details_response.ok else class="hl-string">'end-user'

                            if author_role in [class="hl-string">'agent', class="hl-string">'admin'] and event.get(class="hl-string">'public') and not first_agent_reply_sent:
                                new_event = base_event.copy()
                                new_event[class="hl-string">'ActivityName'] = class="hl-string">'First Agent Public Reply Sent'
                                event_log.append(new_event)
                                first_agent_reply_sent = True

                class=class="hl-string">"hl-comment"># SLA Breach Occurred
                if ticket_metrics.get(class="hl-string">'sla_breach_at'):
                    sla_breach_time = ticket_metrics.get(class="hl-string">'sla_breach_at')
                    new_event = base_event.copy()
                    new_event[class="hl-string">'ActivityName'] = class="hl-string">'SLA Breach Occurred'
                    new_event[class="hl-string">'StartTime'] = sla_breach_time
                    event_log.append(new_event)

            url = data.get(class="hl-string">'after_url')
            if not data.get(class="hl-string">'end_of_stream', False):
                time.sleep(1) class=class="hl-string">"hl-comment"># Be a good API citizen
            else:
                url = None

        except requests.exceptions.RequestException as e:
            print(fclass="hl-string">'An error occurred: {e}')
            break

    class=class="hl-string">"hl-comment"># Write to CSV
    print(fclass="hl-string">'Writing {len(event_log)} events to {CSV_FILENAME}...')
    with open(CSV_FILENAME, class="hl-string">'w', newline=class="hl-string">'', encoding=class="hl-string">'utf-8') as csvfile:
        writer = csv.DictWriter(csvfile, fieldnames=CSV_HEADERS)
        writer.writeheader()
        class=class="hl-string">"hl-comment"># Sort events by ticket and then by time for chronological order
        sorted_log = sorted(event_log, key=lambda x: (x[class="hl-string">'ServiceRequest'], x[class="hl-string">'StartTime']))
        for row in sorted_log:
            writer.writerow(row)
    print(class="hl-string">'Extraction complete.')

if __name__ == class="hl-string">'__main__':
    main()

import csv
import json
import os
import sys
import time
from datetime import datetime, timezone
from typing import Any, Dict, Iterable, List, Optional, Tuple

import requests

ZENDESK_SUBDOMAIN = os.environ.get(class="hl-string">"ZENDESK_SUBDOMAIN", class="hl-string">"[Configure based on your system]")
ZENDESK_EMAIL = os.environ.get(class="hl-string">"ZENDESK_EMAIL", class="hl-string">"[Configure based on your system]")
ZENDESK_API_TOKEN = os.environ.get(class="hl-string">"ZENDESK_API_TOKEN", class="hl-string">"[Configure based on your system]")
START_TIME = os.environ.get(class="hl-string">"START_TIME", class="hl-string">"[Configure based on your system]")
OUTPUT_FILE = os.environ.get(class="hl-string">"OUTPUT_FILE", class="hl-string">"zendesk_processmind_event_log.csv")
SOURCE_SYSTEM = class="hl-string">"Zendesk Support"

REQUIRED_COLUMNS = [
    class="hl-string">"ServiceRequest",
    class="hl-string">"ActivityName",
    class="hl-string">"StartTime",
    class="hl-string">"SourceSystem",
    class="hl-string">"LastDataUpdate",
]
RECOMMENDED_COLUMNS = [
    class="hl-string">"AssignedAgent",
    class="hl-string">"ServiceRequestType",
    class="hl-string">"Priority",
    class="hl-string">"CommunicationChannel",
    class="hl-string">"SlaTargetResolutionTime",
    class="hl-string">"IsSlaBreached",
]
ALL_COLUMNS = REQUIRED_COLUMNS + RECOMMENDED_COLUMNS


def fail(message: str) -> None:
    raise RuntimeError(message)


def parse_start_time(value: str) -> int:
    if value.isdigit():
        return int(value)
    try:
        return int(datetime.fromisoformat(value.replace(class="hl-string">"Z", class="hl-string">"+00:00")).timestamp())
    except ValueError as exc:
        raise RuntimeError(class="hl-string">"START_TIME must be a Unix timestamp or ISO 8601 timestamp") from exc


def iso_timestamp(value: Any) -> Optional[str]:
    if value is None or value == class="hl-string">"":
        return None
    if isinstance(value, (int, float)):
        return datetime.fromtimestamp(value, tz=timezone.utc).isoformat().replace(class="hl-string">"+00:00", class="hl-string">"Z")
    text = str(value)
    if text.isdigit():
        return datetime.fromtimestamp(int(text), tz=timezone.utc).isoformat().replace(class="hl-string">"+00:00", class="hl-string">"Z")
    return text


def now_utc() -> str:
    return datetime.now(timezone.utc).isoformat().replace(class="hl-string">"+00:00", class="hl-string">"Z")


def api_get(session: requests.Session, path: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
    url = fclass="hl-string">"https://{ZENDESK_SUBDOMAIN}.zendesk.com{path}"
    while True:
        response = session.get(url, params=params, timeout=60)
        if response.status_code == 429:
            retry_after = int(response.headers.get(class="hl-string">"Retry-After", class="hl-string">"5"))
            time.sleep(max(retry_after, 1))
            continue
        response.raise_for_status()
        return response.json()


def paged_get(session: requests.Session, path: str, params: Optional[Dict[str, Any]] = None) -> Iterable[Dict[str, Any]]:
    next_url: Optional[str] = fclass="hl-string">"https://{ZENDESK_SUBDOMAIN}.zendesk.com{path}"
    next_params = params
    while next_url:
        while True:
            response = session.get(next_url, params=next_params, timeout=60)
            if response.status_code == 429:
                retry_after = int(response.headers.get(class="hl-string">"Retry-After", class="hl-string">"5"))
                time.sleep(max(retry_after, 1))
                continue
            response.raise_for_status()
            payload = response.json()
            break
        for item in payload.get(class="hl-string">"tickets", payload.get(class="hl-string">"audits", payload.get(class="hl-string">"comments", payload.get(class="hl-string">"results", [])))):
            yield item
        next_url = payload.get(class="hl-string">"next_page")
        next_params = None


def get_user_name(session: requests.Session, user_id: Any, cache: Dict[str, str]) -> Optional[str]:
    if not user_id:
        return None
    key = str(user_id)
    if key in cache:
        return cache[key]
    try:
        user = api_get(session, fclass="hl-string">"/api/v2/users/{key}.json").get(class="hl-string">"user", {})
        name = user.get(class="hl-string">"name") or user.get(class="hl-string">"email") or key
        cache[key] = str(name)
        return cache[key]
    except requests.HTTPError:
        cache[key] = key
        return key


def get_group_name(session: requests.Session, group_id: Any, cache: Dict[str, str]) -> Optional[str]:
    if not group_id:
        return None
    key = str(group_id)
    if key in cache:
        return cache[key]
    try:
        group = api_get(session, fclass="hl-string">"/api/v2/groups/{key}.json").get(class="hl-string">"group", {})
        name = group.get(class="hl-string">"name") or key
        cache[key] = str(name)
        return cache[key]
    except requests.HTTPError:
        cache[key] = key
        return key


def ticket_field(ticket: Dict[str, Any], names: List[str]) -> Any:
    for name in names:
        if ticket.get(name) is not None:
            return ticket.get(name)
    return None


def base_row(ticket: Dict[str, Any], activity: str, event_time: Any, last_update: str, user_cache: Dict[str, str]) -> Dict[str, Any]:
    return {
        class="hl-string">"ServiceRequest": str(ticket.get(class="hl-string">"id")),
        class="hl-string">"ActivityName": activity,
        class="hl-string">"StartTime": iso_timestamp(event_time),
        class="hl-string">"SourceSystem": SOURCE_SYSTEM,
        class="hl-string">"LastDataUpdate": last_update,
        class="hl-string">"AssignedAgent": get_user_name(session, ticket.get(class="hl-string">"assignee_id"), user_cache),
        class="hl-string">"ServiceRequestType": ticket.get(class="hl-string">"type"),
        class="hl-string">"Priority": ticket.get(class="hl-string">"priority"),
        class="hl-string">"CommunicationChannel": ticket.get(class="hl-string">"via", {}).get(class="hl-string">"channel") if isinstance(ticket.get(class="hl-string">"via"), dict) else ticket.get(class="hl-string">"via"),
        class="hl-string">"SlaTargetResolutionTime": ticket.get(class="hl-string">"sla_target_resolution_time"),
        class="hl-string">"IsSlaBreached": ticket.get(class="hl-string">"is_sla_breached"),
    }


def add_event(events: List[Dict[str, Any]], seen: set, ticket: Dict[str, Any], activity: str, event_time: Any, last_update: str, user_cache: Dict[str, str], source_key: str) -> None:
    if event_time is None:
        return
    key = (str(ticket.get(class="hl-string">"id")), activity, source_key, str(event_time))
    if key in seen:
        return
    seen.add(key)
    events.append(base_row(ticket, activity, event_time, last_update, user_cache))


def audit_events(session: requests.Session, ticket_id: Any) -> List[Dict[str, Any]]:
    return list(paged_get(session, fclass="hl-string">"/api/v2/tickets/{ticket_id}/audits.json"))


def audit_events_for_field(audit: Dict[str, Any], field_names: List[str]) -> List[Tuple[str, Any, Any]]:
    output: List[Tuple[str, Any, Any]] = []
    for event in audit.get(class="hl-string">"events", []):
        if event.get(class="hl-string">"type") != class="hl-string">"Change":
            continue
        field = str(event.get(class="hl-string">"field_name") or event.get(class="hl-string">"field") or class="hl-string">"")
        if field in field_names:
            output.append((field, event.get(class="hl-string">"previous_value"), event.get(class="hl-string">"value")))
    return output


def is_customer_actor(audit: Dict[str, Any], ticket: Dict[str, Any]) -> bool:
    author_id = audit.get(class="hl-string">"author_id")
    requester_id = ticket.get(class="hl-string">"requester_id")
    return author_id is not None and requester_id is not None and str(author_id) == str(requester_id)


def is_public_comment(event: Dict[str, Any]) -> bool:
    return event.get(class="hl-string">"type") in {class="hl-string">"Comment", class="hl-string">"VoiceComment"} and event.get(class="hl-string">"public") is not False


def is_automated_acknowledgment(event: Dict[str, Any]) -> bool:
    body = str(event.get(class="hl-string">"body") or event.get(class="hl-string">"html_body") or class="hl-string">"").lower()
    marker = str(event.get(class="hl-string">"via", {}).get(class="hl-string">"channel") if isinstance(event.get(class="hl-string">"via"), dict) else event.get(class="hl-string">"via") or class="hl-string">"").lower()
    return class="hl-string">"acknowledg" in body or class="hl-string">"received your request" in body or marker in {class="hl-string">"trigger", class="hl-string">"automation", class="hl-string">"system"}


def extract_ticket_events(session: requests.Session, ticket: Dict[str, Any], last_update: str, user_cache: Dict[str, str], group_cache: Dict[str, str], seen: set) -> List[Dict[str, Any]]:
    events: List[Dict[str, Any]] = []
    ticket_id = ticket.get(class="hl-string">"id")
    audits = audit_events(session, ticket_id)
    audits.sort(key=lambda item: (str(item.get(class="hl-string">"created_at") or class="hl-string">""), str(item.get(class="hl-string">"id") or class="hl-string">"")))
    add_event(events, seen, ticket, class="hl-string">"Service Request Created", ticket.get(class="hl-string">"created_at"), last_update, user_cache, class="hl-string">"ticket_created")
    first_public_agent_reply = False
    first_ack = False
    for audit in audits:
        audit_time = audit.get(class="hl-string">"created_at")
        audit_id = str(audit.get(class="hl-string">"id") or audit_time)
        for event in audit.get(class="hl-string">"events", []):
            event_type = event.get(class="hl-string">"type")
            if event_type in {class="hl-string">"Comment", class="hl-string">"VoiceComment"} and is_public_comment(event):
                if not first_ack and is_automated_acknowledgment(event):
                    add_event(events, seen, ticket, class="hl-string">"Initial Acknowledgment Sent", audit_time, last_update, user_cache, audit_id + class="hl-string">":ack")
                    first_ack = True
                if not is_customer_actor(audit, ticket) and not first_public_agent_reply and not is_automated_acknowledgment(event):
                    add_event(events, seen, ticket, class="hl-string">"First Agent Public Reply Sent", audit_time, last_update, user_cache, audit_id + class="hl-string">":first_agent_reply")
                    first_public_agent_reply = True
                if is_customer_actor(audit, ticket):
                    add_event(events, seen, ticket, class="hl-string">"Customer Responded", audit_time, last_update, user_cache, audit_id + class="hl-string">":customer_reply")
            if event_type == class="hl-string">"Change":
                field = str(event.get(class="hl-string">"field_name") or event.get(class="hl-string">"field") or class="hl-string">"")
                previous = event.get(class="hl-string">"previous_value")
                value = event.get(class="hl-string">"value")
                if field in {class="hl-string">"type", class="hl-string">"priority", class="hl-string">"custom_fields"}:
                    add_event(events, seen, ticket, class="hl-string">"Request Categorized And Prioritized", audit_time, last_update, user_cache, audit_id + class="hl-string">":category_priority:" + field)
                if field in {class="hl-string">"assignee_id", class="hl-string">"assignee"} and value not in (None, class="hl-string">"", 0, class="hl-string">"0"):
                    add_event(events, seen, ticket, class="hl-string">"Request Assigned To Agent", audit_time, last_update, user_cache, audit_id + class="hl-string">":assignee")
                if field in {class="hl-string">"group_id", class="hl-string">"group"} and previous not in (None, class="hl-string">"", 0, class="hl-string">"0") and value not in (None, class="hl-string">"", 0, class="hl-string">"0") and str(previous) != str(value):
                    get_group_name(session, value, group_cache)
                    add_event(events, seen, ticket, class="hl-string">"Internal Escalation Triggered", audit_time, last_update, user_cache, audit_id + class="hl-string">":group")
                if field == class="hl-string">"status":
                    old_status = str(previous or class="hl-string">"").lower()
                    new_status = str(value or class="hl-string">"").lower()
                    if new_status == class="hl-string">"pending":
                        add_event(events, seen, ticket, class="hl-string">"Information Requested From Customer", audit_time, last_update, user_cache, audit_id + class="hl-string">":pending")
                    if new_status == class="hl-string">"solved":
                        add_event(events, seen, ticket, class="hl-string">"Service Request Resolved", audit_time, last_update, user_cache, audit_id + class="hl-string">":solved")
                    if old_status == class="hl-string">"solved" and new_status == class="hl-string">"open":
                        add_event(events, seen, ticket, class="hl-string">"Service Request Reopened", audit_time, last_update, user_cache, audit_id + class="hl-string">":reopened")
                    if new_status == class="hl-string">"closed":
                        add_event(events, seen, ticket, class="hl-string">"Service Request Closed", audit_time, last_update, user_cache, audit_id + class="hl-string">":closed")
                if field in {class="hl-string">"satisfaction_survey_sent", class="hl-string">"csat_survey_sent"} and str(value).lower() in {class="hl-string">"true", class="hl-string">"1", class="hl-string">"yes", class="hl-string">"sent"}:
                    add_event(events, seen, ticket, class="hl-string">"Satisfaction Survey Sent", audit_time, last_update, user_cache, audit_id + class="hl-string">":survey_sent")
                if field in {class="hl-string">"sla_breached", class="hl-string">"is_sla_breached"} and str(value).lower() in {class="hl-string">"true", class="hl-string">"1", class="hl-string">"yes"}:
                    row = base_row(ticket, class="hl-string">"SLA Breach Occurred", audit_time, last_update, user_cache)
                    row[class="hl-string">"IsSlaBreached"] = True
                    key = (str(ticket_id), class="hl-string">"SLA Breach Occurred", audit_id, str(audit_time))
                    if key not in seen:
                        seen.add(key)
                        events.append(row)
    try:
        satisfaction = api_get(session, fclass="hl-string">"/api/v2/tickets/{ticket_id}/satisfaction_rating.json").get(class="hl-string">"satisfaction_rating")
    except requests.HTTPError:
        satisfaction = None
    if satisfaction:
        add_event(events, seen, ticket, class="hl-string">"Satisfaction Rating Received", satisfaction.get(class="hl-string">"created_at") or satisfaction.get(class="hl-string">"updated_at"), last_update, user_cache, str(satisfaction.get(class="hl-string">"id") or class="hl-string">"rating"))
    return events


def validate_rows(rows: List[Dict[str, Any]]) -> None:
    activities = {
        class="hl-string">"Service Request Created",
        class="hl-string">"Initial Acknowledgment Sent",
        class="hl-string">"Request Categorized And Prioritized",
        class="hl-string">"Request Assigned To Agent",
        class="hl-string">"First Agent Public Reply Sent",
        class="hl-string">"Information Requested From Customer",
        class="hl-string">"Customer Responded",
        class="hl-string">"Internal Escalation Triggered",
        class="hl-string">"SLA Breach Occurred",
        class="hl-string">"Service Request Resolved",
        class="hl-string">"Service Request Reopened",
        class="hl-string">"Satisfaction Survey Sent",
        class="hl-string">"Satisfaction Rating Received",
        class="hl-string">"Service Request Closed",
    }
    missing_columns = [column for column in ALL_COLUMNS if column not in rows[0]] if rows else ALL_COLUMNS
    if missing_columns:
        fail(class="hl-string">"Missing output columns: " + class="hl-string">", ".join(missing_columns))
    invalid = [row for row in rows if not row.get(class="hl-string">"ServiceRequest") or not row.get(class="hl-string">"ActivityName") or not row.get(class="hl-string">"StartTime") or not row.get(class="hl-string">"SourceSystem") or not row.get(class="hl-string">"LastDataUpdate")]
    if invalid:
        fail(fclass="hl-string">"{len(invalid)} rows are missing required values")
    unexpected = sorted({row[class="hl-string">"ActivityName"] for row in rows} - activities)
    if unexpected:
        fail(class="hl-string">"Unexpected activity names: " + class="hl-string">", ".join(unexpected))


def main() -> None:
    global session
    if ZENDESK_SUBDOMAIN.startswith(class="hl-string">"[") or ZENDESK_EMAIL.startswith(class="hl-string">"[") or ZENDESK_API_TOKEN.startswith(class="hl-string">"[") or START_TIME.startswith(class="hl-string">"["):
        fail(class="hl-string">"Configure ZENDESK_SUBDOMAIN, ZENDESK_EMAIL, ZENDESK_API_TOKEN, and START_TIME")
    start_time = parse_start_time(START_TIME)
    session = requests.Session()
    session.auth = (fclass="hl-string">"{ZENDESK_EMAIL}/token", ZENDESK_API_TOKEN)
    session.headers.update({class="hl-string">"Accept": class="hl-string">"application/json"})
    last_update = now_utc()
    user_cache: Dict[str, str] = {}
    group_cache: Dict[str, str] = {}
    seen = set()
    rows: List[Dict[str, Any]] = []
    tickets = paged_get(session, class="hl-string">"/api/v2/incremental/tickets/cursor.json", {class="hl-string">"start_time": start_time})
    for ticket in tickets:
        rows.extend(extract_ticket_events(session, ticket, last_update, user_cache, group_cache, seen))
    rows.sort(key=lambda row: (row[class="hl-string">"ServiceRequest"], row[class="hl-string">"StartTime"], row[class="hl-string">"ActivityName"]))
    validate_rows(rows)
    with open(OUTPUT_FILE, class="hl-string">"w", newline=class="hl-string">"", encoding=class="hl-string">"utf-8") as output:
        writer = csv.DictWriter(output, fieldnames=ALL_COLUMNS, extrasaction=class="hl-string">"ignore")
        writer.writeheader()
        writer.writerows(rows)
    print(json.dumps({class="hl-string">"output_file": OUTPUT_FILE, class="hl-string">"rows": len(rows), class="hl-string">"last_data_update": last_update}))


if __name__ == class="hl-string">"__main__":
    main()

WITH source_data AS (
    SELECT
        [Your ticket identifier column] AS service_request,
        [Your source event identifier column] AS source_event_id,
        [Your event type column] AS source_event_type,
        [Your field name column] AS source_field_name,
        [Your previous value column] AS previous_value,
        [Your new value column] AS new_value,
        [Your event timestamp column] AS event_time,
        [Your actor type column] AS actor_type,
        [Your actor identifier column] AS actor_id,
        [Your assigned agent column] AS assigned_agent,
        [Your request type column] AS service_request_type,
        [Your priority column] AS priority,
        [Your communication channel column] AS communication_channel,
        [Your SLA target resolution time column] AS sla_target_resolution_time,
        [Your SLA breached flag column] AS is_sla_breached,
        [Your automated response flag column] AS is_automated_response,
        [Your public comment flag column] AS is_public_comment,
        [Your customer reply flag column] AS is_customer_reply,
        [Your survey sent flag column] AS is_survey_sent,
        [Your satisfaction rating column] AS satisfaction_rating,
        [Your source system value] AS source_system,
        [Your extraction timestamp column] AS last_data_update
    FROM [Your normalized Zendesk event source]
    WHERE [Your event timestamp column] >= [Your start date parameter]
      AND [Your event timestamp column] < [Your end date parameter]
),
activity_events AS (
    SELECT service_request, 'Service Request Created' AS activity_name, event_time AS start_time, source_event_id, assigned_agent, service_request_type, priority, communication_channel, sla_target_resolution_time, is_sla_breached, source_system, last_data_update
    FROM source_data
    WHERE source_event_type = 'ticket_created'
    UNION ALL
    SELECT service_request, 'Initial Acknowledgment Sent', event_time, source_event_id, assigned_agent, service_request_type, priority, communication_channel, sla_target_resolution_time, is_sla_breached, source_system, last_data_update
    FROM source_data
    WHERE source_event_type = 'comment_added' AND is_automated_response = TRUE AND is_public_comment = TRUE
    UNION ALL
    SELECT service_request, 'Request Categorized And Prioritized', event_time, source_event_id, assigned_agent, service_request_type, priority, communication_channel, sla_target_resolution_time, is_sla_breached, source_system, last_data_update
    FROM source_data
    WHERE source_event_type = 'ticket_field_changed' AND source_field_name IN ('type', 'category', 'priority')
    UNION ALL
    SELECT service_request, 'Request Assigned To Agent', event_time, source_event_id, assigned_agent, service_request_type, priority, communication_channel, sla_target_resolution_time, is_sla_breached, source_system, last_data_update
    FROM source_data
    WHERE source_event_type = 'ticket_field_changed' AND source_field_name IN ('assignee', 'assigned_agent')
    UNION ALL
    SELECT service_request, 'First Agent Public Reply Sent', event_time, source_event_id, assigned_agent, service_request_type, priority, communication_channel, sla_target_resolution_time, is_sla_breached, source_system, last_data_update
    FROM source_data
    WHERE source_event_type = 'comment_added' AND is_public_comment = TRUE AND actor_type = 'agent' AND is_automated_response = FALSE
    UNION ALL
    SELECT service_request, 'Information Requested From Customer', event_time, source_event_id, assigned_agent, service_request_type, priority, communication_channel, sla_target_resolution_time, is_sla_breached, source_system, last_data_update
    FROM source_data
    WHERE source_event_type = 'ticket_status_changed' AND new_value = 'pending'
    UNION ALL
    SELECT service_request, 'Customer Responded', event_time, source_event_id, assigned_agent, service_request_type, priority, communication_channel, sla_target_resolution_time, is_sla_breached, source_system, last_data_update
    FROM source_data
    WHERE source_event_type = 'comment_added' AND is_public_comment = TRUE AND is_customer_reply = TRUE
    UNION ALL
    SELECT service_request, 'Internal Escalation Triggered', event_time, source_event_id, assigned_agent, service_request_type, priority, communication_channel, sla_target_resolution_time, is_sla_breached, source_system, last_data_update
    FROM source_data
    WHERE source_event_type = 'ticket_field_changed' AND source_field_name IN ('group', 'assigned_group')
    UNION ALL
    SELECT service_request, 'SLA Breach Occurred', event_time, source_event_id, assigned_agent, service_request_type, priority, communication_channel, sla_target_resolution_time, TRUE, source_system, last_data_update
    FROM source_data
    WHERE source_event_type IN ('sla_breach', 'sla_breached') OR is_sla_breached = TRUE
    UNION ALL
    SELECT service_request, 'Service Request Resolved', event_time, source_event_id, assigned_agent, service_request_type, priority, communication_channel, sla_target_resolution_time, is_sla_breached, source_system, last_data_update
    FROM source_data
    WHERE source_event_type = 'ticket_status_changed' AND new_value = 'solved'
    UNION ALL
    SELECT service_request, 'Service Request Reopened', event_time, source_event_id, assigned_agent, service_request_type, priority, communication_channel, sla_target_resolution_time, is_sla_breached, source_system, last_data_update
    FROM source_data
    WHERE source_event_type = 'ticket_status_changed' AND previous_value = 'solved' AND new_value = 'open' AND is_customer_reply = TRUE
    UNION ALL
    SELECT service_request, 'Satisfaction Survey Sent', event_time, source_event_id, assigned_agent, service_request_type, priority, communication_channel, sla_target_resolution_time, is_sla_breached, source_system, last_data_update
    FROM source_data
    WHERE source_event_type IN ('satisfaction_survey_sent', 'csat_survey_sent') OR is_survey_sent = TRUE
    UNION ALL
    SELECT service_request, 'Satisfaction Rating Received', event_time, source_event_id, assigned_agent, service_request_type, priority, communication_channel, sla_target_resolution_time, is_sla_breached, source_system, last_data_update
    FROM source_data
    WHERE source_event_type IN ('satisfaction_rating_received', 'csat_rating_received') OR satisfaction_rating IS NOT NULL
    UNION ALL
    SELECT service_request, 'Service Request Closed', event_time, source_event_id, assigned_agent, service_request_type, priority, communication_channel, sla_target_resolution_time, is_sla_breached, source_system, last_data_update
    FROM source_data
    WHERE source_event_type = 'ticket_status_changed' AND new_value = 'closed'
),
deduplicated_events AS (
    SELECT
        service_request,
        activity_name,
        start_time,
        assigned_agent,
        service_request_type,
        priority,
        communication_channel,
        sla_target_resolution_time,
        is_sla_breached,
        source_system,
        last_data_update,
        ROW_NUMBER() OVER (PARTITION BY service_request, activity_name, source_event_id ORDER BY start_time) AS event_rank
    FROM activity_events
)
SELECT
    service_request AS ServiceRequest,
    activity_name AS ActivityName,
    start_time AS StartTime,
    source_system AS SourceSystem,
    last_data_update AS LastDataUpdate,
    assigned_agent AS AssignedAgent,
    service_request_type AS ServiceRequestType,
    priority AS Priority,
    communication_channel AS CommunicationChannel,
    sla_target_resolution_time AS SlaTargetResolutionTime,
    is_sla_breached AS IsSlaBreached
FROM deduplicated_events
WHERE event_rank = 1
ORDER BY service_request, start_time, activity_name;