From c5c3e8fbaa17b4998c390311c70d179900f44248 Mon Sep 17 00:00:00 2001 From: Simbiot Harness Date: Mon, 5 Oct 2026 14:17:03 +0000 Subject: [PATCH] =?UTF-8?q?feat:=20=D0=B2=D1=85=D0=BE=D0=B4=D1=8F=D1=89?= =?UTF-8?q?=D0=B8=D0=B5=20documents/photo=20=D0=BD=D0=B0=20=D0=B4=D0=B8?= =?UTF-8?q?=D1=81=D0=BA=20=D0=BF=D1=80=D0=B8=20=D0=BF=D1=80=D0=B8=D1=91?= =?UTF-8?q?=D0=BC=D0=B5=20(t-261005135853-186b42)=20=E2=80=94=20=D0=B3?= =?UTF-8?q?=D0=B5=D0=B9=D1=82=D0=B2=D0=B5=D0=B9=20=D1=81=D0=BA=D0=B0=D1=87?= =?UTF-8?q?=D0=B8=D0=B2=D0=B0=D0=B5=D1=82=20=D0=B4=D0=BE=D0=BA=D1=83=D0=BC?= =?UTF-8?q?=D0=B5=D0=BD=D1=82/=D1=84=D0=BE=D1=82=D0=BE=20(getFile=E2=86=92?= =?UTF-8?q?download,=20=D0=B0=D1=82=D0=BE=D0=BC=D0=B0=D1=80=D0=BD=D0=BE=20?= =?UTF-8?q?=D1=87=D0=B5=D1=80=D0=B5=D0=B7=20.part)=20=D0=B2=20inbox/files/?= =?UTF-8?q?.=20(paths.inbox=5Ffiles=5Fdir,=20SIMB?= =?UTF-8?q?IOT=5FINBOX=5FFILES=5FDIR),=20=D0=BF=D1=83=D1=82=D1=8C+=D0=B8?= =?UTF-8?q?=D0=BC=D1=8F+=D1=80=D0=B0=D0=B7=D0=BC=D0=B5=D1=80+mime=20=D0=B2?= =?UTF-8?q?=20MessageReceived;=20=D0=B8=D0=B4=D0=B5=D0=BC=D0=BF=D0=BE?= =?UTF-8?q?=D1=82=D0=B5=D0=BD=D1=82=D0=BD=D0=BE=20=D0=BF=D0=BE=20file=5Fun?= =?UTF-8?q?ique=5Fid,=20=D1=81=D0=B1=D0=BE=D0=B9=20getFile=20(=D0=B8=D1=81?= =?UTF-8?q?=D1=82=D1=91=D0=BA=D1=88=D0=B8=D0=B9=20=D1=84=D0=B0=D0=B9=D0=BB?= =?UTF-8?q?)=20=D0=BD=D0=B5=20=D0=B2=D0=B0=D0=BB=D0=B8=D1=82=20=D0=BF?= =?UTF-8?q?=D0=BE=D0=BB=D0=BB=D0=B5=D1=80=20=E2=80=94=20file=5Fid=20=D0=BE?= =?UTF-8?q?=D1=81=D1=82=D0=B0=D1=91=D1=82=D1=81=D1=8F;=20=D1=84=D0=BE?= =?UTF-8?q?=D1=82=D0=BE=20=E2=80=94=20=D0=BC=D0=B0=D0=BA=D1=81=D0=B8=D0=BC?= =?UTF-8?q?=D0=B0=D0=BB=D1=8C=D0=BD=D1=8B=D0=B9=20=D1=80=D0=B0=D0=B7=D0=BC?= =?UTF-8?q?=D0=B5=D1=80;=20=D1=82=D0=B5=D1=81=D1=82=2022=20PASS=20+=20?= =?UTF-8?q?=D1=80=D0=B5=D0=B3=D1=80=D0=B5=D1=81=D1=81=20voice/seen/batch?= =?UTF-8?q?=20=D0=B7=D0=B5=D0=BB=D1=91=D0=BD=D1=8B=D0=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- harness/paths.py | 17 ++++ harness/tg_inbox.py | 99 +++++++++++++++++++++ tests/test_tg_documents.py | 174 +++++++++++++++++++++++++++++++++++++ 3 files changed, 290 insertions(+) create mode 100644 tests/test_tg_documents.py diff --git a/harness/paths.py b/harness/paths.py index 76845c4..c7dce89 100644 --- a/harness/paths.py +++ b/harness/paths.py @@ -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: diff --git a/harness/tg_inbox.py b/harness/tg_inbox.py index 6661bfb..a778c32 100644 --- a/harness/tg_inbox.py +++ b/harness/tg_inbox.py @@ -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/. и пишет путь в 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) на diff --git a/tests/test_tg_documents.py b/tests/test_tg_documents.py new file mode 100644 index 0000000..456c3d5 --- /dev/null +++ b/tests/test_tg_documents.py @@ -0,0 +1,174 @@ +# -*- coding: utf-8 -*- +"""t-261005135853: входящие documents/photo → на диск, путь в событии. + +Покрытие: + 1. unit: _file_ext — имя/mime/мусор/пусто; + 2. e2e: MessageReceived(document) → getFile/download (моки) → файл в + inbox/files/.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)