av/jellybit
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Логирование: ревью всего кода и рефакторинг в соответствии с конвенциями

Browse Source
This commit is contained in:
av
2026-06-28 20:13:40 +03:00
parent c739a20749
commit 9cfccc7b4a
24 changed files with 473 additions and 203 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/worker/discover.go
+4 -4
View File
@@ -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)
}
internal/worker/review.go
+48 -35
View File
@@ -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)
}
}
}
internal/worker/worker.go
+37 -15
View File
@@ -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
}