world/db.py

from __future__ import annotations

import json
import sqlite3
from dataclasses import dataclass, field
from pathlib import Path


@dataclass
class IssueRow:
    issue_id: int
    number: int
    repo: str
    title: str
    body: str
    author_login: str
    is_pr: bool
    tier: int
    is_synthetic_spam: bool
    spam_pattern: str | None
    truth_action: str
    truth_labels: list[str]
    truth_close_reason: str | None
    truth_assignee: str | None
    state: str


@dataclass
class ContributorRow:
    login: str
    public_pr_count: int
    public_issue_count: int
    account_age_days: int
    hidden_reputation: float
    hidden_is_spammer: bool


@dataclass
class CommentRow:
    comment_id: int
    issue_id: int
    author_login: str
    body: str
    created_at: str


def _row_to_issue(r: sqlite3.Row) -> IssueRow:
    return IssueRow(
        issue_id=r["issue_id"],
        number=r["number"],
        repo=r["repo"],
        title=r["title"],
        body=r["body"] or "",
        author_login=r["author_login"],
        is_pr=bool(r["is_pr"]),
        tier=r["tier"],
        is_synthetic_spam=bool(r["is_synthetic_spam"]),
        spam_pattern=r["spam_pattern"],
        truth_action=r["truth_action"],
        truth_labels=json.loads(r["truth_labels"] or "[]"),
        truth_close_reason=r["truth_close_reason"],
        truth_assignee=r["truth_assignee"],
        state=r["state"],
    )


class RepoDB:
    def __init__(self, path: str | Path):
        self.path = str(path)
        self._con = sqlite3.connect(self.path, check_same_thread=False)
        self._con.row_factory = sqlite3.Row
        self._labels_cache: dict[str, list[str]] = {}

    def labels_for_repo(self, repo: str) -> list[str]:
        if repo in self._labels_cache:
            return self._labels_cache[repo]
        rows = self._con.execute(
            "SELECT name FROM labels WHERE repo=? ORDER BY name", (repo,)
        ).fetchall()
        out = [r["name"] for r in rows]
        self._labels_cache[repo] = out
        return out

    def issues_for_tier(self, repo: str, tiers: list[int], limit: int) -> list[IssueRow]:
        placeholders = ",".join("?" * len(tiers))
        rows = self._con.execute(
            f"SELECT * FROM issues WHERE repo=? AND tier IN ({placeholders}) "
            f"AND is_synthetic_spam=0 ORDER BY RANDOM() LIMIT ?",
            (repo, *tiers, limit),
        ).fetchall()
        return [_row_to_issue(r) for r in rows]

    def contributor(self, login: str) -> ContributorRow | None:
        r = self._con.execute(
            "SELECT * FROM contributors WHERE login=?", (login,)
        ).fetchone()
        if not r:
            return None
        return ContributorRow(
            login=r["login"],
            public_pr_count=r["public_pr_count"],
            public_issue_count=r["public_issue_count"],
            account_age_days=r["account_age_days"],
            hidden_reputation=r["hidden_reputation"],
            hidden_is_spammer=bool(r["hidden_is_spammer"]),
        )

    def comments_for_issue(self, issue_id: int, limit: int = 6) -> list[CommentRow]:
        rows = self._con.execute(
            "SELECT * FROM comments WHERE issue_id=? ORDER BY created_at LIMIT ?",
            (issue_id, limit),
        ).fetchall()
        return [
            CommentRow(
                comment_id=r["comment_id"],
                issue_id=r["issue_id"],
                author_login=r["author_login"],
                body=r["body"],
                created_at=r["created_at"],
            )
            for r in rows
        ]

    def search_history(self, repo: str, query: str, limit: int = 5) -> list[IssueRow]:
        like = f"%{query.strip()[:60]}%"
        rows = self._con.execute(
            "SELECT * FROM issues WHERE repo=? AND (title LIKE ? OR body LIKE ?) LIMIT ?",
            (repo, like, like, limit),
        ).fetchall()
        return [_row_to_issue(r) for r in rows]

    def count_issues(self, repo: str) -> int:
        r = self._con.execute(
            "SELECT COUNT(*) AS c FROM issues WHERE repo=? AND is_synthetic_spam=0", (repo,)
        ).fetchone()
        return r["c"] if r else 0