质量管理数据模板

import requests
import pandas as pd
from datetime import datetime
import json

class=class="hl-string">"hl-comment"># --- CONFIGURATION SETTINGS --- #
class=class="hl-string">"hl-comment"># Replace with your ETQ Reliance instance details
ETQ_BASE_URL = class="hl-string">"https://[your_etq_instance_url]/reliance/api"
API_USERNAME = class="hl-string">"[your_api_username]"
API_PASSWORD = class="hl-string">"[your_api_password]"

class=class="hl-string">"hl-comment"># Specify the Application and View ID for Quality Events
APPLICATION_ID = class="hl-string">"[Your_Quality_Events_Application_ID]"
VIEW_ID = class="hl-string">"[Your_Quality_Events_View_ID]"

class=class="hl-string">"hl-comment"># Define the date range for the extraction (YYYY-MM-DD format)
START_DATE = class="hl-string">"2023-01-01"
END_DATE = class="hl-string">"2023-12-31"

class=class="hl-string">"hl-comment"># API settings
PAGE_SIZE = 100

class=class="hl-string">"hl-comment"># --- ACTIVITY MAPPING CONFIGURATION --- #
class=class="hl-string">"hl-comment"># This section requires customization based on your specific ETQ Reliance configuration.
class=class="hl-string">"hl-comment"># Map ETQ field changes to the desired activity names.
class=class="hl-string">"hl-comment"># Format: class="hl-string">'FieldName': {(class="hl-string">'OldValue', class="hl-string">'NewValue'): class="hl-string">'ActivityName'}
class=class="hl-string">"hl-comment"># Use class="hl-string">'*' as a wildcard for OldValue or NewValue if the change itself is the trigger.
ACTIVITY_MAPPING_CONFIG = {
    class=class="hl-string">"hl-comment"># Example mappings based on a class="hl-string">'Status' field
    class="hl-string">'ds_status': {
        (None, class="hl-string">'New'): class="hl-string">'Quality Issue Identified',
        (class="hl-string">'New', class="hl-string">'Under Assessment'): class="hl-string">'Initial Assessment Performed',
        (class="hl-string">'Under Assessment', class="hl-string">'Under Investigation'): class="hl-string">'Investigation Initiated',
        (class="hl-string">'Investigation Complete', class="hl-string">'CAPA Proposed'): class="hl-string">'Corrective Action Plan Proposed',
        (class="hl-string">'CAPA Proposed', class="hl-string">'CAPA Approved'): class="hl-string">'Corrective Action Plan Approved',
        (class="hl-string">'CAPA Proposed', class="hl-string">'CAPA Rejected'): class="hl-string">'Corrective Action Plan Rejected',
        (class="hl-string">'CAPA Approved', class="hl-string">'Implementation in Progress'): class="hl-string">'Corrective Action Implemented',
        (class="hl-string">'Implementation Complete', class="hl-string">'Effectiveness Verification'): class="hl-string">'Effectiveness of Action Verified',
        (class="hl-string">'Effectiveness Verification', class="hl-string">'Closed'): class="hl-string">'Final Review Performed',
        (class="hl-string">'Effectiveness Verification', class="hl-string">'Re-investigate'): class="hl-string">'Effectiveness Verification Failed',
        (class="hl-string">'*', class="hl-string">'Closed'): class="hl-string">'Quality Event Closed',
        (class="hl-string">'*', class="hl-string">'Canceled'): class="hl-string">'Quality Event Canceled'
    },
    class=class="hl-string">"hl-comment"># Example mappings based on completion of form sections (field changes from blank to non-blank)
    class="hl-string">'dt_categorization_complete_date': {
        (None, class="hl-string">'*'): class="hl-string">'Issue Categorized And Prioritized'
    },
    class="hl-string">'dt_rca_complete_date': {
        (None, class="hl-string">'*'): class="hl-string">'Root Cause Analysis Performed'
    },
    class="hl-string">'dt_pa_identified_date': {
        (None, class="hl-string">'*'): class="hl-string">'Preventive Action Identified'
    },
    class="hl-string">'dt_pa_implemented_date': {
        (None, class="hl-string">'*'): class="hl-string">'Preventive Action Implemented'
    },
    class="hl-string">'dt_stakeholder_notified_date': {
        (None, class="hl-string">'*'): class="hl-string">'Stakeholders Notified'
    }
}

def get_etq_token(session):
    class="hl-string">"""Authenticates with ETQ API and returns a session token."""
    auth_url = fclass="hl-string">"{ETQ_BASE_URL}/v1/login"
    credentials = {class="hl-string">'username': API_USERNAME, class="hl-string">'password': API_PASSWORD}
    try:
        response = session.post(auth_url, json=credentials)
        response.raise_for_status()
        return response.json().get(class="hl-string">'token')
    except requests.exceptions.RequestException as e:
        print(fclass="hl-string">"Error during authentication: {e}")
        return None

