How to Handle a Toxic Engineering Manager and Get a Promotion in 6 Months

import json
import os
import datetime
from typing import List, Dict, Optional
from dataclasses import dataclass, asdict

# Data class to represent a single high-impact work item
@dataclass
class WorkItem:
    id: str
    title: str
    description: str
    start_date: datetime.date
    end_date: Optional[datetime.date]
    business_value_usd: float  # Estimated annual recurring revenue impact
    technical_scope: List[str]  # e.g., ["backend", "api", "postgres"]
    stakeholders: List[str]  # e.g., ["product", "sales", "support"]
    evidence_links: List[str]  # e.g., PR links, Grafana dashboards

class ImpactTracker:
    """Persists work items, calculates aggregate impact, and exports promotion-ready reports."""

    def __init__(self, storage_path: str = "impact_data.json"):
        self.storage_path = storage_path
        self.work_items: List[WorkItem] = []
        self._load_existing_items()

    def _load_existing_items(self) -> None:
        """Load existing work items from JSON storage, handle missing/corrupt files."""
        if not os.path.exists(self.storage_path):
            return
        try:
            with open(self.storage_path, "r") as f:
                raw_items = json.load(f)
                for item in raw_items:
                    # Convert ISO format strings back to date objects
                    item["start_date"] = datetime.date.fromisoformat(item["start_date"])
                    if item["end_date"]:
                        item["end_date"] = datetime.date.fromisoformat(item["end_date"])
                    self.work_items.append(WorkItem(**item))
        except json.JSONDecodeError as e:
            print(f"Corrupt storage file: {e}. Starting with empty work item list.")
        except Exception as e:
            print(f"Failed to load work items: {e}. Starting fresh.")

    def add_work_item(self, item: WorkItem) -> None:
        """Add a new work item, validate inputs, and persist to storage."""
        if item.business_value_usd < 0:
            raise ValueError("Business value cannot be negative")
        if item.end_date and item.end_date < item.start_date:
            raise ValueError("End date cannot be before start date")
        self.work_items.append(item)
        self._persist_items()

    def _persist_items(self) -> None:
        """Serialize work items to JSON, handle write errors."""
        try:
            with open(self.storage_path, "w") as f:
                # Convert date objects to ISO format for JSON serialization
                serialized = []
                for item in self.work_items:
                    item_dict = asdict(item)
                    item_dict["start_date"] = item_dict["start_date"].isoformat()
                    item_dict["end_date"] = item_dict["end_date"].isoformat() if item_dict["end_date"] else None
                    serialized.append(item_dict)
                json.dump(serialized, f, indent=2)
        except IOError as e:
            print(f"Failed to write to storage: {e}")

    def calculate_quarterly_impact(self) -> float:
        """Sum business value of all items completed in the last 90 days."""
        cutoff = datetime.date.today() - datetime.timedelta(days=90)
        total = 0.0
        for item in self.work_items:
            if item.end_date and item.end_date >= cutoff:
                total += item.business_value_usd
        return total

    def export_promotion_packet(self, output_path: str = "promotion_packet.md") -> None:
        """Generate a markdown promotion packet with all work items and aggregate metrics."""
        try:
            with open(output_path, "w") as f:
                f.write(f"# Promotion Packet: {datetime.date.today().isoformat()}\n\n")
                f.write(f"## Aggregate Impact\n")
                f.write(f"- Total Work Items: {len(self.work_items)}\n")
                f.write(f"- Quarterly Business Value: ${self.calculate_quarterly_impact():,.2f}\n")
                f.write(f"- Technical Scopes Covered: {len(set(scope for item in self.work_items for scope in item.technical_scope))}\n\n")
                f.write(f"## Work Item Details\n")
                for item in sorted(self.work_items, key=lambda x: x.end_date or datetime.date.max):
                    f.write(f"### {item.title}\n")
                    f.write(f"- **ID**: {item.id}\n")
                    f.write(f"- **Dates**: {item.start_date} to {item.end_date or 'In Progress'}\n")
                    f.write(f"- **Business Value**: ${item.business_value_usd:,.2f}\n")
                    f.write(f"- **Scope**: {', '.join(item.technical_scope)}\n")
                    f.write(f"- **Stakeholders**: {', '.join(item.stakeholders)}\n")
                    f.write(f"- **Evidence**: {', '.join(item.evidence_links)}\n")
                    f.write(f"- **Description**: {item.description}\n\n")
        except IOError as e:
            print(f"Failed to export promotion packet: {e}")

