frishi/threetwo-acquisition-service
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Autodownload loop #4

Merged
rishighan merged 14 commits from autodownload-loop into main 2025-02-17 20:42:22 +00:00
Conversation 0 Commits 14 Files Changed 5 +50 -40
1 changed files with 50 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit ecdc3845cb - Show all commits

90
services/comicprocessor.service.ts
View File

@@ -24,7 +24,7 @@ export default class ComicProcessorService extends Service {
private socketIOInstance: any; private socketIOInstance: any;
private kafkaProducer: any; private kafkaProducer: any;
private prowlarrResultsMap: Map<string, any> = new Map(); private prowlarrResultsMap: Map<string, any> = new Map();
private airDCPPSearchResults: Map<string, any[]> = new Map(); private airDCPPSearchResults: Map<number, any[]> = new Map();
private issuesToSearch: any = []; private issuesToSearch: any = [];
// @ts-ignore // @ts-ignore
@@ -84,10 +84,10 @@ export default class ComicProcessorService extends Service {
// 3. Orchestrate the query // 3. Orchestrate the query
const dcppSearchQuery = { const dcppSearchQuery = {
query: { query: {
pattern: `${volume.name.replace( pattern: `${volume.name
/#/g, .replace(/[^\w\s]/g, "")
"", .replace(/\s+/g, " ")
)} ${inferredIssueNumber} ${inferredYear}`, .trim()}`,
extensions: ["cbz", "cbr", "cb7"], extensions: ["cbz", "cbr", "cb7"],
}, },
hub_urls: hubs, hub_urls: hubs,
@@ -109,51 +109,57 @@ export default class ComicProcessorService extends Service {
namespace: "/automated", namespace: "/automated",
}); });
const prowlarrResults = await this.broker.call("prowlarr.search", { // const prowlarrResults = await this.broker.call("prowlarr.search", {
prowlarrQuery: { // prowlarrQuery: {
port: "9696", // port: "9696",
apiKey: "c4f42e265fb044dc81f7e88bd41c3367", // apiKey: "c4f42e265fb044dc81f7e88bd41c3367",
offset: 0, // offset: 0,
categories: [7030], // categories: [7030],
query: `${volume.name} ${issue.issueNumber} ${year}`, // query: `${volume.name} ${issue.issueNumber} ${year}`,
host: "localhost", // host: "localhost",
limit: 100, // limit: 100,
type: "search", // type: "search",
indexerIds: [2], // indexerIds: [2],
}, // },
}); // });
//
this.logger.info( // this.logger.info(
"Prowlarr search results:", // "Prowlarr search results:",
JSON.stringify(prowlarrResults, null, 4), // JSON.stringify(prowlarrResults, null, 4),
); // );
// Store prowlarr results in map using unique key // Store prowlarr results in map using unique key
const key = `${volume.name}-${issue.issueNumber}-${year}`; // const key = `${volume.name}-${issue.issueNumber}-${year}`;
this.prowlarrResultsMap.set(key, prowlarrResults); // this.prowlarrResultsMap.set(key, prowlarrResults);
} }
} catch (error) { } catch (error) {
this.logger.error("Error processing job:", error); this.logger.error("Error processing job:", error);
} }
}, },
produceResultsToKafka: async (dcppResults: any, prowlarrResults: any) => { produceResultsToKafka: async () => {
const results = { dcppResults, prowlarrResults };
try { try {
/*
Kafka messages need to be in a format that can be serialized to JSON, and a Map is not directly serializable in a way that retains its structure, hence we use Object.fromEntries
*/
await this.kafkaProducer.send({ await this.kafkaProducer.send({
topic: "comic-search-results", topic: "comic-search-results",
messages: [{ value: JSON.stringify(results) }], messages: [
{
value: JSON.stringify(
Object.fromEntries(this.airDCPPSearchResults),
),
},
],
}); });
this.logger.info( console.log(`Produced results to Kafka.`);
"Produced results to Kafka:",
JSON.stringify(results, null, 4),
);
// socket event for UI // socket event for UI
await this.broker.call("socket.broadcast", { await this.broker.call("socket.broadcast", {
namespace: "/", namespace: "/",
event: "searchResultsAvailable", event: "searchResultsAvailable",
args: [ args: [
{ {
dcppResults, bokya: Object.fromEntries(this.airDCPPSearchResults),
}, },
], ],
}); });
@@ -230,26 +236,30 @@ export default class ComicProcessorService extends Service {
this.socketIOInstance.on( this.socketIOInstance.on(
"searchResultUpdated", "searchResultUpdated",
async ({ updatedResult, instanceId }: SearchResultPayload) => { async ({ updatedResult, instanceId }: SearchResultPayload) => {
this.logger.info(
"Received search result update:",
JSON.stringify(updatedResult, null, 4),
);
const resultsForInstance = this.airDCPPSearchResults.get(instanceId); const resultsForInstance = this.airDCPPSearchResults.get(instanceId);
if (resultsForInstance) { if (resultsForInstance) {
const toReplaceIndex = resultsForInstance.findIndex( const toReplaceIndex = resultsForInstance.findIndex(
(element: any) => element.id === updatedResult.result.id, (element: any) => element.id === updatedResult.result.id,
); );
if (toReplaceIndex !== -1) { if (toReplaceIndex !== -1) {
// Replace the existing result with the updated result
resultsForInstance[toReplaceIndex] = updatedResult.result; resultsForInstance[toReplaceIndex] = updatedResult.result;
// Optionally, update the map with the modified array
this.airDCPPSearchResults.set(instanceId, resultsForInstance);
} }
} }
}, },
); );
// Handle searchComplete event // Handle searchComplete event
this.socketIOInstance.on("searchComplete", async (instanceId: string) => { this.socketIOInstance.on("searchesSent", async (data: any) => {
this.logger.info(`Search complete for instance ID ${instanceId}`); this.logger.info(
await this.produceResultsToKafka(instanceId); `Search complete for query: "${data.searchInfo.query.pattern}"`,
);
await this.produceResultsToKafka();
}); });
}, },
async stopped() { async stopped() {