Skip to content

Automation

Higher-level helpers built on the client: monitoring the current user's torrents, cross-seeding a release to another tracker, and cross-uploading one. These are exported from the top-level pygazelle package.

Monitoring

Watch the current user's uploaded and snatched torrents for deletion or trump. Construct a monitor with client.monitor(...).

pygazelle.TorrentMonitor 

TorrentMonitor(
    client: GazelleClient,
    *,
    sources: tuple[UserTorrentType, ...] = (
        "uploaded",
        "snatched",
    ),
    page_size: int = 50,
)

Watches the current user's torrent lists and reports deletions/trumps.

Stateless from the caller's view: each poll() returns the changes since the previous snapshot. The monitor runs no loop, timer, or callbacks — the caller controls cadence. Build one via client.monitor(...).

Source code in src/pygazelle/monitoring.py 
30
31
32
33
34
35
36
37
38
39
40
41
def __init__(
    self,
    client: GazelleClient,
    *,
    sources: tuple[UserTorrentType, ...] = ("uploaded", "snatched"),
    page_size: int = 50,
) -> None:
    self._client = client
    self._sources = tuple(sources)
    self._page_size = page_size
    self._user_id: int | None = None
    self._snapshot: MonitorSnapshot | None = None

poll async 

poll() -> list[TorrentChangeEvent]
Source code in src/pygazelle/monitoring.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
async def poll(self) -> list[TorrentChangeEvent]:
    if self._user_id is None:
        self._user_id = (await self._client.user.me()).id

    current = MonitorSnapshot()
    for source in self._sources:
        current.sources[source] = await self._fetch_source(self._user_id, source)

    if self._snapshot is None:
        self._snapshot = current
        return []

    events: list[TorrentChangeEvent] = []
    for source in self._sources:
        prior = self._snapshot.sources.get(source, {})
        now = current.sources.get(source, {})
        for torrent_id, entry in prior.items():
            if torrent_id not in now:
                events.append(await self._classify(source, entry))

    # Atomic commit: only advance the snapshot once every fetch + lookup
    # above succeeded. A mid-poll error leaves the prior snapshot intact.
    self._snapshot = current
    return events

dump_state 

dump_state() -> dict[str, object] | None

A json-serializable snapshot, or None before the first poll.

Source code in src/pygazelle/monitoring.py 
127
128
129
130
131
def dump_state(self) -> dict[str, object] | None:
    """A json-serializable snapshot, or None before the first poll."""
    if self._snapshot is None:
        return None
    return self._snapshot.model_dump(mode="json")

load_state 

load_state(state: dict[str, object]) -> None

Restore a snapshot previously produced by dump_state().

Source code in src/pygazelle/monitoring.py 
133
134
135
def load_state(self, state: dict[str, object]) -> None:
    """Restore a snapshot previously produced by ``dump_state()``."""
    self._snapshot = MonitorSnapshot.model_validate(state)

pygazelle.TorrentChangeEvent

Bases: GazelleModel

A classified change to a previously-watched torrent.

kind instance-attribute 

kind: ChangeKind

source instance-attribute 

source: str

torrent_id instance-attribute 

torrent_id: int

group_id instance-attribute 

group_id: int

name instance-attribute 

name: str

replacement_torrent_id class-attribute instance-attribute 

replacement_torrent_id: int | None = None

Cross-seed

Find the same release on a target tracker (metadata match + strict file-list verification) and fetch its .torrent.

pygazelle.cross_seed async 

cross_seed(
    source_client: GazelleClient,
    source_torrent_id: int,
    target_client: GazelleClient,
    *,
    max_deep_checks: int = DEFAULT_MAX_DEEP_CHECKS,
) -> CrossSeedResult | None

Find source_torrent_id (on source_client) on target_client and return the target's .torrent. Returns None when the source has no file list or no candidate exactly matches.

Source code in src/pygazelle/crossseed.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
async def cross_seed(
    source_client: GazelleClient,
    source_torrent_id: int,
    target_client: GazelleClient,
    *,
    max_deep_checks: int = DEFAULT_MAX_DEEP_CHECKS,
) -> CrossSeedResult | None:
    """Find ``source_torrent_id`` (on ``source_client``) on ``target_client`` and
    return the target's .torrent. Returns None when the source has no file list
    or no candidate exactly matches.
    """
    source = await source_client.torrents.get(source_torrent_id)
    if not source.files:
        logger.info("cross-seed: source torrent %d has no file list", source_torrent_id)
        return None

    candidates = await find_candidates(source, target_client, max_deep_checks=max_deep_checks)
    for candidate in candidates:
        if verify_match(source, candidate):
            torrent_file = await target_client.torrents.download(candidate.id)
            return CrossSeedResult(
                match=candidate,
                torrent_file=torrent_file,
                source_torrent_id=source_torrent_id,
                target_torrent_id=candidate.id,
            )
    return None

