From 9cfccc7b4a2224aa4589d15c480932d7c7e92caa Mon Sep 17 00:00:00 2001 From: Anton Vakhrushev Date: Sun, 28 Jun 2026 20:13:40 +0300 Subject: [PATCH] =?UTF-8?q?=D0=9B=D0=BE=D0=B3=D0=B8=D1=80=D0=BE=D0=B2?= =?UTF-8?q?=D0=B0=D0=BD=D0=B8=D0=B5:=20=D1=80=D0=B5=D0=B2=D1=8C=D1=8E=20?= =?UTF-8?q?=D0=B2=D1=81=D0=B5=D0=B3=D0=BE=20=D0=BA=D0=BE=D0=B4=D0=B0=20?= =?UTF-8?q?=D0=B8=20=D1=80=D0=B5=D1=84=D0=B0=D0=BA=D1=82=D0=BE=D1=80=D0=B8?= =?UTF-8?q?=D0=BD=D0=B3=20=D0=B2=20=D1=81=D0=BE=D0=BE=D1=82=D0=B2=D0=B5?= =?UTF-8?q?=D1=82=D1=81=D1=82=D0=B2=D0=B8=D0=B8=20=D1=81=20=D0=BA=D0=BE?= =?UTF-8?q?=D0=BD=D0=B2=D0=B5=D0=BD=D1=86=D0=B8=D1=8F=D0=BC=D0=B8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cmd/jellybit/main.go | 12 ++++- cmd/jellybit/serve.go | 2 +- docs/conventions/logging.md | 50 ++++++++++++++++---- internal/httpapi/httpapi.go | 21 ++++++--- internal/httpapi/review.go | 14 +++--- internal/ingest/ingest.go | 27 +++++++---- internal/jellyfin/jellyfin.go | 14 ++++-- internal/layout/layout.go | 26 ++++++----- internal/llm/openai.go | 37 ++++++++------- internal/logctx/logctx.go | 39 ++++++++++++++++ internal/logging/ext.go | 65 ++++++++++++++++++++++++++ internal/logging/logging.go | 22 ++++++++- internal/metadata/http.go | 51 ++++++++++---------- internal/metadata/tmdb.go | 9 ++-- internal/metadata/tvdb.go | 37 ++++++++------- internal/metadata/tvmaze.go | 8 ++-- internal/naming/prompt.go | 10 ++-- internal/qbt/qbt.go | 60 ++++++++++++++++++------ internal/recognize/metadata.go | 7 ++- internal/recognize/recognize.go | 10 ++-- internal/tgbot/bot.go | 12 ++--- internal/worker/discover.go | 8 ++-- internal/worker/review.go | 83 +++++++++++++++++++-------------- internal/worker/worker.go | 52 +++++++++++++++------ 24 files changed, 473 insertions(+), 203 deletions(-) create mode 100644 internal/logctx/logctx.go create mode 100644 internal/logging/ext.go diff --git a/cmd/jellybit/main.go b/cmd/jellybit/main.go index 57da01a..7a92c53 100644 --- a/cmd/jellybit/main.go +++ b/cmd/jellybit/main.go @@ -11,6 +11,8 @@ package main import ( "os" "strings" + + "git.vakhrushev.me/av/jellybit/internal/logging" ) func main() { @@ -38,7 +40,15 @@ func main() { os.Exit(2) } if err != nil { - _, _ = os.Stderr.WriteString("fatal: " + err.Error() + "\n") + if cmd == "serve" { + // Фатальный сбой старта сервиса — структурный лог ERROR (как в проде), + // затем ненулевой код возврата. + logging.NewStderr().Error("fatal startup", "command", cmd, "error", err) + } else { + // Диагностические CLI (add/recognize/healthcheck) — человекочитаемый + // stderr, это пользовательский вывод, а не лог сервиса. + _, _ = os.Stderr.WriteString("fatal: " + err.Error() + "\n") + } os.Exit(1) } } diff --git a/cmd/jellybit/serve.go b/cmd/jellybit/serve.go index 09c98ae..804d1f6 100644 --- a/cmd/jellybit/serve.go +++ b/cmd/jellybit/serve.go @@ -174,7 +174,7 @@ func runServe(args []string) error { } api, terr := tgbotapi.NewBotAPIWithClient(cfg.Telegram.Token, tgbotapi.APIEndpoint, tgClient) if terr != nil { - logger.Error("telegram bot disabled: cannot connect", "err", terr) + logger.Error("telegram bot disabled, cannot connect", "error", terr) } else { bot := tgbot.New(api, ingestor, wrk, tgbot.Config{ AllowedUserIDs: cfg.Telegram.AllowedUserIDs, diff --git a/docs/conventions/logging.md b/docs/conventions/logging.md index 2971bb4..988e1e6 100644 --- a/docs/conventions/logging.md +++ b/docs/conventions/logging.md @@ -36,6 +36,10 @@ log.Info("download accepted", "download_id", id, "media_type", "movie") log.Info(fmt.Sprintf("download %s accepted as movie", id)) ``` +- `msg` — чистая категория без неймспейс-префикса: `recognition done`, а не + `recognize: done`. Подсистему выносим в поле `capability` + (`ingest`/`recognition`/`file-layout`/`review`), не в текст. + ## Уровни Принцип: уровень — это **адресат** («кому сообщение»), а не «насколько @@ -43,8 +47,8 @@ log.Info(fmt.Sprintf("download %s accepted as movie", id)) | Уровень | Кому и когда | Примеры в jellybit | |---|---|---| -| `DEBUG` | разработчику при отладке; в проде выключен | healthcheck-эндпоинты, тела запросов/ответов внешних API, промежуточные шаги распознавания | -| `INFO` | команде, аудит постфактум | приём загрузки, распознан фильм/сериал, раскладка выполнена, старт процессов, **каждый вызов внешнего сервиса** (старт/успех) | +| `DEBUG` | разработчику при отладке; в проде выключен | healthcheck-эндпоинты, поллинг статуса в qBittorrent, авто-рефреш UI, тела запросов/ответов внешних API, промежуточные шаги распознавания | +| `INFO` | команде, аудит постфактум | приём загрузки, распознан фильм/сериал, раскладка выполнена, старт процессов, **событийный вызов внешнего сервиса** (по реальному действию) | | `WARN` | команде, «может стать проблемой» | retry внешнего вызова, низкая уверенность распознавания (ушло в ревью), приближение к лимиту | | `ERROR` | команде, в техдолг / разбор | внешний сервис недоступен после ретраев, операция загрузки не выполнена, необработанная ошибка | @@ -56,6 +60,13 @@ log.Info(fmt.Sprintf("download %s accepted as movie", id)) не «может» — это `INFO`. - Меняется адресат — меняется уровень. Невалидный ввод от пользователя — это `DEBUG` (норма, команде разбирать нечего), а не `ERROR`. +- **Событийное → INFO, рутинно-частое → DEBUG.** Операция, срабатывающая по + реальному действию/изменению (приём загрузки, добавление торрента, вызов + LLM, раскладка), идёт на `INFO`. Повторяющаяся служебная операция, + которую запускает таймер/поллинг и которая сама по себе не несёт события + (healthcheck, поллинг статуса в qBittorrent, авто-рефреш UI), — на + `DEBUG`: на `INFO` она зашумляет аудит. Такие записи смотрят редко, при + предметной отладке (DEBUG включают точечно). - `slog` не разделяет CRITICAL/FATAL — фатальный сбой на старте логируем `ERROR` и завершаем процесс (ненулевой код возврата). @@ -120,11 +131,20 @@ log.Error(err.Error()) Правила: -- Ошибку передаём полем `"error", err` — не склеиваем в `msg`. -- В коде оборачиваем с контекстом (`fmt.Errorf("…: %w", err)`); логируем - развёрнутую ошибку один раз — в точке, где решено «дальше не пробрасываем». -- **Не** логировать одну ошибку дважды по цепочке: либо логируешь и гасишь, - либо оборачиваешь и пробрасываешь — не оба сразу. +- Ошибку передаём полем `"error", err` — не склеиваем в `msg`. Ключ — + `error` (как по умолчанию в zap/zerolog; единый ключ важнее краткости). +- Идиома Go — **либо лог, либо возврат, не оба**. Промежуточные слои только + оборачивают и возвращают (`fmt.Errorf("…: %w", err)`), не логируя — + контекст накапливается в цепочке `%w`. +- Логируем ошибку **один раз — на границе доменного слоя** (use-case + `Ingest`, стадии воркера), которая определяет исход операции: полем + `error`, уровень `ERROR`. В Go логирует этот единый чокпоинт, а не каждый + транспорт — так транспорты остаются тонкими. +- Транспорты (HTTP/web/Telegram) переводят возвращённую ошибку в свой ответ + (статус, сообщение пользователю) и **не логируют** её повторно — иначе + один сбой даёт дубли. +- Телеметрия внешнего вызова (`ext.*`, см. ниже) — отдельная запись о + поведении зависимости, не дубль доменной ошибки. - Глушить ошибку без лога — только с однострочным комментарием «почему». ## Внешние сервисы (обязательно логируем все вызовы) @@ -141,9 +161,15 @@ log.Error(err.Error()) Уровни вызова: -- `INFO` — старт и успешный результат (трафик низкий, шум допустим); +- `INFO` — успешный **событийный** вызов (по реальному действию: добавление + торрента, вызов LLM, рефреш Jellyfin, поиск в метабазе); +- `DEBUG` — успешный **рутинно-частый** вызов (поллинг статуса + `torrents/info`/`torrents/files`, авто-рефреш) — см. правило «событийное → + INFO, рутинно-частое → DEBUG» в разделе «Уровни»; - `WARN` — попытка не удалась, делаем retry; -- `ERROR` — ретраи исчерпаны / сервис недоступен. +- `ERROR` — ретраи исчерпаны / сервис недоступен (сетевой сбой/таймаут). + Завершённый HTTP-ответ с 4xx — это успех на транспортном уровне (`Success` + с `ext.status_code`); решение «это ошибка» принимает доменный вызывающий. Тело запроса/ответа — только на `DEBUG` и **после** вычистки секретов (см. «Безопасность»). @@ -151,7 +177,11 @@ log.Error(err.Error()) ## HTTP и healthcheck - Входящие HTTP-запросы логируем с полями `http.method`, `http.route`, - `http.status_code`, `duration_ms`. + `http.status_code`, `duration_ms`, `transport` (`http`/`web`/`telegram`). +- Для корреляции HTTP-запроса допустим `request_id` (напр. chi `RequestID`) — + это отдельный слой от корреляции загрузки по `download_id` и не противоречит + отказу от `trace_id`. Если запрос порождает загрузку — связь даёт + `download_id` в её записях. - **Эндпоинты healthcheck/liveness/readiness логируем на `DEBUG`** — их дёргают периодически, на `INFO` они забивают аудит шумом. В проде (базовый уровень `INFO`) они не пишутся. diff --git a/internal/httpapi/httpapi.go b/internal/httpapi/httpapi.go index 3679ff7..1b661c1 100644 --- a/internal/httpapi/httpapi.go +++ b/internal/httpapi/httpapi.go @@ -135,7 +135,7 @@ type downloadView struct { func (s *server) handleIndex(w http.ResponseWriter, r *http.Request) { downloads, err := s.deps.Reader.ListDownloads(r.Context()) if err != nil { - s.deps.Logger.Error("list downloads", "err", err) + s.deps.Logger.Error("list downloads", "error", err) http.Error(w, "internal error", http.StatusInternalServerError) return } @@ -145,7 +145,7 @@ func (s *server) handleIndex(w http.ResponseWriter, r *http.Request) { } w.Header().Set("Content-Type", "text/html; charset=utf-8") if err := s.index.Execute(w, view); err != nil { - s.deps.Logger.Error("render index", "err", err) + s.deps.Logger.Error("render index", "error", err) } } @@ -269,7 +269,9 @@ func (s *server) apiCommand(w http.ResponseWriter, r *http.Request, cmd func(con return } if err := cmd(r.Context(), id); err != nil { - s.deps.Logger.Warn("api command failed", "path", r.URL.Path, "id", id, "err", err) + // Тонкий транспорт: возвращённую use-case'ом/воркером ошибку переводим в + // HTTP-статус и не логируем повторно (доменный слой уже залогировал, а + // невалидный ввод — это норма, разбирать команде нечего). writeJSON(w, http.StatusConflict, errJSON(err)) return } @@ -352,10 +354,17 @@ func requestLogger(logger *slog.Logger) func(http.Handler) http.Handler { next.ServeHTTP(ww, r) + // http.route — низкокардинальный шаблон маршрута (chi), а не + // конкретный путь; при отсутствии шаблона падаем на путь. + route := chi.RouteContext(r.Context()).RoutePattern() + if route == "" { + route = r.URL.Path + } logger.Log(r.Context(), requestLogLevel(r), "http request", - "method", r.Method, - "path", r.URL.Path, - "status", ww.Status(), + "transport", "http", + "http.method", r.Method, + "http.route", route, + "http.status_code", ww.Status(), "bytes", ww.BytesWritten(), "duration_ms", time.Since(start).Milliseconds(), "request_id", middleware.GetReqID(r.Context()), diff --git a/internal/httpapi/review.go b/internal/httpapi/review.go index 11ff477..85af8a0 100644 --- a/internal/httpapi/review.go +++ b/internal/httpapi/review.go @@ -82,7 +82,7 @@ func (s *server) handleReview(w http.ResponseWriter, r *http.Request) { http.Error(w, "задача не найдена", http.StatusNotFound) return } - s.deps.Logger.Error("review data", "id", id, "err", err) + s.deps.Logger.Error("review data", "id", id, "error", err) http.Error(w, "internal error", http.StatusInternalServerError) return } @@ -140,7 +140,7 @@ func (s *server) handleReview(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "text/html; charset=utf-8") if err := s.review.Execute(w, view); err != nil { - s.deps.Logger.Error("render review", "err", err) + s.deps.Logger.Error("render review", "error", err) } } @@ -153,7 +153,8 @@ func (s *server) handleApply(w http.ResponseWriter, r *http.Request) { return } if err := s.deps.Reviewer.Apply(r.Context(), id); err != nil { - s.deps.Logger.Warn("review action failed", "action", "apply", "id", id, "err", err) + // Тонкий транспорт: ошибку воркера переводим в ответ, не логируя + // повторно (доменный слой уже залогировал реальный сбой). redirectReview(w, r, id, err.Error()) return } @@ -220,7 +221,6 @@ func (s *server) handleDefer(w http.ResponseWriter, r *http.Request) { return } if err := s.deps.Reviewer.Defer(r.Context(), id); err != nil { - s.deps.Logger.Warn("review action failed", "action", "defer", "id", id, "err", err) redirectReview(w, r, id, err.Error()) return } @@ -234,7 +234,6 @@ func (s *server) handleUndo(w http.ResponseWriter, r *http.Request) { return } if err := s.deps.Reviewer.Undo(r.Context(), id); err != nil { - s.deps.Logger.Warn("review action failed", "action", "undo", "id", id, "err", err) redirectErr(w, r, err.Error()) return } @@ -250,7 +249,6 @@ func (s *server) handleRelink(w http.ResponseWriter, r *http.Request) { return } if err := s.deps.Reviewer.Relink(r.Context(), id); err != nil { - s.deps.Logger.Warn("review action failed", "action", "relink", "id", id, "err", err) redirectErr(w, r, err.Error()) return } @@ -266,8 +264,8 @@ func (s *server) reviewAction(w http.ResponseWriter, r *http.Request, fn func(co return } if err := fn(r.Context(), id); err != nil { - s.deps.Logger.Warn("review action failed", - "action", r.URL.Path, "id", id, "err", err) + // Тонкий транспорт: ошибку переводим в ?err= на странице ревью, не + // логируя повторно (доменный слой/валидация — не дело транспорта). redirectReview(w, r, id, err.Error()) return } diff --git a/internal/ingest/ingest.go b/internal/ingest/ingest.go index 6ab3e4f..cffe13a 100644 --- a/internal/ingest/ingest.go +++ b/internal/ingest/ingest.go @@ -9,11 +9,15 @@ import ( "log/slog" "strings" + "git.vakhrushev.me/av/jellybit/internal/logctx" "git.vakhrushev.me/av/jellybit/internal/magnet" "git.vakhrushev.me/av/jellybit/internal/qbt" "git.vakhrushev.me/av/jellybit/internal/store" ) +// capIngest — стадия приёма для поля capability в логах. +const capIngest = "ingest" + // Store — нужная ingest часть хранилища. type Store interface { FindActiveByInfohash(ctx context.Context, infohash string) (*store.Download, error) @@ -77,11 +81,16 @@ func (s *Service) Ingest(ctx context.Context, req Request) (Result, error) { return Result{}, fmt.Errorf("ingest: %w", err) } + // Scoped-логгер стадии приёма: download_id допишется после CreateDownload. + // Кладём в ctx, чтобы внешние клиенты (qBittorrent, LLM-namer) дописывали + // ключи корреляции к своим ext.*-записям сами. + log := s.log.With("capability", capIngest, "infohash", info.Infohash) + ctx = logctx.With(ctx, log) + if existing, err := s.store.FindActiveByInfohash(ctx, info.Infohash); err != nil { return Result{}, fmt.Errorf("ingest: lookup active: %w", err) } else if existing != nil { - s.log.Info("ingest: attached to active download", - "download_id", existing.ID, "infohash", info.Infohash, "state", existing.State) + log.Info("download attached to active", "download_id", existing.ID, "state", existing.State) return Result{ DownloadID: existing.ID, Infohash: info.Infohash, @@ -111,6 +120,8 @@ func (s *Service) Ingest(ctx context.Context, req Request) (Result, error) { if err != nil { return Result{}, fmt.Errorf("ingest: create download: %w", err) } + log = log.With("download_id", id) + ctx = logctx.With(ctx, log) addErr := s.qbt.Add(ctx, qbt.AddRequest{ URLs: []string{source}, @@ -119,19 +130,19 @@ func (s *Service) Ingest(ctx context.Context, req Request) (Result, error) { Rename: rename, }) if addErr != nil { - s.log.Warn("ingest: qbittorrent add failed, marking download failed", - "download_id", id, "infohash", info.Infohash, "err", addErr) + // Граница доменной операции приёма: логируем исход один раз (ERROR). + // Поведение самого вызова qBittorrent уже залогировал клиент (ext.*) — + // это разные факты, не дубль. + log.Error("download accept failed", "error", addErr) // Задача уже в БД — помечаем failed, чтобы worker её не подхватил. if setErr := s.store.SetDownloadState(ctx, id, store.StateFailed, "qbit_add", addErr.Error()); setErr != nil { - s.log.Error("ingest: failed to mark download failed after qbit error", - "download_id", id, "err", setErr) + log.Error("mark download failed after qbit error failed", "error", setErr) } return Result{DownloadID: id, Infohash: info.Infohash, State: store.StateFailed}, fmt.Errorf("ingest: add to qbittorrent: %w", addErr) } - s.log.Info("ingest: download accepted", - "download_id", id, "infohash", info.Infohash, "category", s.cfg.Category) + log.Info("download accepted", "category", s.cfg.Category) return Result{ DownloadID: id, Infohash: info.Infohash, diff --git a/internal/jellyfin/jellyfin.go b/internal/jellyfin/jellyfin.go index 54e9c0d..2dd951e 100644 --- a/internal/jellyfin/jellyfin.go +++ b/internal/jellyfin/jellyfin.go @@ -13,6 +13,9 @@ import ( "net/url" "strings" "time" + + "git.vakhrushev.me/av/jellybit/internal/logctx" + "git.vakhrushev.me/av/jellybit/internal/logging" ) const defaultTimeout = 10 * time.Second @@ -77,17 +80,22 @@ func (c *Client) RefreshLibraries(ctx context.Context) error { } req.Header.Set("X-Emby-Token", c.apiKey) - start := time.Now() + log := logctx.FromOr(ctx, c.log) + call := logging.ExtCall{Service: logging.ServiceJellyfin, Operation: "library/refresh", Start: time.Now()} resp, err := c.hc.Do(req) if err != nil { + call.Failure(log, err) return fmt.Errorf("jellyfin: refresh: %w", err) } defer func() { _ = resp.Body.Close() }() + call.Status = resp.StatusCode body, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<10)) if resp.StatusCode < 200 || resp.StatusCode >= 300 { - return fmt.Errorf("jellyfin: refresh: status %d body %q", + err := fmt.Errorf("jellyfin: refresh: status %d body %q", resp.StatusCode, strings.TrimSpace(string(body))) + call.Failure(log, err) + return err } - c.log.Info("jellyfin: library refresh triggered", "duration", time.Since(start)) + call.Success(log) return nil } diff --git a/internal/layout/layout.go b/internal/layout/layout.go index eb81580..49267ea 100644 --- a/internal/layout/layout.go +++ b/internal/layout/layout.go @@ -21,6 +21,8 @@ import ( "os" "path/filepath" "syscall" + + "git.vakhrushev.me/av/jellybit/internal/logctx" ) // MediaType — вид контента. @@ -230,7 +232,8 @@ var ErrCollision = errors.New("layout: target collision") // доводит начатое. При коллизии (цель занята чужим файлом) возвращает // ErrCollision, не перезаписывая. Если хардлинк невозможен (разные ФС или ФС // не поддерживает link) — фолбэк на копирование файла с предупреждением в лог. -func (l *Layouter) Apply(_ context.Context, links []Link) ([]Result, error) { +func (l *Layouter) Apply(ctx context.Context, links []Link) ([]Result, error) { + log := logctx.FromOr(ctx, l.log) results := make([]Result, 0, len(links)) for _, ln := range links { root := l.movies @@ -244,13 +247,13 @@ func (l *Layouter) Apply(_ context.Context, links []Link) ([]Result, error) { return results, fmt.Errorf("layout: mkdir %q: %w", filepath.Dir(ln.Dst), err) } - status, err := l.linkOne(ln.Src, ln.Dst) + status, err := l.linkOne(log, ln.Src, ln.Dst) if err != nil { - l.log.Error("layout: link failed", - "src", ln.Src, "dst", ln.Dst, "kind", ln.Kind, "err", err) + log.Error("layout link failed", + "src", ln.Src, "dst", ln.Dst, "kind", ln.Kind, "error", err) return results, err } - l.log.Debug("layout: link applied", + log.Debug("layout link applied", "src", ln.Src, "dst", ln.Dst, "kind", ln.Kind, "status", status) results = append(results, Result{Link: ln, Status: status}) } @@ -259,7 +262,7 @@ func (l *Layouter) Apply(_ context.Context, links []Link) ([]Result, error) { // linkOne создаёт одну ссылку, разбирая «уже существует» и невозможность // хардлинка (фолбэк на копирование). -func (l *Layouter) linkOne(src, dst string) (LinkStatus, error) { +func (l *Layouter) linkOne(log *slog.Logger, src, dst string) (LinkStatus, error) { err := os.Link(src, dst) if err == nil { return StatusLinked, nil @@ -279,8 +282,8 @@ func (l *Layouter) linkOne(src, dst string) (LinkStatus, error) { // раскладку — копируем файл и предупреждаем: диск дублируется, но // задача доходит до конца. dst здесь заведомо отсутствует (иначе был бы // fs.ErrExist выше). - l.log.Warn("layout: hardlink unsupported, falling back to file copy", - "src", src, "dst", dst, "err", err) + log.Warn("layout hardlink unsupported, file copy fallback", + "src", src, "dst", dst, "error", err) if cerr := copyFile(src, dst); cerr != nil { return "", fmt.Errorf("layout: copy fallback %q → %q: %w", src, dst, cerr) } @@ -364,7 +367,8 @@ func sameFile(src, dst string) (bool, error) { // Undo удаляет ссылки и подчищает опустевшие каталоги. Снимает только пути // строго под библиотеками (источник недосягаем). Отсутствующая цель — не // ошибка (идемпотентно). Возвращает число удалённых ссылок. -func (l *Layouter) Undo(_ context.Context, links []Link) (int, error) { +func (l *Layouter) Undo(ctx context.Context, links []Link) (int, error) { + log := logctx.FromOr(ctx, l.log) removed := 0 for _, ln := range links { root := l.movies @@ -378,11 +382,11 @@ func (l *Layouter) Undo(_ context.Context, links []Link) (int, error) { if errors.Is(err, fs.ErrNotExist) { continue } - l.log.Error("layout: undo remove failed", "dst", ln.Dst, "err", err) + log.Error("layout undo remove failed", "dst", ln.Dst, "error", err) return removed, fmt.Errorf("layout: undo remove %q: %w", ln.Dst, err) } removed++ - l.log.Debug("layout: link removed", "dst", ln.Dst) + log.Debug("layout link removed", "dst", ln.Dst) pruneEmptyDirs(filepath.Dir(ln.Dst), root) } return removed, nil diff --git a/internal/llm/openai.go b/internal/llm/openai.go index 1099ba5..dd2bb49 100644 --- a/internal/llm/openai.go +++ b/internal/llm/openai.go @@ -11,6 +11,9 @@ import ( "net/url" "strings" "time" + + "git.vakhrushev.me/av/jellybit/internal/logctx" + "git.vakhrushev.me/av/jellybit/internal/logging" ) const ( @@ -117,7 +120,7 @@ func (c *openAICompat) Complete(ctx context.Context, req Request) (Response, err return Response{}, fmt.Errorf("llm: marshal request: %w", err) } - var lastErr error + log := logctx.FromOr(ctx, c.log) for attempt := 1; attempt <= maxAttempts; attempt++ { if attempt > 1 { if err := c.wait(ctx, attempt); err != nil { @@ -125,31 +128,31 @@ func (c *openAICompat) Complete(ctx context.Context, req Request) (Response, err } } - c.log.Debug("llm: request", - "endpoint", c.endpoint, "model", c.model, - "attempt", attempt, "max_attempts", maxAttempts) - start := time.Now() + call := logging.ExtCall{ + Service: logging.ServiceLLM, + Operation: "chat.completions", + Start: time.Now(), + Attempt: attempt, + } resp, retryable, err := c.do(ctx, body) if err == nil { - c.log.Debug("llm: response ok", - "model", resp.Model, "attempt", attempt, - "duration", time.Since(start), + call.Success(log, "model", resp.Model, "total_tokens", resp.Usage.TotalTokens, "cost", resp.Usage.Cost) return resp, nil } - lastErr = err if !retryable { - c.log.Error("llm: request failed (non-retryable)", - "model", c.model, "attempt", attempt, "duration", time.Since(start), "err", err) + call.Failure(log, err, "model", c.model) return Response{}, err } - c.log.Warn("llm: request failed, will retry", - "model", c.model, "attempt", attempt, "max_attempts", maxAttempts, - "duration", time.Since(start), "err", err) + if attempt < maxAttempts { + call.Retry(log, err, "model", c.model) + continue + } + // Последняя попытка тоже неуспешна — ретраи исчерпаны. + call.Failure(log, err, "model", c.model) + return Response{}, fmt.Errorf("llm: exhausted %d attempts: %w", maxAttempts, err) } - c.log.Error("llm: all attempts exhausted", - "model", c.model, "max_attempts", maxAttempts, "err", lastErr) - return Response{}, fmt.Errorf("llm: exhausted %d attempts: %w", maxAttempts, lastErr) + return Response{}, fmt.Errorf("llm: exhausted %d attempts", maxAttempts) } func (c *openAICompat) buildRequest(req Request) chatRequest { diff --git a/internal/logctx/logctx.go b/internal/logctx/logctx.go new file mode 100644 index 0000000..15355f6 --- /dev/null +++ b/internal/logctx/logctx.go @@ -0,0 +1,39 @@ +// Package logctx переносит scoped-логдер загрузки через context.Context. +// +// На приёме/реконсиляции загрузки заводим логгер с её ключами +// (download_id [+ infohash], capability) и кладём в ctx; стадии (скачивание → +// распознавание → раскладка → ревью) достают его через From и пишут с уже +// дописанными ключами — без ручного доклеивания download_id в каждый вызов. +package logctx + +import ( + "context" + "log/slog" +) + +type ctxKey struct{} + +// With возвращает ctx со вложенным логгером. +func With(ctx context.Context, log *slog.Logger) context.Context { + return context.WithValue(ctx, ctxKey{}, log) +} + +// From достаёт логгер из ctx; при отсутствии — slog.Default(). +func From(ctx context.Context) *slog.Logger { + return FromOr(ctx, slog.Default()) +} + +// FromOr достаёт логгер из ctx; при отсутствии — fallback (или slog.Default(), +// если fallback nil). Удобно во внешних клиентах: внутри стадии загрузки вернёт +// scoped-логгер с download_id, при автономном вызове — собственный логгер клиента. +func FromOr(ctx context.Context, fallback *slog.Logger) *slog.Logger { + if ctx != nil { + if log, ok := ctx.Value(ctxKey{}).(*slog.Logger); ok && log != nil { + return log + } + } + if fallback != nil { + return fallback + } + return slog.Default() +} diff --git a/internal/logging/ext.go b/internal/logging/ext.go new file mode 100644 index 0000000..d7309a3 --- /dev/null +++ b/internal/logging/ext.go @@ -0,0 +1,65 @@ +package logging + +import ( + "log/slog" + "time" +) + +// Имена внешних сервисов для поля ext.service. +const ( + ServiceQBittorrent = "qbittorrent" + ServiceJellyfin = "jellyfin" + ServiceLLM = "llm" + ServiceTMDB = "tmdb" + ServiceTVDB = "tvdb" + ServiceTVMaze = "tvmaze" +) + +// ExtCall описывает один вызов внешнего сервиса для логирования по конвенции +// (поля ext.*). Логгер передаётся аргументом — обычно scoped-логгер загрузки, +// чтобы запись о вызове несла download_id/capability стадии. +type ExtCall struct { + Service string // ext.service: qbittorrent/jellyfin/llm/tmdb/tvdb/tvmaze + Operation string // ext.operation: логическая операция (torrents/add, chat.completions, search/movie) + Start time.Time // начало вызова → duration_ms + Status int // ext.status_code: HTTP-код ответа; 0 — не писать (нет кода) + Attempt int // номер попытки; >0 — пишем поле retry (поле и метод Retry конфликтовали бы) +} + +func (c ExtCall) attrs(extra ...any) []any { + a := make([]any, 0, 10+len(extra)) + a = append(a, + "ext.service", c.Service, + "ext.operation", c.Operation, + "duration_ms", time.Since(c.Start).Milliseconds(), + ) + if c.Status > 0 { + a = append(a, "ext.status_code", c.Status) + } + if c.Attempt > 0 { + a = append(a, "retry", c.Attempt) + } + return append(a, extra...) +} + +// Success логирует успешный вызов внешнего сервиса (INFO) — одна запись на вызов. +func (c ExtCall) Success(log *slog.Logger, extra ...any) { + log.Info("external call", c.attrs(extra...)...) +} + +// SuccessDebug — успешный вызов на DEBUG: для частых служебных вызовов +// (поллинг qBittorrent и т.п.), которые на INFO забивали бы аудит шумом, как и +// healthcheck. Сам факт вызова логируется, но в проде (INFO) не пишется. +func (c ExtCall) SuccessDebug(log *slog.Logger, extra ...any) { + log.Debug("external call", c.attrs(extra...)...) +} + +// Retry логирует неудачную попытку, после которой будет повтор (WARN). +func (c ExtCall) Retry(log *slog.Logger, err error, extra ...any) { + log.Warn("external call retry", c.attrs(append([]any{"error", err}, extra...)...)...) +} + +// Failure логирует окончательную неудачу вызова / недоступность сервиса (ERROR). +func (c ExtCall) Failure(log *slog.Logger, err error, extra ...any) { + log.Error("external call failed", c.attrs(append([]any{"error", err}, extra...)...)...) +} diff --git a/internal/logging/logging.go b/internal/logging/logging.go index 7ecfaa0..e29c6ee 100644 --- a/internal/logging/logging.go +++ b/internal/logging/logging.go @@ -5,11 +5,12 @@ import ( "log/slog" "os" "strings" + "time" ) // New возвращает slog-логгер с указанным уровнем и форматом ("json"|"text"). func New(level, format string) *slog.Logger { - opts := &slog.HandlerOptions{Level: parseLevel(level)} + opts := &slog.HandlerOptions{Level: parseLevel(level), ReplaceAttr: utcTime} var handler slog.Handler if strings.EqualFold(format, "text") { @@ -20,6 +21,25 @@ func New(level, format string) *slog.Logger { return slog.New(handler) } +// utcTime приводит метку времени к UTC. JSONHandler сериализует time.Time в +// RFC3339 с долями секунды; в UTC суффикс — Z. Бизнес-логика остаётся в +// Europe/Moscow, UTC — только в логах (явный TZ, инвариант проекта не нарушен). +func utcTime(groups []string, a slog.Attr) slog.Attr { + if len(groups) == 0 && a.Key == slog.TimeKey { + if t, ok := a.Value.Any().(time.Time); ok { + a.Value = slog.TimeValue(t.UTC()) + } + } + return a +} + +// NewStderr — JSON-логгер в stderr (UTC) для фатальных ошибок старта сервиса, +// когда основной логгер ещё не собран (конфиг не прочитан). Уровень не +// ограничиваем — пишем сам факт фатального сбоя. +func NewStderr() *slog.Logger { + return slog.New(slog.NewJSONHandler(os.Stderr, &slog.HandlerOptions{ReplaceAttr: utcTime})) +} + func parseLevel(level string) slog.Level { switch strings.ToLower(level) { case "debug": diff --git a/internal/metadata/http.go b/internal/metadata/http.go index e4736b0..379fd1b 100644 --- a/internal/metadata/http.go +++ b/internal/metadata/http.go @@ -10,6 +10,9 @@ import ( "net/http" "net/url" "time" + + "git.vakhrushev.me/av/jellybit/internal/logctx" + "git.vakhrushev.me/av/jellybit/internal/logging" ) const defaultTimeout = 10 * time.Second @@ -38,8 +41,9 @@ func newHTTPClient(proxy string, timeout time.Duration) (*http.Client, error) { const maxBody = 4 << 20 // 4 MiB — потолок на тело ответа // getJSON выполняет GET и декодирует JSON-ответ в out. headers — опц. -// дополнительные заголовки (напр. Authorization). -func getJSON(ctx context.Context, hc *http.Client, log *slog.Logger, rawURL string, headers map[string]string, out any) error { +// дополнительные заголовки (напр. Authorization). service/operation — поля +// ext.* для телеметрии вызова. +func getJSON(ctx context.Context, hc *http.Client, log *slog.Logger, service, operation, rawURL string, headers map[string]string, out any) error { req, err := http.NewRequestWithContext(ctx, http.MethodGet, rawURL, nil) if err != nil { return fmt.Errorf("metadata: build request: %w", err) @@ -48,11 +52,11 @@ func getJSON(ctx context.Context, hc *http.Client, log *slog.Logger, rawURL stri for k, v := range headers { req.Header.Set(k, v) } - return doJSON(hc, log, req, out) + return doJSON(ctx, hc, log, service, operation, req, out) } // postJSON выполняет POST с JSON-телом и декодирует ответ. -func postJSON(ctx context.Context, hc *http.Client, log *slog.Logger, rawURL string, body, out any) error { +func postJSON(ctx context.Context, hc *http.Client, log *slog.Logger, service, operation, rawURL string, body, out any) error { payload, err := json.Marshal(body) if err != nil { return fmt.Errorf("metadata: marshal body: %w", err) @@ -63,46 +67,39 @@ func postJSON(ctx context.Context, hc *http.Client, log *slog.Logger, rawURL str } req.Header.Set("Content-Type", "application/json") req.Header.Set("Accept", "application/json") - return doJSON(hc, log, req, out) + return doJSON(ctx, hc, log, service, operation, req, out) } -// doJSON выполняет запрос и декодирует ответ, логируя исход. В лог идут только -// host и path (без query) — у TMDB api_key передаётся query-параметром, его -// нельзя светить в логах. -func doJSON(hc *http.Client, log *slog.Logger, req *http.Request, out any) error { - if log == nil { - log = slog.Default() - } - start := time.Now() +// doJSON выполняет запрос и декодирует ответ, логируя исход телеметрией ext.* +// (логическая operation вместо URL: у TMDB api_key передаётся query-параметром, +// его нельзя светить в логах). Логгер берётся из ctx (scoped-логгер загрузки), +// при отсутствии — переданный fallback. +func doJSON(ctx context.Context, hc *http.Client, log *slog.Logger, service, operation string, req *http.Request, out any) error { + log = logctx.FromOr(ctx, log) + call := logging.ExtCall{Service: service, Operation: operation, Start: time.Now()} resp, err := hc.Do(req) if err != nil { - log.Warn("metadata: request failed", - "method", req.Method, "host", req.URL.Host, "path", req.URL.Path, - "duration", time.Since(start), "err", err) + call.Failure(log, err) return fmt.Errorf("metadata: request: %w", err) } defer func() { _ = resp.Body.Close() }() + call.Status = resp.StatusCode raw, err := io.ReadAll(io.LimitReader(resp.Body, maxBody)) if err != nil { - log.Warn("metadata: read body failed", - "host", req.URL.Host, "path", req.URL.Path, "err", err) + call.Failure(log, err) return fmt.Errorf("metadata: read body: %w", err) } if resp.StatusCode != http.StatusOK { - log.Warn("metadata: non-ok status", - "method", req.Method, "host", req.URL.Host, "path", req.URL.Path, - "status", resp.StatusCode, "duration", time.Since(start)) - return fmt.Errorf("metadata: status %d: %s", resp.StatusCode, snippet(raw)) + err := fmt.Errorf("metadata: status %d: %s", resp.StatusCode, snippet(raw)) + call.Failure(log, err) + return err } if err := json.Unmarshal(raw, out); err != nil { - log.Warn("metadata: decode failed", - "host", req.URL.Host, "path", req.URL.Path, "err", err) + call.Failure(log, err) return fmt.Errorf("metadata: decode: %w (body: %s)", err, snippet(raw)) } - log.Debug("metadata: request ok", - "method", req.Method, "host", req.URL.Host, "path", req.URL.Path, - "duration", time.Since(start)) + call.Success(log) return nil } diff --git a/internal/metadata/tmdb.go b/internal/metadata/tmdb.go index ff25e8c..17c62e9 100644 --- a/internal/metadata/tmdb.go +++ b/internal/metadata/tmdb.go @@ -9,6 +9,8 @@ import ( "strconv" "strings" "time" + + "git.vakhrushev.me/av/jellybit/internal/logging" ) const tmdbDefaultBaseURL = "https://api.themoviedb.org/3" @@ -81,12 +83,11 @@ func (t *TMDB) Search(ctx context.Context, q Query) ([]Candidate, error) { return nil, fmt.Errorf("metadata: tmdb: unknown type %q", q.Type) } - t.log.Debug("tmdb: search", "type", q.Type, "title", q.Title, "year", q.Year) var resp tmdbSearchResp - if err := getJSON(ctx, t.hc, t.log, t.baseURL+path+"?"+params.Encode(), nil, &resp); err != nil { + op := strings.TrimPrefix(path, "/") // search/movie | search/tv + if err := getJSON(ctx, t.hc, t.log, logging.ServiceTMDB, op, t.baseURL+path+"?"+params.Encode(), nil, &resp); err != nil { return nil, fmt.Errorf("tmdb search: %w", err) } - t.log.Debug("tmdb: search done", "title", q.Title, "results", len(resp.Results)) out := make([]Candidate, 0, len(resp.Results)) for _, r := range resp.Results { @@ -116,7 +117,7 @@ type tmdbTVResp struct { func (t *TMDB) SeasonEpisodeCounts(ctx context.Context, id string) (map[int]int, error) { params := url.Values{"api_key": {t.apiKey}} var resp tmdbTVResp - if err := getJSON(ctx, t.hc, t.log, t.baseURL+"/tv/"+url.PathEscape(id)+"?"+params.Encode(), nil, &resp); err != nil { + if err := getJSON(ctx, t.hc, t.log, logging.ServiceTMDB, "tv", t.baseURL+"/tv/"+url.PathEscape(id)+"?"+params.Encode(), nil, &resp); err != nil { return nil, fmt.Errorf("tmdb tv %s: %w", id, err) } out := make(map[int]int, len(resp.Seasons)) diff --git a/internal/metadata/tvdb.go b/internal/metadata/tvdb.go index 0c7371b..32204fb 100644 --- a/internal/metadata/tvdb.go +++ b/internal/metadata/tvdb.go @@ -12,6 +12,9 @@ import ( "strings" "sync" "time" + + "git.vakhrushev.me/av/jellybit/internal/logctx" + "git.vakhrushev.me/av/jellybit/internal/logging" ) const tvdbDefaultBaseURL = "https://api4.thetvdb.com/v4" @@ -70,8 +73,8 @@ func (t *TVDB) login(ctx context.Context) (string, error) { Token string `json:"token"` } `json:"data"` } - t.log.Debug("tvdb: login (fetching bearer token)") - if err := postJSON(ctx, t.hc, t.log, t.baseURL+"/login", + // Тело запроса содержит apikey — postJSON его не логирует (только ext.*). + if err := postJSON(ctx, t.hc, t.log, logging.ServiceTVDB, "login", t.baseURL+"/login", map[string]string{"apikey": t.apiKey}, &resp); err != nil { return "", fmt.Errorf("tvdb login: %w", err) } @@ -83,24 +86,26 @@ func (t *TVDB) login(ctx context.Context) (string, error) { } // get делает авторизованный GET; при 401 один раз перелогинивается. -func (t *TVDB) get(ctx context.Context, path string, out any) error { +// operation — логическая операция для поля ext.operation. +func (t *TVDB) get(ctx context.Context, operation, path string, out any) error { token, err := t.login(ctx) if err != nil { return err } - status, raw, err := t.rawGet(ctx, path, token) + status, raw, err := t.rawGet(ctx, operation, path, token) if err != nil { return err } if status == http.StatusUnauthorized { - t.log.Warn("tvdb: token expired, re-login", "path", path) + // Рутинное обновление протухшего токена — DEBUG (не «может стать проблемой»). + logctx.FromOr(ctx, t.log).Debug("tvdb token expired, re-login") t.mu.Lock() t.token = "" // сбрасываем протухший токен t.mu.Unlock() if token, err = t.login(ctx); err != nil { return err } - if status, raw, err = t.rawGet(ctx, path, token); err != nil { + if status, raw, err = t.rawGet(ctx, operation, path, token); err != nil { return err } } @@ -113,28 +118,28 @@ func (t *TVDB) get(ctx context.Context, path string, out any) error { return nil } -func (t *TVDB) rawGet(ctx context.Context, path, token string) (int, []byte, error) { +func (t *TVDB) rawGet(ctx context.Context, operation, path, token string) (int, []byte, error) { req, err := http.NewRequestWithContext(ctx, http.MethodGet, t.baseURL+path, nil) if err != nil { return 0, nil, fmt.Errorf("tvdb: build request: %w", err) } req.Header.Set("Authorization", "Bearer "+token) req.Header.Set("Accept", "application/json") - start := time.Now() + log := logctx.FromOr(ctx, t.log) + call := logging.ExtCall{Service: logging.ServiceTVDB, Operation: operation, Start: time.Now()} resp, err := t.hc.Do(req) if err != nil { - t.log.Warn("tvdb: request failed", - "host", req.URL.Host, "path", req.URL.Path, "duration", time.Since(start), "err", err) + call.Failure(log, err) return 0, nil, fmt.Errorf("tvdb: request: %w", err) } defer func() { _ = resp.Body.Close() }() + call.Status = resp.StatusCode raw, err := io.ReadAll(io.LimitReader(resp.Body, maxBody)) if err != nil { - t.log.Warn("tvdb: read body failed", "host", req.URL.Host, "path", req.URL.Path, "err", err) + call.Failure(log, err) return 0, nil, fmt.Errorf("tvdb: read body: %w", err) } - t.log.Debug("tvdb: request done", - "host", req.URL.Host, "path", req.URL.Path, "status", resp.StatusCode, "duration", time.Since(start)) + call.Success(log) return resp.StatusCode, raw, nil } @@ -156,12 +161,10 @@ func (t *TVDB) Search(ctx context.Context, q Query) ([]Candidate, error) { if q.Year > 0 { params.Set("year", strconv.Itoa(q.Year)) } - t.log.Debug("tvdb: search", "type", q.Type, "title", q.Title, "year", q.Year) var resp tvdbSearchResp - if err := t.get(ctx, "/search?"+params.Encode(), &resp); err != nil { + if err := t.get(ctx, "search", "/search?"+params.Encode(), &resp); err != nil { return nil, fmt.Errorf("tvdb search: %w", err) } - t.log.Debug("tvdb: search done", "title", q.Title, "results", len(resp.Data)) out := make([]Candidate, 0, len(resp.Data)) for _, r := range resp.Data { if r.TVDBID == "" { @@ -189,7 +192,7 @@ type tvdbExtendedResp struct { // SeasonEpisodeCounts считает число серий по сезонам из расширенных данных. func (t *TVDB) SeasonEpisodeCounts(ctx context.Context, id string) (map[int]int, error) { var resp tvdbExtendedResp - if err := t.get(ctx, "/series/"+url.PathEscape(id)+"/extended?meta=episodes&short=true", &resp); err != nil { + if err := t.get(ctx, "series/extended", "/series/"+url.PathEscape(id)+"/extended?meta=episodes&short=true", &resp); err != nil { return nil, fmt.Errorf("tvdb series %s: %w", id, err) } out := map[int]int{} diff --git a/internal/metadata/tvmaze.go b/internal/metadata/tvmaze.go index fddf587..f53b24c 100644 --- a/internal/metadata/tvmaze.go +++ b/internal/metadata/tvmaze.go @@ -9,6 +9,8 @@ import ( "strconv" "strings" "time" + + "git.vakhrushev.me/av/jellybit/internal/logging" ) const tvmazeDefaultBaseURL = "https://api.tvmaze.com" @@ -68,11 +70,9 @@ func (t *TVMaze) Search(ctx context.Context, q Query) ([]Candidate, error) { Show tvmazeShow `json:"show"` } rawURL := t.baseURL + "/search/shows?q=" + url.QueryEscape(q.Title) - t.log.Debug("tvmaze: search", "title", q.Title) - if err := getJSON(ctx, t.hc, t.log, rawURL, nil, &resp); err != nil { + if err := getJSON(ctx, t.hc, t.log, logging.ServiceTVMaze, "search/shows", rawURL, nil, &resp); err != nil { return nil, fmt.Errorf("tvmaze search: %w", err) } - t.log.Debug("tvmaze: search done", "title", q.Title, "results", len(resp)) out := make([]Candidate, 0, len(resp)) for _, r := range resp { @@ -104,7 +104,7 @@ type tvmazeEpisode struct { func (t *TVMaze) SeasonEpisodeCounts(ctx context.Context, id string) (map[int]int, error) { var eps []tvmazeEpisode rawURL := t.baseURL + "/shows/" + url.PathEscape(id) + "/episodes" - if err := getJSON(ctx, t.hc, t.log, rawURL, nil, &eps); err != nil { + if err := getJSON(ctx, t.hc, t.log, logging.ServiceTVMaze, "shows/episodes", rawURL, nil, &eps); err != nil { return nil, fmt.Errorf("tvmaze episodes %s: %w", id, err) } out := map[int]int{} diff --git a/internal/naming/prompt.go b/internal/naming/prompt.go index 3af385d..173d5de 100644 --- a/internal/naming/prompt.go +++ b/internal/naming/prompt.go @@ -7,6 +7,7 @@ import ( "strings" "git.vakhrushev.me/av/jellybit/internal/llm" + "git.vakhrushev.me/av/jellybit/internal/logctx" ) // systemPrompt инструктирует модель вытащить из вольного контекста короткое @@ -56,6 +57,7 @@ func (n *Namer) extractViaLLM(ctx context.Context, contextText, hint string) (ex {Role: llm.RoleUser, Content: user}, } + log := logctx.FromOr(ctx, n.log) for attempt := 1; attempt <= n.attempts; attempt++ { resp, err := n.provider.Complete(ctx, llm.Request{ Messages: msgs, @@ -63,9 +65,9 @@ func (n *Namer) extractViaLLM(ctx context.Context, contextText, hint string) (ex Temperature: &temp, }) if err != nil { - // Транспортная ошибка/таймаут: дальше пробовать смысла нет — - // уходим в фолбек, приём не валим. - n.log.Warn("naming: llm complete failed, will fall back", "err", err) + // Транспортная ошибка/таймаут залогирована клиентом LLM (ext.*); + // здесь — доменное решение «уходим в фолбек, приём не валим». + log.Debug("naming llm failed, using fallback") return extracted{}, false } @@ -73,7 +75,7 @@ func (n *Namer) extractViaLLM(ctx context.Context, contextText, hint string) (ex if perr == nil { return ex, true } - n.log.Warn("naming: unparsed llm response", "attempt", attempt, "err", perr) + log.Warn("naming llm response unparsed", "attempt", attempt, "error", perr) msgs = append(msgs, llm.Message{Role: llm.RoleAssistant, Content: resp.Content}, llm.Message{Role: llm.RoleUser, Content: "Ответ невалиден: " + perr.Error() + diff --git a/internal/qbt/qbt.go b/internal/qbt/qbt.go index e6fe7ff..10e4e52 100644 --- a/internal/qbt/qbt.go +++ b/internal/qbt/qbt.go @@ -21,6 +21,9 @@ import ( "strings" "sync" "time" + + "git.vakhrushev.me/av/jellybit/internal/logctx" + "git.vakhrushev.me/av/jellybit/internal/logging" ) // Config — параметры подключения к qBittorrent WebUI. @@ -119,19 +122,25 @@ func (c *Client) login(ctx context.Context) error { } req.Header.Set("Content-Type", "application/x-www-form-urlencoded") req.Header.Set("Referer", c.base.String()) // qBit проверяет Referer/Host + log := logctx.FromOr(ctx, c.log) + call := logging.ExtCall{Service: logging.ServiceQBittorrent, Operation: "auth/login", Start: time.Now()} resp, err := c.hc.Do(req) if err != nil { + call.Failure(log, err) return fmt.Errorf("qbittorrent login: %w", err) } defer func() { _ = resp.Body.Close() }() + call.Status = resp.StatusCode body, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<10)) if resp.StatusCode != http.StatusOK || strings.TrimSpace(string(body)) != "Ok." { - c.log.Error("qbittorrent: login failed", - "status", resp.StatusCode, "body", strings.TrimSpace(string(body))) - return fmt.Errorf("qbittorrent login failed: status %d body %q", + // Тело логина не содержит секретов (qBit отвечает "Ok."/"Fails."), но + // учётные данные (логин/пароль) в лог не идут — только факт неуспеха. + err := fmt.Errorf("qbittorrent login failed: status %d body %q", resp.StatusCode, strings.TrimSpace(string(body))) + call.Failure(log, err) + return err } - c.log.Debug("qbittorrent: login ok", "user", c.user) + call.Success(log, "authenticated", true) return nil } @@ -148,7 +157,7 @@ func (c *Client) do(ctx context.Context, build func() (*http.Request, error)) (* } if resp.StatusCode == http.StatusForbidden { _ = resp.Body.Close() - c.log.Debug("qbittorrent: session expired (403), re-login") + logctx.FromOr(ctx, c.log).Debug("qbittorrent session expired, re-login") if err := c.login(ctx); err != nil { return nil, err } @@ -193,6 +202,8 @@ func (c *Client) Add(ctx context.Context, ar AddRequest) error { contentType := mw.FormDataContentType() payload := buf.Bytes() + log := logctx.FromOr(ctx, c.log) + call := logging.ExtCall{Service: logging.ServiceQBittorrent, Operation: "torrents/add", Start: time.Now()} resp, err := c.do(ctx, func() (*http.Request, error) { req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.endpoint("/api/v2/torrents/add"), bytes.NewReader(payload)) @@ -204,27 +215,33 @@ func (c *Client) Add(ctx context.Context, ar AddRequest) error { return req, nil }) if err != nil { + call.Failure(log, err, "category", ar.Category) return fmt.Errorf("qbittorrent add: %w", err) } defer func() { _ = resp.Body.Close() }() + call.Status = resp.StatusCode body, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<10)) if resp.StatusCode != http.StatusOK { - return fmt.Errorf("qbittorrent add: status %d body %q", + err := fmt.Errorf("qbittorrent add: status %d body %q", resp.StatusCode, strings.TrimSpace(string(body))) + call.Failure(log, err, "category", ar.Category) + return err } if strings.TrimSpace(string(body)) == "Fails." { - c.log.Error("qbittorrent: add rejected", - "category", ar.Category, "urls", len(ar.URLs), "torrents", len(ar.Torrents)) - return fmt.Errorf("qbittorrent add: rejected (Fails.)") + err := fmt.Errorf("qbittorrent add: rejected (Fails.)") + call.Failure(log, err, "category", ar.Category, + "urls", len(ar.URLs), "torrents", len(ar.Torrents)) + return err } - c.log.Info("qbittorrent: torrent added", - "category", ar.Category, "save_path", ar.SavePath, + call.Success(log, "category", ar.Category, "save_path", ar.SavePath, "urls", len(ar.URLs), "torrents", len(ar.Torrents), "paused", ar.Paused) return nil } // Torrents возвращает задачи указанной категории (пустая — все). func (c *Client) Torrents(ctx context.Context, category string) ([]Torrent, error) { + log := logctx.FromOr(ctx, c.log) + call := logging.ExtCall{Service: logging.ServiceQBittorrent, Operation: "torrents/info", Start: time.Now()} resp, err := c.do(ctx, func() (*http.Request, error) { u := c.endpoint("/api/v2/torrents/info") if category != "" { @@ -233,19 +250,25 @@ func (c *Client) Torrents(ctx context.Context, category string) ([]Torrent, erro return http.NewRequestWithContext(ctx, http.MethodGet, u, nil) }) if err != nil { + call.Failure(log, err) return nil, fmt.Errorf("qbittorrent info: %w", err) } defer func() { _ = resp.Body.Close() }() + call.Status = resp.StatusCode if resp.StatusCode != http.StatusOK { body, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<10)) - return nil, fmt.Errorf("qbittorrent info: status %d body %q", + err := fmt.Errorf("qbittorrent info: status %d body %q", resp.StatusCode, strings.TrimSpace(string(body))) + call.Failure(log, err) + return nil, err } var ts []Torrent if err := json.NewDecoder(resp.Body).Decode(&ts); err != nil { + call.Failure(log, err) return nil, fmt.Errorf("decode qbittorrent info: %w", err) } - c.log.Debug("qbittorrent: torrents fetched", "category", category, "count", len(ts)) + // Поллинг частый — на DEBUG, чтобы не зашумлять INFO (как healthcheck). + call.SuccessDebug(log, "category", category, "count", len(ts)) return ts, nil } @@ -253,23 +276,30 @@ func (c *Client) Torrents(ctx context.Context, category string) ([]Torrent, erro // включая корневую папку для многофайловых раздач, и размеры). Нужен // распознаванию как один из сигналов; абсолютный путь — join(save_path, Name). func (c *Client) Files(ctx context.Context, hash string) ([]File, error) { + log := logctx.FromOr(ctx, c.log) + call := logging.ExtCall{Service: logging.ServiceQBittorrent, Operation: "torrents/files", Start: time.Now()} resp, err := c.do(ctx, func() (*http.Request, error) { u := c.endpoint("/api/v2/torrents/files?hash=" + url.QueryEscape(hash)) return http.NewRequestWithContext(ctx, http.MethodGet, u, nil) }) if err != nil { + call.Failure(log, err) return nil, fmt.Errorf("qbittorrent files: %w", err) } defer func() { _ = resp.Body.Close() }() + call.Status = resp.StatusCode if resp.StatusCode != http.StatusOK { body, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<10)) - return nil, fmt.Errorf("qbittorrent files: status %d body %q", + err := fmt.Errorf("qbittorrent files: status %d body %q", resp.StatusCode, strings.TrimSpace(string(body))) + call.Failure(log, err) + return nil, err } var fs []File if err := json.NewDecoder(resp.Body).Decode(&fs); err != nil { + call.Failure(log, err) return nil, fmt.Errorf("decode qbittorrent files: %w", err) } - c.log.Debug("qbittorrent: files fetched", "hash", hash, "count", len(fs)) + call.SuccessDebug(log, "count", len(fs)) return fs, nil } diff --git a/internal/recognize/metadata.go b/internal/recognize/metadata.go index 9ddd812..beb3cb5 100644 --- a/internal/recognize/metadata.go +++ b/internal/recognize/metadata.go @@ -5,6 +5,7 @@ import ( "strings" "unicode" + "git.vakhrushev.me/av/jellybit/internal/logctx" "git.vakhrushev.me/av/jellybit/internal/metadata" ) @@ -38,7 +39,9 @@ func (r *Recognizer) matchMetadata(ctx context.Context, plan Plan) (*Match, []me for _, p := range r.providers { cands, err := p.Search(ctx, metadata.Query{Type: mt, Title: searchTitle, Year: plan.Year}) if err != nil { - r.log.Warn("recognize: metadata search failed", "provider", p.Name(), "err", err) + // Сам вызов провайдера залогирован клиентом (ext.*-ERROR); здесь — + // доменное решение «пропускаем провайдера, пробуем следующий». + logctx.FromOr(ctx, r.log).Debug("metadata provider skipped", "provider", p.Name()) continue } @@ -73,7 +76,7 @@ func (r *Recognizer) buildMatch(ctx context.Context, p metadata.Provider, c meta if got, err := p.SeasonEpisodeCounts(ctx, c.ID); err == nil { counts = got } else { - r.log.Warn("recognize: episode counts failed", "provider", p.Name(), "id", c.ID, "err", err) + logctx.FromOr(ctx, r.log).Debug("metadata episode counts skipped", "provider", p.Name(), "id", c.ID) } } prov, pid := CandidateTag(c) diff --git a/internal/recognize/recognize.go b/internal/recognize/recognize.go index a32ec27..da45aaf 100644 --- a/internal/recognize/recognize.go +++ b/internal/recognize/recognize.go @@ -22,6 +22,7 @@ import ( "log/slog" "git.vakhrushev.me/av/jellybit/internal/llm" + "git.vakhrushev.me/av/jellybit/internal/logctx" "git.vakhrushev.me/av/jellybit/internal/metadata" ) @@ -188,6 +189,7 @@ func New(provider LLM, providers []metadata.Provider, cfg Config, log *slog.Logg // error (наверху решат retry/failed). Неразобранный после ретраев ответ — // не ошибка, а Result с решением review (см. recognition.md). func (r *Recognizer) Recognize(ctx context.Context, in Input) (Result, error) { + log := logctx.FromOr(ctx, r.log) pre := preParse(in.Name) msgs := buildMessages(in, pre, r.maxFiles) @@ -214,8 +216,8 @@ func (r *Recognizer) Recognize(ctx context.Context, in Input) (Result, error) { if parseErr == nil { break } - r.log.Warn("recognize: unparsed llm response", - "attempt", attempts, "err", parseErr) + log.Warn("recognition llm response unparsed", + "attempt", attempts, "error", parseErr) // Просим модель исправиться, повторяя схему и ошибку. msgs = append(msgs, llm.Message{Role: llm.RoleAssistant, Content: raw}, @@ -246,8 +248,8 @@ func (r *Recognizer) Recognize(ctx context.Context, in Input) (Result, error) { } dec := decide(plan, pre, match, len(r.providers) > 0, r.threshold) - r.log.Info("recognize: done", - "type", plan.Type, "title", plan.Title, "year", plan.Year, + log.Info("recognition done", + "media_type", plan.Type, "title", plan.Title, "year", plan.Year, "files", len(plan.Files), "attempts", attempts, "matched", match != nil, "candidates", len(candidates), "auto", dec.Auto, "reasons", len(dec.Reasons)) diff --git a/internal/tgbot/bot.go b/internal/tgbot/bot.go index fa75c34..c57478d 100644 --- a/internal/tgbot/bot.go +++ b/internal/tgbot/bot.go @@ -116,7 +116,7 @@ func (b *Bot) handleMessage(ctx context.Context, m *tgbotapi.Message) { return } if !b.allowed[m.From.ID] { - b.log.Warn("telegram: denied user", "user_id", m.From.ID, "username", m.From.UserName) + b.log.Warn("telegram denied user", "user_id", m.From.ID, "username", m.From.UserName) b.send(m.Chat.ID, "Доступ запрещён.", nil) return } @@ -214,7 +214,7 @@ func (b *Bot) handleCallback(ctx context.Context, cq *tgbotapi.CallbackQuery) { func (b *Bot) refreshCard(ctx context.Context, chatID int64, msgID int, id int64) { rd, err := b.reviewer.ReviewData(ctx, id) if err != nil { - b.log.Warn("telegram: refresh card failed", "download_id", id, "err", err) + b.log.Warn("telegram refresh card failed", "download_id", id, "error", err) return } text, kb := b.renderCard(rd) @@ -225,7 +225,7 @@ func (b *Bot) refreshCard(ctx context.Context, chatID int64, msgID int, id int64 edit = tgbotapi.NewEditMessageText(chatID, msgID, text) } if _, err := b.api.Send(edit); err != nil { - b.log.Warn("telegram: edit card failed", "download_id", id, "err", err) + b.log.Warn("telegram edit card failed", "download_id", id, "error", err) } } @@ -235,7 +235,7 @@ func (b *Bot) refreshCard(ctx context.Context, chatID int64, msgID int, id int64 func (b *Bot) Notify(ctx context.Context, downloadID int64, event worker.NotifyEvent) { rd, err := b.reviewer.ReviewData(ctx, downloadID) if err != nil { - b.log.Warn("telegram: notify review data", "download_id", downloadID, "err", err) + b.log.Warn("telegram notify review data", "download_id", downloadID, "error", err) return } var text string @@ -260,13 +260,13 @@ func (b *Bot) send(chatID int64, text string, kb *tgbotapi.InlineKeyboardMarkup) msg.ReplyMarkup = *kb } if _, err := b.api.Send(msg); err != nil { - b.log.Warn("telegram: send failed", "chat_id", chatID, "err", err) + b.log.Warn("telegram send failed", "chat_id", chatID, "error", err) } } func (b *Bot) answer(callbackID, text string) { if _, err := b.api.Request(tgbotapi.NewCallback(callbackID, text)); err != nil { - b.log.Warn("telegram: answer callback failed", "err", err) + b.log.Warn("telegram answer callback failed", "error", err) } } diff --git a/internal/worker/discover.go b/internal/worker/discover.go index e8f175c..1504aa6 100644 --- a/internal/worker/discover.go +++ b/internal/worker/discover.go @@ -40,7 +40,7 @@ func (w *Worker) adopt(ctx context.Context, t qbt.Torrent) { } exists, err := w.store.ExistsByInfohash(ctx, infohash) if err != nil { - w.log.Warn("discover: exists check failed", "infohash", infohash, "err", err) + w.log.Warn("discover exists check failed", "capability", capIngest, "infohash", infohash, "error", err) return } if exists { @@ -61,11 +61,11 @@ func (w *Worker) adopt(ctx context.Context, t qbt.Torrent) { if ex, _ := w.store.ExistsByInfohash(ctx, infohash); ex { return } - w.log.Error("discover: adopt failed", "infohash", infohash, "err", err) + w.log.Error("discover adopt failed", "capability", capIngest, "infohash", infohash, "error", err) return } - w.log.Info("discover: adopted torrent", - "download_id", id, "infohash", infohash, "name", t.Name, + w.log.Info("discover adopted torrent", + "capability", capIngest, "download_id", id, "infohash", infohash, "name", t.Name, "category", t.Category, "tags", t.Tags) } diff --git a/internal/worker/review.go b/internal/worker/review.go index dac315e..1e2b714 100644 --- a/internal/worker/review.go +++ b/internal/worker/review.go @@ -11,6 +11,7 @@ import ( "strings" "git.vakhrushev.me/av/jellybit/internal/layout" + "git.vakhrushev.me/av/jellybit/internal/logctx" "git.vakhrushev.me/av/jellybit/internal/metadata" "git.vakhrushev.me/av/jellybit/internal/qbt" "git.vakhrushev.me/av/jellybit/internal/recognize" @@ -37,7 +38,7 @@ func (w *Worker) recognizePending(ctx context.Context) { pending, err := w.store.ListDownloadsByState(ctx, store.StateCompleted, store.StateRecognizing) w.mu.Unlock() if err != nil { - w.log.Warn("recognize: list pending failed", "err", err) + w.log.Warn("recognition list pending failed", "error", err) return } for _, d := range pending { @@ -54,13 +55,14 @@ func (w *Worker) recognizeOne(ctx context.Context, id int64) { d, err := w.store.GetDownload(ctx, id) if err != nil { w.mu.Unlock() - w.log.Warn("recognize: get download", "download_id", id, "err", err) + w.log.Warn("recognition get download failed", "download_id", id, "error", err) return } if d.State != store.StateCompleted && d.State != store.StateRecognizing { w.mu.Unlock() return } + ctx = w.scoped(ctx, capRecognize, id, d.Infohash.String) if d.State == store.StateCompleted { w.transition(ctx, *d, store.StateRecognizing, "", "") } @@ -68,8 +70,9 @@ func (w *Worker) recognizeOne(ctx context.Context, id int64) { result, savePath, err := w.runRecognize(ctx, *d) if err != nil { - // Не смогли получить сигналы или вызвать LLM — уходим в review с - // причиной, человек перезапустит подсказкой. + // Граница доменной стадии распознавания: логируем исход один раз (ERROR), + // дальше уходим в review с причиной — человек перезапустит подсказкой. + logctx.From(ctx).Error("recognition failed", "error", err) result = recognize.Result{Decision: recognize.Decision{ Reasons: []string{"распознавание не удалось: " + err.Error()}, }} @@ -121,9 +124,10 @@ func (w *Worker) runRecognize(ctx context.Context, d store.Download) (recognize. // finishRecognition сохраняет попытку распознавания и двигает задачу. В Ф3 // метабазы выключены → авто-раскладки не делаем, всегда уходим в review. func (w *Worker) finishRecognition(ctx context.Context, id int64, res recognize.Result, savePath string) { + log := logctx.From(ctx) planJSON, err := json.Marshal(res.Plan) if err != nil { - w.log.Error("recognize: marshal plan", "download_id", id, "err", err) + log.Error("recognition marshal plan failed", "error", err) planJSON = []byte("{}") } @@ -157,24 +161,23 @@ func (w *Worker) finishRecognition(ctx context.Context, id int64, res recognize. d, err := w.store.GetDownload(ctx, id) if err != nil { - w.log.Warn("recognize: reload download", "download_id", id, "err", err) + log.Warn("recognition reload download failed", "error", err) return } if d.State != store.StateRecognizing { // За время вызова LLM задачу увели (cancel/defer) — результат не нужен. - w.log.Info("recognize: result discarded, state changed", - "download_id", id, "state", d.State) + log.Info("recognition result discarded", "state", d.State) return } recID, err := w.store.CreateRecognition(ctx, rec, res.Decision.Reasons) if err != nil { - w.log.Error("recognize: persist", "download_id", id, "err", err) + log.Error("recognition persist failed", "error", err) return } // Кандидаты базы — для ручного выбора в review. if cands := toStoreCandidates(recID, res.Candidates); len(cands) > 0 { if err := w.store.CreateCandidates(ctx, cands); err != nil { - w.log.Warn("recognize: persist candidates", "download_id", id, "err", err) + log.Warn("recognition persist candidates failed", "error", err) } } @@ -186,10 +189,10 @@ func (w *Worker) finishRecognition(ctx context.Context, id int64, res recognize. forceReview := overrides[ovrForceReview] == "1" if res.Decision.Auto && !forceReview && w.layouter != nil { plan := applyOverrides(res.Plan, overrides) - w.transition(ctx, *d, store.StateLinking, "", "") - if err := w.linkPlan(ctx, d, plan, tag, savePath); err != nil { - w.log.Warn("recognize: auto-apply failed, left for review", - "download_id", id, "err", err) + lctx := w.scoped(ctx, capFileLayout, id, d.Infohash.String) + w.transition(lctx, *d, store.StateLinking, "", "") + if err := w.linkPlan(lctx, d, plan, tag, savePath); err != nil { + logctx.From(lctx).Warn("auto-apply failed, left for review", "error", err) } return } @@ -200,7 +203,7 @@ func (w *Worker) finishRecognition(ctx context.Context, id int64, res recognize. func (w *Worker) overridesOrNil(ctx context.Context, id int64) map[string]string { o, err := w.store.ListOverrides(ctx, id) if err != nil { - w.log.Warn("recognize: list overrides", "download_id", id, "err", err) + logctx.From(ctx).Warn("recognition list overrides failed", "error", err) return nil } return o @@ -224,6 +227,7 @@ func (w *Worker) Apply(ctx context.Context, id int64) error { if d.State != store.StateReview && d.State != store.StateDeferred { return fmt.Errorf("apply: download %d is in state %s (expected review/deferred)", id, d.State) } + ctx = w.scoped(ctx, capFileLayout, id, d.Infohash.String) plan, tag, err := w.effectivePlan(ctx, id) if err != nil { @@ -282,7 +286,7 @@ func (w *Worker) linkPlan(ctx context.Context, d *store.Download, plan recognize } w.transition(ctx, *d, store.StateDone, "", "") - w.log.Info("apply: linked", "download_id", d.ID, "batch", batch, "links", len(fl)) + logctx.From(ctx).Info("layout linked", "batch", batch, "links", len(fl)) return nil } @@ -324,8 +328,9 @@ func (w *Worker) Relink(ctx context.Context, id int64) error { if err := w.store.SetOverride(ctx, id, ovrForceReview, "1"); err != nil { return fmt.Errorf("relink: %w", err) } + ctx = w.scoped(ctx, capReview, id, d.Infohash.String) w.transition(ctx, *d, store.StateRecognizing, "", "") - w.log.Info("relink: re-recognizing download", "download_id", id, "from", d.State) + logctx.From(ctx).Info("relink re-recognizing", "from", d.State) return nil } @@ -340,7 +345,8 @@ func (w *Worker) Rerecognize(ctx context.Context, id int64) error { if err != nil { return err } - w.log.Info("review: re-recognizing without hint", "download_id", id) + ctx = w.scoped(ctx, capReview, id, d.Infohash.String) + logctx.From(ctx).Info("review re-recognizing without hint") w.transition(ctx, *d, store.StateRecognizing, "", "") return nil } @@ -358,10 +364,11 @@ func (w *Worker) Refine(ctx context.Context, id int64, hint string) error { if err != nil { return err } + ctx = w.scoped(ctx, capReview, id, d.Infohash.String) if err := w.store.AddHint(ctx, id, hint); err != nil { return fmt.Errorf("refine: %w", err) } - w.log.Info("review: hint added, re-recognizing", "download_id", id, "hint", hint) + logctx.From(ctx).Info("review hint added", "hint", hint) w.transition(ctx, *d, store.StateRecognizing, "", "") return nil } @@ -379,6 +386,7 @@ func (w *Worker) SetType(ctx context.Context, id int64, mediaType string) error if err != nil { return err } + ctx = w.scoped(ctx, capReview, id, d.Infohash.String) if err := w.store.SetOverride(ctx, id, ovrMediaType, mediaType); err != nil { return fmt.Errorf("set type: %w", err) } @@ -403,7 +411,8 @@ func (w *Worker) IgnoreFile(ctx context.Context, id int64, src string) error { w.mu.Lock() defer w.mu.Unlock() - if _, err := w.requireReviewable(ctx, id, "ignore"); err != nil { + d, err := w.requireReviewable(ctx, id, "ignore") + if err != nil { return err } overrides, err := w.store.ListOverrides(ctx, id) @@ -418,7 +427,7 @@ func (w *Worker) IgnoreFile(ctx context.Context, id int64, src string) error { if err := w.store.SetOverride(ctx, id, ovrIgnoredFiles, string(b)); err != nil { return fmt.Errorf("ignore: %w", err) } - w.log.Info("review: file ignored", "download_id", id, "src", src) + logctx.From(w.scoped(ctx, capReview, id, d.Infohash.String)).Info("review file ignored", "src", src) return nil } @@ -434,6 +443,7 @@ func (w *Worker) Defer(ctx context.Context, id int64) error { if d.State.IsTerminal() { return fmt.Errorf("defer: download %d is terminal (%s)", id, d.State) } + ctx = w.scoped(ctx, capReview, id, d.Infohash.String) w.transition(ctx, *d, store.StateDeferred, "", "") return nil } @@ -454,6 +464,7 @@ func (w *Worker) Undo(ctx context.Context, id int64) error { if d.State != store.StateDone { return fmt.Errorf("undo: download %d is in state %s (expected done)", id, d.State) } + ctx = w.scoped(ctx, capFileLayout, id, d.Infohash.String) batch, err := w.store.LatestBatchID(ctx, id) if err != nil { return fmt.Errorf("undo: %w", err) @@ -477,7 +488,7 @@ func (w *Worker) Undo(ctx context.Context, id int64) error { return fmt.Errorf("undo: %w", err) } w.transition(ctx, *d, store.StateReverted, "", "") - w.log.Info("undo: reverted", "download_id", id, "batch", batch, "removed", n) + logctx.From(ctx).Info("layout reverted", "batch", batch, "removed", n) return nil } @@ -502,7 +513,8 @@ func (w *Worker) ChooseCandidate(ctx context.Context, id, candidateID int64) err w.mu.Lock() defer w.mu.Unlock() - if _, err := w.requireReviewable(ctx, id, "choose candidate"); err != nil { + d, err := w.requireReviewable(ctx, id, "choose candidate") + if err != nil { return err } rec, err := w.store.GetCurrentRecognition(ctx, id) @@ -532,8 +544,8 @@ func (w *Worker) ChooseCandidate(ctx context.Context, id, candidateID int64) err if err := w.store.SetCandidateChosen(ctx, rec.ID, candidateID); err != nil { return fmt.Errorf("choose candidate: %w", err) } - w.log.Info("review: candidate chosen", - "download_id", id, "provider", cand.Provider, "provider_id", cand.ProviderID) + logctx.From(w.scoped(ctx, capReview, id, d.Infohash.String)).Info("review candidate chosen", + "provider", cand.Provider, "provider_id", cand.ProviderID) return nil } @@ -552,7 +564,8 @@ func (w *Worker) SetProviderID(ctx context.Context, id int64, provider, provider w.mu.Lock() defer w.mu.Unlock() - if _, err := w.requireReviewable(ctx, id, "set provider"); err != nil { + d, err := w.requireReviewable(ctx, id, "set provider") + if err != nil { return err } if err := w.store.SetOverride(ctx, id, ovrProvider, provider); err != nil { @@ -561,8 +574,8 @@ func (w *Worker) SetProviderID(ctx context.Context, id int64, provider, provider if err := w.store.SetOverride(ctx, id, ovrProviderID, providerID); err != nil { return fmt.Errorf("set provider: %w", err) } - w.log.Info("review: provider set manually", - "download_id", id, "provider", provider, "provider_id", providerID) + logctx.From(w.scoped(ctx, capReview, id, d.Infohash.String)).Info("review provider set", + "provider", provider, "provider_id", providerID) return nil } @@ -571,7 +584,8 @@ func (w *Worker) ClearProvider(ctx context.Context, id int64) error { w.mu.Lock() defer w.mu.Unlock() - if _, err := w.requireReviewable(ctx, id, "clear provider"); err != nil { + d, err := w.requireReviewable(ctx, id, "clear provider") + if err != nil { return err } if err := w.store.SetOverride(ctx, id, ovrProvider, "none"); err != nil { @@ -580,7 +594,7 @@ func (w *Worker) ClearProvider(ctx context.Context, id int64) error { if err := w.store.SetOverride(ctx, id, ovrProviderID, ""); err != nil { return fmt.Errorf("clear provider: %w", err) } - w.log.Info("review: provider cleared (no metadata base)", "download_id", id) + logctx.From(w.scoped(ctx, capReview, id, d.Infohash.String)).Info("review provider cleared") return nil } @@ -605,6 +619,7 @@ func (w *Worker) ReviewData(ctx context.Context, id int64) (*ReviewData, error) if err != nil { return nil, fmt.Errorf("review data: %w", err) } + log := logctx.From(w.scoped(ctx, capReview, id, d.Infohash.String)) rec, err := w.store.GetCurrentRecognition(ctx, id) if err != nil { return nil, fmt.Errorf("review data: %w", err) @@ -627,14 +642,13 @@ func (w *Worker) ReviewData(ctx context.Context, id int64) (*ReviewData, error) if cands, cerr := w.store.ListCandidatesByRecognition(ctx, rec.ID); cerr == nil { rd.Candidates = cands } else { - w.log.Debug("review data: list candidates failed (skipped)", - "download_id", id, "err", cerr) + log.Debug("review data list candidates failed", "error", cerr) } } if rec != nil && rec.Plan.Valid { var plan recognize.Plan if err := json.Unmarshal([]byte(rec.Plan.String), &plan); err != nil { - w.log.Warn("review data: unmarshal plan failed", "download_id", id, "err", err) + log.Warn("review data unmarshal plan failed", "error", err) } else { plan = applyOverrides(plan, overrides) rd.Plan = plan @@ -645,8 +659,7 @@ func (w *Worker) ReviewData(ctx context.Context, id int64) (*ReviewData, error) if links, lerr := w.layouter.BuildLinks(toLayoutPlan(plan, "", tag)); lerr == nil { rd.Preview = links } else { - w.log.Debug("review data: build preview failed (skipped)", - "download_id", id, "err", lerr) + log.Debug("review data build preview failed", "error", lerr) } } } diff --git a/internal/worker/worker.go b/internal/worker/worker.go index 73e6761..6f5a157 100644 --- a/internal/worker/worker.go +++ b/internal/worker/worker.go @@ -20,11 +20,21 @@ import ( "time" "git.vakhrushev.me/av/jellybit/internal/layout" + "git.vakhrushev.me/av/jellybit/internal/logctx" "git.vakhrushev.me/av/jellybit/internal/qbt" "git.vakhrushev.me/av/jellybit/internal/recognize" "git.vakhrushev.me/av/jellybit/internal/store" ) +// Стадии (capability) — адресуют запись к подсистеме при корреляции по +// download_id. Уровень логов от стадии не зависит. +const ( + capIngest = "ingest" // приём/скачивание (поллинг, reconcile, переходы) + capRecognize = "recognition" // распознавание фильма/сериала + capFileLayout = "file-layout" // раскладка хардлинками + capReview = "review" // ручные команды ревью +) + // Store — нужная worker часть хранилища. type Store interface { ListDownloadsByState(ctx context.Context, states ...store.State) ([]store.Download, error) @@ -147,6 +157,17 @@ func defaultBatchID() string { return fmt.Sprintf("b-%d", time.Now().UnixNano()) } +// scoped кладёт в ctx scoped-логгер загрузки (capability + download_id +// [+ infohash]); стадии и внешние клиенты достают его из ctx и дописывают эти +// ключи на каждую запись сами — без ручного доклеивания download_id. +func (w *Worker) scoped(ctx context.Context, capability string, id int64, infohash string) context.Context { + log := w.log.With("capability", capability, "download_id", id) + if infohash != "" { + log = log.With("infohash", infohash) + } + return logctx.With(ctx, log) +} + // Run крутит цикл поллинга до отмены ctx. func (w *Worker) Run(ctx context.Context) { w.log.Info("worker started", "poll_interval", w.cfg.PollInterval, "category", w.cfg.Category) @@ -167,7 +188,7 @@ func (w *Worker) Run(ctx context.Context) { func (w *Worker) pollOnce(ctx context.Context) { if err := w.Poll(ctx); err != nil { - w.log.Warn("poll failed", "err", err) + w.log.Warn("poll failed", "error", err) } // Ф3: распознаём завершённые загрузки (и перезапускаем по подсказке). if w.recognizer != nil { @@ -209,7 +230,7 @@ func (w *Worker) Poll(ctx context.Context) error { t, ok := byHash[strings.ToLower(d.Infohash.String)] if !ok { w.log.Warn("active download not found in qbittorrent", - "download_id", d.ID, "infohash", d.Infohash.String) + "capability", capIngest, "download_id", d.ID, "infohash", d.Infohash.String) continue } w.reconcile(ctx, d, t) @@ -220,6 +241,7 @@ func (w *Worker) Poll(ctx context.Context) error { // reconcile двигает одну задачу по состоянию её торрента. Вызывается под // w.mu. func (w *Worker) reconcile(ctx context.Context, d store.Download, t qbt.Torrent) { + ctx = w.scoped(ctx, capIngest, d.ID, d.Infohash.String) switch classify(t.State) { case classReady: w.transition(ctx, d, store.StateCompleted, "", "") @@ -237,7 +259,7 @@ func (w *Worker) reconcile(ctx context.Context, d store.Download, t qbt.Torrent) func (w *Worker) checkTimeouts(ctx context.Context, d store.Download, t qbt.Torrent) { created, err := d.CreatedTime() if err != nil { - w.log.Warn("cannot parse created_at", "download_id", d.ID, "value", d.CreatedAt, "err", err) + logctx.From(ctx).Warn("cannot parse created_at", "value", d.CreatedAt, "error", err) return } age := w.now().Sub(created) @@ -254,13 +276,14 @@ func (w *Worker) checkTimeouts(ctx context.Context, d store.Download, t qbt.Torr // transition пишет новое состояние и логирует переход. func (w *Worker) transition(ctx context.Context, d store.Download, state store.State, code, msg string) { + // FromOr, а не From: если вызывающий не завёл scoped-логгер, падаем на + // w.log (настроенный), а не на slog.Default(). + log := logctx.FromOr(ctx, w.log) if err := w.store.SetDownloadState(ctx, d.ID, state, code, msg); err != nil { - w.log.Error("state transition failed", - "download_id", d.ID, "from", d.State, "to", state, "err", err) + log.Error("state transition failed", "from", d.State, "to", state, "error", err) return } - w.log.Info("state transition", - "download_id", d.ID, "from", d.State, "to", state, "code", code) + log.Info("state transition", "from", d.State, "to", state, "code", code) // Пинги — неблокирующе и в отдельном контексте: вызов уходит в сеть, а // мы под w.mu (Notify читает состояние уже после освобождения замка). @@ -277,12 +300,11 @@ func (w *Worker) transition(ctx context.Context, d store.Download, state store.S // новые файлы быстрее появились в проигрывателе. Тоже неблокирующе и вне // w.mu; недоступность Jellyfin не влияет на состояние задачи. if w.scanner != nil && state == store.StateDone { - id := d.ID - go func() { - if err := w.scanner.RefreshLibraries(context.Background()); err != nil { - w.log.Warn("jellyfin: library refresh failed", "download_id", id, "err", err) - } - }() + // Скан Jellyfin — неблокирующе и вне w.mu, в фоновом ctx со scoped-логгером + // (download_id для корреляции ext.*-записи клиента). Недоступность Jellyfin + // на задачу не влияет; ошибку вызова логирует сам клиент (ext.*), здесь гасим. + gctx := w.scoped(context.Background(), capFileLayout, d.ID, d.Infohash.String) + go func() { _ = w.scanner.RefreshLibraries(gctx) }() } } @@ -302,7 +324,7 @@ func (w *Worker) Cancel(ctx context.Context, id int64) error { if err := w.store.SetDownloadState(ctx, id, store.StateCancelled, "", ""); err != nil { return fmt.Errorf("cancel: %w", err) } - w.log.Info("download cancelled", "download_id", id, "from", d.State) + logctx.From(w.scoped(ctx, capReview, id, d.Infohash.String)).Info("download cancelled", "from", d.State) return nil } @@ -331,7 +353,7 @@ func (w *Worker) Retry(ctx context.Context, id int64) error { if err := w.store.SetDownloadState(ctx, id, store.StateDownloading, "", ""); err != nil { return fmt.Errorf("retry: %w", err) } - w.log.Info("download retried", "download_id", id, "from", d.State) + logctx.From(w.scoped(ctx, capReview, id, d.Infohash.String)).Info("download retried", "from", d.State) return nil }