您的质量管理数据模板

import requests
import pandas as pd
from datetime import datetime
import io

class=class="hl-string">"hl-comment"># --- Configuration: Update these values --- #
VAULT_DNS = class="hl-string">"[your_vault_dns].veevavault.com"  class=class="hl-string">"hl-comment"># Example: yourcompany.veevavault.com
USERNAME = class="hl-string">"[your_api_username]"
PASSWORD = class="hl-string">"[your_api_password]"
API_VERSION = class="hl-string">"v23.1"
OUTPUT_FILENAME = class="hl-string">"quality_event_log.csv"
class=class="hl-string">"hl-comment"># --- End of Configuration --- #

def get_session_id():
    class="hl-string">"""Authenticates with Veeva Vault API and returns a session ID."""
    auth_url = fclass="hl-string">"https://{VAULT_DNS}/api/{API_VERSION}/auth"
    headers = {class="hl-string">"Content-Type": class="hl-string">"application/x-www-form-urlencoded", class="hl-string">"Accept": class="hl-string">"application/json"}
    data = {class="hl-string">"username": USERNAME, class="hl-string">"password": PASSWORD}
    
    try:
        response = requests.post(auth_url, headers=headers, data=data)
        response.raise_for_status() class=class="hl-string">"hl-comment"># Raises an exception for bad responses (4xx or 5xx)
        auth_data = response.json()
        if auth_data.get(class="hl-string">"responseStatus") == class="hl-string">"SUCCESS":
            print(class="hl-string">"Authentication successful.")
            return auth_data[class="hl-string">"sessionId"]
        else:
            print(fclass="hl-string">"Authentication failed: {auth_data.get('responseMessage')}")
            return None
    except requests.exceptions.RequestException as e:
        print(fclass="hl-string">"An error occurred during authentication: {e}")
        return None

def run_vql_query(session_id, query):
    class="hl-string">"""Runs a VQL query, handles pagination, and returns all results as a pandas DataFrame."""
    query_url = fclass="hl-string">"https://{VAULT_DNS}/api/{API_VERSION}/query"
    headers = {class="hl-string">"Authorization": session_id, class="hl-string">"Accept": class="hl-string">"application/json"}
    all_results = []
    
    params = {class="hl-string">'q': query}
    
    page_num = 1
    while True:
        print(fclass="hl-string">"Fetching page {page_num}...")
        try:
            response = requests.post(query_url, headers=headers, params=params)
            response.raise_for_status()
            query_response = response.json()
            
            if query_response.get(class="hl-string">"responseStatus") != class="hl-string">"SUCCESS":
                print(fclass="hl-string">"Query failed: {query_response.get('responseDetails', {}).get('message')}")
                break

            all_results.extend(query_response.get(class="hl-string">'data', []))
            
            if class="hl-string">'next_page' in query_response.get(class="hl-string">'responseDetails', {}):
                next_page_url = query_response[class="hl-string">'responseDetails'][class="hl-string">'next_page']
                query_url = fclass="hl-string">"https://{VAULT_DNS}{next_page_url}"
                params = {}
                page_num += 1
            else:
                break
        except requests.exceptions.RequestException as e:
            print(fclass="hl-string">"An error occurred during query execution: {e}")
            break

    if not all_results:
        return pd.DataFrame()
    
    class=class="hl-string">"hl-comment"># Convert list of dictionaries to DataFrame
    class=class="hl-string">"hl-comment"># Veeva API returns data as {class="hl-string">'field_name': class="hl-string">'value'}. We need to flatten this.
    flat_data = [{key: item[key] for key in item} for item in all_results]
    return pd.DataFrame(flat_data)


