"""Черновик задачи из ссылки на багрепорт BugTrail (ТЗ 3.22, 0.92).
Ссылка из панели трекера (`{panel}/r/{share_token}`) публична by design: share-токен
в пути и есть авторизация. Отдельного доступа к BugTrail не нужно, токен у нас нигде
не хранится и не настраивается — только вынимается из вставленной ссылки.
Ходим за отчётом **с бэкенда** (у трекера узкий CORS, а share-токен незачем лишний раз
светить в браузере). Хост, к которому обращаемся, берётся **только из настроек**: из
ссылки вынимается лишь share-токен, и он проверяется по charset. Подставить
произвольный адрес нечем — это весь SSRF-барьер фичи (ср. services/favicon.py).
Отсюда же берётся текст задачи: подписи шагов — слово в слово из русской локали самого
BugTrail (`packages/web/src/i18n/ru.json`), чтобы задача в gntodo читалась так же, как
отчёт в панели.
"""
from __future__ import annotations
import json
import re
from typing import Any, Literal
from urllib.parse import urlsplit
import httpx
from pydantic import BaseModel, ConfigDict, Field
from app.config import get_settings
CONNECT_TIMEOUT = 3.0
READ_TIMEOUT = 4.0
TOTAL_TIMEOUT = 6.0
# Тело отчёта — не картинка: в ReportOut входит mouse_track (до 5000 точек у записи
# экрана), поэтому «толстое» тело тут норма, а не признак подделки. Но граница нужна.
MAX_BYTES = 1_000_000
# Потолки на то, что подставляем в описание: лимита на описание в БД нет вовсе, и
# патологический отчёт не должен раздувать задачу (карточка списка рендерит начало).
MAX_STEPS = 50
MAX_IMAGES = 20
MAX_DESCRIPTION = 20_000
_UA = "gntodo/1.0 (bugtrail import)"
# token_urlsafe(16) — 22 символа [A-Za-z0-9_-]; берём с запасом: чужая ссылка всё
# равно упирается в белый список хостов, а битый токен вернёт 404
_TOKEN_RE = re.compile(r"^[A-Za-z0-9_-]{1,64}$")
# Подписи типов шагов — из report.stepTypes русской локали BugTrail
_STEP_LABELS = {
"click": "Клик",
"input": "Ввод",
"url_change": "Смена URL",
"scroll": "Прокрутка",
"navigation": "Переход",
"note": "Заметка",
"screenshot": "Скриншот",
"console": "Ошибка консоли",
}
FetchStatus = Literal["ok", "not_found", "unavailable"]
class _Step(BaseModel):
model_config = ConfigDict(extra="ignore")
type: str = ""
offset_ms: int = 0
data: dict[str, Any] = Field(default_factory=dict)
screenshot_attachment_id: str | None = None
class _Attachment(BaseModel):
model_config = ConfigDict(extra="ignore")
file_id: str = ""
kind: str = ""
filename: str = ""
mime: str = ""
size: int = 0
class Report(BaseModel):
"""Ответ трекера (ReportDetailOut) — только то, что попадает в задачу.
Модель внешнего контракта, поэтому локальная и снисходительная: всё
необязательное (в BugTrail эти же поля бывают `null`), лишнее игнорируем.
"""
model_config = ConfigDict(extra="ignore")
title: str | None = None
description: str | None = None
page_url: str | None = None
page_title: str | None = None
share_token: str | None = None
element: dict[str, Any] | None = None
steps: list[_Step] | None = None
attachments: list[_Attachment] | None = None
class BugTrailDraft(BaseModel):
"""Черновик задачи: заполняем форму, ничего не создаём."""
title: str
description: str
def parse_share_token(url: str) -> str | None:
"""Share-токен из ссылки на отчёт; None — это не ссылка BugTrail.
Принимаем обе формы: человеческую `{panel}/r/{token}` и API-шную
`{api}/api/reports/{token}`. Хост обязан совпасть с настроенным (панель или
API) — иначе ссылку не разбираем вовсе.
"""
settings = get_settings()
if not settings.bugtrail_enabled:
return None
split = urlsplit(url.strip())
if split.scheme not in ("http", "https"):
return None
if split.netloc.lower() not in {_host(settings.bugtrail_url), _host(settings.bugtrail_panel)}:
return None
parts = [part for part in split.path.split("/") if part]
if len(parts) == 2 and parts[0] == "r":
token = parts[1]
elif len(parts) == 3 and parts[:2] == ["api", "reports"]:
token = parts[2]
else:
return None
return token if _TOKEN_RE.match(token) else None
def _host(base: str) -> str:
"""`https://host:port/` → `host:port` в нижнем регистре — для сверки хостов."""
return urlsplit(base).netloc.lower()
def _text(value: Any) -> str:
"""Строка из внешних данных; всё прочее (None, числа, словари) — пусто."""
return value.strip() if isinstance(value, str) else ""
async def _read_limited(resp: httpx.Response, limit: int) -> bytes | None:
"""Тело ответа не длиннее limit; None — превысило (ср. favicon._read_limited)."""
buf = bytearray()
async for chunk in resp.aiter_bytes():
buf += chunk
if len(buf) > limit:
return None
return bytes(buf)
async def _get_json(client: httpx.AsyncClient, url: str) -> tuple[Any, FetchStatus]:
"""JSON по url и что из этого вышло: ok / not_found / unavailable."""
try:
async with client.stream("GET", url, follow_redirects=False) as resp:
if resp.status_code == 404:
return None, "not_found"
if resp.status_code != 200:
return None, "unavailable"
body = await _read_limited(resp, MAX_BYTES)
except httpx.HTTPError:
return None, "unavailable"
if body is None:
return None, "unavailable"
try:
return json.loads(body), "ok"
except ValueError: # не JSON — трекер ответил не тем
return None, "unavailable"
async def fetch_report(token: str) -> tuple[Report | None, FetchStatus]:
"""Отчёт по share-токену. Клиент создаётся здесь — тесты бьют по `_get_json`."""
settings = get_settings()
timeout = httpx.Timeout(TOTAL_TIMEOUT, connect=CONNECT_TIMEOUT, read=READ_TIMEOUT)
url = f"{settings.bugtrail_url.rstrip('/')}/api/reports/{token}"
async with httpx.AsyncClient(timeout=timeout, headers={"User-Agent": _UA}) as client:
payload, status = await _get_json(client, url)
if status != "ok":
return None, status
try:
return Report.model_validate(payload), "ok"
except ValueError: # не наш контракт — считаем трекер недоступным
return None, "unavailable"
def build_draft(report: Report, *, api_base: str, panel_base: str, token: str) -> BugTrailDraft:
"""Черновик задачи из отчёта: заголовок и markdown-описание.
`api_base` — для картинок (`/api/reports/by-token/...`), `panel_base` — для
человеческой ссылки на отчёт. В dev это разные адреса, путать нельзя.
"""
share = report.share_token or token
title = _text(report.title) or _text(report.page_title)
blocks: list[str] = []
if description := _text(report.description):
blocks.append(description)
steps = report.steps or []
if steps:
shots = _shots_by_step(report, api_base=api_base, token=share)
lines: list[str] = []
for index, step in enumerate(steps[:MAX_STEPS]):
lines.append(_step_line(step, index + 1))
if shot := shots.get(step.screenshot_attachment_id or ""):
lines.append(shot)
if len(steps) > MAX_STEPS:
lines.append(f"… ещё {len(steps) - MAX_STEPS}")
blocks.append("**Записанные шаги**\n\n" + "\n".join(lines))
elif element := _element_parts(report.element):
blocks.append(f"**Элемент:** {element}")
if page := _page_line(report):
blocks.append(page)
images, files = _attachments(report, api_base=api_base, token=share)
if images or files:
blocks.append("**Вложения**\n\n" + "\n".join([*images, *files]))
blocks.append(f"**Источник:** BugTrail — {panel_base}/r/{share}")
return BugTrailDraft(title=title or "Багрепорт BugTrail", description=_cut(blocks))
def _step_line(step: _Step, number: int) -> str:
"""Строка шага — как в панели BugTrail (`aiPrompt.ts:111`, `stepLine`)."""
data = step.data
parts: list[str] = []
if element := _element_parts(data.get("element")):
parts.append(element)
if value := _text(data.get("value")):
parts.append(f"«{value}»")
if isinstance(length := data.get("value_length"), int):
parts.append(f"({length} симв.)")
from_url, to_url = _text(data.get("from_url")), _text(data.get("to_url"))
if from_url and to_url:
parts.append(f"{from_url} → {to_url}")
elif to_url:
parts.append(to_url)
x, y = data.get("x"), data.get("y")
if isinstance(x, int | float) and isinstance(y, int | float):
parts.append(f"({x}, {y})")
if text := _text(data.get("text")):
parts.append(text)
label = _STEP_LABELS.get(step.type, step.type)
tail = f": {' '.join(parts)}" if parts else ""
return f"{number}. {step.offset_ms / 1000:.1f} с — {label}{tail}"
def _element_parts(element: Any) -> str:
"""`<button> «Войти» #submit` — тег, подпись, селектор."""
if not isinstance(element, dict):
return ""
parts: list[str] = []
if tag := _text(element.get("tag")):
parts.append(f"<{tag}>")
if snippet := _text(element.get("text_snippet")):
parts.append(f"«{snippet}»")
if selector := _text(element.get("unique_selector")) or _text(element.get("selector")):
parts.append(selector)
return " ".join(parts)
def _page_line(report: Report) -> str:
"""`**Страница:** Вход — http://…` — что из отчёта есть, то и пишем."""
url, title = _text(report.page_url), _text(report.page_title)
if url and title:
return f"**Страница:** {title} — {url}"
if url or title:
return f"**Страница:** {url or title}"
return ""
def _file_url(api_base: str, token: str, file_id: str) -> str:
"""Публичная ссылка на файл вложения: токен в пути и есть авторизация."""
return f"{api_base}/api/reports/by-token/{token}/files/{file_id}"
def _shots_by_step(report: Report, *, api_base: str, token: str) -> dict[str, str]:
"""Картинка, привязанная к шагу (`screenshot_attachment_id`), — под своим шагом."""
images = {a.file_id for a in report.attachments or [] if (a.mime or "").startswith("image/")}
result: dict[str, str] = {}
for step in report.steps or []:
shot = step.screenshot_attachment_id or ""
if shot in images:
result[shot] = f"})"
return result
def _attachments(
report: Report, *, api_base: str, token: str
) -> tuple[list[str], list[str]]:
"""(картинки, прочие файлы) — то, что не встало к шагам."""
inlined = set(_shots_by_step(report, api_base=api_base, token=token))
images: list[str] = []
files: list[str] = []
for attachment in report.attachments or []:
if not attachment.file_id or attachment.file_id in inlined:
continue
url = _file_url(api_base, token, attachment.file_id)
mime = attachment.mime or ""
if mime.startswith("image/"):
if len(images) < MAX_IMAGES:
images.append(f"")
elif mime.startswith("video/"):
files.append(f"- Запись экрана: {url}")
else:
files.append(f"- {_text(attachment.filename) or 'файл'}: {url}")
return images, files
def _cut(blocks: list[str]) -> str:
"""Собрать блоки в описание и подрезать по потолку (лимита в БД нет)."""
text = "\n\n".join(block for block in blocks if block)
if len(text) <= MAX_DESCRIPTION:
return text
head = text[:MAX_DESCRIPTION]
cut = head.rfind("\n")
if cut > 0:
head = head[:cut]
return head + "\n\n… отчёт большой, показано начало"
