How to Switch from Developer to Product Manager with Reforge and Coursera in 2026

import os
import json
import requests
import pandas as pd
from datetime import datetime, timedelta
from typing import Dict, List, Optional
import matplotlib.pyplot as plt
import time

# Reforge API base URL (canonical web link: https://github.com/reforge/api-docs)
REFORGE_API_BASE = \"https://api.reforge.com/v1\"
# Coursera API base URL (canonical web link: https://github.com/coursera/api-docs)
COURSERA_API_BASE = \"https://api.coursera.org/api/catalog.v1\"

class PMTransitionTracker:
    def __init__(self, reforge_api_key: str, coursera_oauth_token: str):
        \"\"\"Initialize tracker with API credentials stored in env vars for security\"\"\"
        self.reforge_api_key = reforge_api_key or os.getenv(\"REFORGE_API_KEY\")
        self.coursera_oauth_token = coursera_oauth_token or os.getenv(\"COURSERA_OAUTH_TOKEN\")
        if not self.reforge_api_key:
            raise ValueError(\"Reforge API key not found. Set REFORGE_API_KEY env var.\")
        if not self.coursera_oauth_token:
            raise ValueError(\"Coursera OAuth token not found. Set COURSERA_OAUTH_TOKEN env var.\")

        self.progress_data = {
            \"reforge\": {},
            \"coursera\": {},
            \"milestones\": [],
            \"last_synced\": None
        }
        self.session = requests.Session()
        self.session.headers.update({
            \"User-Agent\": \"PMTransitionTracker/1.0 (senior-dev@company.com)\",
            \"Reforge-API-Key\": self.reforge_api_key,
            \"Authorization\": f\"Bearer {self.coursera_oauth_token}\"
        })

    def sync_reforge_progress(self) -> Dict:
        \"\"\"Fetch Reforge PM Core certification progress, handle rate limits and 404s\"\"\"
        try:
            response = self.session.get(
                f\"{REFORGE_API_BASE}/certifications/pm-core/enrollments/me\",
                timeout=10
            )
            # Reforge returns 429 for rate limits, retry after 60s
            if response.status_code == 429:
                retry_after = int(response.headers.get(\"Retry-After\", 60))
                print(f\"Reforge rate limit hit. Retrying after {retry_after}s.\")
                time.sleep(retry_after)
                return self.sync_reforge_progress()
            response.raise_for_status()
            data = response.json()
            self.progress_data[\"reforge\"] = {
                \"completion_pct\": data.get(\"completion_percentage\", 0),
                \"modules_completed\": data.get(\"completed_modules\", []),
                \"certification_status\": data.get(\"status\", \"not_enrolled\"),
                \"last_updated\": datetime.utcnow().isoformat()
            }
            return self.progress_data[\"reforge\"]
        except requests.exceptions.RequestException as e:
            print(f\"Failed to sync Reforge progress: {str(e)}\")
            return {}
        except json.JSONDecodeError:
            print(\"Reforge API returned invalid JSON\")
            return {}

    def sync_coursera_progress(self, course_ids: List[str]) -> Dict:
        \"\"\"Fetch Coursera course progress for specified course IDs (Google PM Cert courses)\"\"\"
        try:
            progress = {}
            for course_id in course_ids:
                response = self.session.get(
                    f\"{COURSERA_API_BASE}/courses/{course_id}/enrollments/me\",
                    timeout=10
                )
                if response.status_code == 404:
                    print(f\"Coursera course {course_id} not found. Check course ID.\")
                    continue
                response.raise_for_status()
                course_data = response.json()
                progress[course_id] = {
                    \"completion_pct\": course_data.get(\"completionPercentage\", 0),
                    \"assignments_completed\": course_data.get(\"completedAssignments\", 0),
                    \"total_assignments\": course_data.get(\"totalAssignments\", 0),
                    \"last_updated\": datetime.utcnow().isoformat()
                }
            self.progress_data[\"coursera\"] = progress
            return progress
        except requests.exceptions.RequestException as e:
            print(f\"Failed to sync Coursera progress: {str(e)}\")
            return {}

    def calculate_total_completion(self) -> float:
        \"\"\"Weight Reforge (60%) and Coursera (40%) to get total transition readiness\"\"\"
        reforge_pct = self.progress_data.get(\"reforge\", {}).get(\"completion_pct\", 0)
        coursera_pct = 0
        coursera_courses = self.progress_data.get(\"coursera\", {})
        if coursera_courses:
            total_coursera = sum(c.get(\"completion_pct\", 0) for c in coursera_courses.values())
            coursera_pct = total_coursera / len(coursera_courses)
        return (reforge_pct * 0.6) + (coursera_pct * 0.4)

    def generate_progress_report(self, output_path: str = \"pm_transition_progress.json\"):
        \"\"\"Save progress to JSON and generate a matplotlib chart\"\"\"
        self.progress_data[\"last_synced\"] = datetime.utcnow().isoformat()
        self.progress_data[\"total_completion_pct\"] = self.calculate_total_completion()
        # Save JSON report
        with open(output_path, \"w\") as f:
            json.dump(self.progress_data, f, indent=2)
        # Generate chart
        labels = [\"Reforge PM Core\", \"Coursera Google PM Cert\", \"Total Readiness\"]
        sizes = [
            self.progress_data.get(\"reforge\", {}).get(\"completion_pct\", 0),
            sum(c.get(\"completion_pct\", 0) for c in self.progress_data.get(\"coursera\", {}).values()) / max(len(self.progress_data.get(\"coursera\", {})), 1),
            self.progress_data[\"total_completion_pct\"]
        ]
        plt.bar(labels, sizes, color=[\"#FF6B6B\", \"#4ECDC4\", \"#45B7D1\"])
        plt.title(\"PM Transition Progress Tracker\")
        plt.ylabel(\"Completion Percentage\")
        plt.ylim(0, 100)
        for i, v in enumerate(sizes):
            plt.text(i, v + 1, f\"{v:.1f}%\", ha=\"center\")
        plt.savefig(\"pm_progress_chart.png\")
        print(f\"Report saved to {output_path} and chart to pm_progress_chart.png\")

if __name__ == \"__main__\":
    # Initialize tracker (credentials from env vars for security)
    try:
        tracker = PMTransitionTracker(
            reforge_api_key=None,  # Loaded from REFORGE_API_KEY env var
            coursera_oauth_token=None  # Loaded from COURSERA_OAUTH_TOKEN env var
        )
        # Sync Reforge progress
        print(\"Syncing Reforge PM Core progress...\")
        reforge_progress = tracker.sync_reforge_progress()
        print(f\"Reforge completion: {reforge_progress.get('completion_pct', 0)}%\")
        # Coursera Google PM Cert course IDs (2026 catalog)
        google_pm_course_ids = [
            \"google-project-management-professional-certificate\",
            \"agile-project-management\",
            \"product-management-fundamentals\"
        ]
        print(\"Syncing Coursera progress...\")
        coursera_progress = tracker.sync_coursera_progress(google_pm_course_ids)
        print(f\"Coursera average completion: {sum(c.get('completion_pct',0) for c in coursera_progress.values()) / max(len(coursera_progress),1):.1f}%\")
        # Generate report
        tracker.generate_progress_report()
        print(f\"Total transition readiness: {tracker.calculate_total_completion():.1f}%\")
    except Exception as e:
        print(f\"Fatal error: {str(e)}\")

import spacy
import requests
from typing import Dict, List
import json
from collections import Counter

# Load spaCy small English model (download with python -m spacy download en_core_web_sm)
try:
    nlp = spacy.load(\"en_core_web_sm\")
except OSError:
    raise OSError(\"spaCy en_core_web_sm model not found. Run: python -m spacy download en_core_web_sm\")

# PM skills mapped to Reforge/Coursera modules
SKILL_MODULE_MAP = {
    \"user research\": [\"reforge/user-research\", \"coursera/pm-fundamentals\"],
    \"product strategy\": [\"reforge/product-strategy\", \"coursera/pm-capstone\"],
    \"backlog prioritization\": [\"reforge/prioritization\", \"coursera/agile-pm\"],
    \"rice scoring\": [\"reforge/prioritization\", \"coursera/pm-fundamentals\"],
    \"stakeholder management\": [\"reforge/stakeholder-mgmt\", \"coursera/pm-capstone\"],
    \"technical debt communication\": [\"reforge/technical-pm\", \"coursera/technical-pm\"]
}

class PMJobGapAnalyzer:
    def __init__(self, completed_modules: List[str]):
        \"\"\"Initialize with list of completed Reforge/Coursera module IDs\"\"\"
        self.completed_modules = completed_modules
        self.skill_counts = Counter()

    def extract_skills_from_jd(self, jd_url: str) -> List[str]:
        \"\"\"Fetch job description from URL and extract PM-related skills\"\"\"
        try:
            response = requests.get(jd_url, timeout=10)
            response.raise_for_status()
            jd_text = response.text
            # Process text with spaCy
            doc = nlp(jd_text.lower())
            # Extract noun chunks and match to known skills
            extracted_skills = []
            for chunk in doc.noun_chunks:
                chunk_text = chunk.text.strip()
                for skill in SKILL_MODULE_MAP.keys():
                    if skill in chunk_text:
                        extracted_skills.append(skill)
                        self.skill_counts[skill] += 1
            return list(set(extracted_skills))
        except requests.exceptions.RequestException as e:
            print(f\"Failed to fetch job description: {str(e)}\")
            return []
        except Exception as e:
            print(f\"Error processing job description: {str(e)}\")
            return []

    def generate_gap_analysis(self, extracted_skills: List[str]) -> Dict:
        \"\"\"Compare extracted skills to completed modules, output gaps\"\"\"
        gaps = []
        matched = []
        for skill in extracted_skills:
            required_modules = SKILL_MODULE_MAP.get(skill, [])
            # Check if any required module is completed
            if any(module in self.completed_modules for module in required_modules):
                matched.append(skill)
            else:
                gaps.append({
                    \"skill\": skill,
                    \"required_modules\": required_modules,
                    \"course_link\": f\"https://github.com/reforge/pm-core-curriculum/{required_modules[0].split('/')[1]}\" if \"reforge\" in required_modules[0] else f\"https://github.com/coursera/google-pm-professional-certificate/{required_modules[0].split('/')[1]}\"
                })
        return {
            \"total_skills_required\": len(extracted_skills),
            \"matched_skills\": matched,
            \"gap_skills\": gaps,
            \"match_pct\": (len(matched) / max(len(extracted_skills), 1)) * 100
        }

    def print_gap_report(self, gap_analysis: Dict):
        \"\"\"Print formatted gap report to console\"\"\"
        print(f\"\\n=== PM Job Gap Analysis ===\")
        print(f\"Total skills required: {gap_analysis['total_skills_required']}\")
        print(f\"Matched skills: {len(gap_analysis['matched_skills'])} ({gap_analysis['match_pct']:.1f}%)\")
        print(f\"\\nTop 5 required skills: {self.skill_counts.most_common(5)}\")
        print(f\"\\nGap skills to address:\")
        for gap in gap_analysis[\"gap_skills\"]:
            print(f\"- {gap['skill']}: Required modules {gap['required_modules']}\")
            print(f\"  Course link: {gap['course_link']}\")

if __name__ == \"__main__\":
    # List of completed modules (from your progress tracker)
    completed_modules = [
        \"reforge/user-research\",
        \"coursera/pm-fundamentals\",
        \"reforge/prioritization\"
    ]
    analyzer = PMJobGapAnalyzer(completed_modules)
    # Example PM job description URL (2026 PM role at Stripe)
    jd_url = \"https://stripe.com/jobs/pm-technical-products\"
    print(\"Extracting skills from job description...\")
    extracted_skills = analyzer.extract_skills_from_jd(jd_url)
    print(f\"Extracted {len(extracted_skills)} skills: {extracted_skills}\")
    gap_analysis = analyzer.generate_gap_analysis(extracted_skills)
    analyzer.print_gap_report(gap_analysis)

import git
import json
from typing import Dict, List
from datetime import datetime

# RICE scoring weights (Reforge recommended weights)
RICE_WEIGHTS = {
    \"reach\": 1.0,
    \"impact\": 2.0,
    \"confidence\": 1.5,
    \"effort\": -1.0  # Lower effort = higher score
}

class PMPortfolioGenerator:
    def __init__(self, repo_path: str, rice_scores: Dict[str, Dict]):
        \"\"\"Initialize with path to git repo and pre-calculated RICE scores for projects\"\"\"
        try:
            self.repo = git.Repo(repo_path)
        except git.exc.InvalidGitRepositoryError:
            raise ValueError(f\"Invalid git repository at {repo_path}\")
        self.rice_scores = rice_scores
        self.portfolio_items = []

    def extract_commits_for_project(self, project_name: str, start_date: str, end_date: str) -> List[Dict]:
        \"\"\"Extract commits for a project between two dates, filter by commit message keywords\"\"\"
        try:
            start = datetime.strptime(start_date, \"%Y-%m-%d\")
            end = datetime.strptime(end_date, \"%Y-%m-%d\")
            commits = []
            for commit in self.repo.iter_commits():
                commit_date = datetime.fromtimestamp(commit.committed_date)
                if start <= commit_date <= end:
                    # Filter commits mentioning the project name
                    if project_name.lower() in commit.message.lower():
                        commits.append({
                            \"hash\": commit.hexsha,
                            \"message\": commit.message,
                            \"date\": commit_date.isoformat(),
                            \"author\": commit.author.name
                        })
            return commits
        except ValueError as e:
            print(f\"Invalid date format: {str(e)}. Use YYYY-MM-DD.\")
            return []
        except Exception as e:
            print(f\"Error extracting commits: {str(e)}\")
            return []

    def generate_portfolio_item(self, project_name: str, user_story: str, start_date: str, end_date: str) -> Dict:
        \"\"\"Generate a single portfolio item with RICE score and commit history\"\"\"
        commits = self.extract_commits_for_project(project_name, start_date, end_date)
        rice_score = self.rice_scores.get(project_name, {
            \"reach\": 0,
            \"impact\": 0,
            \"confidence\": 0,
            \"effort\": 0
        })
        # Calculate weighted RICE score
        weighted_score = (
            rice_score[\"reach\"] * RICE_WEIGHTS[\"reach\"] +
            rice_score[\"impact\"] * RICE_WEIGHTS[\"impact\"] +
            rice_score[\"confidence\"] * RICE_WEIGHTS[\"confidence\"] +
            rice_score[\"effort\"] * RICE_WEIGHTS[\"effort\"]
        )
        portfolio_item = {
            \"project_name\": project_name,
            \"user_story\": user_story,
            \"dates\": f\"{start_date} to {end_date}\",
            \"rice_score\": weighted_score,
            \"rice_breakdown\": rice_score,
            \"commit_count\": len(commits),
            \"key_commits\": [c[\"message\"] for c in commits[:5]]  # Top 5 commits
        }
        self.portfolio_items.append(portfolio_item)
        return portfolio_item

    def export_portfolio(self, output_path: str = \"pm_portfolio.json\"):
        \"\"\"Export portfolio to JSON and generate HTML page\"\"\"
        with open(output_path, \"w\") as f:
            json.dump(self.portfolio_items, f, indent=2)
        # Generate simple HTML page
        html = \"\"\"PM Portfolio\"\"\"
        for item in self.portfolio_items:
            html += f\"\"\"{item['project_name']}User Story: {item['user_story']}RICE Score: {item['rice_score']:.1f}Key Commits: {len(item['key_commits'])}\"\"\"
        html += \"\"\"\"\"\"
        with open(\"pm_portfolio.html\", \"w\") as f:
            f.write(html)
        print(f\"Portfolio exported to {output_path} and pm_portfolio.html\")

if __name__ == \"__main__\":
    # Example RICE scores for past projects (calculated using Reforge framework)
    rice_scores = {
        \"checkout-latency-fix\": {
            \"reach\": 10000,  # 10k monthly active users
            \"impact\": 3,     # 3x impact on retention
            \"confidence\": 80,  # 80% confidence
            \"effort\": 2       # 2 weeks effort
        }
    }
    try:
        generator = PMPortfolioGenerator(
            repo_path=\".\",  # Current directory git repo
            rice_scores=rice_scores
        )
        # Generate portfolio item for checkout latency fix
        portfolio_item = generator.generate_portfolio_item(
            project_name=\"checkout-latency-fix\",
            user_story=\"As a customer, I want checkout to load in under 200ms so I don't abandon my cart\",
            start_date=\"2026-01-01\",
            end_date=\"2026-02-01\"
        )
        print(f\"Generated portfolio item for {portfolio_item['project_name']} with RICE score {portfolio_item['rice_score']:.1f}\")
        generator.export_portfolio()
    except Exception as e:
        print(f\"Fatal error: {str(e)}\")

def calculate_rice(reach: int, impact: int, confidence: int, effort: int) -> float:
    \"\"\"Calculate RICE score using Reforge recommended weights\"\"\"
    return (reach * 1.0) + (impact * 2.0) + (confidence * 1.5) + (effort * -1.0)

# Example: Checkout latency fix
rice = calculate_rice(reach=10000, impact=3, confidence=80, effort=2)
print(f\"RICE Score: {rice}\")  # Output: 10124.0

import git
repo = git.Repo(\".\")
commits = list(repo.iter_commits(\"main\", max_count=100))
for commit in commits:
    print(f\"{commit.hexsha[:7]}: {commit.message[:50]}\")

import requests
import os

REFORGE_API_KEY = os.getenv(\"REFORGE_API_KEY\")
response = requests.get(
    \"https://api.reforge.com/v1/alumni\",
    headers={\"Reforge-API-Key\": REFORGE_API_KEY},
    params={\"company\": \"Stripe\", \"role\": \"Product Manager\"}
)
alumni = response.json()
for person in alumni:
    print(f\"{person['name']}: {person['email']}\")