pip install stix2 requests jinja2

1. Direction     → Define PIRs, scope collection
2. Collection    → Gather data from sources (feeds, OSINT, HUMINT, sensors)
3. Processing    → Normalize, deduplicate, enrich raw data
4. Analysis      → Apply analytic techniques, produce assessments
5. Dissemination → Distribute to stakeholders via appropriate channels/TLP
6. Feedback      → Measure effectiveness, refine requirements

PIR-001: Which threat actors are currently targeting [sector] in [region]?
  Priority: HIGH
  Stakeholder: CISO, SOC Manager
  Collection Sources: ISAC feeds, vendor reports, dark web monitoring
  Update Frequency: Monthly
  Dissemination: TLP:AMBER — Executive threat brief

import json
from datetime import datetime

def create_pir(
    pir_id: str,
    question: str,
    priority: str,
    stakeholder: str,
    sources: list[str],
    cadence: str,
    tlp: str,
) -> dict:
    """Create a Priority Intelligence Requirement."""
    return {
        "pir_id": pir_id,
        "question": question,
        "priority": priority,
        "stakeholder": stakeholder,
        "collection_sources": sources,
        "update_cadence": cadence,
        "dissemination_tlp": tlp,
        "created": datetime.now().isoformat(),
        "status": "active",
        "last_answered": None,
    }

def create_collection_plan(pirs: list[dict]) -> dict:
    """Generate collection plan from PIRs."""
    sources_needed = set()
    for pir in pirs:
        sources_needed.update(pir["collection_sources"])

    return {
        "plan_id": f"CP-{datetime.now().strftime('%Y%m%d')}",
        "pir_count": len(pirs),
        "sources_required": sorted(sources_needed),
        "collection_tasks": [
            {"pir": p["pir_id"], "sources": p["collection_sources"],
             "cadence": p["update_cadence"], "priority": p["priority"]}
            for p in pirs
        ],
    }

def track_collection(source: str, pir_id: str, data_collected: int) -> dict:
    """Track collection activity against PIR."""
    return {
        "source": source,
        "pir_id": pir_id,
        "timestamp": datetime.now().isoformat(),
        "data_collected": data_collected,
        "status": "collected",
    }

def assess_collection_gaps(pirs: list[dict], collection_log: list[dict]) -> dict:
    """Identify PIRs with insufficient collection."""
    answered_pirs = {c["pir_id"] for c in collection_log}
    gaps = [p for p in pirs if p["pir_id"] not in answered_pirs]
    return {
        "total_pirs": len(pirs),
        "answered": len(answered_pirs),
        "gaps": [{"pir_id": p["pir_id"], "question": p["question"]} for p in gaps],
    }

def create_intelligence_product(
    title: str,
    product_type: str,
    tlp: str,
    key_findings: list[str],
    confidence: str,
    recommendations: list[str],
    pir_addressed: list[str],
) -> dict:
    """Create structured intelligence product."""
    return {
        "title": title,
        "type": product_type,
        "tlp": tlp,
        "produced": datetime.now().isoformat(),
        "confidence": confidence,
        "key_findings": key_findings,
        "recommendations": recommendations,
        "pirs_addressed": pir_addressed,
        "key_assumptions": [],
        "alternative_hypotheses": [],
        "limitations": [],
    }

def disseminate(product: dict, recipients: list[str], channel: str) -> dict:
    """Track dissemination of intelligence product."""
    return {
        "product_title": product["title"],
        "tlp": product["tlp"],
        "channel": channel,
        "recipients": recipients,
        "disseminated_at": datetime.now().isoformat(),
        "feedback_requested": True,
    }

def validate_tlp_compliance(product: dict, recipients: list[str]) -> dict:
    """Validate TLP marking matches recipient scope."""
    tlp = product["tlp"].upper()
    issues = []
    if tlp == "TLP:RED" and len(recipients) > 5:
        issues.append("TLP:RED should only go to named recipients")
    if tlp == "TLP:AMBER+STRICT" and any("external" in r for r in recipients):
        issues.append("TLP:AMBER+STRICT restricted to organization")
    return {"compliant": len(issues) == 0, "issues": issues}

def calculate_program_metrics(
    pirs: list[dict],
    products: list[dict],
    feedback: list[dict],
) -> dict:
    """Calculate CTI program effectiveness metrics."""
    pirs_addressed = set()
    for p in products:
        pirs_addressed.update(p.get("pirs_addressed", []))

    positive_feedback = sum(1 for f in feedback if f.get("useful", False))

    return {
        "metrics_period": datetime.now().strftime("%Y-Q%q" if False else "%Y-%m"),
        "pir_coverage": f"{len(pirs_addressed)}/{len(pirs)}",
        "products_produced": len(products),
        "stakeholder_satisfaction": (
            round(positive_feedback / max(len(feedback), 1) * 100, 1)
        ),
        "mean_time_to_produce_hours": 0,  # Calculate from timestamps
        "feedback_count": len(feedback),
        "recommendations": [
            "Review unanswered PIRs for collection gaps",
            "Survey stakeholders on product relevance",
            "Track time-to-detection for tactical IOCs",
        ],
    }