model.py

import json
import logging
from datetime import datetime
from typing import Dict, List, Optional

# --- Constants ---
ALERT_THRESHOLD = 0.75  # Example threshold, adjust as needed
QUALITY_WEIGHT = 0.4
TIMELINESS_WEIGHT = 0.3
SAFETY_WEIGHT = 0.15
COMMUNICATION_WEIGHT = 0.15
# --- Helper Functions ---
def calculate_quality_score(details: str) -> float:
    """
    Calculates a quality score based on work completion details.

    Args:
        details: Text description of the work completed.

    Returns:
        A score between 0 and 1, where 1 is the highest quality.
    """
    # Placeholder for NLP-based quality assessment
    # Replace with your actual NLP model (e.g., sentiment analysis, keyword extraction)
    # Example: Simple keyword matching
    keywords = ["excellent", "high quality", "flawless", "best practice", "exceeded expectations"]
    score = 0.5  # Default score
    for keyword in keywords:
        if keyword in details.lower():
            score = 1.0
            break
    return score
def calculate_timeliness_score(delay_reports: str) -> float:
    """
    Calculates a timeliness score based on delay reports.

    Args:
        delay_reports: Text describing any delays.

    Returns:
        A score between 0 and 1, where 1 indicates no delays.
    """
    # Placeholder for timeliness assessment
    # Replace with your actual logic (e.g., count delays, severity analysis)
    if not delay_reports:
        return 1.0  # No delays reported
    else:
        #very basic example
        num_delays = len(delay_reports.split(','))
        return max(0, 1 - (num_delays * 0.1)) # Deduct 0.1 per delay, but minimum 0
    return 0.7 #default
def calculate_safety_score(incident_logs: str) -> float:
    """
    Calculates a safety score based on incident logs.

    Args:
        incident_logs: Text describing any safety incidents.

    Returns:
        A score between 0 and 1, where 1 indicates no incidents.
    """
    # Placeholder for safety assessment
    # Replace with your actual logic (e.g., severity scoring, incident type analysis)
    if not incident_logs:
        return 1.0  # No incidents reported
    else:
        return 0.8
def calculate_communication_score() -> float:
    """
    Calculates a communication score.  Since the input data doesn't directly
    contain communication data, this is a placeholder.  You'll need to
    determine how to derive this score (e.g., from separate feedback logs,
    response times, etc.).

    Returns:
       A score between 0 and 1.
    """
    return 0.9  # Placeholder - Replace with your actual communication scoring logic
def calculate_final_score(quality: float, timeliness: float, safety: float, communication: float) -> float:
    """
    Calculates the final performance score using weighted average.

    Args:
        quality: Quality score.
        timeliness: Timeliness score.
        safety: Safety score.
        communication: Communication score.

    Returns:
        The final performance score (between 0 and 1).
    """
    return (
        QUALITY_WEIGHT * quality +
        TIMELINESS_WEIGHT * timeliness +
        SAFETY_WEIGHT * safety +
        COMMUNICATION_WEIGHT * communication
    )
def analyze_trends(vendor_id: str, scores: List[Dict]) -> float:
    """
    Analyzes performance trends for a vendor.  This is a placeholder, as
    you'll need to store historical data to implement this properly.

    Args:
        vendor_id: The ID of the vendor.
        scores: A list of historical scores for the vendor.  In a real
                implementation, this would come from a database.

    Returns:
        A trend deviation value (e.g., 0 for stable, >0 for improving, <0 for declining).
    """
    # Placeholder:  In a real implementation, fetch historical data and compare
    # to the current score.  For this example, we'll return 0 (no trend).
    if not scores:
        return 0.0
    latest_score = scores[-1]['final_score']
    if len(scores) > 1:
        previous_score = scores[-2]['final_score']
        return latest_score - previous_score
    return 0.0
def get_historical_scores(vendor_id: str) -> List[Dict]:
    """
    Retrieves historical scores for a vendor from a data source.
    THIS IS A PLACEHOLDER.  You will need to replace this with
    actual database or data retrieval logic.  This example
    returns an empty list.

    Args:
        vendor_id: The ID of the vendor.

    Returns:
        A list of dictionaries, where each dictionary represents a historical score record.
        Example:
        [
            {'month': '2024-01', 'final_score': 0.85},
            {'month': '2024-02', 'final_score': 0.90},
            {'month': '2024-03', 'final_score': 0.88},
        ]
    """
    # Placeholder: Replace with database query or data retrieval
    return []
