b5adfdf9b5
Модуль attachments (fmt_size, ref_names, resolve, state, get); mirror.get_issue отдаёт attachments, description_attachment_ids, attachment_ids комментариев и скрывает fetch_attachment из commands. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
78 lines
3.8 KiB
Python
78 lines
3.8 KiB
Python
"""Вложения задач для интерфейсов: ссылки на них в вики-разметке текста, формат размера и состояние (data-model
|
|
«Скачанные вложения», hub-api «Ссылки на вложения в тексте»). Состояние вычисляется при чтении; модуль только читает
|
|
базу и хранилище файлов, очередь и HTTP не знает."""
|
|
import re
|
|
|
|
from . import files
|
|
|
|
_IMAGE = re.compile(r"!([^!|\n]+)(?:\|[^!\n]*)?!")
|
|
_LINK = re.compile(r"\[(?:[^\]|\n]*\|)?\^([^\]\n]+)\]")
|
|
|
|
|
|
def fmt_size(size: int | None) -> str:
|
|
if size is None:
|
|
return "размер неизвестен"
|
|
if size < 1024:
|
|
return f"{size} Б"
|
|
if size < 1024 ** 2:
|
|
return f"{size / 1024:.1f} КБ".replace(".", ",")
|
|
return f"{size / 1024 ** 2:.1f} МБ".replace(".", ",")
|
|
|
|
|
|
def ref_names(text: str | None) -> list[str]:
|
|
"""Имена вложений из текста в порядке появления, без повторов; внешние картинки (`://`) пропускаются."""
|
|
if not text:
|
|
return []
|
|
found = [(m.start(), m.group(1)) for m in _IMAGE.finditer(text) if "://" not in m.group(1)]
|
|
found += [(m.start(), m.group(1)) for m in _LINK.finditer(text)]
|
|
names: list[str] = []
|
|
for _, name in sorted(found, key=lambda f: f[0]):
|
|
if name not in names:
|
|
names.append(name)
|
|
return names
|
|
|
|
|
|
def _newest_key(item: dict):
|
|
aid = str(item["attachment_id"])
|
|
return item.get("created_at") or "", int(aid) if aid.isdigit() else -1, aid
|
|
|
|
|
|
def resolve(names: list[str], items: list[dict]) -> list[str]:
|
|
"""attachment_id по точному имени; при повторе имени — самое новое; имени нет среди вложений — пропуск."""
|
|
out: list[str] = []
|
|
for name in names:
|
|
same = [i for i in items if i["filename"] == name]
|
|
if same:
|
|
aid = str(max(same, key=_newest_key)["attachment_id"])
|
|
if aid not in out:
|
|
out.append(aid)
|
|
return out
|
|
|
|
|
|
def state(conn, files_dir, attachment_id: str) -> dict:
|
|
"""Состояние вложения: ready / fetching / failed / none. files_dir=None — файла в хранилище нет."""
|
|
cmds = conn.execute(
|
|
"SELECT id, batch_id, status, error FROM command WHERE type = 'fetch_attachment' "
|
|
"AND json_extract(params, '$.attachment_id') = ? ORDER BY id DESC", (attachment_id,)).fetchall()
|
|
last = cmds[0] if cmds else None
|
|
row = conn.execute("SELECT file FROM attachment_file WHERE attachment_id = ?", (attachment_id,)).fetchone()
|
|
if row is not None and files_dir is not None and files.exists(files_dir, row["file"]):
|
|
return {"state": "ready", "error": None, "command_id": last["id"] if last else None,
|
|
"batch_id": last["batch_id"] if last else None}
|
|
live = next((c for c in cmds if c["status"] in ("queued", "claimed")), None)
|
|
if live is not None:
|
|
return {"state": "fetching", "error": None, "command_id": live["id"], "batch_id": live["batch_id"]}
|
|
if last is not None and last["status"] in ("failed", "cancelled"):
|
|
return {"state": "failed", "error": last["error"], "command_id": last["id"], "batch_id": last["batch_id"]}
|
|
return {"state": "none", "error": None, "command_id": last["id"] if last else None,
|
|
"batch_id": last["batch_id"] if last else None}
|
|
|
|
|
|
def get(conn, files_dir, attachment_id) -> dict | None:
|
|
row = conn.execute(
|
|
"SELECT attachment_id, key, filename, size, mime_type, author, created FROM issue_attachment WHERE attachment_id = ?",
|
|
(attachment_id,)).fetchone()
|
|
if row is None:
|
|
return None
|
|
return {**dict(row), **state(conn, files_dir, attachment_id)}
|