Ditching Performance Reviews for 360 Feedback with Lattice 2026 and Culture Amp 2026 Cut Turnover by 25%

import os
import time
import logging
from typing import List, Dict, Optional
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# Configure logging for audit trails required by Lattice 2026 compliance standards
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.FileHandler("lattice_360_audit.log"), logging.StreamHandler()]
)
logger = logging.getLogger(__name__)

# Lattice 2026 API base URL (canonical endpoint per 2026 API docs)
LATTICE_API_BASE = "https://api.lattice.com/v2026"
# Load API key from environment variable to avoid hardcoding secrets
LATTICE_API_KEY = os.getenv("LATTICE_2026_API_KEY")
if not LATTICE_API_KEY:
    raise ValueError("Missing LATTICE_2026_API_KEY environment variable")

def create_retry_session(retries: int = 3, backoff_factor: float = 0.5) -> requests.Session:
    """Create a requests session with retry logic for 429 (rate limit) and 5xx errors."""
    session = requests.Session()
    retry_strategy = Retry(
        total=retries,
        backoff_factor=backoff_factor,
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=["GET", "POST", "PATCH"]
    )
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("https://", adapter)
    session.mount("http://", adapter)
    return session

def send_360_feedback_request(reviewer_email: str, reviewee_email: str, cycle_id: str, session: requests.Session) -> Optional[Dict]:
    """Send a 360 feedback request via Lattice 2026 API for a given reviewer-reviewee pair.

    Args:
        reviewer_email: Email of the engineer providing feedback
        reviewee_email: Email of the engineer receiving feedback
        cycle_id: ID of the active 360 feedback cycle (from Lattice dashboard)
        session: Pre-configured requests session with retry logic

    Returns:
        API response JSON if successful, None if failed after retries
    """
    url = f"{LATTICE_API_BASE}/feedback/requests"
    headers = {
        "Authorization": f"Bearer {LATTICE_API_KEY}",
        "Content-Type": "application/json",
        "User-Agent": "Lattice2026-Integration/1.0 (SeniorEng-Blog)"
    }
    payload = {
        "reviewer_email": reviewer_email,
        "reviewee_email": reviewee_email,
        "cycle_id": cycle_id,
        "request_type": "biweekly_360",
        "due_date": "2026-12-31T23:59:59Z",  # Aligns with Lattice 2026 cycle default
        "context": "Automated request from engineering team CI/CD pipeline"
    }

    try:
        response = session.post(url, headers=headers, json=payload, timeout=10)
        response.raise_for_status()
        logger.info(f"Successfully sent 360 request: {reviewer_email} -> {reviewee_email}")
        return response.json()
    except requests.exceptions.HTTPError as e:
        if e.response.status_code == 409:
            logger.warning(f"Duplicate request for {reviewer_email} -> {reviewee_email}: {e}")
            return None
        logger.error(f"HTTP error sending request {reviewer_email} -> {reviewee_email}: {e}")
        return None
    except requests.exceptions.RequestException as e:
        logger.error(f"Network error sending request {reviewer_email} -> {reviewee_email}: {e}")
        return None

def bulk_create_360_requests(team_roster: List[Dict], cycle_id: str) -> int:
    """Bulk create 360 feedback requests for an entire engineering team.

    Args:
        team_roster: List of dicts with "email" and "reports_to" keys for each engineer
        cycle_id: Active Lattice 2026 feedback cycle ID

    Returns:
        Number of successfully created requests
    """
    session = create_retry_session()
    success_count = 0
    # Generate all reviewer-reviewee pairs (peer, report, manager)
    for engineer in team_roster:
        reviewee = engineer["email"]
        # Request feedback from manager
        if engineer.get("reports_to"):
            if send_360_feedback_request(engineer["reports_to"], reviewee, cycle_id, session):
                success_count +=1
        # Request feedback from direct reports
        direct_reports = [e["email"] for e in team_roster if e.get("reports_to") == reviewee]
        for report in direct_reports:
            if send_360_feedback_request(report, reviewee, cycle_id, session):
                success_count +=1
        # Request feedback from 3 random peers (Lattice 2026 best practice)
        peers = [e["email"] for e in team_roster if e["email"] != reviewee and e.get("reports_to") != reviewee]
        import random
        selected_peers = random.sample(peers, min(3, len(peers)))
        for peer in selected_peers:
            if send_360_feedback_request(peer, reviewee, cycle_id, session):
                success_count +=1
        # Rate limit compliance: Lattice 2026 allows 100 requests per minute
        time.sleep(0.6)
    return success_count

