Allow configuration of producer urls (#203)

* Allow configuration of urls in scrapers by mounting the scrapers.json file over the one in the container

* version bump

deployment/docker/config/qbit/qbittorrent.conf

@@ -14,12 +14,14 @@ program=
 [BitTorrent]
 Session\AnonymousModeEnabled=true
 Session\BTProtocol=TCP
+Session\ConnectionSpeed=150
 Session\DefaultSavePath=/downloads/
 Session\ExcludedFileNames=
-Session\MaxActiveCheckingTorrents=5
-Session\MaxActiveDownloads=10
+Session\MaxActiveCheckingTorrents=20
+Session\MaxActiveDownloads=20
 Session\MaxActiveTorrents=50
 Session\MaxActiveUploads=50
+Session\MaxConcurrentHTTPAnnounces=1000
 Session\MaxConnections=2000
 Session\Port=6881
 Session\QueueingSystemEnabled=true

deployment/docker/docker-compose.yaml

@@ -94,7 +94,7 @@ services:
         condition: service_healthy
     env_file: stack.env
     hostname: knightcrawler-addon
-    image: gabisonfire/knightcrawler-addon:2.0.20
+    image: gabisonfire/knightcrawler-addon:2.0.23
     labels:
       logging: promtail
     networks:
@@ -117,7 +117,7 @@ services:
       redis:
         condition: service_healthy
     env_file: stack.env
-    image: gabisonfire/knightcrawler-consumer:2.0.20
+    image: gabisonfire/knightcrawler-consumer:2.0.23
     labels:
       logging: promtail
     networks:
@@ -138,7 +138,7 @@ services:
       redis:
         condition: service_healthy
     env_file: stack.env
-    image: gabisonfire/knightcrawler-debrid-collector:2.0.20
+    image: gabisonfire/knightcrawler-debrid-collector:2.0.23
     labels:
       logging: promtail
     networks:
@@ -152,7 +152,7 @@ services:
       migrator:
         condition: service_completed_successfully
     env_file: stack.env
-    image: gabisonfire/knightcrawler-metadata:2.0.20
+    image: gabisonfire/knightcrawler-metadata:2.0.23
     networks:
       - knightcrawler-network
     restart: "no"
@@ -163,7 +163,7 @@ services:
       postgres:
         condition: service_healthy
     env_file: stack.env
-    image: gabisonfire/knightcrawler-migrator:2.0.20
+    image: gabisonfire/knightcrawler-migrator:2.0.23
     networks:
       - knightcrawler-network
     restart: "no"
@@ -182,7 +182,7 @@ services:
       redis:
         condition: service_healthy
     env_file: stack.env
-    image: gabisonfire/knightcrawler-producer:2.0.20
+    image: gabisonfire/knightcrawler-producer:2.0.23
     labels:
       logging: promtail
     networks:
@@ -207,7 +207,7 @@ services:
     deploy:
       replicas: ${QBIT_REPLICAS:-0}
     env_file: stack.env
-    image: gabisonfire/knightcrawler-qbit-collector:2.0.20
+    image: gabisonfire/knightcrawler-qbit-collector:2.0.23
     labels:
       logging: promtail
     networks:

deployment/docker/src/components/knightcrawler.yaml

@@ -20,7 +20,7 @@ x-depends: &knightcrawler-app-depends
 
 services:
   metadata:
-    image: gabisonfire/knightcrawler-metadata:2.0.20
+    image: gabisonfire/knightcrawler-metadata:2.0.23
     env_file: ../../.env
     networks:
       - knightcrawler-network
@@ -30,7 +30,7 @@ services:
         condition: service_completed_successfully
 
   migrator:
-    image: gabisonfire/knightcrawler-migrator:2.0.20
+    image: gabisonfire/knightcrawler-migrator:2.0.23
     env_file: ../../.env
     networks:
       - knightcrawler-network
@@ -40,7 +40,7 @@ services:
         condition: service_healthy
 
   addon:
-    image: gabisonfire/knightcrawler-addon:2.0.20
+    image: gabisonfire/knightcrawler-addon:2.0.23
     <<: [*knightcrawler-app, *knightcrawler-app-depends]
     restart: unless-stopped
     hostname: knightcrawler-addon
@@ -48,22 +48,22 @@ services:
       - "7000:7000"
 
   consumer:
-    image: gabisonfire/knightcrawler-consumer:2.0.20
+    image: gabisonfire/knightcrawler-consumer:2.0.23
     <<: [*knightcrawler-app, *knightcrawler-app-depends]
     restart: unless-stopped
 
   debridcollector:
-    image: gabisonfire/knightcrawler-debrid-collector:2.0.20
+    image: gabisonfire/knightcrawler-debrid-collector:2.0.23
     <<: [*knightcrawler-app, *knightcrawler-app-depends]
     restart: unless-stopped
         
   producer:
-    image: gabisonfire/knightcrawler-producer:2.0.20
+    image: gabisonfire/knightcrawler-producer:2.0.23
     <<: [*knightcrawler-app, *knightcrawler-app-depends]
     restart: unless-stopped
 
   qbitcollector:
-    image: gabisonfire/knightcrawler-qbit-collector:2.0.20
+    image: gabisonfire/knightcrawler-qbit-collector:2.0.23
     <<: [*knightcrawler-app, *knightcrawler-app-depends]
     restart: unless-stopped
     depends_on:

deployment/docker/stack.env

@@ -32,6 +32,7 @@ COLLECTOR_DEBRID_ENABLED=true
 COLLECTOR_REAL_DEBRID_API_KEY=
 QBIT_HOST=http://qbittorrent:8080
 QBIT_TRACKERS_URL=https://raw.githubusercontent.com/ngosang/trackerslist/master/trackers_all_http.txt
+QBIT_CONCURRENCY=8
 
 # Number of replicas for the qBittorrent collector and qBitTorrent client. Should be 0 or 1.
 QBIT_REPLICAS=0

src/migrator/migrations/014_file_subtitle_infohash_fileidx_index.sql

@@ -0,0 +1,43 @@
+-- Drop Duplicate Files in Files Table
+DELETE FROM public.files
+WHERE id NOT IN (
+    SELECT MAX(id)
+    FROM public.files
+    GROUP BY "infoHash", "fileIndex"
+);
+
+-- Add Index to files table
+DO $$
+BEGIN
+    IF NOT EXISTS (
+        SELECT 1
+        FROM pg_constraint
+        WHERE conname = 'files_unique_infohash_fileindex'
+    ) THEN
+        ALTER TABLE public.files
+            ADD CONSTRAINT files_unique_infohash_fileindex UNIQUE ("infoHash", "fileIndex");
+    END IF;
+END $$;
+
+
+-- Drop Duplicate subtitles in Subtitles Table
+DELETE FROM public.subtitles
+WHERE id NOT IN (
+    SELECT MAX(id)
+    FROM public.subtitles
+    GROUP BY "infoHash", "fileIndex"
+);
+
+-- Add Index to subtitles table
+DO $$
+    BEGIN
+        IF NOT EXISTS (
+            SELECT 1
+            FROM pg_constraint
+            WHERE conname = 'subtitles_unique_infohash_fileindex'
+        ) THEN
+            ALTER TABLE public.subtitles
+                ADD CONSTRAINT subtitles_unique_infohash_fileindex UNIQUE ("infoHash", "fileIndex");
+        END IF;
+    END $$;
+

src/producer/src/Configuration/scrapers.json

@@ -4,27 +4,34 @@
       {
         "Name": "SyncEzTvJob",
         "IntervalSeconds": 60,
-        "Enabled": true
+        "Enabled": true,
+        "Url": "https://eztv1.xyz/ezrss.xml",
+        "XmlNamespace": "http://xmlns.ezrss.it/0.1/"
       },
       {
         "Name": "SyncNyaaJob",
         "IntervalSeconds": 60,
-        "Enabled": true
+        "Enabled": true,
+        "Url": "https://nyaa.si/?page=rss&c=1_2&f=0",
+        "XmlNamespace": "https://nyaa.si/xmlns/nyaa"
       },
       {
         "Name": "SyncTpbJob",
         "IntervalSeconds": 60,
-        "Enabled": true
+        "Enabled": true,
+        "Url": "https://apibay.org/precompiled/data_top100_recent.json"
       },
       {
         "Name": "SyncYtsJob",
         "IntervalSeconds": 60,
-        "Enabled": true
+        "Enabled": true,
+        "Url": "https://yts.am/rss"
       },
       {
         "Name": "SyncTgxJob",
         "IntervalSeconds": 60,
-        "Enabled": true
+        "Enabled": true,
+        "Url": "https://tgx.rs/rss"
       },
       {
         "Name": "SyncDmmJob",

src/producer/src/Features/CrawlerSupport/BaseJsonCrawler.cs

@@ -6,6 +6,12 @@ public abstract class BaseJsonCrawler(IHttpClientFactory httpClientFactory, ILog
 
     protected virtual async Task Execute(string collectionName)
     {
+        if (string.IsNullOrWhiteSpace(Url))
+        {
+            logger.LogWarning("No URL provided for {Source} crawl", Source);
+            return;
+        }
+        
         logger.LogInformation("Starting {Source} crawl", Source);
 
         using var client = httpClientFactory.CreateClient("Scraper");

src/producer/src/Features/CrawlerSupport/BaseXmlCrawler.cs

@@ -4,6 +4,12 @@ public abstract class BaseXmlCrawler(IHttpClientFactory httpClientFactory, ILogg
 {
     public override async Task Execute()
     {
+        if (string.IsNullOrWhiteSpace(Url))
+        {
+            logger.LogWarning("No URL provided for {Source} crawl", Source);
+            return;
+        }
+        
         logger.LogInformation("Starting {Source} crawl", Source);
 
         using var client = httpClientFactory.CreateClient(Literals.CrawlerClient);

src/producer/src/Features/CrawlerSupport/Scraper.cs

@@ -7,4 +7,8 @@ public class Scraper
     public int IntervalSeconds { get; set; } = 60;
 
     public bool Enabled { get; set; } = true;
+    
+    public string? Url { get; set; }
+    
+    public string? XmlNamespace { get; set; }
 }

src/producer/src/Features/Crawlers/EzTv/EzTvCrawler.cs

@@ -1,11 +1,10 @@
 namespace Producer.Features.Crawlers.EzTv;
 
-public class EzTvCrawler(IHttpClientFactory httpClientFactory, ILogger<EzTvCrawler> logger, IDataStorage storage) : BaseXmlCrawler(httpClientFactory, logger, storage)
+public class EzTvCrawler(IHttpClientFactory httpClientFactory, ILogger<EzTvCrawler> logger, IDataStorage storage, ScrapeConfiguration scrapeConfiguration) : BaseXmlCrawler(httpClientFactory, logger, storage)
 {
-    protected override string Url => "https://eztv1.xyz/ezrss.xml";
+    protected override string Url => scrapeConfiguration.Scrapers.FirstOrDefault(x => x.Name.Equals("SyncEzTvJob", StringComparison.OrdinalIgnoreCase))?.Url ?? string.Empty;
     protected override string Source => "EZTV";
-
-    private static readonly XNamespace XmlNamespace = "http://xmlns.ezrss.it/0.1/";
+    private XNamespace XmlNamespace => scrapeConfiguration.Scrapers.FirstOrDefault(x => x.Name.Equals("SyncEzTvJob", StringComparison.OrdinalIgnoreCase))?.Url ?? string.Empty;
 
     protected override IReadOnlyDictionary<string, string> Mappings =>
         new Dictionary<string, string>

src/producer/src/Features/Crawlers/Nyaa/NyaaCrawler.cs

@@ -1,11 +1,10 @@
 namespace Producer.Features.Crawlers.Nyaa;
 
-public class NyaaCrawler(IHttpClientFactory httpClientFactory, ILogger<NyaaCrawler> logger, IDataStorage storage) : BaseXmlCrawler(httpClientFactory, logger, storage)
+public class NyaaCrawler(IHttpClientFactory httpClientFactory, ILogger<NyaaCrawler> logger, IDataStorage storage, ScrapeConfiguration scrapeConfiguration) : BaseXmlCrawler(httpClientFactory, logger, storage)
 {
-    protected override string Url => "https://nyaa.si/?page=rss&c=1_2&f=0";
+    protected override string Url => scrapeConfiguration.Scrapers.FirstOrDefault(x => x.Name.Equals("SyncNyaaJob", StringComparison.OrdinalIgnoreCase))?.Url ?? string.Empty;
     protected override string Source => "Nyaa";
-
-    private static readonly XNamespace XmlNamespace = "https://nyaa.si/xmlns/nyaa";
+    private XNamespace XmlNamespace => scrapeConfiguration.Scrapers.FirstOrDefault(x => x.Name.Equals("SyncNyaaJob", StringComparison.OrdinalIgnoreCase))?.Url ?? string.Empty;
 
     protected override IReadOnlyDictionary<string, string> Mappings =>
         new Dictionary<string, string>

src/producer/src/Features/Crawlers/Tgx/TgxCrawler.cs

@@ -1,13 +1,13 @@
 namespace Producer.Features.Crawlers.Tgx;
 
-public partial class TgxCrawler(IHttpClientFactory httpClientFactory, ILogger<TgxCrawler> logger, IDataStorage storage) : BaseXmlCrawler(httpClientFactory, logger, storage)
+public partial class TgxCrawler(IHttpClientFactory httpClientFactory, ILogger<TgxCrawler> logger, IDataStorage storage, ScrapeConfiguration scrapeConfiguration) : BaseXmlCrawler(httpClientFactory, logger, storage)
 {
     [GeneratedRegex(@"Size:\s+(.+?)\s+Added")]
     private static partial Regex SizeStringExtractor();
     [GeneratedRegex(@"(?i)\b(\d+(\.\d+)?)\s*([KMGT]?B)\b", RegexOptions.None, "en-GB")]
     private static partial Regex SizeStringParser();
 
-    protected override string Url => "https://tgx.rs/rss";
+    protected override string Url => scrapeConfiguration.Scrapers.FirstOrDefault(x => x.Name.Equals("SyncTgxJob", StringComparison.OrdinalIgnoreCase))?.Url ?? string.Empty;
 
     protected override string Source => "TorrentGalaxy";
     protected override IReadOnlyDictionary<string, string> Mappings

src/producer/src/Features/Crawlers/Tpb/TpbCrawler.cs

@@ -1,8 +1,8 @@
 namespace Producer.Features.Crawlers.Tpb;
 
-public class TpbCrawler(IHttpClientFactory httpClientFactory, ILogger<TpbCrawler> logger, IDataStorage storage) : BaseJsonCrawler(httpClientFactory, logger, storage)
+public class TpbCrawler(IHttpClientFactory httpClientFactory, ILogger<TpbCrawler> logger, IDataStorage storage, ScrapeConfiguration scrapeConfiguration) : BaseJsonCrawler(httpClientFactory, logger, storage)
 {
-    protected override string Url => "https://apibay.org/precompiled/data_top100_recent.json";
+    protected override string Url => scrapeConfiguration.Scrapers.FirstOrDefault(x => x.Name.Equals("SyncTpbJob", StringComparison.OrdinalIgnoreCase))?.Url ?? string.Empty;
 
     protected override string Source => "TPB";
     

src/producer/src/Features/Crawlers/Yts/YtsCrawler.cs

@@ -1,9 +1,8 @@
 namespace Producer.Features.Crawlers.Yts;
 
-public class YtsCrawler(IHttpClientFactory httpClientFactory, ILogger<YtsCrawler> logger, IDataStorage storage) : BaseXmlCrawler(httpClientFactory, logger, storage)
+public class YtsCrawler(IHttpClientFactory httpClientFactory, ILogger<YtsCrawler> logger, IDataStorage storage, ScrapeConfiguration scrapeConfiguration) : BaseXmlCrawler(httpClientFactory, logger, storage)
 {
-    protected override string Url => "https://yts.am/rss";
-
+    protected override string Url => scrapeConfiguration.Scrapers.FirstOrDefault(x => x.Name.Equals("SyncYtsJob", StringComparison.OrdinalIgnoreCase))?.Url ?? string.Empty;
     protected override string Source => "YTS";
     protected override IReadOnlyDictionary<string, string> Mappings
         => new Dictionary<string, string>

src/qbit-collector/Extensions/ServiceCollectionExtensions.cs

@@ -44,6 +44,7 @@ public static class ServiceCollectionExtensions
     {
         var rabbitConfiguration = services.LoadConfigurationFromEnv<RabbitMqConfiguration>();
         var redisConfiguration = services.LoadConfigurationFromEnv<RedisConfiguration>();
+        var qbitConfiguration = services.LoadConfigurationFromEnv<QbitConfiguration>();
 
         services.AddStackExchangeRedisCache(
             option =>
@@ -80,8 +81,8 @@ public static class ServiceCollectionExtensions
                     e.ConfigureConsumer<WriteQbitMetadataConsumer>(context);
                     e.ConfigureConsumer<PerformQbitMetadataRequestConsumer>(context);
                     e.ConfigureSaga<QbitMetadataSagaState>(context);
-                    e.ConcurrentMessageLimit = 5;
-                    e.PrefetchCount = 5;
+                    e.ConcurrentMessageLimit = qbitConfiguration.Concurrency;
+                    e.PrefetchCount = qbitConfiguration.Concurrency;
                 });
             });
         });
@@ -98,7 +99,7 @@ public static class ServiceCollectionExtensions
                     cfg.UseTimeout(
                         timeout =>
                         {
-                            timeout.Timeout = TimeSpan.FromMinutes(1);
+                            timeout.Timeout = TimeSpan.FromMinutes(3);
                         });
                 })
             .RedisRepository(redisConfiguration.ConnectionString, options =>
@@ -110,7 +111,7 @@ public static class ServiceCollectionExtensions
     {
         var qbitConfiguration = services.LoadConfigurationFromEnv<QbitConfiguration>();
         var client = new QBittorrentClient(new(qbitConfiguration.Host));
-        client.Timeout = TimeSpan.FromSeconds(10);
+        client.Timeout = TimeSpan.FromSeconds(20);
 
         services.AddSingleton<IQBittorrentClient>(client);
     }

src/qbit-collector/Features/Qbit/QBitRequestProcessor.cs

@@ -1,6 +1,6 @@
 namespace QBitCollector.Features.Qbit;
 
-public class QbitRequestProcessor(IQBittorrentClient client, ITrackersService trackersService, ILogger<QbitRequestProcessor> logger)
+public class QbitRequestProcessor(IQBittorrentClient client, ITrackersService trackersService, ILogger<QbitRequestProcessor> logger, QbitConfiguration configuration)
 {
     public async Task<IReadOnlyList<TorrentContent>?> ProcessAsync(string infoHash, CancellationToken cancellationToken = default)
     {
@@ -14,7 +14,7 @@ public class QbitRequestProcessor(IQBittorrentClient client, ITrackersService tr
 
         using var timeoutCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
 
-        timeoutCts.CancelAfter(TimeSpan.FromSeconds(30));
+        timeoutCts.CancelAfter(TimeSpan.FromSeconds(60));
 
         try
         {
@@ -30,7 +30,7 @@ public class QbitRequestProcessor(IQBittorrentClient client, ITrackersService tr
                     break;
                 }
 
-                await Task.Delay(TimeSpan.FromSeconds(1), timeoutCts.Token);
+                await Task.Delay(TimeSpan.FromMilliseconds(200), timeoutCts.Token);
             }
         }
         catch (OperationCanceledException) when (timeoutCts.IsCancellationRequested)

src/qbit-collector/Features/Qbit/QbitConfiguration.cs

@@ -5,7 +5,10 @@ public class QbitConfiguration
     private const string Prefix = "QBIT";
     private const string HOST_VARIABLE = "HOST";
     private const string TRACKERS_URL_VARIABLE = "TRACKERS_URL";
+    private const string CONCURRENCY_VARIABLE = "CONCURRENCY";
     
     public string? Host { get; init; } = Prefix.GetRequiredEnvironmentVariableAsString(HOST_VARIABLE);
     public string? TrackersUrl { get; init; } = Prefix.GetRequiredEnvironmentVariableAsString(TRACKERS_URL_VARIABLE);
+    
+    public int Concurrency { get; init; } = Prefix.GetEnvironmentVariableAsInt(CONCURRENCY_VARIABLE, 8);
 }

src/qbit-collector/Features/Worker/WriteQbitMetadataConsumer.cs

@@ -5,6 +5,12 @@ public class WriteQbitMetadataConsumer(IRankTorrentName rankTorrentName, IDataSt
     public async Task Consume(ConsumeContext<WriteQbitMetadata> context)
     {
         var request = context.Message;
+        
+        if (request.Metadata.Metadata.Count == 0)
+        {
+            await context.Publish(new QbitMetadataWritten(request.Metadata, false));
+            return;
+        }
 
         var torrentFiles = QbitMetaToTorrentMeta.MapMetadataToFilesCollection(
             rankTorrentName, request.Torrent, request.ImdbId, request.Metadata.Metadata, logger);

src/shared/Dapper/DapperDataStorage.cs

@@ -152,7 +152,8 @@ public class DapperDataStorage(PostgresConfiguration configuration, RabbitMqConf
                     INSERT INTO files
                         ("infoHash", "fileIndex", title, "size", "imdbId", "imdbSeason", "imdbEpisode", "kitsuId", "kitsuEpisode", "createdAt", "updatedAt")
                     VALUES
-                        (@InfoHash, @FileIndex, @Title, @Size, @ImdbId, @ImdbSeason, @ImdbEpisode, @KitsuId, @KitsuEpisode, Now(), Now());
+                        (@InfoHash, @FileIndex, @Title, @Size, @ImdbId, @ImdbSeason, @ImdbEpisode, @KitsuId, @KitsuEpisode, Now(), Now())
+                    ON CONFLICT ("infoHash", "fileIndex") DO NOTHING;
                     """;
                 
                 await connection.ExecuteAsync(query, files);
@@ -167,7 +168,8 @@ public class DapperDataStorage(PostgresConfiguration configuration, RabbitMqConf
                     INSERT INTO subtitles
                         ("infoHash", "fileIndex", "fileId", "title")
                     VALUES
-                        (@InfoHash, @FileIndex, @FileId, @Title);
+                        (@InfoHash, @FileIndex, @FileId, @Title)
+                    ON CONFLICT ("infoHash", "fileIndex") DO NOTHING;
                     """;
                 
                 await connection.ExecuteAsync(query, subtitles);