maxerton
b4cabbf146
Сервер запускается через uv run --script, ключи запрашиваются при установке плагина (userConfig) вместо .env. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
445 lines
18 KiB
Python
445 lines
18 KiB
Python
"""Тонкий async-клиент Trello REST API."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import os
|
|
import re
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
|
|
TRELLO_API_BASE = "https://api.trello.com/1"
|
|
|
|
_SHORT_LINK_RE = re.compile(r"trello\.com/c/([A-Za-z0-9]+)")
|
|
|
|
_ATTACHMENT_FIELDS = "id,name,url,mimeType,bytes,isUpload,date"
|
|
|
|
# Больше этого вложение не качаем — неожиданно тяжёлый файл не должен
|
|
# забивать диск и контекст.
|
|
MAX_ATTACHMENT_BYTES = 25 * 1024 * 1024
|
|
|
|
|
|
class TrelloConfigError(RuntimeError):
|
|
"""Не заданы обязательные переменные окружения для Trello API."""
|
|
|
|
|
|
def _credentials() -> tuple[str, str]:
|
|
key = os.environ.get("TRELLO_API_KEY")
|
|
token = os.environ.get("TRELLO_TOKEN")
|
|
if not key or not token:
|
|
raise TrelloConfigError(
|
|
"TRELLO_API_KEY / TRELLO_TOKEN не заданы (проверь .env)"
|
|
)
|
|
return key, token
|
|
|
|
|
|
def parse_short_link(card_url: str) -> str:
|
|
"""Достаёт shortLink из ссылки вида https://trello.com/c/abc123/45-title."""
|
|
match = _SHORT_LINK_RE.search(card_url)
|
|
if not match:
|
|
raise ValueError(f"Не похоже на ссылку на карточку Trello: {card_url!r}")
|
|
return match.group(1)
|
|
|
|
|
|
async def get_card(card_url: str) -> dict:
|
|
"""Возвращает {id, name, desc, url, board_id, list_id, labels, attachments, comments}.
|
|
|
|
comments — see get_comments(): список {author, text, date}, для контекста задачи.
|
|
labels — [{name, color}]. attachments — только список (см. _attachment_summary),
|
|
содержимое тянется отдельно через get_attachment().
|
|
"""
|
|
key, token = _credentials()
|
|
short_link = parse_short_link(card_url)
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
response = await client.get(
|
|
f"{TRELLO_API_BASE}/cards/{short_link}",
|
|
params={
|
|
"key": key,
|
|
"token": token,
|
|
"fields": "name,desc,shortUrl,idBoard,idList,labels",
|
|
"attachments": "true",
|
|
"attachment_fields": _ATTACHMENT_FIELDS,
|
|
},
|
|
)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
|
|
comments = await get_comments(short_link)
|
|
|
|
return {
|
|
"id": data["id"],
|
|
"name": data["name"],
|
|
"desc": data["desc"],
|
|
"url": data["shortUrl"],
|
|
"board_id": data["idBoard"],
|
|
"list_id": data["idList"],
|
|
"labels": [
|
|
{"name": label.get("name") or "", "color": label.get("color")}
|
|
for label in data.get("labels") or []
|
|
],
|
|
"attachments": [_attachment_summary(item) for item in data.get("attachments") or []],
|
|
"comments": comments,
|
|
}
|
|
|
|
|
|
def _attachment_kind(attachment: dict) -> str:
|
|
"""card — ссылка на другую карточку Trello, image/file — загруженный файл, link — прочая ссылка."""
|
|
if _SHORT_LINK_RE.search(attachment.get("url") or ""):
|
|
return "card"
|
|
if attachment.get("isUpload"):
|
|
mime = attachment.get("mimeType") or ""
|
|
return "image" if mime.startswith("image/") else "file"
|
|
return "link"
|
|
|
|
|
|
def _attachment_summary(attachment: dict) -> dict:
|
|
return {
|
|
"id": attachment["id"],
|
|
"name": attachment.get("name") or "",
|
|
"kind": _attachment_kind(attachment),
|
|
"mime": attachment.get("mimeType") or None,
|
|
"size": attachment.get("bytes"),
|
|
"date": attachment.get("date"),
|
|
"url": attachment.get("url"),
|
|
}
|
|
|
|
|
|
def _attachments_dir() -> Path:
|
|
return Path(
|
|
os.environ.get("TRELLO_ATTACHMENTS_DIR")
|
|
or Path(tempfile.gettempdir()) / "trello-attachments"
|
|
)
|
|
|
|
|
|
def _safe_filename(name: str) -> str:
|
|
cleaned = re.sub(r"[^\w.\-]+", "_", name, flags=re.UNICODE).strip("._")
|
|
return cleaned[:100] or "attachment"
|
|
|
|
|
|
async def get_attachment(card_id: str, attachment_id: str) -> dict:
|
|
"""Метаданные одного вложения + содержимое, если это загруженный файл.
|
|
|
|
Для kind image/file файл скачивается в TRELLO_ATTACHMENTS_DIR (по
|
|
умолчанию <tmp>/trello-attachments/<card_id>/) — в ответе path и data
|
|
(bytes). Для card/link ничего не качается: как с ними быть, решает
|
|
вызывающий код.
|
|
"""
|
|
key, token = _credentials()
|
|
|
|
async with httpx.AsyncClient(follow_redirects=True) as client:
|
|
response = await client.get(
|
|
f"{TRELLO_API_BASE}/cards/{card_id}/attachments/{attachment_id}",
|
|
params={"key": key, "token": token, "fields": _ATTACHMENT_FIELDS},
|
|
)
|
|
response.raise_for_status()
|
|
attachment = _attachment_summary(response.json())
|
|
|
|
if attachment["kind"] not in ("image", "file"):
|
|
return attachment
|
|
|
|
if (attachment["size"] or 0) > MAX_ATTACHMENT_BYTES:
|
|
raise ValueError(
|
|
f"Вложение {attachment['name']!r} больше "
|
|
f"{MAX_ATTACHMENT_BYTES // (1024 * 1024)} МБ — не скачиваю"
|
|
)
|
|
|
|
# Загруженные файлы Trello отдаёт только с OAuth-заголовком,
|
|
# key/token в query для download не работают. Заголовок с токеном —
|
|
# только на домены Trello (при редиректе на чужой домен httpx его снимает).
|
|
host = httpx.URL(attachment["url"]).host
|
|
if not (host == "trello.com" or host.endswith(".trello.com")):
|
|
raise ValueError(f"Вложение ведёт не на Trello ({host}) — не скачиваю")
|
|
download = await client.get(
|
|
attachment["url"],
|
|
headers={
|
|
"Authorization": f'OAuth oauth_consumer_key="{key}", oauth_token="{token}"'
|
|
},
|
|
)
|
|
download.raise_for_status()
|
|
data = download.content
|
|
|
|
path = _attachments_dir() / card_id / f"{attachment_id}-{_safe_filename(attachment['name'])}"
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_bytes(data)
|
|
|
|
return {**attachment, "size": len(data), "path": str(path), "data": data}
|
|
|
|
|
|
async def get_comments(card_id: str, limit: int = 20) -> list[dict]:
|
|
"""Комментарии карточки в хронологическом порядке (старые → новые).
|
|
|
|
card_id — полный id или shortLink, Trello API принимает оба.
|
|
"""
|
|
key, token = _credentials()
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
response = await client.get(
|
|
f"{TRELLO_API_BASE}/cards/{card_id}/actions",
|
|
params={
|
|
"key": key,
|
|
"token": token,
|
|
"filter": "commentCard",
|
|
"limit": limit,
|
|
},
|
|
)
|
|
response.raise_for_status()
|
|
actions = response.json()
|
|
|
|
return [
|
|
{
|
|
"author": (action.get("memberCreator") or {}).get("fullName") or "?",
|
|
"text": action["data"]["text"],
|
|
"date": action["date"],
|
|
}
|
|
for action in reversed(actions)
|
|
]
|
|
|
|
|
|
async def add_comment(card_id: str, text: str) -> dict:
|
|
"""Постит комментарий к карточке, возвращает id созданного комментария."""
|
|
key, token = _credentials()
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
response = await client.post(
|
|
f"{TRELLO_API_BASE}/cards/{card_id}/actions/comments",
|
|
params={"key": key, "token": token, "text": text},
|
|
)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
|
|
return {"id": data["id"]}
|
|
|
|
|
|
def create_targets() -> list[str]:
|
|
"""Столбцы, в которые разрешено создавать карточки (TRELLO_CREATE_LISTS).
|
|
|
|
Через запятую, пробелы по краям не важны: "Бэклог, Баги с прода".
|
|
"""
|
|
raw = os.environ.get("TRELLO_CREATE_LISTS", "")
|
|
return [name.strip() for name in raw.split(",") if name.strip()]
|
|
|
|
|
|
def _board_id() -> str:
|
|
board_id = os.environ.get("TRELLO_BOARD_ID")
|
|
if not board_id:
|
|
raise TrelloConfigError("TRELLO_BOARD_ID не задан (проверь .env)")
|
|
return board_id
|
|
|
|
|
|
async def resolve_create_list(list_name: str) -> dict:
|
|
"""Находит на доске столбец из разрешённых, возвращает {id, name}.
|
|
|
|
Отказывает, если столбца нет в TRELLO_CREATE_LISTS — даже если на доске
|
|
он существует: писать можно только туда, куда разрешил пользователь.
|
|
"""
|
|
allowed = {name.casefold(): name for name in create_targets()}
|
|
if list_name.strip().casefold() not in allowed:
|
|
raise ValueError(
|
|
f"Столбец {list_name!r} не разрешён для создания карточек. "
|
|
f"Разрешены (TRELLO_CREATE_LISTS): {', '.join(allowed.values()) or 'ни одного'}"
|
|
)
|
|
|
|
key, token = _credentials()
|
|
async with httpx.AsyncClient() as client:
|
|
response = await client.get(
|
|
f"{TRELLO_API_BASE}/boards/{_board_id()}/lists",
|
|
params={"key": key, "token": token, "fields": "name", "filter": "open"},
|
|
)
|
|
response.raise_for_status()
|
|
lists = response.json()
|
|
|
|
for item in lists:
|
|
if item["name"].strip().casefold() == list_name.strip().casefold():
|
|
return {"id": item["id"], "name": item["name"]}
|
|
|
|
raise ValueError(f"Столбец {list_name!r} не найден на доске TRELLO_BOARD_ID")
|
|
|
|
|
|
# Что переносится из шаблона в новую карточку: участники, метки, чек-листы и
|
|
# прочее. Описание не копируем — его пишет Claude. Обложку Trello через
|
|
# idCardSource не копирует — её ставим отдельно (см. create_card).
|
|
_TEMPLATE_KEEP = "attachments,checklists,customFields,labels,members,stickers"
|
|
|
|
|
|
async def list_templates() -> list[dict]:
|
|
"""Все карточки-шаблоны доски TRELLO_BOARD_ID, включая архивные.
|
|
|
|
[{id, name, list, desc, members: [имя], cover: {color, size, brightness} | None,
|
|
labels: [{name, color}], checklists: [{name, items}]}]
|
|
Шаблоны на доске лежат в архиве, но используются — поэтому filter=all
|
|
(без него Trello их не отдаёт). Карточек с filter=all тысячи, так что
|
|
сначала берём их с минимумом полей, а чек-листы — только у шаблонов.
|
|
"""
|
|
key, token = _credentials()
|
|
auth = {"key": key, "token": token}
|
|
board_id = _board_id()
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
cards_response = await client.get(
|
|
f"{TRELLO_API_BASE}/boards/{board_id}/cards",
|
|
params={
|
|
**auth,
|
|
"fields": "name,desc,idList,labels,isTemplate,idMembers,cover",
|
|
"filter": "all",
|
|
},
|
|
)
|
|
cards_response.raise_for_status()
|
|
lists_response = await client.get(
|
|
f"{TRELLO_API_BASE}/boards/{board_id}/lists",
|
|
params={**auth, "fields": "name", "filter": "all"},
|
|
)
|
|
lists_response.raise_for_status()
|
|
members_response = await client.get(
|
|
f"{TRELLO_API_BASE}/boards/{board_id}/members",
|
|
params={**auth, "fields": "fullName"},
|
|
)
|
|
members_response.raise_for_status()
|
|
|
|
templates = [card for card in cards_response.json() if card.get("isTemplate")]
|
|
checklist_responses = await asyncio.gather(
|
|
*(
|
|
client.get(
|
|
f"{TRELLO_API_BASE}/cards/{card['id']}/checklists",
|
|
params={**auth, "fields": "name", "checkItem_fields": "name"},
|
|
)
|
|
for card in templates
|
|
)
|
|
)
|
|
for response in checklist_responses:
|
|
response.raise_for_status()
|
|
|
|
list_names = {item["id"]: item["name"] for item in lists_response.json()}
|
|
member_names = {item["id"]: item["fullName"] for item in members_response.json()}
|
|
return [
|
|
{
|
|
"id": card["id"],
|
|
"name": card["name"],
|
|
"list": list_names.get(card["idList"]),
|
|
"desc": card.get("desc") or "",
|
|
"members": [member_names.get(member_id, member_id) for member_id in card.get("idMembers") or []],
|
|
"cover": _template_cover(card.get("cover") or {}),
|
|
"labels": [
|
|
{"name": label.get("name") or "", "color": label.get("color")}
|
|
for label in card.get("labels") or []
|
|
],
|
|
"checklists": [
|
|
{
|
|
"name": checklist["name"],
|
|
"items": [item["name"] for item in checklist.get("checkItems") or []],
|
|
}
|
|
for checklist in checklists.json()
|
|
],
|
|
}
|
|
for card, checklists in zip(templates, checklist_responses)
|
|
]
|
|
|
|
|
|
def _template_cover(cover: dict) -> dict | None:
|
|
"""Цветная обложка шаблона в виде, пригодном для PUT /cards/{id}; None — без цвета."""
|
|
if not cover.get("color"):
|
|
return None
|
|
return {key: cover[key] for key in ("color", "size", "brightness") if cover.get(key)}
|
|
|
|
|
|
async def list_labels() -> list[dict]:
|
|
"""Метки доски [{name, color, uses}], самые используемые первыми; безымянные пропускаются."""
|
|
key, token = _credentials()
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
response = await client.get(
|
|
f"{TRELLO_API_BASE}/boards/{_board_id()}/labels",
|
|
params={"key": key, "token": token, "fields": "name,color,uses", "limit": 1000},
|
|
)
|
|
response.raise_for_status()
|
|
|
|
labels = [
|
|
{"id": label["id"], "name": label["name"], "color": label.get("color"), "uses": label.get("uses", 0)}
|
|
for label in response.json()
|
|
if (label.get("name") or "").strip()
|
|
]
|
|
return sorted(labels, key=lambda label: -label["uses"])
|
|
|
|
|
|
async def resolve_labels(names: list[str]) -> list[dict]:
|
|
"""Метки доски по названиям (без учёта регистра) — или ошибка с неизвестными."""
|
|
by_name: dict[str, dict] = {}
|
|
for label in await list_labels():
|
|
# При дублях названий берём самую используемую — список уже отсортирован.
|
|
by_name.setdefault(label["name"].strip().casefold(), label)
|
|
|
|
found, missing = [], []
|
|
for name in names:
|
|
label = by_name.get(name.strip().casefold())
|
|
if label is None:
|
|
missing.append(name)
|
|
elif label not in found:
|
|
found.append(label)
|
|
if missing:
|
|
raise ValueError(
|
|
f"Меток нет на доске: {', '.join(map(repr, missing))}. "
|
|
"Список — trello_list_labels."
|
|
)
|
|
return found
|
|
|
|
|
|
async def resolve_template(template_name: str) -> dict:
|
|
"""Шаблон по названию (без учёта регистра) — или ошибка со списком доступных."""
|
|
templates = await list_templates()
|
|
wanted = template_name.strip().casefold()
|
|
for template in templates:
|
|
if template["name"].strip().casefold() == wanted:
|
|
return template
|
|
available = ", ".join(t["name"] for t in templates) or "ни одного"
|
|
raise ValueError(f"Шаблон {template_name!r} не найден. Есть: {available}")
|
|
|
|
|
|
async def create_card(
|
|
list_id: str,
|
|
name: str,
|
|
desc: str,
|
|
template: dict | None = None,
|
|
label_ids: list[str] | None = None,
|
|
) -> dict:
|
|
"""Создаёт карточку в столбце, возвращает {id, name, url}.
|
|
|
|
template — шаблон из list_templates: карточка создаётся его копией
|
|
(участники, метки, чек-листы…) и получает его обложку. label_ids —
|
|
метки, которые добавляются к меткам шаблона.
|
|
"""
|
|
key, token = _credentials()
|
|
auth = {"key": key, "token": token}
|
|
payload = {"idList": list_id, "name": name, "desc": desc}
|
|
if template:
|
|
payload |= {"idCardSource": template["id"], "keepFromSource": _TEMPLATE_KEEP}
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
response = await client.post(f"{TRELLO_API_BASE}/cards", params=auth, json=payload)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
|
|
# Копия шаблона может унаследовать его описание, флаг isTemplate или
|
|
# архивность (шаблоны на доске архивные) — добиваем до того, что
|
|
# подтвердил пользователь.
|
|
fixes = {}
|
|
if data.get("closed"):
|
|
fixes["closed"] = False
|
|
if data.get("desc", desc) != desc:
|
|
fixes["desc"] = desc
|
|
if data.get("isTemplate"):
|
|
fixes["isTemplate"] = False
|
|
if template and template.get("cover"):
|
|
fixes["cover"] = template["cover"]
|
|
extra_labels = [label_id for label_id in label_ids or [] if label_id not in data.get("idLabels", [])]
|
|
if extra_labels:
|
|
fixes["idLabels"] = ",".join([*data.get("idLabels", []), *extra_labels])
|
|
if fixes:
|
|
fix_response = await client.put(
|
|
f"{TRELLO_API_BASE}/cards/{data['id']}", params=auth, json=fixes
|
|
)
|
|
fix_response.raise_for_status()
|
|
|
|
return {"id": data["id"], "name": data["name"], "url": data["shortUrl"]}
|