Legal Risk Assessment

#!/usr/bin/env python3
"""
Legal Risk Report Generator

Generates a formatted risk assessment memo in markdown from a risk register
JSON file. Includes risk matrix visualization, distribution summary,
top risks, recommended actions, and monitoring plan.

Usage:
    python risk_report_generator.py --input risk_register.json
    python risk_report_generator.py --input risk_register.json --output memo.md
    python risk_report_generator.py --input risk_register.json --json
"""

import argparse
import json
import sys
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional


SEVERITY_LABELS: Dict[int, str] = {1: "Negligible", 2: "Minor", 3: "Moderate", 4: "Major", 5: "Critical"}
LIKELIHOOD_LABELS: Dict[int, str] = {1: "Remote", 2: "Unlikely", 3: "Possible", 4: "Likely", 5: "Almost Certain"}
ACTIONS: Dict[str, str] = {"GREEN": "Accept and document. Monitor quarterly.", "YELLOW": "Assign owner. Implement controls. Review monthly.", "ORANGE": "Escalate to senior counsel. Develop contingency plan. Consider outside counsel.", "RED": "Immediate escalation. Assemble response team. Engage outside counsel. Report to board."}
MONITORING: Dict[str, str] = {"GREEN": "Quarterly review", "YELLOW": "Monthly review", "ORANGE": "Bi-weekly review", "RED": "Weekly review (daily during active response)"}


def get_risk_level(score: int) -> str:
    """Return risk level color based on score."""
    if score <= 4: return "GREEN"
    elif score <= 9: return "YELLOW"
    elif score <= 15: return "ORANGE"
    return "RED"


def get_action(level: str) -> str:
    return ACTIONS.get(level, "Unknown")


def get_monitoring_frequency(level: str) -> str:
    return MONITORING.get(level, "Unknown")


def load_register(filepath: str) -> List[Dict[str, Any]]:
    """Load risk register from JSON file."""
    path = Path(filepath)
    if not path.exists():
        print(f"Error: File not found: {filepath}", file=sys.stderr)
        sys.exit(1)
    try:
        with open(path, "r") as f:
            data = json.load(f)
    except json.JSONDecodeError as e:
        print(f"Error: Invalid JSON: {e}", file=sys.stderr)
        sys.exit(1)

    if isinstance(data, dict) and "risks" in data:
        return data["risks"]
    elif isinstance(data, list):
        return data
    else:
        print("Error: Expected 'risks' array or top-level array", file=sys.stderr)
        sys.exit(1)