def main():
    session_id = get_session_id()
    if not session_id:
        return

    class=class="hl-string">"hl-comment"># Common fields to select in every query for context
    common_fields = (
        class="hl-string">"id, "
        class="hl-string">"name__v, "
        class="hl-string">"investigator__c, "
        class="hl-string">"department_owner__c, "
        class="hl-string">"type_of_quality_event__c, "
        class="hl-string">"status__v, "
        class="hl-string">"target_due_date__c, "
        class="hl-string">"severity__c"
    )

    class=class="hl-string">"hl-comment"># Define VQL queries for each activity
    class=class="hl-string">"hl-comment"># NOTE: Field names like class="hl-string">'triage_completed_date__c' are placeholders for common custom date fields.
    class=class="hl-string">"hl-comment"># Adjust these to match your Veeva Vault configuration.
    activity_queries = {
        class="hl-string">"Quality Event Created": fclass="hl-string">"SELECT {common_fields}, created_date__v FROM quality_event__v WHERE created_date__v > '2023-01-01T00:00:00.000Z'",
        class="hl-string">"Initial Triage Completed": fclass="hl-string">"SELECT {common_fields}, triage_completed_date__c FROM quality_event__v WHERE triage_completed_date__c != null",
        class="hl-string">"Issue Categorized And Prioritized": fclass="hl-string">"SELECT {common_fields}, categorization_date__c FROM quality_event__v WHERE categorization_date__c != null",
        class="hl-string">"Investigation Initiated": fclass="hl-string">"SELECT {common_fields}, investigation_start_date__c FROM quality_event__v WHERE investigation_start_date__c != null",
        class="hl-string">"Root Cause Analysis Performed": fclass="hl-string">"SELECT {common_fields}, rca_completed_date__c FROM quality_event__v WHERE rca_completed_date__c != null",
        class="hl-string">"Corrective Action Plan Proposed": fclass="hl-string">"SELECT {common_fields}, quality_event__vr.created_date__v FROM capa__v WHERE quality_event__vr.created_date__v != null AND action_type__c = 'Corrective'",
        class="hl-string">"Corrective Action Plan Approved": fclass="hl-string">"SELECT {common_fields}, quality_event__vr.approval_date__c FROM capa__v WHERE quality_event__vr.approval_date__c != null AND action_type__c = 'Corrective'",
        class="hl-string">"Corrective Action Implemented": fclass="hl-string">"SELECT {common_fields}, quality_event__vr.implementation_date__c FROM capa__v WHERE quality_event__vr.implementation_date__c != null AND action_type__c = 'Corrective'",
        class="hl-string">"Preventive Action Identified": fclass="hl-string">"SELECT {common_fields}, quality_event__vr.created_date__v FROM capa__v WHERE quality_event__vr.created_date__v != null AND action_type__c = 'Preventive'",
        class="hl-string">"Effectiveness Check Initiated": fclass="hl-string">"SELECT {common_fields}, effectiveness_check_start_date__c FROM quality_event__v WHERE effectiveness_check_start_date__c != null",
        class="hl-string">"Effectiveness Of Action Verified": fclass="hl-string">"SELECT {common_fields}, effectiveness_verified_date__c FROM quality_event__v WHERE effectiveness_verified_date__c != null",
        class="hl-string">"Final Review And Closure": fclass="hl-string">"SELECT {common_fields}, closed_date__c FROM quality_event__v WHERE closed_date__c != null",
        class="hl-string">"Stakeholders Notified Of Resolution": fclass="hl-string">"SELECT {common_fields}, notification_date__c FROM quality_event__v WHERE notification_date__c != null"
    }

    all_events_df = pd.DataFrame()

    for activity_name, query in activity_queries.items():
        print(fclass="hl-string">"\n--- Extracting activity: {activity_name} ---")
        df = run_vql_query(session_id, query)
        
        if df.empty:
            print(fclass="hl-string">"No data found for activity: {activity_name}")
            continue

        print(fclass="hl-string">"Found {len(df)} records for {activity_name}")
        
        class=class="hl-string">"hl-comment"># Standardize columns
        df[class="hl-string">'ActivityName'] = activity_name
        df[class="hl-string">'SourceSystem'] = class="hl-string">'Veeva Vault Quality'
        df[class="hl-string">'LastDataUpdate'] = datetime.utcnow().isoformat() + class="hl-string">'Z'
        
        class=class="hl-string">"hl-comment"># Find the correct timestamp column based on the activity name
        class=class="hl-string">"hl-comment"># This logic assumes the last field in the SELECT statement is the event time
        timestamp_field = query.split(class="hl-string">' FROM')[0].split(class="hl-string">',')[-1].strip()
        if class="hl-string">'.' in timestamp_field:
            timestamp_field = timestamp_field.split(class="hl-string">'.')[-1]

        df.rename(columns={
            class="hl-string">'name__v': class="hl-string">'QualityEvent',
            timestamp_field: class="hl-string">'EventTime',
            class="hl-string">'investigator__c': class="hl-string">'AssignedInvestigator',
            class="hl-string">'department_owner__c': class="hl-string">'AssignedDepartment',
            class="hl-string">'type_of_quality_event__c': class="hl-string">'QualityEventType',
            class="hl-string">'status__v': class="hl-string">'QualityEventStatus',
            class="hl-string">'target_due_date__c': class="hl-string">'TargetResolutionDate',
            class="hl-string">'severity__c': class="hl-string">'SeverityLevel'
        }, inplace=True)
        
        class=class="hl-string">"hl-comment"># For CAPA queries, the Quality Event name is nested
        if class="hl-string">'quality_event__vr.name__v' in df.columns:
            df.rename(columns={class="hl-string">'quality_event__vr.name__v': class="hl-string">'QualityEvent'}, inplace=True)

        all_events_df = pd.concat([all_events_df, df], ignore_index=True)
    
    class=class="hl-string">"hl-comment"># Define the final column structure
    final_columns = [
        class="hl-string">'QualityEvent', class="hl-string">'ActivityName', class="hl-string">'EventTime', class="hl-string">'SourceSystem', class="hl-string">'LastDataUpdate',
        class="hl-string">'AssignedInvestigator', class="hl-string">'AssignedDepartment', class="hl-string">'QualityEventType',
        class="hl-string">'QualityEventStatus', class="hl-string">'TargetResolutionDate', class="hl-string">'SeverityLevel'
    ]
    
    class=class="hl-string">"hl-comment"># Select and reorder columns, filling missing ones with None
    output_df = pd.DataFrame(columns=final_columns)
    for col in final_columns:
        if col in all_events_df.columns:
            output_df[col] = all_events_df[col]
        else:
            output_df[col] = None
    
    class=class="hl-string">"hl-comment"># Drop rows where EventTime is NaT (Not a Time)
    output_df.dropna(subset=[class="hl-string">'EventTime'], inplace=True)

    class=class="hl-string">"hl-comment"># Sort by case and event time for readability
    output_df.sort_values(by=[class="hl-string">'QualityEvent', class="hl-string">'EventTime'], inplace=True)
    
    output_df.to_csv(OUTPUT_FILENAME, index=False)
    print(fclass="hl-string">"\nExtraction complete. Event log saved to {OUTPUT_FILENAME}")

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

