|
# retoor <retoor@molodetz.nl>
|
|
|
|
import json
|
|
import logging
|
|
import re
|
|
from dataclasses import dataclass
|
|
|
|
import httpx
|
|
|
|
from devplacepy import stealth
|
|
from devplacepy.config import INTERNAL_GATEWAY_URL
|
|
from devplacepy.services.gitea.config import HTTP_TIMEOUT_SECONDS, GiteaConfig
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
MAX_TOKENS = 1200
|
|
TITLE_MAX = 200
|
|
BODY_MAX = 60000
|
|
|
|
SYSTEM_PROMPT = (
|
|
"You are a senior triage engineer for DevPlace, a social network for software "
|
|
"developers. You rewrite raw user issue reports into clear, consistent, high-quality "
|
|
"issue tickets for a Gitea tracker. Preserve every concrete fact the reporter gave; "
|
|
"never invent versions, stack traces, or steps that were not provided. Keep a calm, "
|
|
"technical, professional tone. Do not add greetings or sign-offs.\n\n"
|
|
"Return ONLY a JSON object with exactly two string keys: \"title\" and \"body\".\n"
|
|
"- title: a concise, specific, imperative summary under 120 characters.\n"
|
|
"- body: GitHub-flavored markdown using EXACTLY these sections, in order, each as a "
|
|
"level-2 heading:\n"
|
|
"## Summary\n## Steps to Reproduce\n## Expected Behaviour\n## Actual Behaviour\n"
|
|
"## Environment\n"
|
|
"Under each heading put what the report supports; if a section is unknown write "
|
|
"'Not provided.' Use a numbered list under Steps to Reproduce."
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class EnhancedTicket:
|
|
title: str
|
|
body: str
|
|
enhanced: bool
|
|
|
|
|
|
def _fallback(title: str, description: str) -> EnhancedTicket:
|
|
body = (
|
|
f"## Summary\n{title}\n\n"
|
|
f"## Steps to Reproduce\n{description}\n\n"
|
|
"## Expected Behaviour\nNot provided.\n\n"
|
|
"## Actual Behaviour\nNot provided.\n\n"
|
|
"## Environment\nNot provided.\n"
|
|
)
|
|
return EnhancedTicket(title=title[:TITLE_MAX], body=body[:BODY_MAX], enhanced=False)
|
|
|
|
|
|
def _parse(text: str) -> tuple[str, str] | None:
|
|
match = re.search(r"\{.*\}", text, re.DOTALL)
|
|
if not match:
|
|
return None
|
|
try:
|
|
payload = json.loads(match.group())
|
|
except (ValueError, TypeError):
|
|
return None
|
|
title = str(payload.get("title", "")).strip()
|
|
body = str(payload.get("body", "")).strip()
|
|
if not title or not body:
|
|
return None
|
|
return title[:TITLE_MAX], body[:BODY_MAX]
|
|
|
|
|
|
async def enhance_ticket(
|
|
title: str, description: str, config: GiteaConfig
|
|
) -> EnhancedTicket:
|
|
title = title.strip()
|
|
description = description.strip()
|
|
if not config.ai_enhance:
|
|
return _fallback(title, description)
|
|
|
|
payload = {
|
|
"model": config.ai_model,
|
|
"messages": [
|
|
{"role": "system", "content": SYSTEM_PROMPT},
|
|
{
|
|
"role": "user",
|
|
"content": f"Reporter title: {title}\n\nReporter description:\n{description}",
|
|
},
|
|
],
|
|
"max_tokens": MAX_TOKENS,
|
|
"temperature": 0.2,
|
|
}
|
|
headers = {"Content-Type": "application/json"}
|
|
if config.ai_key:
|
|
headers["Authorization"] = f"Bearer {config.ai_key}"
|
|
|
|
try:
|
|
async with stealth.stealth_async_client(timeout=HTTP_TIMEOUT_SECONDS) as client:
|
|
response = await client.post(
|
|
INTERNAL_GATEWAY_URL, json=payload, headers=headers
|
|
)
|
|
response.raise_for_status()
|
|
content = (
|
|
response.json()
|
|
.get("choices", [{}])[0]
|
|
.get("message", {})
|
|
.get("content", "")
|
|
)
|
|
except (httpx.HTTPError, ValueError, KeyError, IndexError) as exc:
|
|
logger.warning("Issue ticket enhancement failed, using fallback: %s", exc)
|
|
return _fallback(title, description)
|
|
|
|
parsed = _parse(content)
|
|
if not parsed:
|
|
logger.warning("Issue ticket enhancement returned unparseable content")
|
|
return _fallback(title, description)
|
|
new_title, new_body = parsed
|
|
logger.info("Enhanced issue ticket '%s' -> '%s'", title[:60], new_title[:60])
|
|
return EnhancedTicket(title=new_title, body=new_body, enhanced=True)
|