84 lines
4.5 KiB
Python
84 lines
4.5 KiB
Python
"""Вложения задач для интерфейсов: ссылки на них в вики-разметке текста, формат размера и состояние (data-model
|
||
«Скачанные вложения», hub-api «Ссылки на вложения в тексте»). Состояние вычисляется при чтении; модуль только читает
|
||
базу и хранилище файлов, очередь и HTTP не знает."""
|
||
import re
|
||
|
||
from ...core import files
|
||
|
||
# lookahead: совпадения перекрываются, поэтому одиночный «!» ранее в тексте не съедает открывающий «!» настоящей ссылки;
|
||
# после открывающего «!» пробела нет (как в Jira), иначе закрывающий «!» одной картинки породил бы ложное имя
|
||
_IMAGE = re.compile(r"(?=!(?!\s)([^!|\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} Б"
|
||
# десятые — в целых, округление half up (как в вебе, fmtSize); единица — по округлённому значению:
|
||
# 1048575 Б — «1,0 МБ», а не «1024,0 КБ»
|
||
kb = (size * 10 + 512) // 1024
|
||
if kb < 10240:
|
||
return f"{kb // 10},{kb % 10} КБ"
|
||
mb = (size * 10 + 524288) // 1048576
|
||
return f"{mb // 10},{mb % 10} МБ"
|
||
|
||
|
||
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)}
|