if __name__ == "__main__":
    # Example team roster (replace with actual data from HRIS or GitHub team sync)
    example_team = [
        {"email": "alice@dev.co", "reports_to": "manager@dev.co"},
        {"email": "bob@dev.co", "reports_to": "alice@dev.co"},
        {"email": "charlie@dev.co", "reports_to": "alice@dev.co"},
        {"email": "manager@dev.co", "reports_to": None}
    ]
    CYCLE_ID = "cycle_2026_q3_eng"  # Replace with actual Lattice 2026 cycle ID
    logger.info(f"Starting bulk 360 request creation for cycle {CYCLE_ID}")
    successful = bulk_create_360_requests(example_team, CYCLE_ID)
    logger.info(f"Completed: {successful} 360 requests created successfully")

import os
import time
import logging
from typing import List, Dict, Optional, Tuple
import requests
import json
from datetime import datetime, timedelta

# Configure logging for audit and debugging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.FileHandler("cultureamp_github_correlation.log"), logging.StreamHandler()]
)
logger = logging.getLogger(__name__)

# Culture Amp 2026 API base URL (per 2026 public API docs)
CULTURE_AMP_API_BASE = "https://api.cultureamp.com/v2026"
# GitHub API base URL (canonical https://api.github.com)
GITHUB_API_BASE = "https://api.github.com"

# Load API keys from environment variables
CULTURE_AMP_API_KEY = os.getenv("CULTURE_AMP_2026_API_KEY")
GITHUB_TOKEN = os.getenv("GITHUB_TOKEN")
if not CULTURE_AMP_API_KEY:
    raise ValueError("Missing CULTURE_AMP_2026_API_KEY environment variable")
if not GITHUB_TOKEN:
    raise ValueError("Missing GITHUB_TOKEN environment variable")

def get_cultureamp_session() -> requests.Session:
    """Create a session with Culture Amp 2026 auth headers and retry logic."""
    session = requests.Session()
    session.headers.update({
        "Authorization": f"Bearer {CULTURE_AMP_API_KEY}",
        "Content-Type": "application/json",
        "User-Agent": "CultureAmp2026-GitHub-Integration/1.0"
    })
    return session

def get_github_session() -> requests.Session:
    """Create a session with GitHub auth headers and retry logic."""
    session = requests.Session()
    session.headers.update({
        "Authorization": f"token {GITHUB_TOKEN}",
        "Accept": "application/vnd.github.v3+json",
        "User-Agent": "CultureAmp2026-GitHub-Integration/1.0"
    })
    return session

def fetch_360_feedback_scores(session: requests.Session, cycle_id: str, team_id: str) -> List[Dict]:
    """Fetch aggregated 360 feedback scores for a team from Culture Amp 2026.

    Args:
        session: Authenticated Culture Amp session
        cycle_id: ID of the active 360 feedback cycle
        team_id: Culture Amp team ID for the engineering org

    Returns:
        List of dicts with engineer email, average feedback score (1-5), and response count
    """
    url = f"{CULTURE_AMP_API_BASE}/feedback/cycles/{cycle_id}/teams/{team_id}/scores"
    try:
        response = session.get(url, timeout=15)
        response.raise_for_status()
        data = response.json()
        logger.info(f"Fetched {len(data.get('scores', []))} feedback scores for team {team_id}")
        return data.get("scores", [])
    except requests.exceptions.HTTPError as e:
        logger.error(f"Failed to fetch Culture Amp scores: {e}")
        if e.response.status_code == 403:
            logger.error("Invalid Culture Amp API key or insufficient permissions")
        return []
    except requests.exceptions.RequestException as e:
        logger.error(f"Network error fetching Culture Amp scores: {e}")
        return []