pygazelle.cross_seed_sync 

cross_seed_sync(
    source_client: GazelleSyncClient,
    source_torrent_id: int,
    target_client: GazelleSyncClient,
    *,
    max_deep_checks: int | None = None,
) -> CrossSeedResult | None

Synchronous cross-seed: runs the async cross_seed on the source client's background loop and returns the result (or None) directly, no await.

Source code in src/pygazelle/sync.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
def cross_seed_sync(
    source_client: GazelleSyncClient,
    source_torrent_id: int,
    target_client: GazelleSyncClient,
    *,
    max_deep_checks: int | None = None,
) -> CrossSeedResult | None:
    """Synchronous cross-seed: runs the async cross_seed on the source client's
    background loop and returns the result (or None) directly, no await.
    """
    from .crossseed import DEFAULT_MAX_DEEP_CHECKS, cross_seed

    return source_client._bg.run(  # pyright: ignore[reportPrivateUsage]
        cross_seed(
            source_client._async,  # pyright: ignore[reportPrivateUsage]
            source_torrent_id,
            target_client._async,  # pyright: ignore[reportPrivateUsage]
            max_deep_checks=DEFAULT_MAX_DEEP_CHECKS if max_deep_checks is None else max_deep_checks,
        )
    )

pygazelle.find_candidates async 

find_candidates(
    source: Torrent,
    target_client: GazelleClient,
    *,
    max_deep_checks: int = DEFAULT_MAX_DEEP_CHECKS,
) -> list[Torrent]

Search the target tracker by the source's artist/album, cheaply pre-filter candidates on format/encoding/media/size/file_count, then fetch the full file list for each survivor (bounded by max_deep_checks).

Source code in src/pygazelle/crossseed.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
async def find_candidates(
    source: Torrent,
    target_client: GazelleClient,
    *,
    max_deep_checks: int = DEFAULT_MAX_DEEP_CHECKS,
) -> list[Torrent]:
    """Search the target tracker by the source's artist/album, cheaply pre-filter
    candidates on format/encoding/media/size/file_count, then fetch the full
    file list for each survivor (bounded by ``max_deep_checks``).
    """
    artist = source.group.artists[0].name if source.group and source.group.artists else None
    album = source.group.name if source.group else None

    results = []
    if artist:
        params: dict[str, str | int] = {"artistname": artist}
        if album:
            params["groupname"] = album
        results = await target_client.torrents.search("", **params)
    if not results and album:
        results = await target_client.torrents.search("", groupname=album)

    candidate_ids: list[int] = []
    for group in results:
        for row in group.torrents:
            if (
                row.format == source.format
                and row.encoding == source.encoding
                and row.media == source.media
                and row.size == source.size
                and row.file_count == source.file_count
            ):
                candidate_ids.append(row.torrent_id)

    if len(candidate_ids) > max_deep_checks:
        logger.warning(
            "cross-seed: %d candidates exceed the deep-check cap (%d); checking only the first %d",
            len(candidate_ids),
            max_deep_checks,
            max_deep_checks,
        )
        candidate_ids = candidate_ids[:max_deep_checks]

    candidates: list[Torrent] = []
    for cid in candidate_ids:
        try:
            candidates.append(await target_client.torrents.get(cid))
        except GazelleNotFoundError:
            continue
    return candidates

pygazelle.verify_match 

verify_match(source: Torrent, candidate: Torrent) -> bool

Strict cross-seed match: identical top-level folder and identical sorted (path, size) file list. Empty file lists never match.

Source code in src/pygazelle/crossseed.py 
27
28
29
30
31
32
33
34
35
36
37
def verify_match(source: Torrent, candidate: Torrent) -> bool:
    """Strict cross-seed match: identical top-level folder and identical sorted
    (path, size) file list. Empty file lists never match.
    """
    if source.file_path != candidate.file_path:
        return False
    source_files = sorted((f.path, f.size) for f in source.files)
    if not source_files:
        return False
    candidate_files = sorted((f.path, f.size) for f in candidate.files)
    return source_files == candidate_files

pygazelle.CrossSeedResult dataclass 

CrossSeedResult(
    match: Torrent,
    torrent_file: bytes,
    source_torrent_id: int,
    target_torrent_id: int,
    confidence: Literal["exact"] = "exact",
)

A confirmed cross-seed match plus the target tracker's .torrent bytes.

