|
# retoor <retoor@molodetz.nl>
|
|
|
|
import logging
|
|
from datetime import datetime, timezone
|
|
|
|
from devplacepy.database import get_table
|
|
from devplacepy.utils import generate_uid
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
TICKETS = "bug_tickets"
|
|
COMMENT_AUTHORS = "bug_comment_authors"
|
|
|
|
|
|
def _now() -> str:
|
|
return datetime.now(timezone.utc).isoformat()
|
|
|
|
|
|
def record_ticket(
|
|
number: int,
|
|
author_uid: str,
|
|
original_title: str,
|
|
original_description: str,
|
|
enhanced_title: str,
|
|
html_url: str,
|
|
status: str,
|
|
) -> str:
|
|
uid = generate_uid()
|
|
get_table(TICKETS).insert(
|
|
{
|
|
"uid": uid,
|
|
"gitea_number": int(number),
|
|
"author_uid": author_uid,
|
|
"original_title": original_title[:200],
|
|
"original_description": original_description[:5000],
|
|
"enhanced_title": enhanced_title[:200],
|
|
"html_url": html_url,
|
|
"last_status": status,
|
|
"last_comment_count": 0,
|
|
"created_at": _now(),
|
|
"updated_at": _now(),
|
|
"deleted_at": None,
|
|
"deleted_by": None,
|
|
}
|
|
)
|
|
logger.info("Recorded bug ticket #%s by %s", number, author_uid)
|
|
return uid
|
|
|
|
|
|
def get_ticket(number: int) -> dict | None:
|
|
return get_table(TICKETS).find_one(gitea_number=int(number), deleted_at=None)
|
|
|
|
|
|
def author_uid_for_issue(number: int) -> str | None:
|
|
row = get_ticket(number)
|
|
return row["author_uid"] if row else None
|
|
|
|
|
|
def author_map(numbers: list[int]) -> dict[int, str]:
|
|
if not numbers:
|
|
return {}
|
|
table = get_table(TICKETS)
|
|
ids = [int(n) for n in numbers]
|
|
rows = table.find(table.table.columns.gitea_number.in_(ids), deleted_at=None)
|
|
return {int(row["gitea_number"]): row["author_uid"] for row in rows}
|
|
|
|
|
|
def tracked_tickets() -> list[dict]:
|
|
return list(get_table(TICKETS).find(deleted_at=None, order_by=["gitea_number"]))
|
|
|
|
|
|
def update_ticket_cache(number: int, status: str, comment_count: int) -> None:
|
|
row = get_ticket(number)
|
|
if not row:
|
|
return
|
|
get_table(TICKETS).update(
|
|
{
|
|
"uid": row["uid"],
|
|
"last_status": status,
|
|
"last_comment_count": int(comment_count),
|
|
"updated_at": _now(),
|
|
},
|
|
["uid"],
|
|
)
|
|
|
|
|
|
def record_comment_author(comment_id: int, number: int, author_uid: str) -> str:
|
|
uid = generate_uid()
|
|
get_table(COMMENT_AUTHORS).insert(
|
|
{
|
|
"uid": uid,
|
|
"gitea_comment_id": int(comment_id),
|
|
"gitea_number": int(number),
|
|
"author_uid": author_uid,
|
|
"created_at": _now(),
|
|
"deleted_at": None,
|
|
"deleted_by": None,
|
|
}
|
|
)
|
|
return uid
|
|
|
|
|
|
def comment_author_map(comment_ids: list[int]) -> dict[int, str]:
|
|
if not comment_ids:
|
|
return {}
|
|
table = get_table(COMMENT_AUTHORS)
|
|
ids = [int(cid) for cid in comment_ids]
|
|
rows = table.find(
|
|
table.table.columns.gitea_comment_id.in_(ids), deleted_at=None
|
|
)
|
|
return {int(row["gitea_comment_id"]): row["author_uid"] for row in rows}
|
|
|
|
|
|
def local_comment_ids(number: int) -> set[int]:
|
|
rows = get_table(COMMENT_AUTHORS).find(gitea_number=int(number), deleted_at=None)
|
|
return {int(row["gitea_comment_id"]) for row in rows}
|