Приём: добавление загрузки по .torrent-файлу
Принимаем .torrent как загруженные байты — через файл-пикер в веб-форме и Telegram-документ, наряду с magnet. Файл несёт полные метаданные: работает там, где magnet не резолвится (закрытые трекеры, без DHT), и даёт максимум контекста для распознавания без сети. - internal/torrent: парсер поверх anacrolix/torrent/metainfo — инфохэш(и) (v1 SHA1 исходных байтов info; v2 BEP52 при наличии) + Context() из имени, дерева файлов, размера, трекеров. Извлечение файлов панико-безопасно (недоверенный вход). - Персистентность байтов: таблица-спутник download_torrent (миграция 0009); пишется в транзакции создания загрузки, только на ветке создания (не при дедупе). Байты живут весь срок строки — нужны для повторного добавления при retry. - ingest: Request.TorrentData/TorrentName, диспетч парсера; source_ref — человекочитаемый референс (имя раздачи/файла), не адрес добавления. - worker: общий sourceAddParts ветвит по source_type в ОБОИХ add-путях — processCatched и Retry (torrent добавляется файлом, не magnet-хешем). - Транспорты: multipart-форма с файл-пикером (деградация без JS) и приём Telegram-документа (скачивание с редактированием токена из ошибок — секрет не в логи; обработка до ветки pending/текста). Разработка по OpenSpec (SDD): change torrent-file-ingest, два чекпоинта ревью (дизайн до кода, код до архива) сабагентами; дельты влиты в спеки, change архивирован. Ручная проверка на живом qBittorrent (7.3) — за деплоем. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Claude Opus 4.8
+87
-25
@@ -15,6 +15,7 @@ import (
|
||||
"git.vakhrushev.me/av/jellybit/internal/logctx"
|
||||
"git.vakhrushev.me/av/jellybit/internal/magnet"
|
||||
"git.vakhrushev.me/av/jellybit/internal/store"
|
||||
"git.vakhrushev.me/av/jellybit/internal/torrent"
|
||||
)
|
||||
|
||||
// capIngest — стадия приёма для поля capability в логах.
|
||||
@@ -28,7 +29,9 @@ type Store interface {
|
||||
// CreateDownloadIfNoActive атомарно проверяет инвариант «одна активная
|
||||
// загрузка на infohash» и заводит задачу; вернувшаяся existing ≠ nil —
|
||||
// дедуп на активную задачу (недостающие хеши вызова метод доносит сам).
|
||||
CreateDownloadIfNoActive(ctx context.Context, d *store.Download, hashes []string) (*store.Download, error)
|
||||
// torrentBlob (для source_type=torrent) пишется в той же транзакции только
|
||||
// на ветке создания; при дедупе не пишется. Для magnet — nil.
|
||||
CreateDownloadIfNoActive(ctx context.Context, d *store.Download, hashes []string, torrentBlob []byte) (*store.Download, error)
|
||||
// AddInfohashes доносит задаче недостающие хеши (guarded). Нужен на
|
||||
// быстром дедуп-пути, который не доходит до CreateDownloadIfNoActive.
|
||||
AddInfohashes(ctx context.Context, downloadID string, hashes []string) error
|
||||
@@ -45,10 +48,13 @@ func New(st Store, log *slog.Logger) *Service {
|
||||
return &Service{store: st, log: log}
|
||||
}
|
||||
|
||||
// Request — входной запрос приёма.
|
||||
// Request — входной запрос приёма. Источник задаётся либо строкой (magnet), либо
|
||||
// байтами `.torrent` (TorrentData); при непустом TorrentData он в приоритете.
|
||||
type Request struct {
|
||||
Source string // пока — magnet-ссылка
|
||||
Context string // подсказка для распознавания (опц.)
|
||||
Source string // magnet-ссылка (когда TorrentData пуст)
|
||||
TorrentData []byte // байты .torrent-файла (опц.); при наличии — источник torrent
|
||||
TorrentName string // имя загруженного файла (опц.); фолбек для source_ref
|
||||
Context string // подсказка для распознавания (опц.)
|
||||
}
|
||||
|
||||
// Result — итог приёма.
|
||||
@@ -64,39 +70,39 @@ type Result struct {
|
||||
// `catched` и сразу возвращает результат. Добавление в qBittorrent и вывод
|
||||
// имени выполняет worker (см. download-tracking).
|
||||
func (s *Service) Ingest(ctx context.Context, req Request) (Result, error) {
|
||||
source := strings.TrimSpace(req.Source)
|
||||
info, err := magnet.Parse(source)
|
||||
src, err := s.parse(req)
|
||||
if err != nil {
|
||||
// Ф1: поддержан только magnet. .torrent/url — следующий заход.
|
||||
return Result{}, fmt.Errorf("ingest: parse source: %w", err)
|
||||
return Result{}, err
|
||||
}
|
||||
|
||||
// Scoped-логгер стадии приёма: download_id допишется после CreateDownload.
|
||||
log := s.log.With("capability", capIngest, "infohash", info.Infohash)
|
||||
log := s.log.With("capability", capIngest, "infohash", src.infohashes[0])
|
||||
ctx = logctx.With(ctx, log)
|
||||
|
||||
// Быстрый дедуп-чек; авторитетная (атомарная) проверка — внутри
|
||||
// CreateDownloadIfNoActive ниже. Дедуп — по ЛЮБОМУ из хешей источника:
|
||||
// гибридный magnet несёт и v1, и v2.
|
||||
if existing, err := s.store.FindActiveByInfohash(ctx, info.Infohashes...); err != nil {
|
||||
// гибридный несёт и v1, и v2.
|
||||
if existing, err := s.store.FindActiveByInfohash(ctx, src.infohashes...); err != nil {
|
||||
return Result{}, fmt.Errorf("ingest: lookup active: %w", err)
|
||||
} else if existing != nil {
|
||||
log.Info("download attached to active", "download_id", existing.ID, "state", existing.State)
|
||||
return s.attached(ctx, info, existing), nil
|
||||
return s.attached(ctx, src, existing), nil
|
||||
}
|
||||
|
||||
// Контекст распознавания дополняем фактами из полей самой magnet-ссылки
|
||||
// (dn/xl/tr/xs/kt) — без сети. Пользовательский текст идёт первым.
|
||||
// Контекст распознавания дополняем фактами из полей источника (magnet-поля
|
||||
// или дерево `.torrent`) — без сети. Пользовательский текст идёт первым.
|
||||
// DisplayName пуст: имя выведет worker на шаге добавления (rename действует
|
||||
// только при добавлении, а тут медленный LLM в пути ответа недопустим).
|
||||
d := &store.Download{
|
||||
SourceType: store.SourceMagnet,
|
||||
SourceRef: source,
|
||||
Context: mergeContext(req.Context, info.Context()),
|
||||
SourceType: src.sourceType,
|
||||
SourceRef: src.sourceRef,
|
||||
Context: mergeContext(req.Context, src.synthContext),
|
||||
State: store.StateCatched,
|
||||
}
|
||||
// Все хеши из magnet (гибридный несёт v1 и v2); kind store выведет по длине.
|
||||
existing, err := s.store.CreateDownloadIfNoActive(ctx, d, info.Infohashes)
|
||||
// Все хеши источника (гибрид несёт v1 и v2); kind store выведет по длине.
|
||||
// torrentBlob непуст только для source_type=torrent — пишется в той же
|
||||
// транзакции лишь на ветке создания.
|
||||
existing, err := s.store.CreateDownloadIfNoActive(ctx, d, src.infohashes, src.torrentBlob)
|
||||
if err != nil {
|
||||
return Result{}, fmt.Errorf("ingest: create download: %w", err)
|
||||
}
|
||||
@@ -107,7 +113,7 @@ func (s *Service) Ingest(ctx context.Context, req Request) (Result, error) {
|
||||
log.Info("download attached to active", "download_id", existing.ID, "state", existing.State)
|
||||
return Result{
|
||||
DownloadID: existing.ID,
|
||||
Infohashes: info.Infohashes,
|
||||
Infohashes: src.infohashes,
|
||||
State: existing.State,
|
||||
Deduplicated: true,
|
||||
}, nil
|
||||
@@ -116,11 +122,67 @@ func (s *Service) Ingest(ctx context.Context, req Request) (Result, error) {
|
||||
log.Info("download catched", "download_id", d.ID)
|
||||
return Result{
|
||||
DownloadID: d.ID,
|
||||
Infohashes: info.Infohashes,
|
||||
Infohashes: src.infohashes,
|
||||
State: store.StateCatched,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// MaxTorrentSize — предел размера принимаемого `.torrent` (защита от разбухания
|
||||
// БД и oversized-загрузок). Реальные торрент-файлы много меньше; крупные (много
|
||||
// файлов → много piece-хешей) отсекаются здесь.
|
||||
const MaxTorrentSize = 8 << 20 // 8 MiB
|
||||
|
||||
// parsedSource — нормализованный источник приёма (magnet или .torrent).
|
||||
type parsedSource struct {
|
||||
sourceType store.SourceType
|
||||
sourceRef string // референс для человека/логов (magnet-URI или имя torrent)
|
||||
infohashes []string // все хеши источника; v1 раньше v2
|
||||
synthContext string // синтез контекста из полей источника (без сети)
|
||||
torrentBlob []byte // байты .torrent (для source_type=torrent); иначе nil
|
||||
}
|
||||
|
||||
// parse разбирает источник запроса: при непустом TorrentData — как `.torrent`
|
||||
// (в приоритете), иначе — как magnet. Синтез контекста и хеши берутся из полей
|
||||
// источника, без сети.
|
||||
func (s *Service) parse(req Request) (parsedSource, error) {
|
||||
if len(req.TorrentData) > 0 {
|
||||
if len(req.TorrentData) > MaxTorrentSize {
|
||||
return parsedSource{}, fmt.Errorf("ingest: torrent too large: %d > %d bytes", len(req.TorrentData), MaxTorrentSize)
|
||||
}
|
||||
info, err := torrent.Parse(req.TorrentData)
|
||||
if err != nil {
|
||||
return parsedSource{}, fmt.Errorf("ingest: parse torrent: %w", err)
|
||||
}
|
||||
// SourceRef — человекочитаемый референс (имя раздачи), НЕ адрес
|
||||
// добавления: torrent добавляется байтами (см. worker), не по SourceRef.
|
||||
// Фолбек на имя файла, если у раздачи нет содержательного имени
|
||||
// (пустое или NoName-сентинел "-").
|
||||
ref := strings.TrimSpace(info.DisplayName)
|
||||
if ref == "" || ref == "-" {
|
||||
ref = strings.TrimSpace(req.TorrentName)
|
||||
}
|
||||
return parsedSource{
|
||||
sourceType: store.SourceTorrent,
|
||||
sourceRef: ref,
|
||||
infohashes: info.Infohashes,
|
||||
synthContext: info.Context(),
|
||||
torrentBlob: req.TorrentData,
|
||||
}, nil
|
||||
}
|
||||
|
||||
source := strings.TrimSpace(req.Source)
|
||||
info, err := magnet.Parse(source)
|
||||
if err != nil {
|
||||
return parsedSource{}, fmt.Errorf("ingest: parse source: %w", err)
|
||||
}
|
||||
return parsedSource{
|
||||
sourceType: store.SourceMagnet,
|
||||
sourceRef: source,
|
||||
infohashes: info.Infohashes,
|
||||
synthContext: info.Context(),
|
||||
}, nil
|
||||
}
|
||||
|
||||
// mergeContext склеивает контекст от транспорта с синтезом из полей magnet:
|
||||
// пользовательский текст идёт первым, затем факты из ссылки. Пустые части
|
||||
// опускаются; при пустых обеих — пустая строка (пустой контекст допустим).
|
||||
@@ -140,15 +202,15 @@ func mergeContext(userText, synth string) string {
|
||||
// принести хеш, которого задача ещё не знает; guarded-путь через
|
||||
// CreateDownloadIfNoActive сюда не доходит). Донос — best-effort: конфликт
|
||||
// хеша с другой активной задачей логируется, приём не валится.
|
||||
func (s *Service) attached(ctx context.Context, info magnet.Info, existing *store.Download) Result {
|
||||
if len(info.Infohashes) > len(existing.Infohashes) {
|
||||
if err := s.store.AddInfohashes(ctx, existing.ID, info.Infohashes); err != nil {
|
||||
func (s *Service) attached(ctx context.Context, src parsedSource, existing *store.Download) Result {
|
||||
if len(src.infohashes) > len(existing.Infohashes) {
|
||||
if err := s.store.AddInfohashes(ctx, existing.ID, src.infohashes); err != nil {
|
||||
logctx.FromOr(ctx, s.log).Warn("ingest top-up infohashes failed", "error", err)
|
||||
}
|
||||
}
|
||||
return Result{
|
||||
DownloadID: existing.ID,
|
||||
Infohashes: info.Infohashes,
|
||||
Infohashes: src.infohashes,
|
||||
State: existing.State,
|
||||
Deduplicated: true,
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user