match instance-attribute 

match: Torrent

torrent_file instance-attribute 

torrent_file: bytes

source_torrent_id instance-attribute 

source_torrent_id: int

target_torrent_id instance-attribute 

target_torrent_id: int

confidence class-attribute instance-attribute 

confidence: Literal['exact'] = 'exact'

Cross-upload

Two-phase upload of a release to another tracker: a read-only prepare_upload (map metadata, detect duplicates, build a draft) followed by an explicit, duplicate-gated submit_upload. The caller builds the target .torrent; the library exposes the target announce URL via client.user.announce_url().

pygazelle.prepare_upload async 

prepare_upload(
    source_client: GazelleClient,
    source_torrent_id: int,
    target_client: GazelleClient,
    *,
    torrent_file: bytes,
) -> UploadDraft

Read-only: map metadata + detect duplicates + build an UploadDraft. No write.

Source code in src/pygazelle/crossupload.py 
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
async def prepare_upload(
    source_client: GazelleClient,
    source_torrent_id: int,
    target_client: GazelleClient,
    *,
    torrent_file: bytes,
) -> UploadDraft:
    """Read-only: map metadata + detect duplicates + build an UploadDraft. No write."""
    source = await source_client.torrents.get(source_torrent_id)
    target = _tracker_kind(target_client)
    mapped = map_metadata(source, target)
    try:
        duplicates = await duplicate_check(source, target_client)
    except GazelleError as exc:
        duplicates = []
        mapped.warnings.append(f"duplicate check failed: {exc}")
    return UploadDraft(
        form=mapped.fields,
        unmapped=mapped.unmapped,
        warnings=mapped.warnings,
        duplicates=duplicates,
        torrent_file=torrent_file,
        source_torrent_id=source_torrent_id,
        target_tracker=target,
    )

pygazelle.submit_upload async 

submit_upload(
    target_client: GazelleClient,
    draft: UploadDraft,
    *,
    allow_duplicate: bool = False,
) -> UploadResult

The live write: validate the draft, gate on exact duplicates, then POST action=upload. Refuses (no write) on missing required fields or an unallowed exact duplicate.

Source code in src/pygazelle/crossupload.py 
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
async def submit_upload(
    target_client: GazelleClient,
    draft: UploadDraft,
    *,
    allow_duplicate: bool = False,
) -> UploadResult:
    """The live write: validate the draft, gate on exact duplicates, then POST
    action=upload. Refuses (no write) on missing required fields or an unallowed
    exact duplicate."""
    missing = _missing_required(draft)
    if missing:
        raise ValueError(f"cannot submit: missing required field(s): {', '.join(missing)}")
    if not allow_duplicate and any(d.kind == "exact" for d in draft.duplicates):
        raise ValueError(
            "an exact duplicate exists on the target; pass allow_duplicate=True to override"
        )
    files = [
        ("file_input", ("upload.torrent", draft.torrent_file))
    ]  # VERIFY upload file field name
    data = await target_client._transport.request_write(  # pyright: ignore[reportPrivateUsage]
        "upload", data=dict(draft.form), files=files
    )
    # VERIFY response keys for the new torrent/group id.
    torrent_id = int(data.get("torrentid") or data.get("torrentId") or 0)
    group_id = int(data.get("groupid") or data.get("groupId") or 0)
    return UploadResult(
        torrent_id=torrent_id,
        group_id=group_id,
        url=f"torrents.php?id={group_id}&torrentid={torrent_id}",
    )

pygazelle.prepare_upload_sync 

prepare_upload_sync(
    source_client: GazelleSyncClient,
    source_torrent_id: int,
    target_client: GazelleSyncClient,
    *,
    torrent_file: bytes,
) -> UploadDraft
Source code in src/pygazelle/sync.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
def prepare_upload_sync(
    source_client: GazelleSyncClient,
    source_torrent_id: int,
    target_client: GazelleSyncClient,
    *,
    torrent_file: bytes,
) -> UploadDraft:
    from .crossupload import prepare_upload

    return source_client._bg.run(  # pyright: ignore[reportPrivateUsage]
        prepare_upload(
            source_client._async,  # pyright: ignore[reportPrivateUsage]
            source_torrent_id,
            target_client._async,  # pyright: ignore[reportPrivateUsage]
            torrent_file=torrent_file,
        )
    )

pygazelle.submit_upload_sync 