def enrich_risks(risks: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """Calculate scores and levels for each risk."""
    enriched: List[Dict[str, Any]] = []
    for risk in risks:
        sev: int = risk.get("severity", 1)
        lik: int = risk.get("likelihood", 1)
        score: int = sev * lik
        level: str = get_risk_level(score)
        enriched.append({
            **risk,
            "score": score,
            "level": level,
            "action": get_action(level),
            "monitoring": get_monitoring_frequency(level),
            "severity_label": SEVERITY_LABELS.get(sev, "Unknown"),
            "likelihood_label": LIKELIHOOD_LABELS.get(lik, "Unknown"),
        })
    enriched.sort(key=lambda r: r["score"], reverse=True)
    return enriched


def build_matrix(risks: List[Dict[str, Any]]) -> str:
    """Build ASCII risk matrix showing risk counts per cell."""
    cell_counts: Dict[tuple, int] = {}
    for risk in risks:
        key = (risk["severity"], risk["likelihood"])
        cell_counts[key] = cell_counts.get(key, 0) + 1
    lines: List[str] = ["```", "         L=1    L=2    L=3    L=4    L=5"]
    for sev in range(5, 0, -1):
        row: List[str] = []
        for lik in range(1, 6):
            score = sev * lik
            level = get_risk_level(score)[0]  # G/Y/O/R
            count = cell_counts.get((sev, lik), 0)
            cell = f"{count}x" if count > 0 else f" {level}"
            row.append(f"{score:>2d}({cell})")
        lines.append(f"  S={sev}  {'  '.join(row)}")
    lines.append("  Legend: G=GREEN Y=YELLOW O=ORANGE R=RED  Nx=risk count")
    lines.append("```")
    return "\n".join(lines)


def build_distribution(risks: List[Dict[str, Any]]) -> str:
    """Build risk distribution summary table."""
    dist: Dict[str, int] = {"RED": 0, "ORANGE": 0, "YELLOW": 0, "GREEN": 0}
    for risk in risks:
        dist[risk["level"]] += 1

    total: int = len(risks)
    lines: List[str] = [
        "| Level | Count | Percentage | Action Required |",
        "|-------|-------|------------|-----------------|",
    ]
    for level in ["RED", "ORANGE", "YELLOW", "GREEN"]:
        count = dist[level]
        pct = round(100 * count / total, 1) if total > 0 else 0.0
        action = get_action(level).split(".")[0] + "."
        lines.append(f"| {level} | {count} | {pct}% | {action} |")

    lines.append(f"| **Total** | **{total}** | **100%** | |")
    return "\n".join(lines)


def build_top_risks(risks: List[Dict[str, Any]], limit: int = 10) -> str:
    """Build top risks table."""
    top = risks[:limit]
    lines: List[str] = [
        "| # | Score | Level | Category | Description | Action |",
        "|---|-------|-------|----------|-------------|--------|",
    ]
    for i, risk in enumerate(top, 1):
        desc = risk.get("description", "N/A")
        if len(desc) > 50:
            desc = desc[:47] + "..."
        lines.append(
            f"| {i} | {risk['score']} | {risk['level']} | "
            f"{risk.get('category', 'N/A')} | {desc} | {risk['action'].split('.')[0]}. |"
        )
    return "\n".join(lines)


def build_monitoring_plan(risks: List[Dict[str, Any]]) -> str:
    """Build monitoring plan section."""
    lines: List[str] = [
        "| Risk Level | Review Frequency | Responsible | Reporting |",
        "|------------|-----------------|-------------|-----------|",
        "| RED | Weekly (daily during active response) | General Counsel / Outside Counsel | Board + Executive Team |",
        "| ORANGE | Bi-weekly | Senior Counsel | Legal Leadership |",
        "| YELLOW | Monthly | Assigned Risk Owner | Legal Team |",
        "| GREEN | Quarterly | Legal Operations | Internal Log |",
    ]
    return "\n".join(lines)


def build_category_summary(risks: List[Dict[str, Any]]) -> str:
    """Build category breakdown table."""
    cats: Dict[str, Dict[str, Any]] = {}
    for risk in risks:
        cat = risk.get("category", "Other")
        if cat not in cats:
            cats[cat] = {"count": 0, "total_score": 0, "max_score": 0}
        cats[cat]["count"] += 1
        cats[cat]["total_score"] += risk["score"]
        cats[cat]["max_score"] = max(cats[cat]["max_score"], risk["score"])

    sorted_cats = sorted(cats.items(), key=lambda x: x[1]["max_score"], reverse=True)
    lines: List[str] = [
        "| Category | Count | Avg Score | Max Score |",
        "|----------|-------|-----------|-----------|",
    ]
    for cat, data in sorted_cats:
        avg = round(data["total_score"] / data["count"], 1)
        lines.append(f"| {cat} | {data['count']} | {avg} | {data['max_score']} |")
    return "\n".join(lines)


def generate_memo(risks: List[Dict[str, Any]]) -> str:
    """Generate the full risk assessment memo in markdown."""
    now = datetime.now().strftime("%Y-%m-%d")
    enriched = enrich_risks(risks)

    scores = [r["score"] for r in enriched]
    avg_score = round(sum(scores) / len(scores), 1) if scores else 0
    max_score = max(scores) if scores else 0

    sections: List[str] = []

    # Header
    sections.append(f"# Legal Risk Assessment Memo\n")
    sections.append(f"**Date:** {now}  ")
    sections.append(f"**Total Risks:** {len(enriched)}  ")
    sections.append(f"**Average Score:** {avg_score}  ")
    sections.append(f"**Highest Score:** {max_score}  ")
    sections.append("")

    # Executive Summary
    red_count = sum(1 for r in enriched if r["level"] == "RED")
    orange_count = sum(1 for r in enriched if r["level"] == "ORANGE")
    sections.append("## 1. Executive Summary\n")
    if red_count > 0:
        sections.append(f"**CRITICAL:** {red_count} RED-level risk(s) require immediate escalation and response team activation.\n")
    if orange_count > 0:
        sections.append(f"**ATTENTION:** {orange_count} ORANGE-level risk(s) require senior counsel review and contingency planning.\n")
    if red_count == 0 and orange_count == 0:
        sections.append("No RED or ORANGE risks identified. Current risk posture is within acceptable tolerance.\n")

    # Risk Matrix
    sections.append("## 2. Risk Matrix\n")
    sections.append(build_matrix(enriched))
    sections.append("")

    # Distribution
    sections.append("## 3. Risk Distribution\n")
    sections.append(build_distribution(enriched))
    sections.append("")

    # Category Breakdown
    sections.append("## 4. Category Breakdown\n")
    sections.append(build_category_summary(enriched))
    sections.append("")

    # Top Risks
    sections.append("## 5. Top Risks\n")
    sections.append(build_top_risks(enriched))
    sections.append("")

    # Recommended Actions
    sections.append("## 6. Recommended Actions\n")
    for i, risk in enumerate(enriched, 1):
        if risk["level"] in ("RED", "ORANGE"):
            sections.append(f"### Risk {i}: {risk.get('description', 'N/A')}")
            sections.append(f"- **Score:** {risk['score']} ({risk['level']})")
            sections.append(f"- **Category:** {risk.get('category', 'N/A')}")
            sections.append(f"- **Action:** {risk['action']}")
            sections.append(f"- **Monitoring:** {risk['monitoring']}")
            sections.append("")

    # Monitoring Plan
    sections.append("## 7. Monitoring Plan\n")
    sections.append(build_monitoring_plan(enriched))
    sections.append("")

    # Escalation Summary
    sections.append("## 8. Escalation Summary\n")
    if red_count > 0:
        sections.append("- **Outside counsel engagement:** MANDATORY for RED-level risks")
    if orange_count > 0:
        sections.append("- **Senior counsel review:** REQUIRED for ORANGE-level risks")
    sections.append(f"- **Next review date:** Schedule within {'1 week' if red_count > 0 else '2 weeks' if orange_count > 0 else '1 month'}")
    sections.append("")

    # Footer
    sections.append("---\n")
    sections.append(f"*Generated by Legal Risk Report Generator on {now}*")

    return "\n".join(sections)


def generate_json_metadata(risks: List[Dict[str, Any]]) -> Dict[str, Any]:
    """Generate JSON metadata about the report."""
    enriched = enrich_risks(risks)
    dist: Dict[str, int] = {"RED": 0, "ORANGE": 0, "YELLOW": 0, "GREEN": 0}
    for r in enriched:
        dist[r["level"]] += 1

    scores = [r["score"] for r in enriched]
    return {
        "generated": datetime.now().isoformat(),
        "total_risks": len(enriched),
        "distribution": dist,
        "average_score": round(sum(scores) / len(scores), 1) if scores else 0,
        "max_score": max(scores) if scores else 0,
        "requires_escalation": dist["RED"] > 0,
        "risks": enriched,
    }


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Legal Risk Report Generator — generates risk assessment memos"
    )
    parser.add_argument("--input", required=True, help="Path to risk register JSON file")
    parser.add_argument("--output", type=str, help="Save memo to file (markdown)")
    parser.add_argument("--json", action="store_true", help="Output report metadata as JSON")

    args = parser.parse_args()

    risks = load_register(args.input)
    if not risks:
        print("Error: No risks found in input file", file=sys.stderr)
        sys.exit(1)

    if args.json:
        metadata = generate_json_metadata(risks)
        print(json.dumps(metadata, indent=2))
    else:
        memo = generate_memo(risks)
        if args.output:
            Path(args.output).write_text(memo)
            print(f"Memo written to {args.output}")
        else:
            print(memo)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Legal Risk Scorer

