Qbitmaid

qbit-maid

Development ¹ of qbitmaid was over the course of several months. At first, the project was called qbit-clean and didn't have all the features the project has now. The issue was mainly with my download cache in unraid being filled with torrents I no longer needed to seed². When I would get a notification from the server that the download cache was 95% full I would have to manually go to qbittorrent, sort the torrents by age and remove the ones older than two weeks avoiding torrents I wanted to keep.

This was tedious. Very tedious. So I went off to do more work just to avoid a little.qbitmaid.py is the main file that glues the project together. This was my first project where I heavily abstracted the design. More on this later. First, we'll connect to the API. I used an existing client library that makes this process simpler than writing your own client.

qbitmaid.py³

...
class Qbt:
    def __init__(self):
        """Main object, should be calling functions from qlist.py, qlogging.py and qprocess.py"""
    ...
        #logging in
        try:
            self.tl.info('Connecting to host.')
            self.qbt_client.auth_log_in()
            self.tl.info('Connected.')
        except qbittorrentapi.APIError as e:
            self.tl.exception(e)
            self.po.send_message(e, title="qbit-maid API ERROR")
    ...

Then we use the api to make an list of the torrents:

        # Pulling all torrent data
        self.torrent_list = self.qbt_client.torrents_info()

Next, we "sift" out torrents to be deleted. This was created with a positive sieve meaning we specify positive scenarios. In other words, I know which torrents I want to keep as opposed to the torrents I don't want. Theres pros and cons to both scenarios however in the long term a positive sieve is less work.

qlist.py has functions at the bottom of the file that are referenced in the conditions. This very method of programming made it easy to write unit tests as I went.

if is_preme(torrent['seeding_time'], self.min_age):
            continue

def is_preme(seeding_time, minage):
    if seeding_time <= minage:
        return True

When it comes across an item that meets certain criteria it will skip it. For instance, the example above checks to see if it's too soon to remove a torrent. This is because some trackers require a minimum seed time. If you were to remove a torrent sooner than they require, it could lead to getting kicked.

qlist.py Has a couple jobs:

Tag torrents according to how they should be treated.
Sort

qlist.py

...
def build_tor_list(self):
   while self.torrent_list:
       ...
       if is_tracker_blank(torrent['tracker']):
           ...
           continue
       elif is_cat_ignored(torrent['category'], self.cat_whitelist.values()):
           ...
           continue
       elif is_ignored_tag(self.ignored_tags.values(),torrent['tags']):
           ...
           continue
       if is_tag_blank(torrent['tags']):
           ...
           if is_protected_tracker(torrent['tracker'], self.tracker_whitelist.values()):
               self.qbt_client.torrents_add_tags(self.tracker_protected_tag,torrent['hash'])
           elif is_not_protected_tracker(torrent['tracker'], self.tracker_whitelist.values()):
               self.qbt_client.torrents_add_tags(self.tracker_non_protected_tag,torrent['hash'])
       if is_preme(torrent['seeding_time'], self.min_age):
           continue
       elif is_protected_tracker(torrent['tracker'], self.tracker_whitelist.values()):
           if is_tag_blank(torrent['tags']):
               self.qbt_client.torrents_add_tags(self.tracker_protected_tag,torrent['hash'])
           ...
           self.tracker_list.append(torrent)
       elif is_not_protected_tracker(torrent['tracker'], self.tracker_whitelist.values()):
           if is_tag_blank(torrent['tags']):
               self.qbt_client.torrents_add_tags(self.tracker_non_protected_tag,torrent['hash'])
           ...
           self.tracker_list.append(torrent)

In this case the items I want to keep stays. qlist.py passes the data over to qprocess.py. This is done through about 2 layers of abstraction. Unfortunatley, this type of programming makes it difficult to follow.

qprocess.py has four jobs:

Collect telemetry
Delete torrents if needed
Enable debugging if needed

qprocess.py

def tor_processor(self):
    """Main logic to sort through both self.tracker_nonprotected_list and self.tracker_protected_list
    If torrent meets criteria for deletion, its infohash_v1 will be appended to self.torrent_hash_delete_list
    """
    for canidate in self.tracker_list:
        if self.enable_telemetry:
            header = ['state','ratio','tags','added','hash','name','tracker']
            row = [canidate['state'],canidate['ratio'],canidate["tags"],canidate['added_on'],canidate['infohash_v1'],canidate["name"][0:20],canidate['tracker']]
            write_csv(self.cv,self.telemetry_outfile,header,row)
        ...
        elif is_protected_over_ratio(canidate['ratio'], 1.05, self.tracker_protected_tag, canidate["tags"]):
            if self.use_log:
                self.tl.debug(f'["{canidate["name"][0:20]}..."] is above a 1.05 ratio({canidate["ratio"]}).')
            self.torrent_hash_delete_list.append(canidate['infohash_v1'])
        ...
        elif is_not_protected_tor(self.tracker_non_protected_tag, canidate["tags"]):
            self.torrent_hash_delete_list.append(canidate['infohash_v1'])
        ...
        else:
            if self.enable_dragnet:
                header = ['state','ratio','tags','added','thash','tname','trname']
                row = [canidate['state'],canidate['ratio'],canidate["tags"],canidate['added_on'],canidate['infohash_v1'],canidate["name"][0:20],canidate['tracker']]
                write_csv(self.cv,self.dragnet_outfile,header,row)
            continue

I package this in a docker file:

FROM python:alpine3.18
WORKDIR /
COPY . opt
RUN apk add --no-cache supercronic
RUN pip install requests
RUN pip install qbittorrent-api
RUN chmod +x /opt/entrypoint.sh
CMD ["/opt/entrypoint.sh"]

Then use Drone to package this into a container. This pushes the container to an OCI repo in gitea. The application is configured through a toml file:

[qbittorrent]
host = "192.168.x.x"
port = 8080
username = "user"
password = "pass"

...

[healthcheck]
use_healthcheck = true
healthcheck_url = "https://example.com/ping/<uuid>>"

Using unraid has honestly been a delight. I had some performance issues but that was due to how I was using the storage pool.

unraid

Finally, this same container will run the test cases in test_qbitmaid.py. This is handled by drone. So eachtime I push new code to a development branch on gitea, it creates a container to test and tests the code. Once I see that it has passed, I can merge the code to the main branch.

Final Notes

I have been using this for over 2 years. It was a huge learning experience and my coding practices have evolved over my newer projects. While I did make this for my use mainly, feel free to try it out! If you have any questions, you can open an issue here.

Footnotes

The Source Code can be found here. ↩
Private trackers require you to seed a torrent for a period of time. In my case, I have to seed for about 2 weeks or to a ratio of 1. ↩
Code has been removed for examples in this article. ↩