if __name__ == "__main__":
    # Example usage: track a high-impact API migration
    tracker = ImpactTracker()
    example_item = WorkItem(
        id="api-migration-2024-q3",
        title="Legacy REST API to gRPC Migration",
        description="Migrated 12 legacy REST endpoints to gRPC, reducing p99 latency by 82% and saving $140k/year in infrastructure costs",
        start_date=datetime.date(2024, 7, 1),
        end_date=datetime.date(2024, 9, 15),
        business_value_usd=140000.0,
        technical_scope=["backend", "grpc", "postgres", "load-balancing"],
        stakeholders=["product", "infrastructure", "support"],
        evidence_links=[
            "https://github.com/your-org/api-migration/pull/123",
            "https://grafana.your-org.com/dashboard/api-latency"
        ]
    )
    try:
        tracker.add_work_item(example_item)
        print(f"Added work item. Quarterly impact: ${tracker.calculate_quarterly_impact():,.2f}")
        tracker.export_promotion_packet()
        print("Exported promotion packet to promotion_packet.md")
    except ValueError as e:
        print(f"Invalid work item: {e}")

import os
import requests
import datetime
import json
from typing import List, Dict, Optional
from dotenv import load_dotenv

load_dotenv()  # Load GITHUB_TOKEN and JIRA_TOKEN from .env file