#!/bin/bash

# Veeva Vault Connection Details (Replace with your credentials)
VAULTDNS="[Your Vault DNS e.g., your-vault.veevavault.com]"
USERNAME="[Your Username]"
PASSWORD="[Your Password]"

# Configuration Paths (Update paths as needed)
LOADERDIR="/path/to/vault-loader"
INPUTDIR="/path/to/input/files"
OUTPUTDIR="/path/to/output/files"
LOGFILE="${OUTPUTDIR}/loader_export.log"

# Input VQL Query File
QUERYCSV="${INPUTDIR}/vql_query.csv"

# Raw Output File
OUTPUTCSV="${OUTPUTDIR}/quality_events_raw.csv"

# --- VQL Query Definition ---
# This query extracts a wide table of Quality Event data.
# A separate transformation step is required to pivot this into an event log.
# Field names (`__c`) are examples and must be verified in your Vault.
cat > "$QUERYCSV" <<- EOM
query
"SELECT id, name__v, created_date__v, status__v, type__v, severity_level__c, target_resolution_date__c, assigned_investigator__c, assigned_department__c, initial_triage_completed_date__c, issue_categorized_date__c, investigation_initiated_date__c, rca_performed_date__c, capa_proposed_date__c, capa_approved_date__c, capa_implemented_date__c, pa_identified_date__c, effectiveness_check_initiated_date__c, effectiveness_verified_date__c, final_review_closure_date__c, stakeholder_notification_date__c FROM quality_event__v WHERE created_date__v >= '2023-01-01'"
EOM

# --- Veeva Loader Execution ---
echo "Starting Veeva Vault Quality Event export..."

java -jar "${LOADERDIR}/vault-loader.jar" \
-action EXTRACT \
-V "https://www.veeva.com" \
-vql "$QUERYCSV" \
-outputdir "$OUTPUTDIR" \
-resultsfile "$OUTPUTCSV" \
-u "$USERNAME" \
-p "$PASSWORD" \
-dns "$VAULTDNS" \
-logfile "$LOGFILE"

echo "Export complete. Raw data saved to $OUTPUTCSV"
echo "Log file available at $LOGFILE"