def main():
    class="hl-string">"""Main function to extract, transform, and save event data."""
    with requests.Session() as session:
        token = get_etq_token(session)
        if not token:
            return
        
        session.headers.update({class="hl-string">'Authorization': fclass="hl-string">'Bearer {token}'})
        
        class=class="hl-string">"hl-comment"># Get all quality events within the date range
        all_events = []
        page = 0
        while True:
            search_url = fclass="hl-string">"{ETQ_BASE_URL}/v2/applications/{APPLICATION_ID}/views/{VIEW_ID}/search"
            params = {
                class="hl-string">'page': page,
                class="hl-string">'size': PAGE_SIZE,
                class="hl-string">'query': fclass="hl-string">'creationDate>=\"{START_DATE}T00:00:00Z\" and creationDate<=\"{END_DATE}T23:59:59Z\"'
            }
            try:
                response = session.get(search_url, params=params)
                response.raise_for_status()
                data = response.json()
                all_events.extend(data.get(class="hl-string">'content', []))
                if data.get(class="hl-string">'last', True):
                    break
                page += 1
            except requests.exceptions.RequestException as e:
                print(fclass="hl-string">"Error fetching quality event list: {e}")
                break
        
        print(fclass="hl-string">"Found {len(all_events)} total quality events.")
        event_log = []

        class=class="hl-string">"hl-comment"># Process each event to get its history
        for event in all_events:
            event_id = event.get(class="hl-string">'uid')
            case_id = event.get(class="hl-string">'documentNumber', event_id)
            if not event_id:
                continue

            history_url = fclass="hl-string">"{ETQ_BASE_URL}/v1/applications/{APPLICATION_ID}/documents/{event_id}/history"
            try:
                hist_response = session.get(history_url)
                hist_response.raise_for_status()
                history_data = hist_response.json()
            except requests.exceptions.RequestException as e:
                print(fclass="hl-string">"Could not fetch history for event {case_id}: {e}")
                continue

            class=class="hl-string">"hl-comment"># Extract initial record creation as class="hl-string">'Quality Issue Identified'
            class=class="hl-string">"hl-comment"># Note: This logic assumes the first history entry isnclass="hl-string">'t a creation event, using the main record's creationDate instead.
            class=class="hl-string">"hl-comment"># If your history includes a class="hl-string">'Create' event, adjust accordingly.
            creation_time = event.get(class="hl-string">'creationDate', datetime.now().isoformat())
            event_log.append({
                class="hl-string">'QualityEvent': case_id,
                class="hl-string">'ActivityName': class="hl-string">'Quality Issue Identified',
                class="hl-string">'EventTimestamp': creation_time,
                class="hl-string">'ActionPerformedBy': event.get(class="hl-string">'createdBy', {}).get(class="hl-string">'uid'),
                class="hl-string">'ResponsibleDepartment': event.get(class="hl-string">'values', {}).get(class="hl-string">'[Your_Department_Field]', None),
                class="hl-string">'SeverityLevel': event.get(class="hl-string">'values', {}).get(class="hl-string">'[Your_Severity_Field]', None),
                class="hl-string">'RootCauseCategory': None,
                class="hl-string">'EffectivenessVerificationOutcome': None,
                class="hl-string">'EventEndTime': None
            })

            class=class="hl-string">"hl-comment"># Process history for other activities
            for entry in sorted(history_data, key=lambda x: x[class="hl-string">'date']):
                timestamp = entry[class="hl-string">'date']
                user = entry.get(class="hl-string">'user', {}).get(class="hl-string">'uid')
                for change in entry.get(class="hl-string">'changes', []):
                    field_name = change.get(class="hl-string">'field', {}).get(class="hl-string">'uid')
                    old_value = change.get(class="hl-string">'oldValue')
                    new_value = change.get(class="hl-string">'newValue')

                    if field_name in ACTIVITY_MAPPING_CONFIG:
                        mappings = ACTIVITY_MAPPING_CONFIG[field_name]
                        class=class="hl-string">"hl-comment"># Check for specific value transitions
                        if (old_value, new_value) in mappings:
                            activity = mappings[(old_value, new_value)]
                        class=class="hl-string">"hl-comment"># Check for wildcard class="hl-string">'from anything' to specific value
                        elif (class="hl-string">'*', new_value) in mappings:
                            activity = mappings[(class="hl-string">'*', new_value)]
                        class=class="hl-string">"hl-comment"># Check for wildcard class="hl-string">'from specific value' to anything
                        elif (old_value, class="hl-string">'*') in mappings:
                            activity = mappings[(old_value, class="hl-string">'*')]
                        class=class="hl-string">"hl-comment"># Check for wildcard from class="hl-string">'None' to anything (field populated for first time)
                        elif (None, class="hl-string">'*') in mappings and old_value is None:
                            activity = mappings[(None, class="hl-string">'*')]
                        else:
                            continue class=class="hl-string">"hl-comment"># No match
                        
                        event_log.append({
                            class="hl-string">'QualityEvent': case_id,
                            class="hl-string">'ActivityName': activity,
                            class="hl-string">'EventTimestamp': timestamp,
                            class="hl-string">'ActionPerformedBy': user,
                            class="hl-string">'ResponsibleDepartment': event.get(class="hl-string">'values', {}).get(class="hl-string">'[Your_Department_Field]', None),
                            class="hl-string">'SeverityLevel': event.get(class="hl-string">'values', {}).get(class="hl-string">'[Your_Severity_Field]', None),
                            class="hl-string">'RootCauseCategory': event.get(class="hl-string">'values', {}).get(class="hl-string">'[Your_RootCause_Field]', None),
                            class="hl-string">'EffectivenessVerificationOutcome': event.get(class="hl-string">'values', {}).get(class="hl-string">'[Your_Effectiveness_Field]', None),
                            class="hl-string">'EventEndTime': None
                        })

    if not event_log:
        print(class="hl-string">"No events were processed. The event log is empty.")
        return

    class=class="hl-string">"hl-comment"># Convert to DataFrame and save to CSV
    df = pd.DataFrame(event_log)
    df[class="hl-string">'EventTimestamp'] = pd.to_datetime(df[class="hl-string">'EventTimestamp'])
    df = df.sort_values(by=[class="hl-string">'QualityEvent', class="hl-string">'EventTimestamp'])
    df.to_csv(class="hl-string">'etq_quality_events_log.csv', index=False)
    print(class="hl-string">"Successfully created etq_quality_events_log.csv")

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