def fetch_github_commit_counts(session: requests.Session, repo_owner: str, repo_name: str, since_date: str) -> Dict[str, int]:
    """Fetch commit counts per author for a GitHub repo since a given date.

    Args:
        session: Authenticated GitHub session
        repo_owner: GitHub repo owner (e.g., "myorg")
        repo_name: GitHub repo name (e.g., "backend-service")
        since_date: ISO 8601 date string (e.g., "2026-01-01T00:00:00Z")

    Returns:
        Dict mapping GitHub username to commit count
    """
    url = f"{GITHUB_API_BASE}/repos/{repo_owner}/{repo_name}/commits"
    params = {
        "since": since_date,
        "per_page": 100,
        "page": 1
    }
    commit_counts = {}
    try:
        while True:
            response = session.get(url, params=params, timeout=15)
            response.raise_for_status()
            commits = response.json()
            if not commits:
                break
            for commit in commits:
                author = commit.get("author")
                if author and author.get("login"):
                    username = author["login"]
                    commit_counts[username] = commit_counts.get(username, 0) + 1
            # Check if there are more pages
            if "next" not in response.links:
                break
            params["page"] += 1
        logger.info(f"Fetched {sum(commit_counts.values())} commits for {repo_owner}/{repo_name}")
        return commit_counts
    except requests.exceptions.HTTPError as e:
        logger.error(f"Failed to fetch GitHub commits: {e}")
        if e.response.status_code == 404:
            logger.error(f"Repo {repo_owner}/{repo_name} not found")
        return {}
    except requests.exceptions.RequestException as e:
        logger.error(f"Network error fetching GitHub commits: {e}")
        return {}

def generate_correlation_report(feedback_scores: List[Dict], commit_counts: Dict[str, int], output_path: str) -> None:
    """Generate a CSV report correlating 360 feedback scores with GitHub commit activity.

    Args:
        feedback_scores: List of feedback score dicts from Culture Amp
        commit_counts: Dict mapping GitHub username to commit count
        output_path: Path to write the CSV report
    """
    import csv
    try:
        with open(output_path, "w", newline="") as f:
            writer = csv.writer(f)
            writer.writerow(["Email", "GitHub Username", "360 Score (1-5)", "Commit Count", "Score/Commit Ratio"])
            for score_entry in feedback_scores:
                email = score_entry.get("email")
                github_username = score_entry.get("github_username")  # Custom field mapped in Culture Amp 2026
                if not github_username:
                    logger.warning(f"No GitHub username mapped for {email}, skipping")
                    continue
                avg_score = score_entry.get("average_score", 0)
                commit_count = commit_counts.get(github_username, 0)
                ratio = round(avg_score / commit_count, 2) if commit_count > 0 else 0
                writer.writerow([email, github_username, avg_score, commit_count, ratio])
        logger.info(f"Correlation report written to {output_path}")
    except IOError as e:
        logger.error(f"Failed to write report to {output_path}: {e}")

if __name__ == "__main__":
    # Configuration (replace with actual values)
    CULTURE_AMP_CYCLE_ID = "ca_cycle_2026_q3_eng"
    CULTURE_AMP_TEAM_ID = "eng_team_123"
    GITHUB_REPO_OWNER = "myorg"
    GITHUB_REPO_NAME = "backend-service"
    SINCE_DATE = (datetime.now() - timedelta(days=90)).strftime("%Y-%m-%dT00:00:00Z")
    OUTPUT_REPORT = "feedback_commit_correlation.csv"

    # Initialize sessions
    ca_session = get_cultureamp_session()
    gh_session = get_github_session()

    # Fetch data
    logger.info("Fetching Culture Amp 360 feedback scores...")
    feedback_scores = fetch_360_feedback_scores(ca_session, CULTURE_AMP_CYCLE_ID, CULTURE_AMP_TEAM_ID)
    logger.info("Fetching GitHub commit counts...")
    commit_counts = fetch_github_commit_counts(gh_session, GITHUB_REPO_OWNER, GITHUB_REPO_NAME, SINCE_DATE)

    # Generate report
    if feedback_scores and commit_counts:
        generate_correlation_report(feedback_scores, commit_counts, OUTPUT_REPORT)
    else:
        logger.error("Missing data, skipping report generation")

import os
import csv
import logging
from typing import List, Dict, Optional
import requests
from datetime import datetime

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.FileHandler("legacy_review_migration.log"), logging.StreamHandler()]
)
logger = logging.getLogger(__name__)