class WorkEvidenceScraper:
    """Automates collection of engineering work evidence from GitHub, Jira, and CI systems."""

    def __init__(self, github_org: str, jira_project: str):
        self.github_token = os.getenv("GITHUB_TOKEN")
        self.jira_token = os.getenv("JIRA_TOKEN")
        self.github_org = github_org
        self.jira_project = jira_project
        self.github_headers = {
            "Authorization": f"token {self.github_token}",
            "Accept": "application/vnd.github.v3+json"
        }
        self.jira_headers = {
            "Authorization": f"Bearer {self.jira_token}",
            "Accept": "application/json"
        }
        self._validate_tokens()

    def _validate_tokens(self) -> None:
        """Check that required API tokens are present and valid."""
        if not self.github_token:
            raise ValueError("Missing GITHUB_TOKEN in environment variables")
        if not self.jira_token:
            raise ValueError("Missing JIRA_TOKEN in environment variables")
        # Test GitHub token
        try:
            resp = requests.get("https://api.github.com/user", headers=self.github_headers)
            resp.raise_for_status()
        except requests.exceptions.HTTPError as e:
            raise ValueError(f"Invalid GitHub token: {e}")
        # Test Jira token
        try:
            resp = requests.get(
                f"https://your-jira-instance.atlassian.net/rest/api/3/project/{self.jira_project}",
                headers=self.jira_headers
            )
            resp.raise_for_status()
        except requests.exceptions.HTTPError as e:
            raise ValueError(f"Invalid Jira token: {e}")

    def fetch_github_prs(self, repo: str, days_back: int = 90) -> List[Dict]:
        """Fetch all merged PRs authored by the current user in the last N days."""
        cutoff_date = (datetime.date.today() - datetime.timedelta(days=days_back)).isoformat()
        url = f"https://api.github.com/repos/{self.github_org}/{repo}/pulls"
        params = {
            "state": "closed",
            "sort": "updated",
            "direction": "desc",
            "per_page": 100,
            "page": 1
        }
        all_prs = []
        while True:
            try:
                resp = requests.get(url, headers=self.github_headers, params=params)
                resp.raise_for_status()
                prs = resp.json()
                if not prs:
                    break
                for pr in prs:
                    # Filter for merged PRs by current user, within date range
                    if (pr.get("merged_at") and 
                        pr["merged_at"] >= cutoff_date and 
                        pr["user"]["login"] == self._get_github_username()):
                        all_prs.append({
                            "id": pr["number"],
                            "title": pr["title"],
                            "url": pr["html_url"],
                            "merged_at": pr["merged_at"],
                            "additions": pr["additions"],
                            "deletions": pr["deletions"],
                            "changed_files": pr["changed_files"]
                        })
                params["page"] += 1
            except requests.exceptions.HTTPError as e:
                print(f"Failed to fetch PRs for {repo}: {e}")
                break
        return all_prs

    def _get_github_username(self) -> str:
        """Retrieve the authenticated GitHub username."""
        resp = requests.get("https://api.github.com/user", headers=self.github_headers)
        resp.raise_for_status()
        return resp.json()["login"]

    def fetch_jira_tickets(self, days_back: int = 90) -> List[Dict]:
        """Fetch all completed Jira tickets assigned to the current user in the last N days."""
        cutoff_date = (datetime.date.today() - datetime.timedelta(days=days_back)).isoformat()
        url = f"https://your-jira-instance.atlassian.net/rest/api/3/search"
        params = {
            "jql": f"project={self.jira_project} AND assignee=currentUser() AND status=Done AND updated>= {cutoff_date}",
            "maxResults": 100,
            "startAt": 0
        }
        all_tickets = []
        while True:
            try:
                resp = requests.get(url, headers=self.jira_headers, params=params)
                resp.raise_for_status()
                data = resp.json()
                if not data["issues"]:
                    break
                for issue in data["issues"]:
                    all_tickets.append({
                        "key": issue["key"],
                        "summary": issue["fields"]["summary"],
                        "url": f"https://your-jira-instance.atlassian.net/browse/{issue['key']}",
                        "status": issue["fields"]["status"]["name"],
                        "resolution_date": issue["fields"]["resolutiondate"]
                    })
                params["startAt"] += 100
            except requests.exceptions.HTTPError as e:
                print(f"Failed to fetch Jira tickets: {e}")
                break
        return all_tickets

    def export_evidence_bundle(self, output_path: str = "evidence_bundle.json") -> None:
        """Export all collected evidence to a single JSON file for promotion packets."""
        evidence = {
            "generated_at": datetime.date.today().isoformat(),
            "github_prs": {},
            "jira_tickets": self.fetch_jira_tickets()
        }
        # Fetch PRs for all repos in the org (simplified for example)
        repos = ["api-service", "frontend-app", "infra-terraform"]
        for repo in repos:
            evidence["github_prs"][repo] = self.fetch_github_prs(repo)
        try:
            with open(output_path, "w") as f:
                json.dump(evidence, f, indent=2)
            print(f"Exported evidence bundle to {output_path}")
        except IOError as e:
            print(f"Failed to export evidence bundle: {e}")

if __name__ == "__main__":
    try:
        scraper = WorkEvidenceScraper(github_org="your-org", jira_project="ENG")
        scraper.export_evidence_bundle()
        # Print summary
        print(f"Collected {len(scraper.fetch_jira_tickets())} Jira tickets")
    except ValueError as e:
        print(f"Configuration error: {e}")

import json
import os
from typing import List, Dict, Tuple
from dataclasses import dataclass

@dataclass
class PromotionCriteria:
    """Defines the criteria for a promotion at a given level (e.g., Senior to Staff)."""
    min_work_items: int
    min_quarterly_value_usd: float
    min_technical_scopes: int
    required_stakeholder_types: List[str]  # e.g., ["product", "infrastructure"]
    min_evidence_items: int  # Min number of PRs, tickets, dashboards per work item

