CyberSentinel-AI/reports/action_report.py

from __future__ import annotations

from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

from clients.github_client import GithubClient
from config import GITHUB_KEYWORDS, WATCHED_REPOSITORIES, settings
from database.init_db import init_database
from database.session import SessionFactory, get_engine
from monitors.cve_monitor import CVEMonitor
from monitors.github_monitor import GithubMonitor
from utils.blacklist_manager import BlacklistManager


@dataclass
class ReportData:
    generated_at: datetime
    keywords: list[str] = field(default_factory=list)
    cve_candidates: list[dict[str, Any]] = field(default_factory=list)
    repository_candidates: list[dict[str, Any]] = field(default_factory=list)
    errors: list[str] = field(default_factory=list)


def parse_keywords(raw_keywords: str | None, *, defaults: list[str], limit: int) -> list[str]:
    """Parse comma/newline separated keywords while keeping multi-word phrases."""
    source = raw_keywords or ""
    candidates = source.replace("\n", ",").split(",") if source.strip() else defaults

    normalized: list[str] = []
    seen: set[str] = set()
    for item in candidates:
        keyword = str(item).strip()
        if not keyword:
            continue
        key = keyword.lower()
        if key in seen:
            continue
        seen.add(key)
        normalized.append(keyword)

    return normalized[: max(limit, 0)]


def trim_items(items: list[dict[str, Any]], limit: int) -> list[dict[str, Any]]:
    return list(items[: max(limit, 0)])


def _format_reasons(value: Any) -> str:
    if isinstance(value, list):
        reasons = [str(item) for item in value if str(item).strip()]
    else:
        reasons = [str(value)] if str(value).strip() else []
    return ", ".join(reasons) if reasons else "-"


def _format_score(value: Any) -> str:
    try:
        return f"{float(value):.2f}"
    except (TypeError, ValueError):
        return "-"


def _format_stars(value: Any) -> str:
    try:
        return str(int(value))
    except (TypeError, ValueError):
        return "-"


def build_report_markdown(
    *,
    generated_at: datetime,
    cve_candidates: list[dict[str, Any]],
    repository_candidates: list[dict[str, Any]],
    keywords: list[str],
    errors: list[str] | None = None,
) -> str:
    generated = generated_at.astimezone(timezone.utc).isoformat()
    lines = [
        "# CyberSentinel AI GitHub Actions Report",
        "",
        f"- Generated at: `{generated}`",
        "- Mode: `artifact-only`",
        "- This report is generated by GitHub Actions and is not committed back to the repository.",
        "",
        "## Search configuration",
        "",
        f"- Keywords: {', '.join(keywords) if keywords else '-'}",
        f"- CVE candidates: {len(cve_candidates)}",
        f"- Repository candidates: {len(repository_candidates)}",
        "",
    ]

    if errors:
        lines.extend(["## Warnings", ""])
        for error in errors:
            lines.append(f"- {error}")
        lines.append("")

    lines.extend(
        [
            "## CVE candidates",
            "",
            "| CVE | Repository | Stars | Quality | Reasons |",
            "| --- | --- | ---: | ---: | --- |",
        ]
    )
    if cve_candidates:
        for item in cve_candidates:
            cve_id = str(item.get("cve_id") or "Unknown")
            repo_url = str(item.get("repository_url") or item.get("url") or "")
            cve_cell = f"[{cve_id}]({repo_url})" if repo_url else cve_id
            lines.append(
                "| {cve} | {repo} | {stars} | {score} | {reasons} |".format(
                    cve=cve_cell,
                    repo=repo_url or "-",
                    stars=_format_stars(item.get("stars")),
                    score=_format_score(item.get("quality_score")),
                    reasons=_format_reasons(item.get("quality_reasons")),
                )
            )
    else:
        lines.append("| - | - | - | - | No CVE candidates collected |")

    lines.extend(
        [
            "",
            "## Repository candidates",
            "",
            "| Repository | Keyword | Stars | Quality | Reasons |",
            "| --- | --- | ---: | ---: | --- |",
        ]
    )
    if repository_candidates:
        for item in repository_candidates:
            name = str(item.get("name") or item.get("repository_name") or "Unknown")
            url = str(item.get("url") or item.get("repository_url") or "")
            name_cell = f"[{name}]({url})" if url else name
            lines.append(
                "| {name} | {keyword} | {stars} | {score} | {reasons} |".format(
                    name=name_cell,
                    keyword=str(item.get("search_keyword") or "-"),
                    stars=_format_stars(item.get("stars")),
                    score=_format_score(item.get("quality_score")),
                    reasons=_format_reasons(item.get("quality_reasons")),
                )
            )
    else:
        lines.append("| - | - | - | - | No repository candidates collected |")

    lines.extend(
        [
            "",
            "## Next steps",
            "",
            "- Download this Markdown file from the workflow artifact.",
            "- Review candidates manually before acting on them.",
            "",
        ]
    )
    return "\n".join(lines)


def collect_report_data(
    *,
    max_items: int,
    keyword_limit: int,
    keywords: str | None,
    include_cve: bool,
    include_repositories: bool,
    database_path: str | None,
) -> ReportData:
    selected_keywords = parse_keywords(keywords, defaults=GITHUB_KEYWORDS, limit=keyword_limit)
    data = ReportData(generated_at=datetime.now(timezone.utc), keywords=selected_keywords)

    if not include_cve and not include_repositories:
        return data

    db_path = database_path or settings.database_path
    init_database(db_path)
    engine = get_engine(db_path)
    session_factory = SessionFactory(engine)
    github_client = GithubClient()
    blacklist_manager = BlacklistManager()

    if include_cve:
        try:
            cve_monitor = CVEMonitor(
                engine,
                github_client=github_client,
                blacklist_manager=blacklist_manager,
                session_factory=session_factory,
            )
            fetched = cve_monitor.fetch_latest_cves()
            data.cve_candidates = trim_items(cve_monitor.process_cve_data(fetched), max_items)
        except Exception as exc:  # pragma: no cover - network/runtime dependent
            data.errors.append(f"CVE collection failed: {exc}")

    if include_repositories:
        try:
            github_monitor = GithubMonitor(
                engine,
                github_client=github_client,
                blacklist_manager=blacklist_manager,
                session_factory=session_factory,
                keywords=selected_keywords,
                watched_repositories=WATCHED_REPOSITORIES,
            )
            data.repository_candidates = trim_items(github_monitor.run(interval=1), max_items)
        except Exception as exc:  # pragma: no cover - network/runtime dependent
            data.errors.append(f"Repository collection failed: {exc}")

    return data


def write_report(output_path: str | Path, report_data: ReportData) -> Path:
    path = Path(output_path)
    path.parent.mkdir(parents=True, exist_ok=True)
    markdown = build_report_markdown(
        generated_at=report_data.generated_at,
        cve_candidates=report_data.cve_candidates,
        repository_candidates=report_data.repository_candidates,
        keywords=report_data.keywords,
        errors=report_data.errors,
    )
    path.write_text(markdown, encoding="utf-8")
    return path