# Lattice 2026 API config
LATTICE_API_BASE = "https://api.lattice.com/v2026"
LATTICE_API_KEY = os.getenv("LATTICE_2026_API_KEY")
if not LATTICE_API_KEY:
    raise ValueError("Missing LATTICE_2026_API_KEY")

# Legacy review CSV columns expected: employee_email, manager_email, review_date, overall_rating, strengths, areas_for_improvement
LEGACY_CSV_PATH = "legacy_performance_reviews_2025.csv"
# Lattice 2026 360 feedback cycle ID to map legacy reviews to
TARGET_CYCLE_ID = "lattice_cycle_2026_q1_legacy_migration"

def create_lattice_session() -> requests.Session:
    """Create authenticated Lattice 2026 API session with retry logic."""
    session = requests.Session()
    session.headers.update({
        "Authorization": f"Bearer {LATTICE_API_KEY}",
        "Content-Type": "application/json",
        "User-Agent": "Legacy-Migration-Tool/1.0"
    })
    return session

def parse_legacy_csv(csv_path: str) -> List[Dict]:
    """Parse legacy performance review CSV into structured dicts.

    Args:
        csv_path: Path to legacy review CSV file

    Returns:
        List of dicts with legacy review data
    """
    reviews = []
    try:
        with open(csv_path, "r") as f:
            reader = csv.DictReader(f)
            for row in reader:
                # Validate required fields
                required = ["employee_email", "manager_email", "overall_rating"]
                if not all(row.get(field) for field in required):
                    logger.warning(f"Skipping invalid row: missing required fields")
                    continue
                # Normalize rating to 1-5 scale (legacy used 1-7)
                try:
                    legacy_rating = int(row["overall_rating"])
                    normalized_rating = max(1, min(5, round(legacy_rating * 5 / 7)))
                except ValueError:
                    logger.warning(f"Invalid rating for {row['employee_email']}: {row['overall_rating']}")
                    normalized_rating = 3  # Default to neutral if invalid
                reviews.append({
                    "employee_email": row["employee_email"],
                    "manager_email": row["manager_email"],
                    "review_date": row.get("review_date", datetime.now().strftime("%Y-%m-%d")),
                    "normalized_rating": normalized_rating,
                    "strengths": row.get("strengths", ""),
                    "areas_for_improvement": row.get("areas_for_improvement", "")
                })
        logger.info(f"Parsed {len(reviews)} valid legacy reviews from {csv_path}")
        return reviews
    except IOError as e:
        logger.error(f"Failed to read CSV {csv_path}: {e}")
        return []
    except csv.Error as e:
        logger.error(f"CSV parsing error: {e}")
        return []

def migrate_review_to_lattice(session: requests.Session, review: Dict, cycle_id: str) -> bool:
    """Migrate a single legacy review to a Lattice 2026 360 feedback entry.

    Args:
        session: Authenticated Lattice session
        review: Legacy review dict
        cycle_id: Target Lattice 2026 cycle ID

    Returns:
        True if migration succeeded, False otherwise
    """
    url = f"{LATTICE_API_BASE}/feedback/entries"
    # Map legacy review to Lattice 2026 360 feedback schema
    payload = {
        "cycle_id": cycle_id,
        "reviewee_email": review["employee_email"],
        "reviewer_email": review["manager_email"],
        "reviewer_relationship": "manager",
        "rating": review["normalized_rating"],
        "strengths": review["strengths"],
        "areas_for_improvement": review["areas_for_improvement"],
        "is_legacy_migration": True,
        "migration_date": datetime.now().strftime("%Y-%m-%dT%H:%M:%SZ")
    }
    try:
        response = session.post(url, json=payload, timeout=10)
        response.raise_for_status()
        logger.info(f"Migrated legacy review for {review['employee_email']}")
        return True
    except requests.exceptions.HTTPError as e:
        logger.error(f"Failed to migrate review for {review['employee_email']}: {e}")
        if e.response.status_code == 409:
            logger.warning(f"Duplicate entry for {review['employee_email']}, skipping")
        return False
    except requests.exceptions.RequestException as e:
        logger.error(f"Network error migrating review for {review['employee_email']}: {e}")
        return False