class PromotionPacketValidator:
    """Validates that a promotion packet meets all company-defined criteria."""

    def __init__(self, criteria: PromotionCriteria, packet_path: str = "promotion_packet.md", evidence_path: str = "evidence_bundle.json"):
        self.criteria = criteria
        self.packet_path = packet_path
        self.evidence_path = evidence_path
        self.validation_errors: List[str] = []

    def validate_all(self) -> Tuple[bool, List[str]]:
        """Run all validation checks, return pass/fail and list of errors."""
        self.validation_errors = []
        self._validate_packet_exists()
        self._validate_work_item_count()
        self._validate_quarterly_value()
        self._validate_technical_scope()
        self._validate_stakeholders()
        self._validate_evidence()
        return (len(self.validation_errors) == 0, self.validation_errors)

    def _validate_packet_exists(self) -> None:
        if not os.path.exists(self.packet_path):
            self.validation_errors.append(f"Promotion packet not found at {self.packet_path}")

    def _validate_work_item_count(self) -> None:
        """Count work items in the promotion packet (simplified parsing for example)."""
        try:
            with open(self.packet_path, "r") as f:
                content = f.read()
                # Count work item headers (### Title)
                work_item_count = content.count("### ")
                if work_item_count < self.criteria.min_work_items:
                    self.validation_errors.append(
                        f"Insufficient work items: {work_item_count} found, {self.criteria.min_work_items} required"
                    )
        except IOError as e:
            self.validation_errors.append(f"Failed to read promotion packet: {e}")

    def _validate_quarterly_value(self) -> None:
        """Extract quarterly value from promotion packet and check against criteria."""
        try:
            with open(self.packet_path, "r") as f:
                content = f.read()
                # Find the quarterly value line
                for line in content.split("\n"):
                    if "Quarterly Business Value" in line:
                        # Extract dollar amount
                        value_str = line.split("$")[1].replace(",", "")
                        value = float(value_str)
                        if value < self.criteria.min_quarterly_value_usd:
                            self.validation_errors.append(
                                f"Insufficient quarterly value: ${value:,.2f} found, ${self.criteria.min_quarterly_value_usd:,.2f} required"
                            )
                        return
                self.validation_errors.append("Quarterly Business Value not found in promotion packet")
        except Exception as e:
            self.validation_errors.append(f"Failed to validate quarterly value: {e}")

    def _validate_technical_scope(self) -> None:
        """Count unique technical scopes across all work items."""
        try:
            with open(self.packet_path, "r") as f:
                content = f.read()
                # Extract scope lines
                scopes = set()
                for line in content.split("\n"):
                    if line.startswith("- **Scope**:"):
                        scope_str = line.split(": ")[1]
                        for scope in scope_str.split(", "):
                            scopes.add(scope.strip())
                if len(scopes) < self.criteria.min_technical_scopes:
                    self.validation_errors.append(
                        f"Insufficient technical scopes: {len(scopes)} found, {self.criteria.min_technical_scopes} required"
                    )
        except IOError as e:
            self.validation_errors.append(f"Failed to read promotion packet: {e}")

    def _validate_stakeholders(self) -> None:
        """Check that all required stakeholder types are present."""
        try:
            with open(self.packet_path, "r") as f:
                content = f.read()
                stakeholder_lines = [line for line in content.split("\n") if line.startswith("- **Stakeholders**:")]
                all_stakeholders = set()
                for line in stakeholder_lines:
                    stakeholder_str = line.split(": ")[1]
                    for stakeholder in stakeholder_str.split(", "):
                        all_stakeholders.add(stakeholder.strip())
                missing = [s for s in self.criteria.required_stakeholder_types if s not in all_stakeholders]
                if missing:
                    self.validation_errors.append(f"Missing required stakeholders: {', '.join(missing)}")
        except IOError as e:
            self.validation_errors.append(f"Failed to read promotion packet: {e}")

    def _validate_evidence(self) -> None:
        """Check that evidence bundle exists and has sufficient items per work item."""
        if not os.path.exists(self.evidence_path):
            self.validation_errors.append(f"Evidence bundle not found at {self.evidence_path}")
            return
        try:
            with open(self.evidence_path, "r") as f:
                evidence = json.load(f)
                # Count total PRs and tickets
                total_prs = sum(len(prs) for prs in evidence.get("github_prs", {}).values())
                total_tickets = len(evidence.get("jira_tickets", []))
                total_evidence = total_prs + total_tickets
                # Simplified: assume 1 work item per 2 evidence items
                work_item_count = self._get_work_item_count()
                expected_evidence = work_item_count * self.criteria.min_evidence_items
                if total_evidence < expected_evidence:
                    self.validation_errors.append(
                        f"Insufficient evidence: {total_evidence} items found, {expected_evidence} required"
                    )
        except json.JSONDecodeError as e:
            self.validation_errors.append(f"Corrupt evidence bundle: {e}")

    def _get_work_item_count(self) -> int:
        """Re-count work items from packet."""
        try:
            with open(self.packet_path, "r") as f:
                return f.read().count("### ")
        except IOError:
            return 0

