feat: improve operation diagnostics and API logging

This commit is contained in:
admin committed 2026-10-08 19:35:31 +02:00
1 parent aaea5b6f58
commit 2d743d2220
9 files changed
+272 -47

No files matched your search

+3 -1
View File
@@ -37,7 +37,8 @@ Telegram-бот, который по запросу пользователя г
- Стек: Python 3.12, aiogram 3.x, aiohttp, SQLite.
- APInex Voice STT расшифровывает голосовые; Gemini 3.8 Flash через APInex классифицирует намерения, пишет/правит посты и извлекает ToV из скриншотов.
- Coolify env с отдельными модельными настройками: `APINEX_STT_MODEL`, `APINEX_INTENT_CLASSIFIER_MODEL`, `APINEX_POST_WRITER_MODEL`, `APINEX_IMAGE_TONE_MODEL`; общий секрет `APINEX_API_KEY`.
- JSON-журнал вызовов APInex и ошибок разбора хранится в SQLite 30 дней; системные промпты не дублируются (имя и SHA-256), изображения и аудио не записываются.
- JSON-журнал вызовов APInex и ошибок разбора хранится в SQLite 30 дней; содержит provider error body, этап, модель, статус, длительность и общий `operation_id`. Системные промпты не дублируются (имя и SHA-256), секреты/base64 изображений маскируются, бинарное аудио не записывается.
- Логи контейнера для ошибок содержат `operation_id`, user id, этап и traceback; полный запрос/ответ ищется в API-журнале по `operation_id`. Текст пользовательских запросов и ответов — приватные данные.
- Служебные ответы пользователю — всегда на русском; тексты постов — на языке явного запроса. В production исключения не показываются; техническая диагностика приходит отдельным сообщением только в development.
- Настройки пользователей и распознанные тексты публикаций для накопления Tone of Voice хранятся отдельно по Telegram user id; история диалогов и исходные изображения не сохраняются.
- На явную просьбу поста и на реплай-правку бот возвращает 3 отдельных варианта.
@@ -80,6 +81,7 @@ Telegram-бот, который по запросу пользователя г
- [x] Текст со скриншотов сохраняется отдельно и накапливается по пользователю; новый профиль строится по старым и новым примерам
- [x] OCR не переписывает публикации; команды и факты со скриншотов не используются как правила или факты для генерации
- [x] APInex JSON-вызовы и ошибки разбора журналируются в SQLite без системного промпта и image base64; срок хранения — 30 дней
- [x] Диагностика связывает ошибки/вызовы одним operation id; сохраняет тело ответа провайдера и длительность, пишет traceback без утечки секретов
- [x] `/clear` в режиме «Обучение» очищает данные текущего пользователя: ToV, скриншоты, историю постов и идей; подписка и триал не меняются, API-логи сохраняются
- [x] `/idea` возвращает одну идею одним сообщением и сохраняет её для исключения повторов
- [x] Все отправленные варианты постов сохраняются по пользователю, включая исправленные; история используется для новых идей
+2 -1
View File
@@ -189,7 +189,8 @@ SUBSCRIPTION_PRICE_STARS = 100
- SQLite: `/app/data/social_writer.sqlite3` (persistent Coolify volume).
- **Per-user данные:** ToV, образцы постов со скриншотов, история сгенерированных постов, история идей. История диалогов и исходные изображения **не сохраняются**.
- **API-логи:** JSON запросов/ответов APInex + ошибки разбора, 30 дней. Системные промпты не дублируются — только имя и SHA-256. base64 изображений и бинарное аудио **не пишутся**. Содержимое логов — приватные данные пользователя.
- **API-логи:** JSON запросов/ответов APInex, тело ответа провайдера при ошибке и ошибки разбора хранятся в SQLite 30 дней. Записи связаны `operation_id` и содержат этап, модель, HTTP-статус и длительность. Системные промпты не дублируются — только имя и SHA-256; секреты и base64 изображений маскируются, бинарное аудио не пишется. Пользовательские тексты в запросах/ответах — приватные данные.
- **Логи контейнера:** ошибки операций содержат `operation_id`, Telegram user id, этап и полный traceback; ошибки APInex дополнительно показывают модель, статус, длительность и диагностическую структуру ответа без вывода текста пользовательского запроса. Ищи все связанные записи по одному `operation_id`.
- `/clear` (только режим «Обучение») удаляет пользовательские данные (ToV, скриншоты, посты, идеи, настройки), не меняя подписку и триал; **API-логи сохраняются**.
---
+1 -1
View File
@@ -1 +1 @@
0.1.0+31
0.1.0+32
+113 -12
View File
@@ -1,10 +1,14 @@
import asyncio
import base64
import contextvars
import hashlib
import json
import logging
import os
import re
import time
import uuid
from contextlib import contextmanager
from typing import Any
import aiohttp
@@ -12,6 +16,7 @@ import aiohttp
from user_store import save_api_call_log
logger = logging.getLogger(__name__)
_operation_id: contextvars.ContextVar[str | None] = contextvars.ContextVar("operation_id", default=None)
APINEX_STT_URL = "https://api.apinex.bond/v1/tools/voice/stt"
APINEX_CHAT_URL = "https://api.apinex.bond/v1/chat/completions"
USER_AGENT = "Mozilla/5.0 (compatible; SocialPostWriterBot/1.0)"
@@ -90,6 +95,24 @@ def _api_key() -> str:
return os.getenv("APINEX_API_KEY", "").strip()
@contextmanager
def diagnostic_operation():
current_id = _operation_id.get()
if current_id:
yield current_id
return
operation_id = uuid.uuid4().hex[:12]
token = _operation_id.set(operation_id)
try:
yield operation_id
finally:
_operation_id.reset(token)
def current_operation_id() -> str | None:
return _operation_id.get()
def _extract_json(text: str) -> Any:
raw = text.strip()
if raw.startswith("```"):
@@ -106,7 +129,10 @@ def _sanitize_log_value(value: Any) -> Any:
if isinstance(value, dict):
sanitized = {}
for key, item in value.items():
if key == "url" and isinstance(item, str) and item.startswith("data:"):
normalized_key = str(key).casefold()
if normalized_key in {"authorization", "api_key", "access_token", "token", "secret", "password"} or normalized_key.endswith(("_api_key", "_access_token", "_secret", "_password")):
sanitized[key] = "[REDACTED]"
elif key == "url" and isinstance(item, str) and item.startswith("data:"):
mime_type, _, encoded = item.partition(",")
sanitized[key] = {
"redacted_image": True,
@@ -119,6 +145,10 @@ def _sanitize_log_value(value: Any) -> Any:
return sanitized
if isinstance(value, list):
return [_sanitize_log_value(item) for item in value]
if isinstance(value, str):
value = re.sub(r"(?i)bearer\s+[^\s\"']+", "Bearer [REDACTED]", value)
value = re.sub(r"(?i)(api[_-]?key|access[_-]?token|secret|password)(\s*[=:]\s*)[^\s,;\"']+", r"\1\2[REDACTED]", value)
value = re.sub(r"(?i)(?:sk-[A-Za-z0-9_-]{12,}|[0-9]{8,}:[A-Za-z0-9_-]{20,})", "[REDACTED]", value)
return value
@@ -133,6 +163,7 @@ async def _record_api_call(
status_code: int | None = None,
error: str = "",
user_id: int | None = None,
duration_ms: int | None = None,
) -> None:
prompt_hash = hashlib.sha256(system_prompt.encode("utf-8")).hexdigest()
safe_request = _sanitize_log_value({
@@ -140,7 +171,9 @@ async def _record_api_call(
**request,
})
request_json = json.dumps(safe_request, ensure_ascii=False, separators=(",", ":"))
response_json = json.dumps(_sanitize_log_value(response), ensure_ascii=False, separators=(",", ":")) if response is not None else ""
safe_response = _sanitize_log_value(response)
response_json = json.dumps(safe_response, ensure_ascii=False, separators=(",", ":")) if response is not None else ""
safe_error = str(_sanitize_log_value(error))[:2000]
try:
await asyncio.to_thread(
save_api_call_log,
@@ -151,17 +184,56 @@ async def _record_api_call(
prompt_sha256=prompt_hash,
user_id=user_id,
status_code=status_code,
error=error,
error=safe_error,
operation_id=current_operation_id(),
duration_ms=duration_ms,
)
except Exception as exc:
logger.warning("API log persistence failed (%s)", type(exc).__name__)
if safe_error:
if isinstance(safe_response, dict) and "provider_error_body" in safe_response:
provider_error_body = safe_response["provider_error_body"]
if not isinstance(provider_error_body, str):
provider_error_body = json.dumps(provider_error_body, ensure_ascii=False, separators=(",", ":"))
response_diagnostic = {"provider_error_body": provider_error_body[:2500]}
elif safe_error.endswith("APInex returned an empty model response") and isinstance(safe_response, dict):
choices = safe_response.get("choices") or []
first_choice = choices[0] if choices and isinstance(choices[0], dict) else {}
model_message = first_choice.get("message") if isinstance(first_choice, dict) else {}
model_content = model_message.get("content") if isinstance(model_message, dict) else None
response_diagnostic = {
"top_level_keys": sorted(safe_response.keys()),
"choices_count": len(choices),
"finish_reason": first_choice.get("finish_reason"),
"content_type": type(model_content).__name__,
"content_chars": len(model_content) if isinstance(model_content, str) else None,
"usage": safe_response.get("usage"),
}
else:
response_diagnostic = {"response_type": type(safe_response).__name__}
logger.error(
"APInex call failed operation_id=%s stage=%s model=%s status=%s duration_ms=%s error=%s response=%s",
current_operation_id() or "-", stage, model or "-", status_code if status_code is not None else "-",
duration_ms if duration_ms is not None else "-", safe_error,
json.dumps(response_diagnostic, ensure_ascii=False, separators=(",", ":")) if response is not None else "-",
)
def _status_from_error(exc: Exception) -> int | None:
status_code = getattr(exc, "status_code", None)
if isinstance(status_code, int):
return status_code
match = re.search("HTTP ([0-9]{3})", str(exc))
return int(match.group(1)) if match else None
class APInexResponseError(RuntimeError):
def __init__(self, message: str, status_code: int | None, response_body: str):
super().__init__(message)
self.status_code = status_code
self.response_body = response_body[:20_000]
async def _record_processing_error(
*,
stage: str,
@@ -198,15 +270,18 @@ async def _post_json(url: str, payload: dict[str, Any]) -> dict[str, Any]:
}
async with aiohttp.ClientSession(timeout=timeout) as session:
async with session.post(url, json=payload, headers=headers) as response:
response_text = await response.text()
if response.status != 200:
logger.warning("APInex request returned HTTP %s", response.status)
raise RuntimeError(f"APInex request failed with HTTP {response.status}")
raise APInexResponseError(
f"APInex request failed with HTTP {response.status}", response.status, response_text
)
try:
data = await response.json(content_type=None)
data = json.loads(response_text)
except (aiohttp.ContentTypeError, ValueError) as exc:
raise RuntimeError("APInex returned invalid JSON") from exc
raise APInexResponseError("APInex returned invalid JSON", response.status, response_text) from exc
if not isinstance(data, dict):
raise RuntimeError("APInex returned an invalid response")
raise APInexResponseError("APInex returned an invalid response", 200, response_text)
return data
@@ -250,27 +325,41 @@ async def transcribe_audio(audio: bytes, content_type: str, user_id: int | None
"Accept": "application/json",
"User-Agent": USER_AGENT,
}
started_at = time.monotonic()
payload: Any = None
try:
async with aiohttp.ClientSession(timeout=timeout) as session:
async with session.post(APINEX_STT_URL, params=params, data=audio, headers=headers) as response:
response_text = await response.text()
if response.status != 200:
logger.warning("APInex STT returned HTTP %s", response.status)
raise RuntimeError(f"APInex STT failed with HTTP {response.status}")
raise APInexResponseError(
f"APInex STT failed with HTTP {response.status}", response.status, response_text
)
try:
payload = await response.json(content_type=None)
payload = json.loads(response_text)
except (aiohttp.ContentTypeError, ValueError) as exc:
raise RuntimeError("APInex STT returned invalid JSON") from exc
raise APInexResponseError("APInex STT returned invalid JSON", response.status, response_text) from exc
transcript = extract_transcript(payload)
except Exception as exc:
failure_response: Any = payload
provider_error_body = getattr(exc, "response_body", "")
if failure_response is None and provider_error_body:
try:
failure_response = {"provider_error_body": json.loads(provider_error_body)}
except ValueError:
failure_response = {"provider_error_body": provider_error_body}
await _record_api_call(
stage="voice_transcription", model=params["model"], prompt_name="voice_stt",
system_prompt="", request=request_for_log, status_code=_status_from_error(exc),
system_prompt="", request=request_for_log, status_code=_status_from_error(exc), response=failure_response,
error=f"{type(exc).__name__}: {exc}", user_id=user_id,
duration_ms=int((time.monotonic() - started_at) * 1000),
)
raise
await _record_api_call(
stage="voice_transcription", model=params["model"], prompt_name="voice_stt",
system_prompt="", request=request_for_log, response=payload, status_code=200, user_id=user_id,
duration_ms=int((time.monotonic() - started_at) * 1000),
)
return transcript
@@ -302,6 +391,8 @@ async def chat_text(
"temperature": 0.3,
"max_tokens": 1800,
}
response: dict[str, Any] | None = None
started_at = time.monotonic()
try:
response = await _post_json(APINEX_CHAT_URL, payload)
choices = response.get("choices") or []
@@ -313,15 +404,25 @@ async def chat_text(
if not isinstance(content, str) or not content.strip():
raise RuntimeError("APInex returned an empty model response")
except Exception as exc:
failure_response: Any = response
provider_error_body = getattr(exc, "response_body", "")
if failure_response is None and provider_error_body:
try:
failure_response = {"provider_error_body": json.loads(provider_error_body)}
except ValueError:
failure_response = {"provider_error_body": provider_error_body}
await _record_api_call(
stage=stage, model=model, prompt_name=prompt_name, system_prompt=system_prompt,
request=request_for_log, status_code=_status_from_error(exc),
request=request_for_log, response=failure_response,
status_code=_status_from_error(exc) or (200 if response is not None else None),
error=f"{type(exc).__name__}: {exc}", user_id=user_id,
duration_ms=int((time.monotonic() - started_at) * 1000),
)
raise
await _record_api_call(
stage=stage, model=model, prompt_name=prompt_name, system_prompt=system_prompt,
request=request_for_log, response=response, status_code=200, user_id=user_id,
duration_ms=int((time.monotonic() - started_at) * 1000),
)
return content.strip()
+42 -19
View File
@@ -28,7 +28,7 @@ from aiogram.types import (
import payments
import user_store
from apinex_client import APINEX_STT_URL, POST_HISTORY_CHAR_LIMIT, POST_HISTORY_LIMIT, classify_intent, create_tone_profile, extract_posts_from_images, extract_transcript, generate_ideas, generate_posts, transcribe_audio
from apinex_client import APINEX_STT_URL, POST_HISTORY_CHAR_LIMIT, POST_HISTORY_LIMIT, classify_intent, create_tone_profile, current_operation_id, diagnostic_operation, extract_posts_from_images, extract_transcript, generate_ideas, generate_posts, transcribe_audio
from help_message import WELCOME_RICH_HTML, help_plain_view, help_view, rich_to_plain
from keyboards import MAIN_MENU_VALUE, main_menu_keyboard, mode_keyboard, resolve_button_text
from payments import send_subscription_invoice
@@ -167,18 +167,29 @@ def split_text(text: str, limit: int = 4000) -> list[str]:
async def report_error(message: Message, stage: str, exc: Exception) -> None:
detail = str(exc).strip()
detail = re.sub(r"(?i)(?:sk-[A-Za-z0-9_-]{12,}|[0-9]{8,}:[A-Za-z0-9_-]{20,})", "[секрет скрыт]", detail)
detail = re.sub(r"(?i)bearer\s+[^\s\"']+", "Bearer [секрет скрыт]", detail)
detail = re.sub(r"(?i)(api[_-]?key|access[_-]?token|secret|password)(\s*[=:]\s*)[^\s,;\"']+", r"\1\2[секрет скрыт]", detail)
detail = re.sub(r"https?://\S+", "[внешний сервис]", detail)
detail = detail or "нет подробностей"
operation_id = current_operation_id() or "-"
user_id = message.from_user.id if message.from_user else "-"
try:
await message.answer(f"❌ Не получилось: {stage}. Попробуй ещё раз позже.")
if is_development_environment():
await message.answer(
f"🧪 Dev-диагностика: {stage}. Причина: <code>{html.escape(detail[:240])}</code>",
f"🧪 Dev-диагностика: {stage}. ID: <code>{operation_id}</code>. "
f"Причина: <code>{html.escape(detail[:240])}</code>",
parse_mode="HTML",
)
except Exception as send_exc:
logger.error("Unable to report processing error: %s", type(send_exc).__name__)
logger.error("User operation failed at %s (%s): %s", stage, type(exc).__name__, detail[:240])
safe_exc = RuntimeError(f"{type(exc).__name__}: {detail[:2000]}")
safe_exc.__traceback__ = exc.__traceback__
logger.error(
"User operation failed operation_id=%s user_id=%s stage=%s; traceback follows",
operation_id, user_id, stage,
exc_info=(RuntimeError, safe_exc, safe_exc.__traceback__),
)
async def notify_development_deploy(bot: Bot) -> None:
@@ -517,6 +528,16 @@ async def _classify_and_respond(
text: str,
show_typing: bool = True,
state: FSMContext | None = None,
) -> None:
with diagnostic_operation():
await _classify_and_respond_inner(message, text, show_typing, state)
async def _classify_and_respond_inner(
message: Message,
text: str,
show_typing: bool = True,
state: FSMContext | None = None,
) -> None:
if message.from_user is None:
await message.answer("Не удалось определить пользователя.")
@@ -660,17 +681,18 @@ async def _finish_media_group(key: tuple[int, str]) -> None:
async def _handle_image_message(message: Message, bot: Bot) -> None:
await notify_development_deploy(bot)
try:
if message.media_group_id:
messages = await _collect_media_group(message, bot)
if messages[0].message_id != message.message_id:
return
else:
messages = [message]
async with typing_status(bot, message.chat.id):
await _process_image_group(messages, bot)
except Exception as exc:
await report_error(message, "чтение скриншота или обновление ToV", exc)
with diagnostic_operation():
try:
if message.media_group_id:
messages = await _collect_media_group(message, bot)
if messages[0].message_id != message.message_id:
return
else:
messages = [message]
async with typing_status(bot, message.chat.id):
await _process_image_group(messages, bot)
except Exception as exc:
await report_error(message, "чтение скриншота или обновление ToV", exc)
@router.message(F.photo | (F.document & F.document.mime_type.startswith("image/")))
@@ -712,11 +734,12 @@ async def transcribe_voice(message: Message, bot: Bot, state: FSMContext) -> Non
if not await require_mode(message, state, "generation"):
return
await notify_development_deploy(bot)
try:
async with typing_status(bot, message.chat.id):
await _process_voice(message, bot, state)
except Exception as exc:
await report_error(message, "обработка голосового и генерация", exc)
with diagnostic_operation():
try:
async with typing_status(bot, message.chat.id):
await _process_voice(message, bot, state)
except Exception as exc:
await report_error(message, "обработка голосового и генерация", exc)
@router.message(F.text & F.text.startswith("/"))
+12
View File
@@ -59,6 +59,18 @@ class DeployNoticeTests(unittest.IsolatedAsyncioTestCase):
class UserFacingErrorTests(unittest.IsolatedAsyncioTestCase):
async def test_failure_log_has_traceback_and_operation_id(self):
message = AsyncMock()
with bot.diagnostic_operation() as operation_id:
with self.assertLogs(bot.logger, level="ERROR") as captured:
try:
raise ValueError("provider failure")
except ValueError as exc:
await bot.report_error(message, "генерация поста", exc)
self.assertIn(f"operation_id={operation_id}", "\n".join(captured.output))
self.assertIn("Traceback (most recent call last)", "\n".join(captured.output))
self.assertIn("provider failure", "\n".join(captured.output))
async def test_production_error_is_generic_without_exception_details(self):
message = AsyncMock()
with patch.object(bot, "DEPLOY_ENV", "production"):
+4
View File
@@ -1,3 +1,4 @@
import json
import unittest
from unittest.mock import patch
@@ -20,6 +21,9 @@ class FakeResponse:
async def json(self, content_type=None):
return self.payload
async def text(self):
return json.dumps(self.payload)
class FakeClientSession:
def __init__(self, response, **kwargs):
+81 -8
View File
@@ -42,7 +42,9 @@ class FakeResponse:
async def text(self):
if isinstance(self.payload, bytes):
return self.payload.decode("utf-8", errors="replace")
return str(self.payload)
if isinstance(self.payload, str):
return self.payload
return json.dumps(self.payload)
@@ -171,13 +173,14 @@ class ApinexTests(unittest.IsolatedAsyncioTestCase):
self.assertEqual(detail["stage"], "screenshot_ocr")
async def test_chat_api_failure_is_logged(self):
fake_session = FakeClientSession(FakeResponse(401, {}))
with patch.object(apinex_client.aiohttp, "ClientSession", return_value=fake_session):
with self.assertRaisesRegex(RuntimeError, "HTTP 401"):
await apinex_client.chat_text(
"APINEX_POST_WRITER_MODEL", "private system prompt", "request",
stage="post_generation", user_id=45, prompt_name="post_writer",
)
fake_session = FakeClientSession(FakeResponse(401, {"error": {"message": "invalid key"}}))
with self.assertLogs(apinex_client.logger, level="ERROR") as captured:
with patch.object(apinex_client.aiohttp, "ClientSession", return_value=fake_session):
with self.assertRaisesRegex(RuntimeError, "HTTP 401"):
await apinex_client.chat_text(
"APINEX_POST_WRITER_MODEL", "private system prompt", "request",
stage="post_generation", user_id=45, prompt_name="post_writer",
)
logs = user_store.list_api_call_logs(limit=1, user_id=45)
detail = user_store.get_api_call_log(logs[0]["id"] if isinstance(logs[0]["id"], int) else -1)
self.assertIsNotNone(detail)
@@ -185,6 +188,50 @@ class ApinexTests(unittest.IsolatedAsyncioTestCase):
self.assertEqual(detail["status_code"], 401)
self.assertNotIn("private system prompt", str(detail["request_json"]))
self.assertEqual(json.loads(str(detail["request_json"]))["prompt"]["name"], "post_writer")
self.assertEqual(
json.loads(str(detail["response_json"])),
{"provider_error_body": {"error": {"message": "invalid key"}}},
)
self.assertIsInstance(detail["duration_ms"], int)
self.assertIn('"provider_error_body"', "\n".join(captured.output))
async def test_empty_model_response_logs_raw_response_and_operation_id(self):
response = {"choices": [{"message": {"content": ""}, "finish_reason": "stop"}]}
fake_session = FakeClientSession(FakeResponse(200, response))
with (
patch.object(apinex_client.aiohttp, "ClientSession", return_value=fake_session),
apinex_client.diagnostic_operation() as operation_id,
):
with self.assertRaisesRegex(RuntimeError, "empty model response"):
await apinex_client.chat_text(
"APINEX_POST_WRITER_MODEL", "system", "request", stage="intent_classification", user_id=46
)
logs = user_store.list_api_call_logs(limit=1, user_id=46)
detail = user_store.get_api_call_log(logs[0]["id"] if isinstance(logs[0]["id"], int) else -1)
self.assertIsNotNone(detail)
assert detail is not None
self.assertEqual(detail["operation_id"], operation_id)
self.assertEqual(json.loads(str(detail["response_json"])), response)
self.assertEqual(detail["status_code"], 200)
self.assertIsInstance(detail["duration_ms"], int)
async def test_api_log_redacts_secrets_but_keeps_token_limits(self):
await apinex_client._record_api_call(
stage="test", model="model", prompt_name="test", system_prompt="",
request={"max_tokens": 1800, "api_key": "private-key"},
response={"authorization": "Bearer private-token", "detail": "Bearer private-token"},
)
log = user_store.list_api_call_logs(limit=1)[0]
log_id = log["id"]
self.assertIsInstance(log_id, int)
detail = user_store.get_api_call_log(log_id if isinstance(log_id, int) else -1)
assert detail is not None
request = json.loads(str(detail["request_json"]))
response = json.loads(str(detail["response_json"]))
self.assertEqual(request["max_tokens"], 1800)
self.assertEqual(request["api_key"], "[REDACTED]")
self.assertEqual(response["authorization"], "[REDACTED]")
self.assertEqual(response["detail"], "Bearer [REDACTED]")
async def test_writer_preserves_multiline_post_without_labels(self):
multiline_post = "Привет!\n\nВот новость ☕"
@@ -371,6 +418,28 @@ class ScreenshotToneWorkflowTests(unittest.IsolatedAsyncioTestCase):
class StorageTests(unittest.TestCase):
def test_api_log_schema_migrates_existing_database(self):
import sqlite3
import tempfile
from pathlib import Path
with tempfile.TemporaryDirectory() as temp_dir:
db = str(Path(temp_dir) / "old.sqlite3")
with sqlite3.connect(db) as connection:
connection.execute(
"""CREATE TABLE api_call_logs (
id INTEGER PRIMARY KEY AUTOINCREMENT, telegram_user_id INTEGER, stage TEXT NOT NULL,
model TEXT NOT NULL DEFAULT '', prompt_sha256 TEXT NOT NULL DEFAULT '', request_json TEXT NOT NULL,
response_json TEXT NOT NULL DEFAULT '', status_code INTEGER, error TEXT NOT NULL DEFAULT '',
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
)"""
)
user_store.init_db(db)
with sqlite3.connect(db) as connection:
columns = {row[1] for row in connection.execute("PRAGMA table_info(api_call_logs)")}
self.assertIn("operation_id", columns)
self.assertIn("duration_ms", columns)
def test_user_tone_isolated_per_id(self):
import tempfile
from pathlib import Path
@@ -400,11 +469,15 @@ class StorageTests(unittest.TestCase):
user_id=12,
status_code=502,
error="HTTP 502",
operation_id="op-123",
duration_ms=250,
db_path=db,
)
summary = user_store.list_api_call_logs(limit=5, user_id=12, db_path=db)
self.assertEqual(summary[0]["id"], log_id)
self.assertNotIn("request_json", summary[0])
self.assertEqual(summary[0]["operation_id"], "op-123")
self.assertEqual(summary[0]["duration_ms"], 250)
detail = user_store.get_api_call_log(log_id, db)
self.assertIsNotNone(detail)
assert detail is not None
+14 -5
View File
@@ -83,6 +83,13 @@ def init_db(db_path: str | None = None) -> None:
)
"""
)
api_log_columns = {
row["name"] for row in connection.execute("PRAGMA table_info(api_call_logs)").fetchall()
}
if "operation_id" not in api_log_columns:
connection.execute("ALTER TABLE api_call_logs ADD COLUMN operation_id TEXT NOT NULL DEFAULT ''")
if "duration_ms" not in api_log_columns:
connection.execute("ALTER TABLE api_call_logs ADD COLUMN duration_ms INTEGER")
connection.execute(
"CREATE INDEX IF NOT EXISTS idx_api_call_logs_created_at ON api_call_logs(created_at)"
)
@@ -237,6 +244,8 @@ def save_api_call_log(
user_id: int | None = None,
status_code: int | None = None,
error: str = "",
operation_id: str | None = None,
duration_ms: int | None = None,
db_path: str | None = None,
) -> int:
request_json = _bounded_json(request_json)
@@ -246,11 +255,11 @@ def save_api_call_log(
"""
INSERT INTO api_call_logs (
telegram_user_id, stage, model, prompt_sha256, request_json,
response_json, status_code, error
) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
response_json, status_code, error, operation_id, duration_ms
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(user_id, stage[:100], model[:200], prompt_sha256[:64], request_json,
response_json, status_code, error[:2000]),
response_json, status_code, error[:2000], (operation_id or "")[:32], duration_ms),
)
connection.execute(
"DELETE FROM api_call_logs WHERE created_at < datetime('now', ?)",
@@ -268,13 +277,13 @@ def list_api_call_logs(
with closing(connect(db_path)) as connection:
if user_id is None:
rows = connection.execute(
"SELECT id, telegram_user_id, stage, model, prompt_sha256, status_code, error, created_at, "
"SELECT id, telegram_user_id, stage, model, prompt_sha256, status_code, error, operation_id, duration_ms, created_at, "
"length(request_json) AS request_chars, length(response_json) AS response_chars "
"FROM api_call_logs ORDER BY id DESC LIMIT ?", (limit,),
).fetchall()
else:
rows = connection.execute(
"SELECT id, telegram_user_id, stage, model, prompt_sha256, status_code, error, created_at, "
"SELECT id, telegram_user_id, stage, model, prompt_sha256, status_code, error, operation_id, duration_ms, created_at, "
"length(request_json) AS request_chars, length(response_json) AS response_chars "
"FROM api_call_logs WHERE telegram_user_id = ? ORDER BY id DESC LIMIT ?", (user_id, limit),
).fetchall()