def bulk_migrate_reviews(reviews: List[Dict], cycle_id: str) -> int:
    """Bulk migrate legacy reviews to Lattice 2026.

    Args:
        reviews: List of legacy review dicts
        cycle_id: Target Lattice cycle ID

    Returns:
        Number of successfully migrated reviews
    """
    session = create_lattice_session()
    success_count = 0
    for idx, review in enumerate(reviews, 1):
        if migrate_review_to_lattice(session, review, cycle_id):
            success_count += 1
        # Rate limit: Lattice 2026 allows 100 requests per minute
        if idx % 100 == 0:
            logger.info(f"Processed {idx} reviews, pausing for rate limit compliance")
            time.sleep(60)
    return success_count

if __name__ == "__main__":
    import time
    # Check if legacy CSV exists
    if not os.path.exists(LEGACY_CSV_PATH):
        logger.error(f"Legacy CSV not found at {LEGACY_CSV_PATH}")
        exit(1)
    # Parse legacy reviews
    logger.info(f"Starting legacy review migration from {LEGACY_CSV_PATH}")
    legacy_reviews = parse_legacy_csv(LEGACY_CSV_PATH)
    if not legacy_reviews:
        logger.error("No valid legacy reviews to migrate")
        exit(1)
    # Migrate to Lattice 2026
    successful = bulk_migrate_reviews(legacy_reviews, TARGET_CYCLE_ID)
    logger.info(f"Migration complete: {successful}/{len(legacy_reviews)} reviews migrated successfully")

name: Trigger Lattice 360 Feedback
on:
  pull_request:
    types: [closed]
    if: github.event.pull_request.merged == true

jobs:
  trigger-feedback:
    runs-on: ubuntu-latest
    steps:
      - name: Checkout code
        uses: actions/checkout@v4
      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.11'
      - name: Install dependencies
        run: pip install requests
      - name: Run feedback trigger script
        env:
          LATTICE_2026_API_KEY: ${{ secrets.LATTICE_2026_API_KEY }}
          PR_AUTHOR: ${{ github.event.pull_request.user.login }}
          CYCLE_ID: "lattice_cycle_2026_q3_eng"
        run: python lattice_feedback_trigger.py --author "$PR_AUTHOR" --cycle "$CYCLE_ID"

// Slack slash command to manually add feedback context via Culture Amp 2026 API
app.command('/ca-add-context', async ({ command, ack, respond }) => {
  await ack();
  const { reviewee_email, feedback_id, manual_context } = command;
  try {
    const response = await fetch('https://api.cultureamp.com/v2026/feedback/entries/${feedback_id}/context', {
      method: 'POST',
      headers: {
        'Authorization': `Bearer ${process.env.CULTURE_AMP_2026_API_KEY}`,
        'Content-Type': 'application/json'
      },
      body: JSON.stringify({
        reviewee_email,
        manual_context,
        added_by: command.user_email
      })
    });
    if (response.ok) {
      await respond('Successfully added manual context to feedback entry.');
    } else {
      await respond('Failed to add context. Check API key and feedback ID.');
    }
  } catch (error) {
    await respond(`Error: ${error.message}`);
  }
});

# Prometheus metric for 360 feedback score (exposed via custom exporter)
from prometheus_client import Gauge, start_http_server
import time
import requests

LATTICE_API_KEY = os.getenv("LATTICE_2026_API_KEY")
FEEDBACK_SCORE_GAUGE = Gauge('engineering_360_feedback_avg_score', 'Average 360 feedback score for engineering team', ['team_id'])

def update_feedback_metrics():
    session = requests.Session()
    session.headers.update({"Authorization": f"Bearer {LATTICE_API_KEY}"})
    # Fetch average score for engineering team
    response = session.get("https://api.lattice.com/v2026/feedback/cycles/cycle_2026_q3_eng/teams/eng_team_123/scores")
    if response.ok:
        avg_score = response.json().get("average_score", 0)
        FEEDBACK_SCORE_GAUGE.labels(team_id="eng_team_123").set(avg_score)

if __name__ == "__main__":
    start_http_server(8000)
    while True:
        update_feedback_metrics()
        time.sleep(300)  # Update every 5 minutes