if __name__ == "__main__":
    # Define criteria for Senior to Staff promotion
    criteria = PromotionCriteria(
        min_work_items=5,
        min_quarterly_value_usd=500000.0,
        min_technical_scopes=3,
        required_stakeholder_types=["product", "infrastructure"],
        min_evidence_items=3
    )
    validator = PromotionPacketValidator(criteria)
    passed, errors = validator.validate_all()
    if passed:
        print("✅ Promotion packet meets all criteria!")
    else:
        print("❌ Promotion packet has errors:")
        for error in errors:
            print(f"- {error}")

import os
import requests

OTTER_API_KEY = os.getenv("OTTER_API_KEY")
NOTION_API_KEY = os.getenv("NOTION_API_KEY")
NOTION_DATABASE_ID = "your-notion-database-id"

def export_otter_to_notion():
    # Fetch Otter transcripts from last 7 days
    otter_resp = requests.get(
        "https://api.otter.ai/v1/transcripts",
        headers={"Authorization": f"Bearer {OTTER_API_KEY}"},
        params={"since": "7d"}
    )
    transcripts = otter_resp.json()["data"]
    # Export to Notion
    for t in transcripts:
        requests.post(
            "https://api.notion.com/v1/pages",
            headers={
                "Authorization": f"Bearer {NOTION_API_KEY}",
                "Notion-Version": "2022-06-28"
            },
            json={
                "parent": {"database_id": NOTION_DATABASE_ID},
                "properties": {
                    "Title": {"title": [{"text": {"content": t["title"]}}]},
                    "Date": {"date": {"start": t["created_at"]}},
                    "Tags": {"multi_select": [{"name": "manager-meeting"}]}
                },
                "children": [{"object": "block", "type": "paragraph", "paragraph": {"rich_text": [{"text": {"content": t["text"]}}]}}]
            }
        )

import requests

LATTICE_API_KEY = "your-lattice-api-key"
COMPANY_OKRS = ["increase-annual-revenue-20%", "reduce-infra-costs-15%", "improve-api-reliability-99.99%"]

def check_okr_alignment(work_item_scope: list, work_item_description: str):
    # Simplified: check if any OKR keyword is present in scope or description
    okr_keywords = [okr.split("-") for okr in COMPANY_OKRS]
    for scope in work_item_scope:
        for keyword in okr_keywords:
            if keyword in scope.lower():
                return True
    for word in work_item_description.lower().split():
        for keyword in okr_keywords:
            if keyword in word:
                return True
    return False

name: Weekly Promotion Packet Update
on:
  schedule:
    - cron: "0 9 * * 0"  # Every Sunday at 9am
jobs:
  update-packet:
    runs-on: ubuntu-latest
    steps:
      - 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 python-dotenv
      - name: Run Impact Tracker
        env:
          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
          JIRA_TOKEN: ${{ secrets.JIRA_TOKEN }}
        run: python impact_tracker.py
      - name: Commit updated packet
        uses: actions/git-commit@v4
        with:
          message: "Weekly promotion packet update"
          files: promotion_packet.md evidence_bundle.json
      - name: Push changes
        uses: ad-m/github-push-action@v0.6.0
        with:
          github_token: ${{ secrets.GITHUB_TOKEN }}