Function run_multi_target

openosint/multi_target.py:105–162

Investigate targets in parallel using asyncio.gather(). Parameters ---------- targets: List of OSINT targets (emails, usernames, domains, IPs, names). Maximum ``MAX_TARGETS`` entries. api_key: Anthropic API key. Falls back to ``ANTHROPIC_API_KEY`` env v

(
    targets: list[str],
    api_key: str | None = None,
    is_pdf_disabled: bool = False,
)

async def run_multi_target(
    targets: list[str],
    api_key: str | None = None,
    is_pdf_disabled: bool = False,
) -> str:
    """
    Investigate targets in parallel using asyncio.gather().

    Parameters
    ----------
    targets:
        List of OSINT targets (emails, usernames, domains, IPs, names).
        Maximum ``MAX_TARGETS`` entries.
    api_key:
        Anthropic API key.  Falls back to ``ANTHROPIC_API_KEY`` env var.
    is_pdf_disabled:
        When True, skip PDF generation for the summary report.

    Returns
    -------
    str
        Markdown summary report (also written to ``reports/``).

    Raises
    ------
    ValueError
        If more than ``MAX_TARGETS`` targets are supplied.
    """
    if len(targets) > MAX_TARGETS:
        raise ValueError(
            f"Multi-target investigation supports at most {MAX_TARGETS} targets; "
            f"got {len(targets)}. Split into smaller batches."
        )
    if not targets:
        return "No targets provided."

    reports_dir = Path("reports")
    reports_dir.mkdir(exist_ok=True)
    date_prefix = datetime.now().strftime("%Y-%m-%d")

    tasks = [_investigate_one(target, api_key, reports_dir, date_prefix) for target in targets]
    results: list[tuple[str, AgentResponse]] = await asyncio.gather(*tasks)

    summary = _build_summary(results, date_prefix)

    summary_path = reports_dir / f"{date_prefix}_summary.md"
    summary_path.write_text(summary, encoding="utf-8")
    logger.info("Summary report: %s", summary_path)

    if not is_pdf_disabled:
        try:
            from openosint.pdf_report import generate_pdf_report

            await generate_pdf_report(summary_path)
        except Exception:
            logger.debug("PDF skipped for summary.", exc_info=True)

    return summary