feat: входящие documents/photo на диск при приёме (t-261005135853-186b42) — гейтвей скачивает документ/фото (getFile→download, атомарно через .part) в inbox/files/<file_unique_id>.<ext> (paths.inbox_files_dir, SIMBIOT_INBOX_FILES_DIR), путь+имя+размер+mime в MessageReceived; идемпотентно по file_unique_id, сбой getFile (истёкший файл) не валит поллер — file_id остаётся; фото — максимальный размер; тест 22 PASS + регресс voice/seen/batch зелёный
This commit is contained in:
Simbiot Harness
parent
20e655e4ec
commit
c5c3e8fbaa
3 changed files with 290 additions and 0 deletions
|
|
@ -109,6 +109,23 @@ def inbox_outbox_dir() -> str:
|
|||
return os.path.join(REPO_ROOT, "portal", "inbox", "outbox")
|
||||
|
||||
|
||||
# ── входящие файлы каналов (t-261005135853): documents/photo на диск ────────
|
||||
def inbox_files_dir() -> str:
|
||||
"""Каталог входящих файлов (documents/photo из TG).
|
||||
|
||||
Голосовые живут в voice/ (кэш ASR), а документы/фото — данные уровня
|
||||
инбокса: выписки, сканы, артефакты Алекса. Не задаём SIMBIOT_HOME
|
||||
(legacy) → /var/lib/simbiot-harness/inbox/files — тот же корень, где
|
||||
голосовые (voice_dir legacy), рядом с inbox/ каналов портала.
|
||||
"""
|
||||
env = os.environ.get("SIMBIOT_INBOX_FILES_DIR", "").strip()
|
||||
if env:
|
||||
return env
|
||||
if IS_ISOLATED:
|
||||
return os.path.join(_HOME, "inbox", "files")
|
||||
return "/var/lib/simbiot-harness/inbox/files"
|
||||
|
||||
|
||||
# ── runtime-штампы каналов для дашборда ─────────────────────────────────────
|
||||
def stamp_dir() -> str:
|
||||
if IS_ISOLATED:
|
||||
|
|
|
|||
|
|
@ -140,6 +140,88 @@ def typing_keepalive(chat_ids: list, stop, interval: float = 2.0):
|
|||
pass
|
||||
|
||||
|
||||
# ── входящие документы/фото → диск (t-261005135853) ─────────────────────────
|
||||
# Голосовые качает voice-пайплайн (voice/, кэш ASR), а documents/photo раньше
|
||||
# терялись: file_id есть в событии, файл не скачивался (кейс 4 PDF-выписок
|
||||
# Alex 05.10 — доставал вручную getFile'ом). Теперь гейтвей сам кладёт файл
|
||||
# в inbox/files/<file_unique_id>.<ext> и пишет путь в MessageReceived.
|
||||
_SAFE_EXT_RE = None # ленивый импорт re
|
||||
|
||||
|
||||
def _file_ext(name: str, mime: str = "") -> str:
|
||||
"""Расширение из имени/mime, безопасное (без путей и мусора)."""
|
||||
global _SAFE_EXT_RE
|
||||
if _SAFE_EXT_RE is None:
|
||||
import re
|
||||
_SAFE_EXT_RE = re.compile(r"[^A-Za-z0-9]")
|
||||
ext = ""
|
||||
if name and "." in name:
|
||||
ext = name.rsplit(".", 1)[-1]
|
||||
if not ext and "/" in (mime or ""):
|
||||
sub = mime.split("/", 1)[-1]
|
||||
ext = {"pdf": "pdf", "jpeg": "jpg", "png": "png", "webp": "webp",
|
||||
"octet-stream": "bin"}.get(sub, sub)
|
||||
ext = _SAFE_EXT_RE.sub("", ext)[:8]
|
||||
return ext or "bin"
|
||||
|
||||
|
||||
def save_document(msg: dict):
|
||||
"""Скачать document/photo из сообщения → inbox/files/.
|
||||
|
||||
Возвращает абсолютный путь к файлу или None: сбой скачивания не должен
|
||||
валить поллер — путь просто не попадёт в событие, file_id в нём
|
||||
останется для ручного достёгивания (как было до фикса).
|
||||
Идемпотентно: file_unique_id в имени → повторный апдейт не перекачивает
|
||||
(файл на диске = готовый ответ).
|
||||
"""
|
||||
doc = msg.get("document")
|
||||
photo = msg.get("photo")
|
||||
if not doc and not photo:
|
||||
return None
|
||||
if doc:
|
||||
file_id = doc.get("file_id")
|
||||
uniq = doc.get("file_unique_id") or file_id
|
||||
ext = _file_ext(doc.get("file_name") or "", doc.get("mime_type") or "")
|
||||
else:
|
||||
biggest = photo[-1] # самый большой размер
|
||||
file_id = biggest.get("file_id")
|
||||
uniq = biggest.get("file_unique_id") or file_id
|
||||
ext = "jpg"
|
||||
if not file_id:
|
||||
return None
|
||||
files_dir = _paths.inbox_files_dir()
|
||||
abs_path = os.path.join(files_dir, f"{uniq}.{ext}")
|
||||
if not os.path.exists(abs_path): # кэш: один файл — одна закачка
|
||||
try:
|
||||
_load_env()
|
||||
if not TG_TOKEN:
|
||||
return None
|
||||
resp = _api("getFile", {"file_id": file_id})
|
||||
if not resp.get("ok"):
|
||||
return None
|
||||
fp = (resp.get("result") or {}).get("file_path")
|
||||
if not fp:
|
||||
return None
|
||||
durl = f"https://api.telegram.org/file/bot{TG_TOKEN}/{fp}"
|
||||
with urllib.request.urlopen(durl, timeout=90) as r:
|
||||
blob = r.read()
|
||||
if not blob:
|
||||
return None
|
||||
os.makedirs(files_dir, exist_ok=True)
|
||||
tmp = abs_path + ".part"
|
||||
with open(tmp, "wb") as f:
|
||||
f.write(blob)
|
||||
os.replace(tmp, abs_path) # атомарно: нет наполовину записанных
|
||||
except urllib.error.HTTPError as e:
|
||||
# 400/410 = file истёк у TG — не ошибка поллера, тихий отказ
|
||||
print(f"[tg] getFile HTTP {e.code}: файл {uniq} недоступен", flush=True)
|
||||
return None
|
||||
except Exception as e:
|
||||
print(f"[tg] download {uniq} error: {e}", flush=True)
|
||||
return None
|
||||
return abs_path
|
||||
|
||||
|
||||
def receive_message(update: dict):
|
||||
"""MessageReceived в стор. Идемпотентно по update_id."""
|
||||
msg = update.get("message") or update.get("edited_message") or {}
|
||||
|
|
@ -170,6 +252,15 @@ def receive_message(update: dict):
|
|||
photo = msg.get("photo")
|
||||
if photo and not file_id:
|
||||
file_id = photo[-1].get("file_id") # самый большой размер
|
||||
# t-261005135853: documents/photo сразу на диск при приёме — путь в
|
||||
# событии, файл не теряется (голосовые качает voice-пайплайн отдельно).
|
||||
# Сбой скачивания не валит приём: file_path=None, file_id остаётся.
|
||||
doc_rel = None
|
||||
if kind in ("document", "photo"):
|
||||
try:
|
||||
doc_rel = save_document(msg)
|
||||
except Exception as e:
|
||||
print(f"[tg] save_document error: {e}", flush=True)
|
||||
data = {
|
||||
"update_id": update["update_id"],
|
||||
"chat_id": chat.get("id"),
|
||||
|
|
@ -185,6 +276,14 @@ def receive_message(update: dict):
|
|||
# file_unique_id: ключ аудио-кэша и идемпотентности расшифровки
|
||||
# (file_id ротируется, unique — постоянный для одного файла)
|
||||
"file_unique_id": file.get("file_unique_id") or (photo[-1].get("file_unique_id") if photo else None),
|
||||
# t-261005135853: путь к скачанному файлу (абсолютный, открывается
|
||||
# напрямую с любого уровня) + исходное имя/размер/mime — триаж видит
|
||||
# что за файл, без getFile. None = скачать не удалось (file_id остаётся).
|
||||
"file_path": doc_rel,
|
||||
"file_name": (msg.get("document") or {}).get("file_name"),
|
||||
"file_size": (msg.get("document") or {}).get("file_size")
|
||||
or (photo[-1].get("file_size") if photo else None),
|
||||
"mime_type": (msg.get("document") or {}).get("mime_type"),
|
||||
"duration": duration,
|
||||
"date": msg.get("date"),
|
||||
# reply_to_message_id (t-261002085926): когда Alex отвечает (reply) на
|
||||
|
|
|
|||
174
tests/test_tg_documents.py
Normal file
174
tests/test_tg_documents.py
Normal file
|
|
@ -0,0 +1,174 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
"""t-261005135853: входящие documents/photo → на диск, путь в событии.
|
||||
|
||||
Покрытие:
|
||||
1. unit: _file_ext — имя/mime/мусор/пусто;
|
||||
2. e2e: MessageReceived(document) → getFile/download (моки) → файл в
|
||||
inbox/files/<unique>.pdf, file_path (абсолютный) + file_name + size
|
||||
в проекции inbox;
|
||||
3. e2e: photo (3 размера) → .jpg самого большого размера;
|
||||
4. идемпотентность: повторный апдейт не перекачивает (счётчик getFile);
|
||||
5. сбой getFile (ok=False / HTTP 400) → file_path=None, входящее живо
|
||||
(file_id остаётся для ручного достёгивания).
|
||||
|
||||
Запуск: ./venv/bin/python tests/test_tg_documents.py | pytest tests/test_tg_documents.py
|
||||
Telegram-сеть мокнута (нет реальных file_id), стор — свой в /tmp.
|
||||
"""
|
||||
import io
|
||||
import os
|
||||
import sys
|
||||
import urllib.error
|
||||
|
||||
# ── изоляция: свой стор и свой каталог файлов, прод не трогаем ──
|
||||
_DST = "/tmp/simbiot-test-tg-docs.db"
|
||||
_FILES = "/tmp/simbiot-test-tg-docs-files"
|
||||
for p in (_DST, _DST + "-wal", _DST + "-shm"):
|
||||
if os.path.exists(p):
|
||||
os.unlink(p)
|
||||
os.environ["SIMBIOT_DB"] = _DST
|
||||
os.environ["SIMBIOT_INBOX_FILES_DIR"] = _FILES
|
||||
sys.path.insert(0, "/root/simbiot-harness")
|
||||
|
||||
from harness import es, tg_inbox
|
||||
|
||||
fails = []
|
||||
_calls = {"getFile": 0}
|
||||
|
||||
|
||||
def check(name, cond, extra=""):
|
||||
print(("PASS" if cond else "FAIL"), name, (f"| {extra}" if not cond else ""), flush=True)
|
||||
if not cond:
|
||||
fails.append(name)
|
||||
|
||||
|
||||
# ── моки Telegram: getFile → file_path, urlopen → байты файла ──
|
||||
class _Resp(io.BytesIO):
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, *a):
|
||||
return False
|
||||
|
||||
|
||||
def _fake_api(method, params, timeout=35):
|
||||
if method == "getFile":
|
||||
_calls["getFile"] += 1
|
||||
return {"ok": True, "result": {
|
||||
"file_id": params["file_id"],
|
||||
"file_path": f"documents/{params['file_id']}.pdf"}}
|
||||
return {"ok": True, "result": {}}
|
||||
|
||||
|
||||
def _fake_urlopen(url, timeout=60):
|
||||
if "getFile-fail" in url or "expired" in url:
|
||||
raise urllib.error.HTTPError(url, 400, "Bad Request", {}, io.BytesIO(b""))
|
||||
return _Resp(b"%PDF-1.4 fake-test-blob")
|
||||
|
||||
|
||||
tg_inbox._api = _fake_api
|
||||
tg_inbox._load_env = lambda: setattr(tg_inbox, "TG_TOKEN", "TEST:TOKEN") # токен мока
|
||||
_real_urlopen = urllib.request.urlopen
|
||||
urllib.request.urlopen = lambda url, timeout=60: _fake_urlopen(url, timeout)
|
||||
|
||||
es.init_db()
|
||||
|
||||
|
||||
# ────────────────────────────── 1. _file_ext ────────────────────────────────
|
||||
|
||||
def test_file_ext():
|
||||
check("ext из имени", tg_inbox._file_ext("Alfa-Bank_выписка.PDF") == "PDF"
|
||||
or tg_inbox._file_ext("Alfa-Bank_выписка.PDF").lower() == "pdf",
|
||||
tg_inbox._file_ext("Alfa-Bank_выписка.PDF"))
|
||||
check("ext из mime", tg_inbox._file_ext("", "image/jpeg") == "jpg",
|
||||
tg_inbox._file_ext("", "image/jpeg"))
|
||||
check("мусор вырезан", tg_inbox._file_ext("a.tar.gz/../../x", "") in ("gz", "x", "bin"),
|
||||
tg_inbox._file_ext("a.tar.gz/../../x"))
|
||||
check("пусто → bin", tg_inbox._file_ext("", "") == "bin")
|
||||
|
||||
|
||||
# ─────────────────────── 2. document → файл + путь в событии ────────────────
|
||||
|
||||
def test_document_e2e():
|
||||
upd = {"update_id": 990001, "message": {
|
||||
"message_id": 111, "date": 1791200000,
|
||||
"from": {"id": 1, "first_name": "A"},
|
||||
"chat": {"id": 1},
|
||||
"document": {"file_id": "file-AAA", "file_unique_id": "AgADdoc1",
|
||||
"file_name": "Alfa_Statement.pdf",
|
||||
"mime_type": "application/pdf", "file_size": 123456},
|
||||
"caption": "выписка октябрь",
|
||||
}}
|
||||
mid = tg_inbox.receive_message(upd)
|
||||
p = f"{_FILES}/AgADdoc1.pdf"
|
||||
check("файл в inbox/files", os.path.exists(p), p)
|
||||
check("содержимое скачано", os.path.exists(p) and open(p, "rb").read() == b"%PDF-1.4 fake-test-blob")
|
||||
m = es.run_projections()["inbox"][mid]
|
||||
check("file_path в событии (абс)", m.get("file_path") == p, m.get("file_path"))
|
||||
check("путь открывается напрямую", os.path.exists(m.get("file_path") or ""))
|
||||
check("file_name в событии", m.get("file_name") == "Alfa_Statement.pdf")
|
||||
check("file_size в событии", m.get("file_size") == 123456)
|
||||
check("kind=document", m.get("kind") == "document")
|
||||
check("caption в text", m.get("text") == "выписка октябрь")
|
||||
return upd
|
||||
|
||||
|
||||
# ───────────────────────────── 3. photo → .jpg ───────────────────────────────
|
||||
|
||||
def test_photo_e2e():
|
||||
upd = {"update_id": 990002, "message": {
|
||||
"message_id": 112, "date": 1791200001,
|
||||
"from": {"id": 1, "first_name": "A"},
|
||||
"chat": {"id": 1},
|
||||
"photo": [
|
||||
{"file_id": "ph-sm", "file_unique_id": "UPsm", "file_size": 100, "width": 90, "height": 90},
|
||||
{"file_id": "ph-md", "file_unique_id": "UPmd", "file_size": 900, "width": 320, "height": 320},
|
||||
{"file_id": "ph-lg", "file_unique_id": "UPlg", "file_size": 9900, "width": 1280, "height": 1280},
|
||||
],
|
||||
}}
|
||||
mid = tg_inbox.receive_message(upd)
|
||||
p = f"{_FILES}/UPlg.jpg"
|
||||
check("фото: самый большой размер сохранён", os.path.exists(p), p)
|
||||
m = es.run_projections()["inbox"][mid]
|
||||
check("photo file_path", m.get("file_path") == p, m.get("file_path"))
|
||||
check("photo kind", m.get("kind") == "photo")
|
||||
check("photo file_size", m.get("file_size") == 9900)
|
||||
return upd
|
||||
|
||||
|
||||
# ─────────────────── 4. идемпотентность: повтор не качает ────────────────────
|
||||
|
||||
def test_no_redownload(upd):
|
||||
before = _calls["getFile"]
|
||||
mtime = os.path.getmtime(f"{_FILES}/AgADdoc1.pdf")
|
||||
tg_inbox.receive_message(upd) # тот же update_id → тот же файл
|
||||
check("getFile не рос при повторе", _calls["getFile"] == before,
|
||||
f"{before} → {_calls['getFile']}")
|
||||
check("файл не переписан", os.path.getmtime(f"{_FILES}/AgADdoc1.pdf") == mtime)
|
||||
|
||||
|
||||
# ──────────────────── 5. сбой скачивания не валит приём ─────────────────────
|
||||
|
||||
def test_download_failed():
|
||||
upd = {"update_id": 990003, "message": {
|
||||
"message_id": 113, "date": 1791200002,
|
||||
"from": {"id": 1, "first_name": "A"},
|
||||
"chat": {"id": 1},
|
||||
"document": {"file_id": "getFile-fail-expired", "file_unique_id": "ADdead",
|
||||
"file_name": "old.pdf", "mime_type": "application/pdf"},
|
||||
}}
|
||||
mid = tg_inbox.receive_message(upd)
|
||||
m = es.run_projections()["inbox"][mid]
|
||||
check("file_path=None при сбое", m.get("file_path") is None, m.get("file_path"))
|
||||
check("file_id сохранился (ручной путь)", m.get("file_id") == "getFile-fail-expired")
|
||||
check("входящее живо (не упало)", bool(mid))
|
||||
check("мусора .part не осталось", not any(f.endswith(".part") for f in os.listdir(_FILES)))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
test_file_ext()
|
||||
upd = test_document_e2e()
|
||||
test_photo_e2e()
|
||||
test_no_redownload(upd)
|
||||
test_download_failed()
|
||||
print(f"\n{'ALL PASS' if not fails else 'FAILURES: ' + ', '.join(fails)}")
|
||||
sys.exit(1 if fails else 0)
|
||||
Loading…
Reference in a new issue