strat-chatbot/adapters/outbound/gitea_issues.py

"""Outbound adapter: Gitea issue tracker.

Implements the IssueTracker port using the Gitea REST API.  A single
httpx.AsyncClient is shared across all calls (connection pool reuse); callers
must await close() when the adapter is no longer needed, typically in an
application lifespan handler.
"""

import logging
from typing import Optional

import httpx

from domain.ports import IssueTracker

logger = logging.getLogger(__name__)

_LABEL_TAGS: list[str] = ["rules-gap", "ai-generated", "needs-review"]
_TITLE_MAX_QUESTION_LEN = 80


class GiteaIssueTracker(IssueTracker):
    """Outbound adapter that creates Gitea issues for unanswered questions.

    Args:
        token:    Personal access token with issue-write permission.
        owner:    Repository owner (user or org name).
        repo:     Repository slug.
        base_url: Base URL of the Gitea instance, e.g. "https://gitea.example.com".
                  Trailing slashes are stripped automatically.
    """

    def __init__(
        self,
        token: str,
        owner: str,
        repo: str,
        base_url: str,
    ) -> None:
        self._token = token
        self._owner = owner
        self._repo = repo
        self._base_url = base_url.rstrip("/")

        self._headers = {
            "Authorization": f"token {token}",
            "Content-Type": "application/json",
            "Accept": "application/json",
        }
        self._client = httpx.AsyncClient(
            headers=self._headers,
            timeout=30.0,
        )

    # ------------------------------------------------------------------
    # IssueTracker port implementation
    # ------------------------------------------------------------------

    async def create_unanswered_issue(
        self,
        question: str,
        user_id: str,
        channel_id: str,
        attempted_rules: list[str],
        conversation_id: str,
    ) -> str:
        """Create a Gitea issue for a question the bot could not answer.

        The question is embedded in a fenced code block to prevent markdown
        injection — a user could craft a question that contains headers, links,
        or other markdown syntax that would corrupt the issue layout.

        Returns:
            The HTML URL of the newly created issue.

        Raises:
            RuntimeError: If the Gitea API responds with a non-2xx status code.
        """
        title = self._build_title(question)
        body = self._build_body(
            question, user_id, channel_id, attempted_rules, conversation_id
        )

        logger.info(
            "Creating Gitea issue for unanswered question from user=%s channel=%s",
            user_id,
            channel_id,
        )

        payload: dict = {
            "title": title,
            "body": body,
        }

        url = f"{self._base_url}/repos/{self._owner}/{self._repo}/issues"
        response = await self._client.post(url, json=payload)

        if response.status_code not in (200, 201):
            error_detail = response.text
            logger.error(
                "Gitea API returned %s creating issue: %s",
                response.status_code,
                error_detail,
            )
            raise RuntimeError(
                f"Gitea API error {response.status_code} creating issue: {error_detail}"
            )

        data = response.json()
        html_url: str = data.get("html_url", "")
        logger.info("Created Gitea issue: %s", html_url)
        return html_url

    # ------------------------------------------------------------------
    # Lifecycle
    # ------------------------------------------------------------------

    async def close(self) -> None:
        """Release the underlying HTTP connection pool.

        Call this in an application shutdown handler (e.g. FastAPI lifespan)
        to avoid ResourceWarning on interpreter exit.
        """
        await self._client.aclose()

    # ------------------------------------------------------------------
    # Private helpers
    # ------------------------------------------------------------------

    @staticmethod
    def _build_title(question: str) -> str:
        """Return a short, human-readable issue title."""
        truncated = question[:_TITLE_MAX_QUESTION_LEN]
        suffix = "..." if len(question) > _TITLE_MAX_QUESTION_LEN else ""
        return f"Unanswered rules question: {truncated}{suffix}"

    @staticmethod
    def _build_body(
        question: str,
        user_id: str,
        channel_id: str,
        attempted_rules: list[str],
        conversation_id: str,
    ) -> str:
        """Compose the Gitea issue body with all triage context.

        The question is fenced so that markdown special characters in user
        input cannot alter the issue structure.
        """
        rules_list: str = ", ".join(attempted_rules) if attempted_rules else "None"
        labels_text: str = ", ".join(_LABEL_TAGS)

        return (
            "## Unanswered Question\n\n"
            f"**User:** {user_id}\n\n"
            f"**Channel:** {channel_id}\n\n"
            f"**Conversation ID:** {conversation_id}\n\n"
            "**Question:**\n"
            f"```\n{question}\n```\n\n"
            f"**Searched Rules:** {rules_list}\n\n"
            f"**Labels:** {labels_text}\n\n"
            "**Additional Context:**\n"
            "This question was asked in Discord and the bot could not provide "
            "a confident answer. The rules either don't cover this question or "
            "the information was ambiguous.\n\n"
            "---\n\n"
            "*This issue was automatically created by the Strat-Chatbot.*"
        )