def generate_pdf_certificate(vendor_name: str, month: str, quality: float, timeliness: float, safety: float, communication: float, final_score: float) -> str:
    """
    Generates a PDF certificate (filename) for a vendor's performance.

    Args:
        vendor_name: Name of the vendor.
        month: The month the score is for.
        quality: Quality score.
        timeliness: Timeliness score.
        safety: Safety score.
        communication: Communication score.
        final_score: Final performance score.

    Returns:
        The filename of the generated PDF certificate.
    """
    # Placeholder for PDF generation
    # Replace with your actual PDF generation library (e.g., reportlab, pdfkit)
    # Example:  Create a dummy PDF file
    filename = f"{vendor_name}_Performance_Certificate_{month}.pdf"
    try:
        with open(filename, "w") as f:
            f.write(f"Vendor: {vendor_name}\n")
            f.write(f"Month: {month}\n")
            f.write(f"Quality: {quality:.2f}\n")
            f.write(f"Timeliness: {timeliness:.2f}\n")
            f.write(f"Safety: {safety:.2f}\n")
            f.write(f"Communication: {communication:.2f}\n")
            f.write(f"Final Score: {final_score:.2f}\n")
            f.write("--- Certificate of Performance ---\n")
        logging.info(f"Generated PDF certificate: {filename}")
    except Exception as e:
        logging.error(f"Error generating PDF: {e}")
        return None  # Indicate failure
    return filename  # Return the filename
def send_score_to_salesforce(vendor_id: str, month: str, quality: float, timeliness: float, safety: float, communication: float, final_score: float, certificate_url: str) -> bool:
    """
    Sends the vendor's performance score to Salesforce.
    THIS IS A PLACEHOLDER.  You MUST replace this with your actual
    Salesforce API integration code (e.g., using the Salesforce REST API).

    Args:
        vendor_id: The ID of the vendor.
        month: The month the score is for (YYYY-MM format).
        quality: Quality score.
        timeliness: Timeliness score.
        safety: Safety score.
        communication: Communication score.
        final_score: Final performance score.
        certificate_url: URL where the PDF certificate can be downloaded.

    Returns:
        True if the data was sent successfully, False otherwise.
    """
    # Placeholder for Salesforce API call
    # Replace with your actual Salesforce API integration
    logging.info(f"Sending score to Salesforce for Vendor ID: {vendor_id}, Month: {month}")
    logging.info(f"Quality: {quality}, Timeliness: {timeliness}, Safety: {safety}, Communication: {communication}, Final Score: {final_score}, Certificate URL: {certificate_url}")
    # Simulate a successful API call
    return True
def process_vendor_logs(vendor_logs: List[Dict]) -> List[Dict]:
    """
    Processes vendor logs, calculates scores, and prepares data for Salesforce and PDF.

    Args:
        vendor_logs: A list of dictionaries, where each dictionary represents a vendor log.
            Example:
            [
                {
                    'vendor_id': 'V123',
                    'work_completion_details': 'Work completed on time and to high standards.',
                    'delay_reports': '',
                    'incident_logs': '',
                    'log_date': '2024-07-28'
                },
                {
                    'vendor_id': 'V456',
                    'work_completion_details': 'Work completed with minor issues.',
                    'delay_reports': 'Minor delay due to weather.',
                    'incident_logs': 'Minor safety violation.',
                    'log_date': '2024-07-29'
                },
            ]

    Returns:
        A list of dictionaries, where each dictionary represents a vendor score record
        ready for sending to Salesforce.  Includes the certificate URL.
        Example:
        [
            {
                'vendor_id': 'V123',
                'month': '2024-07',
                'quality': 0.95,
                'timeliness': 1.0,
                'safety': 1.0,
                'communication': 0.9,
                'final_score': 0.96,
                'alert_flag': False,
                'certificate_url': 'https://example.com/certificates/V123_2024-07.pdf'
            },
            {
                'vendor_id': 'V456',
                'month': '2024-07',
                'quality': 0.7,
                'timeliness': 0.8,
                'safety': 0.6,
                'communication': 0.9,
                'final_score': 0.75,
                'alert_flag': True,
                'certificate_url': 'https://example.com/certificates/V456_2024-07.pdf'
            },
        ]
    """
    scores_for_salesforce = []
    for log in vendor_logs:
        vendor_id = log.get('vendor_id')
        work_completion_details = log.get('work_completion_details', '')
        delay_reports = log.get('delay_reports', '')
        incident_logs = log.get('incident_logs', '')
        log_date_str = log.get('log_date')
        try:
            log_date = datetime.strptime(log_date_str, '%Y-%m-%d')
            month = log_date.strftime('%Y-%m')  # Extract %Y-%m
        except ValueError:
            logging.error(f"Invalid date format: {log_date_str} for vendor {vendor_id}. Skipping log.")
            continue  # Skip this log
        quality = calculate_quality_score(work_completion_details)
        timeliness = calculate_timeliness_score(delay_reports)
        safety = calculate_safety_score(incident_logs)
        communication = calculate_communication_score()
        final_score = calculate_final_score(quality, timeliness, safety, communication)
        historical_scores = get_historical_scores(vendor_id) #get historical scores
        trend_deviation = analyze_trends(vendor_id, historical_scores)
        alert_flag = final_score < ALERT_THRESHOLD
        # Generate PDF certificate
        certificate_filename = generate_pdf_certificate(
            vendor_id,  # Use vendor ID as name
            month,
            quality,
            timeliness,
            safety,
            communication,
            final_score
        )
        if certificate_filename:
            certificate_url = f"https://your-domain.com/certificates/{certificate_filename}"  # Replace with your actual URL
        else:
            certificate_url = ""
        scores_for_salesforce.append({
            'vendor_id': vendor_id,
            'month': month,
            'quality': quality,
            'timeliness': timeliness,
            'safety': safety,
            'communication': communication,
            'final_score': final_score,
            'alert_flag': alert_flag,
            'certificate_url': certificate_url,
            'trend_deviation': trend_deviation # Include trend deviation
        })
    return scores_for_salesforce