Calculates legal risk scores using a 5x5 Severity x Likelihood matrix.
Assigns GREEN/YELLOW/ORANGE/RED levels with recommended actions.
Supports single-risk and batch mode from JSON input.

Usage:
    python risk_scorer.py --severity 4 --likelihood 3 --category "Contract" --description "SLA breach"
    python risk_scorer.py --input risks.json --json
    python risk_scorer.py --severity 5 --likelihood 4 --category "Litigation" --description "Patent claim" --json
"""

import argparse
import json
import sys
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple


SEVERITY_LABELS: Dict[int, str] = {1: "Negligible", 2: "Minor", 3: "Moderate", 4: "Major", 5: "Critical"}
SEVERITY_EXPOSURE: Dict[int, str] = {1: "<0.1% revenue", 2: "0.1-1% revenue", 3: "1-5% revenue", 4: "5-15% revenue", 5: ">15% revenue"}
LIKELIHOOD_LABELS: Dict[int, str] = {1: "Remote", 2: "Unlikely", 3: "Possible", 4: "Likely", 5: "Almost Certain"}
LIKELIHOOD_PROBABILITY: Dict[int, str] = {1: "<5%", 2: "5-20%", 3: "20-50%", 4: "50-80%", 5: ">80%"}

VALID_CATEGORIES: List[str] = [
    "Contract", "Regulatory", "Litigation", "IP",
    "Data Privacy", "Employment", "Corporate",
]


def get_risk_level(score: int) -> str:
    """Return risk level color based on score."""
    if score <= 4:
        return "GREEN"
    elif score <= 9:
        return "YELLOW"
    elif score <= 15:
        return "ORANGE"
    else:
        return "RED"


def get_recommended_action(level: str) -> str:
    """Return recommended action for a risk level."""
    actions: Dict[str, str] = {
        "GREEN": "Accept — document risk, monitor quarterly, no immediate action required",
        "YELLOW": "Monitor — assign risk owner, implement controls, review monthly",
        "ORANGE": "Mitigate — escalate to senior counsel, develop contingency plan, consider outside counsel",
        "RED": "Escalate — immediate escalation, assemble response team, engage outside counsel, board reporting",
    }
    return actions.get(level, "Unknown")


def validate_risk(risk: Dict[str, Any]) -> Tuple[bool, str]:
    """Validate a risk entry has required fields and valid values."""
    required = ["severity", "likelihood", "category", "description"]
    for field in required:
        if field not in risk:
            return False, f"Missing required field: {field}"

    sev = risk["severity"]
    lik = risk["likelihood"]

    if not isinstance(sev, int) or sev < 1 or sev > 5:
        return False, f"Severity must be integer 1-5, got: {sev}"
    if not isinstance(lik, int) or lik < 1 or lik > 5:
        return False, f"Likelihood must be integer 1-5, got: {lik}"

    cat = risk["category"]
    if cat not in VALID_CATEGORIES:
        return False, f"Invalid category '{cat}'. Valid: {', '.join(VALID_CATEGORIES)}"

    if not risk["description"].strip():
        return False, "Description cannot be empty"

    return True, ""


def score_risk(risk: Dict[str, Any]) -> Dict[str, Any]:
    """Score a single risk and return enriched result."""
    sev: int = risk["severity"]
    lik: int = risk["likelihood"]
    score: int = sev * lik
    level: str = get_risk_level(score)

    return {
        "description": risk["description"],
        "category": risk["category"],
        "severity": sev,
        "severity_label": SEVERITY_LABELS[sev],
        "severity_exposure": SEVERITY_EXPOSURE[sev],
        "likelihood": lik,
        "likelihood_label": LIKELIHOOD_LABELS[lik],
        "likelihood_probability": LIKELIHOOD_PROBABILITY[lik],
        "score": score,
        "level": level,
        "recommended_action": get_recommended_action(level),
    }


def generate_summary(scored_risks: List[Dict[str, Any]]) -> Dict[str, Any]:
    """Generate summary statistics for a list of scored risks."""
    total: int = len(scored_risks)
    if total == 0:
        return {"total": 0, "distribution": {}, "average_score": 0.0, "max_score": 0}

    distribution: Dict[str, int] = {"GREEN": 0, "YELLOW": 0, "ORANGE": 0, "RED": 0}
    scores: List[int] = []

    for risk in scored_risks:
        distribution[risk["level"]] += 1
        scores.append(risk["score"])

    avg_score: float = sum(scores) / total
    max_score: int = max(scores)
    max_risk = next(r for r in scored_risks if r["score"] == max_score)

    return {
        "total": total,
        "distribution": distribution,
        "average_score": round(avg_score, 1),
        "max_score": max_score,
        "highest_risk": max_risk["description"],
        "red_count": distribution["RED"],
        "orange_count": distribution["ORANGE"],
        "requires_escalation": distribution["RED"] > 0,
    }


def format_human_single(result: Dict[str, Any]) -> str:
    """Format a single risk result for human-readable output."""
    lines: List[str] = [
        "=" * 60,
        "LEGAL RISK ASSESSMENT",
        "=" * 60,
        "",
        f"Description:    {result['description']}",
        f"Category:       {result['category']}",
        "",
        f"Severity:       {result['severity']} — {result['severity_label']} ({result['severity_exposure']})",
        f"Likelihood:     {result['likelihood']} — {result['likelihood_label']} ({result['likelihood_probability']})",
        "",
        f"Risk Score:     {result['score']} / 25",
        f"Risk Level:     {result['level']}",
        "",
        f"Action:         {result['recommended_action']}",
        "",
        "=" * 60,
    ]
    return "\n".join(lines)


def format_human_batch(scored_risks: List[Dict[str, Any]], summary: Dict[str, Any]) -> str:
    """Format batch results for human-readable output."""
    lines: List[str] = [
        "=" * 70,
        "LEGAL RISK REGISTER — SCORED RESULTS",
        f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M')}",
        "=" * 70,
        "",
    ]

    # Summary
    lines.append("SUMMARY")
    lines.append("-" * 40)
    lines.append(f"  Total Risks:      {summary['total']}")
    lines.append(f"  Average Score:    {summary['average_score']}")
    lines.append(f"  Highest Score:    {summary['max_score']}")
    lines.append("")
    lines.append("  Distribution:")
    for level in ["RED", "ORANGE", "YELLOW", "GREEN"]:
        count = summary["distribution"][level]
        bar = "#" * count
        lines.append(f"    {level:8s}  {count:3d}  {bar}")
    lines.append("")

    if summary["requires_escalation"]:
        lines.append("  *** ESCALATION REQUIRED: RED-level risks detected ***")
        lines.append("")

    # Individual risks sorted by score descending
    lines.append("RISKS (sorted by score, highest first)")
    lines.append("-" * 70)

    for i, risk in enumerate(scored_risks, 1):
        lines.append(f"\n  [{i}] {risk['description']}")
        lines.append(f"      Category:    {risk['category']}")
        lines.append(f"      Severity:    {risk['severity']} ({risk['severity_label']})")
        lines.append(f"      Likelihood:  {risk['likelihood']} ({risk['likelihood_label']})")
        lines.append(f"      Score:       {risk['score']}  Level: {risk['level']}")
        lines.append(f"      Action:      {risk['recommended_action']}")

    lines.append("")
    lines.append("=" * 70)
    return "\n".join(lines)


def load_risks_from_file(filepath: str) -> List[Dict[str, Any]]:
    """Load risks from a JSON file."""
    path = Path(filepath)
    if not path.exists():
        print(f"Error: File not found: {filepath}", file=sys.stderr)
        sys.exit(1)

    try:
        with open(path, "r") as f:
            data = json.load(f)
    except json.JSONDecodeError as e:
        print(f"Error: Invalid JSON in {filepath}: {e}", file=sys.stderr)
        sys.exit(1)

    if isinstance(data, dict) and "risks" in data:
        return data["risks"]
    elif isinstance(data, list):
        return data
    else:
        print("Error: JSON must contain a 'risks' array or be a top-level array", file=sys.stderr)
        sys.exit(1)


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Legal Risk Scorer — 5x5 Severity x Likelihood matrix"
    )
    parser.add_argument("--severity", type=int, choices=range(1, 6), help="Severity rating (1-5)")
    parser.add_argument("--likelihood", type=int, choices=range(1, 6), help="Likelihood rating (1-5)")
    parser.add_argument("--category", type=str, help=f"Risk category: {', '.join(VALID_CATEGORIES)}")
    parser.add_argument("--description", type=str, help="Risk description")
    parser.add_argument("--input", type=str, help="JSON file with multiple risks (batch mode)")
    parser.add_argument("--json", action="store_true", help="Output in JSON format")

    args = parser.parse_args()

    # Determine mode
    if args.input:
        # Batch mode
        risks = load_risks_from_file(args.input)
        scored: List[Dict[str, Any]] = []

        for idx, risk in enumerate(risks):
            valid, msg = validate_risk(risk)
            if not valid:
                print(f"Error in risk #{idx + 1}: {msg}", file=sys.stderr)
                sys.exit(1)
            scored.append(score_risk(risk))

        # Sort by score descending
        scored.sort(key=lambda r: r["score"], reverse=True)
        summary = generate_summary(scored)

        if args.json:
            output = {"risks": scored, "summary": summary, "generated": datetime.now().isoformat()}
            print(json.dumps(output, indent=2))
        else:
            print(format_human_batch(scored, summary))

    elif args.severity and args.likelihood and args.category and args.description:
        # Single risk mode
        risk = {
            "severity": args.severity,
            "likelihood": args.likelihood,
            "category": args.category,
            "description": args.description,
        }
        valid, msg = validate_risk(risk)
        if not valid:
            print(f"Error: {msg}", file=sys.stderr)
            sys.exit(1)

        result = score_risk(risk)

        if args.json:
            print(json.dumps(result, indent=2))
        else:
            print(format_human_single(result))
    else:
        parser.error(
            "Provide either --input for batch mode, or all of "
            "--severity, --likelihood, --category, --description for single risk mode"
        )


if __name__ == "__main__":
    main()