submit_upload_sync(
    target_client: GazelleSyncClient,
    draft: UploadDraft,
    *,
    allow_duplicate: bool = False,
) -> UploadResult
Source code in src/pygazelle/sync.py 
165
166
167
168
169
170
171
172
173
174
175
def submit_upload_sync(
    target_client: GazelleSyncClient,
    draft: UploadDraft,
    *,
    allow_duplicate: bool = False,
) -> UploadResult:
    from .crossupload import submit_upload

    return target_client._bg.run(  # pyright: ignore[reportPrivateUsage]
        submit_upload(target_client._async, draft, allow_duplicate=allow_duplicate)  # pyright: ignore[reportPrivateUsage]
    )

pygazelle.map_metadata 

map_metadata(
    source: Torrent, target: TrackerKind
) -> MappedForm

Best-effort map of a source release's metadata to the target upload schema.

Confident fields go to fields; anything unmappable/uncertain is recorded in unmapped + warnings and never guessed.

Source code in src/pygazelle/crossupload.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
def map_metadata(source: Torrent, target: TrackerKind) -> MappedForm:
    """Best-effort map of a source release's metadata to the target upload schema.

    Confident fields go to ``fields``; anything unmappable/uncertain is recorded in
    ``unmapped`` + ``warnings`` and never guessed.
    """
    out = MappedForm()
    group = source.group
    if group is not None:
        out.fields["title"] = group.name
        out.fields["year"] = group.year
        out.fields["artists"] = [a.name for a in group.artists]
        if group.tags:
            out.fields["tags"] = list(group.tags)
            out.warnings.append("tags carried over verbatim; review against the target's tag rules")
        rt = group.release_type
        mapped_rt = RELEASE_TYPE_MAP.get(target, {}).get(rt) if rt is not None else None
        if mapped_rt is not None:
            out.fields["release_type"] = mapped_rt
        else:
            out.unmapped.append("release_type")
            out.warnings.append(f"release type {rt} has no {target} equivalent; set it manually")
    else:
        # No group metadata on the source: flag the group-derived fields for the
        # caller rather than silently omitting them.
        out.unmapped.extend(["title", "year", "artists", "release_type"])
        out.warnings.append(
            "source has no group metadata; title/year/artists/release_type must be set manually"
        )
    out.fields["format"] = source.format
    out.fields["bitrate"] = source.encoding  # Gazelle 'bitrate' carries the encoding value
    out.fields["media"] = source.media
    return out

pygazelle.duplicate_check async 

duplicate_check(
    source: Torrent, target_client: GazelleClient
) -> list[DuplicateMatch]

Search the target for releases matching the source; classify each as an exact duplicate (file-list match) or a possible duplicate (same group/metadata).

Source code in src/pygazelle/crossupload.py 
136
137
138
139
140
141
142
143
144
145
146
async def duplicate_check(source: Torrent, target_client: GazelleClient) -> list[DuplicateMatch]:
    """Search the target for releases matching the source; classify each as an
    exact duplicate (file-list match) or a possible duplicate (same group/metadata)."""
    candidates = await find_candidates(source, target_client)
    matches: list[DuplicateMatch] = []
    for cand in candidates:
        kind: DuplicateKind = "exact" if verify_match(source, cand) else "possible"
        group_id = cand.group.id if cand.group else 0
        name = cand.group.name if cand.group else ""
        matches.append(DuplicateMatch(torrent_id=cand.id, group_id=group_id, kind=kind, name=name))
    return matches

pygazelle.UploadDraft dataclass 

UploadDraft(
    form: dict[str, Any],
    unmapped: list[str],
    warnings: list[str],
    duplicates: list[DuplicateMatch],
    torrent_file: bytes,
    source_torrent_id: int,
    target_tracker: TrackerKind,
)

Mutable: the caller fills form for any unmapped fields before submit.

form instance-attribute 

form: dict[str, Any]

unmapped instance-attribute 

unmapped: list[str]

warnings instance-attribute 

warnings: list[str]

duplicates instance-attribute 

duplicates: list[DuplicateMatch]

torrent_file instance-attribute 

torrent_file: bytes

source_torrent_id instance-attribute 

source_torrent_id: int

target_tracker instance-attribute 

target_tracker: TrackerKind

pygazelle.UploadResult dataclass 

UploadResult(torrent_id: int, group_id: int, url: str)

torrent_id instance-attribute 

torrent_id: int

group_id instance-attribute 

group_id: int

url instance-attribute 

url: str

pygazelle.DuplicateMatch dataclass 

DuplicateMatch(
    torrent_id: int,
    group_id: int,
    kind: DuplicateKind,
    name: str,
)

torrent_id instance-attribute 

torrent_id: int

group_id instance-attribute 

group_id: int

kind instance-attribute 

kind: DuplicateKind

name instance-attribute 

name: str