def get_lowest_score(scores: List[Dict]) -> Optional[Dict]:
    """
    Finds the vendor with the lowest final score.

    Args:
        scores: A list of vendor score dictionaries.

    Returns:
        The dictionary representing the vendor with the lowest score, or None if the
        list is empty.
    """
    if not scores:
        return None
    lowest_score_vendor = scores[0]
    for vendor_score in scores:
        if vendor_score['final_score'] < lowest_score_vendor['final_score']:
            lowest_score_vendor = vendor_score
    return lowest_score_vendor
def send_alert_message(vendor_id: str, month: str, final_score: float):
    """
    Sends an alert message to the vendor with the lowest score.
    THIS IS A PLACEHOLDER.  You will need to replace this with your actual
    email sending or notification logic (e.g., using a library like `smtplib`
    or an API like SendGrid, or by triggering a Salesforce alert).

    Args:
        vendor_id: The ID of the vendor.
        month: The month of the score.
        final_score: The vendor's final score.
    """
    # Placeholder for alert message sending
    # Replace with your actual email or notification logic
    alert_message = (
        f"Subject: Performance Alert - Subcontractor {vendor_id} - {month}\n\n"
        f"Your performance score for {month} is {final_score:.2f}, which is below the acceptable threshold.\n"
        f"Please review your performance and take corrective action."
    )
    logging.warning(f"Sending alert message: {alert_message}")
    #  Send email (replace with real email sending)
    # send_email(recipient="vendor_email@example.com", message=alert_message)
def process_and_send_data(vendor_logs: List[Dict]) -> bool:
    """
    Processes vendor logs, sends data to Salesforce, and handles alerts.

    Args:
        vendor_logs:  A list of dictionaries, where each dictionary represents a vendor log.


    Returns:
        True if the entire process was successful, False otherwise.
    """
    try:
        scores_for_salesforce = process_vendor_logs(vendor_logs)
        if not scores_for_salesforce:
            logging.warning("No scores to send to Salesforce.")
            return True  # Consider this a "successful" no-op
        # Send data to Salesforce
        all_salesforce_updates_successful = True #track
        for score_data in scores_for_salesforce:
            salesforce_success = send_score_to_salesforce(
                vendor_id=score_data['vendor_id'],
                month=score_data['month'],
                quality=score_data['quality'],
                timeliness=score_data['timeliness'],
                safety=score_data['safety'],
                communication=score_data['communication'],
                final_score=score_data['final_score'],
                certificate_url=score_data['certificate_url']
            )
            if not salesforce_success:
                all_salesforce_updates_successful = False
                logging.error(f"Failed to send score for vendor {score_data['vendor_id']} to Salesforce.")
        if not all_salesforce_updates_successful:
            return False
        # Handle alerts
        lowest_scoring_vendor = get_lowest_score(scores_for_salesforce)
        if lowest_scoring_vendor and lowest_scoring_vendor['alert_flag']:
            send_alert_message(
                vendor_id=lowest_scoring_vendor['vendor_id'],
                month=lowest_scoring_vendor['month'],
                final_score=lowest_scoring_vendor['final_score']
            )
        return True
    except Exception as e:
        logging.error(f"Error processing and sending data: {e}")
        return False
if __name__ == "__main__":
    # --- Example Usage (for testing) ---
    #  Replace this with actual data from Salesforce/your vendor portal
    sample_vendor_logs = [
        {
            'vendor_id': 'V123',
            'work_completion_details': 'Excellent work, completed ahead of schedule.  The team demonstrated high quality and followed best practices.',
            'delay_reports': '',
            'incident_logs': '',
            'log_date': '2024-07-28'
        },
        {
            'vendor_id': 'V456',
            'work_completion_details': 'Work completed with minor issues; some rework was required.',
            'delay_reports': 'Minor delay due to weather conditions.',
            'incident_logs': 'Minor safety violation: worker not wearing proper PPE.',
            'log_date': '2024-07-29'
        },
         {
            'vendor_id': 'V789',
            'work_completion_details': 'Work completed on time.',
            'delay_reports': 'Major delay due to lack of materials',
            'incident_logs': '',
            'log_date': '2024-07-29'
        },
    ]
    # Set up basic logging (replace with your desired configuration)
    logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
    success = process_and_send_data(sample_vendor_logs)
    if success:
        logging.info("Data processing and sending successful.")
    else:
        logging.error("Data processing and sending failed.")

 # Set up basic logging (replace with your desired configuration)
    logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
    success = process_and_send_data(sample_vendor_logs)
    if success:
        logging.info("Data processing and sending successful.")
    else:
        logging.error("Data processing and sending failed.")