mirror of
https://github.com/Viren070/AIOStreams.git
synced 2025-12-01 23:14:04 +01:00
feat: add dynamic fetching strategy with redesigned strategy card and fix parallel groups
closes #410
This commit is contained in:
@@ -370,6 +370,12 @@ export const UserDataSchema = z.object({
|
||||
// })
|
||||
// )
|
||||
// .optional(),
|
||||
dynamicAddonFetching: z
|
||||
.object({
|
||||
enabled: z.boolean().optional(),
|
||||
condition: z.string().min(1).max(200).optional(),
|
||||
})
|
||||
.optional(),
|
||||
groups: z
|
||||
.object({
|
||||
enabled: z.boolean().optional(),
|
||||
|
||||
@@ -607,6 +607,26 @@ export abstract class StreamExpressionEngine {
|
||||
}
|
||||
}
|
||||
|
||||
export class ExitConditionEvaluator extends StreamExpressionEngine {
|
||||
constructor(
|
||||
private totalStreams: ParsedStream[],
|
||||
private totalTimeTaken: number
|
||||
) {
|
||||
super();
|
||||
this.parser.consts.totalStreams = this.totalStreams;
|
||||
this.parser.consts.totalTimeTaken = this.totalTimeTaken;
|
||||
}
|
||||
|
||||
async evaluate(condition: string) {
|
||||
return await this.evaluateCondition(condition);
|
||||
}
|
||||
|
||||
static async testEvaluate(condition: string) {
|
||||
const parser = new ExitConditionEvaluator([], 0);
|
||||
return await parser.evaluate(condition);
|
||||
}
|
||||
}
|
||||
|
||||
export class GroupConditionEvaluator extends StreamExpressionEngine {
|
||||
private previousStreams: ParsedStream[];
|
||||
private totalStreams: ParsedStream[];
|
||||
|
||||
@@ -7,7 +7,10 @@ import {
|
||||
getTimeTakenSincePoint,
|
||||
} from '../utils/index.js';
|
||||
import { Wrapper } from '../wrapper.js';
|
||||
import { GroupConditionEvaluator } from '../parser/streamExpression.js';
|
||||
import {
|
||||
ExitConditionEvaluator,
|
||||
GroupConditionEvaluator,
|
||||
} from '../parser/streamExpression.js';
|
||||
import StreamFilter from './filterer.js';
|
||||
import StreamPrecompute from './precomputer.js';
|
||||
import StreamDeduplicator from './deduplicator.js';
|
||||
@@ -135,7 +138,7 @@ class StreamFetcher {
|
||||
(s) => s.type !== constants.ERROR_STREAM_TYPE
|
||||
),
|
||||
errors: addonErrors,
|
||||
statistics: statisticStream,
|
||||
statistic: statisticStream,
|
||||
timeTaken: Date.now() - start,
|
||||
};
|
||||
} catch (error) {
|
||||
@@ -155,7 +158,6 @@ class StreamFetcher {
|
||||
return {
|
||||
success: false as const,
|
||||
errors: [addonErrors],
|
||||
statistics: [],
|
||||
timeTaken: 0,
|
||||
streams: [],
|
||||
};
|
||||
@@ -171,7 +173,9 @@ class StreamFetcher {
|
||||
|
||||
const groupStreams = results.flatMap((r) => r.streams);
|
||||
const groupErrors = results.flatMap((r) => r.errors);
|
||||
const groupStatistics = results.flatMap((r) => r.statistics);
|
||||
const groupStatistics = results
|
||||
.flatMap((r) => r.statistic)
|
||||
.filter((s) => s !== undefined);
|
||||
|
||||
const filteredStreams = await this.deduplicate.deduplicate(
|
||||
await this.filter.filter(groupStreams, type, id)
|
||||
@@ -190,7 +194,62 @@ class StreamFetcher {
|
||||
};
|
||||
|
||||
// If groups are configured, handle group-based fetching
|
||||
if (
|
||||
if (this.userData.dynamicAddonFetching?.enabled) {
|
||||
const condition = this.userData.dynamicAddonFetching.condition;
|
||||
if (!condition) {
|
||||
throw new Error('Dynamic addon fetching condition is not set');
|
||||
}
|
||||
|
||||
await new Promise<void>((resolve) => {
|
||||
let activePromises = addons.length;
|
||||
if (activePromises === 0) {
|
||||
resolve();
|
||||
return;
|
||||
}
|
||||
|
||||
const checkExit = async () => {
|
||||
const timeTaken = Date.now() - start;
|
||||
const evaluator = new ExitConditionEvaluator(allStreams, timeTaken);
|
||||
const shouldExit = await evaluator.evaluate(condition);
|
||||
|
||||
if (shouldExit) {
|
||||
logger.info(
|
||||
`Exit condition met with results from ${addons.length - activePromises} addons. (${activePromises} addons still fetching) Returning results.`
|
||||
);
|
||||
resolve();
|
||||
}
|
||||
};
|
||||
|
||||
addons.forEach((addon) => {
|
||||
fetchFromAddon(addon)
|
||||
.then(async (result) => {
|
||||
allStreams.push(...result.streams);
|
||||
allErrors.push(...result.errors);
|
||||
if (result.statistic) {
|
||||
allStatisticStreams.push(result.statistic);
|
||||
}
|
||||
await checkExit();
|
||||
})
|
||||
.catch((error) => {
|
||||
logger.error(
|
||||
`Unhandled error from fetchFromAddon for ${getAddonName(addon)}:`,
|
||||
error
|
||||
);
|
||||
allErrors.push({
|
||||
title: `[❌] ${getAddonName(addon)}`,
|
||||
description:
|
||||
error instanceof Error ? error.message : String(error),
|
||||
});
|
||||
})
|
||||
.finally(() => {
|
||||
activePromises--;
|
||||
if (activePromises === 0) {
|
||||
resolve();
|
||||
}
|
||||
});
|
||||
});
|
||||
});
|
||||
} else if (
|
||||
this.userData.groups?.groupings &&
|
||||
this.userData.groups.groupings.length > 0 &&
|
||||
this.userData.groups.enabled !== false
|
||||
@@ -219,6 +278,7 @@ class StreamFetcher {
|
||||
const groupAddons = addons.filter(
|
||||
(addon) => addon.preset.id && group.addons.includes(addon.preset.id)
|
||||
);
|
||||
if (groupAddons.length === 0) return Promise.resolve(null);
|
||||
logger.info(
|
||||
`Queueing parallel fetch for group with ${groupAddons.length} addons.`
|
||||
);
|
||||
@@ -226,48 +286,52 @@ class StreamFetcher {
|
||||
});
|
||||
|
||||
for (let i = 0; i < this.userData.groups.groupings.length; i++) {
|
||||
const groupResult = await groupPromises[i];
|
||||
const group = this.userData.groups.groupings[i];
|
||||
const groupPromise = groupPromises[i];
|
||||
|
||||
if (i === 0) {
|
||||
const groupResult = await groupPromise;
|
||||
if (!groupResult) continue;
|
||||
allStreams.push(...groupResult.streams);
|
||||
allErrors.push(...groupResult.errors);
|
||||
allStatisticStreams.push(...groupResult.statistics);
|
||||
totalTimeTaken = groupResult.totalTime;
|
||||
previousGroupStreams = groupResult.streams;
|
||||
previousGroupTimeTaken = groupResult.totalTime;
|
||||
continue;
|
||||
}
|
||||
// For groups other than the first, check their condition
|
||||
const group = this.userData.groups.groupings[i];
|
||||
if (!group.condition || !group.addons.length) continue;
|
||||
|
||||
const evaluator = new GroupConditionEvaluator(
|
||||
previousGroupStreams,
|
||||
allStreams,
|
||||
previousGroupTimeTaken,
|
||||
totalTimeTaken,
|
||||
queryType
|
||||
);
|
||||
const shouldIncludeAndContinue = await evaluator.evaluate(
|
||||
group.condition
|
||||
);
|
||||
|
||||
if (shouldIncludeAndContinue) {
|
||||
logger.info(
|
||||
`Condition met for parallel group ${i + 1}, awaiting its streams and continuing.`
|
||||
);
|
||||
const groupResult = await groupPromise;
|
||||
if (!groupResult) continue;
|
||||
allStreams.push(...groupResult.streams);
|
||||
allErrors.push(...groupResult.errors);
|
||||
allStatisticStreams.push(...groupResult.statistics);
|
||||
totalTimeTaken = Math.max(totalTimeTaken, groupResult.totalTime);
|
||||
previousGroupStreams = groupResult.streams;
|
||||
previousGroupTimeTaken = groupResult.totalTime;
|
||||
} else {
|
||||
// For groups other than the first, check their condition
|
||||
if (!group.condition || !group.addons.length) continue;
|
||||
|
||||
const evaluator = new GroupConditionEvaluator(
|
||||
previousGroupStreams,
|
||||
allStreams,
|
||||
previousGroupTimeTaken,
|
||||
totalTimeTaken,
|
||||
queryType
|
||||
logger.info(
|
||||
`Condition not met for parallel group ${i + 1}, skipping remaining groups.`
|
||||
);
|
||||
const shouldIncludeAndContinue = await evaluator.evaluate(
|
||||
group.condition
|
||||
);
|
||||
|
||||
if (shouldIncludeAndContinue) {
|
||||
logger.info(
|
||||
`Condition met for parallel group ${i + 1}, including streams and continuing.`
|
||||
);
|
||||
allStreams.push(...groupResult.streams);
|
||||
allErrors.push(...groupResult.errors);
|
||||
allStatisticStreams.push(...groupResult.statistics);
|
||||
totalTimeTaken = Math.max(totalTimeTaken, groupResult.totalTime);
|
||||
previousGroupStreams = groupResult.streams;
|
||||
previousGroupTimeTaken = groupResult.totalTime;
|
||||
} else {
|
||||
logger.info(
|
||||
`Condition not met for parallel group ${i + 1}, skipping remaining groups.`
|
||||
);
|
||||
// exit early.
|
||||
break;
|
||||
}
|
||||
// exit early.
|
||||
break;
|
||||
}
|
||||
}
|
||||
} else {
|
||||
|
||||
@@ -24,6 +24,7 @@ import {
|
||||
} from './index.js';
|
||||
import { ZodError } from 'zod';
|
||||
import {
|
||||
ExitConditionEvaluator,
|
||||
GroupConditionEvaluator,
|
||||
StreamSelector,
|
||||
} from '../parser/streamExpression.js';
|
||||
@@ -364,6 +365,19 @@ export async function validateConfig(
|
||||
}
|
||||
}
|
||||
|
||||
if (
|
||||
config.dynamicAddonFetching?.condition &&
|
||||
config.dynamicAddonFetching.enabled
|
||||
) {
|
||||
try {
|
||||
await ExitConditionEvaluator.testEvaluate(
|
||||
config.dynamicAddonFetching.condition
|
||||
);
|
||||
} catch (error) {
|
||||
throw new Error(`Invalid dynamic addon fetching condition: ${error}`);
|
||||
}
|
||||
}
|
||||
|
||||
// validate excluded filter condition
|
||||
const streamExpressions = [
|
||||
...(config.excludedStreamExpressions ?? []),
|
||||
|
||||
@@ -429,7 +429,7 @@ function Content() {
|
||||
{userData.presets.length > 0 && <CatalogSettingsCard />}
|
||||
|
||||
{userData.presets.length > 0 && mode === 'pro' && (
|
||||
<AddonGroupCard />
|
||||
<AddonFetchingBehaviorCard />
|
||||
)}
|
||||
</PageWrapper>
|
||||
)}
|
||||
@@ -1186,8 +1186,13 @@ function AddonFilterPopover({
|
||||
);
|
||||
}
|
||||
|
||||
function AddonGroupCard() {
|
||||
function AddonFetchingBehaviorCard() {
|
||||
const { userData, setUserData } = useUserData();
|
||||
const [mode, setMode] = useState(() => {
|
||||
if (userData.dynamicAddonFetching?.enabled) return 'dynamic';
|
||||
if (userData.groups?.enabled) return 'groups';
|
||||
return 'default';
|
||||
});
|
||||
|
||||
// Helper function to get presets that are not in any group except the current one
|
||||
const getAvailablePresets = (currentGroupIndex: number) => {
|
||||
@@ -1213,23 +1218,15 @@ function AddonGroupCard() {
|
||||
updates: Partial<{ addons: string[]; condition: string }>
|
||||
) => {
|
||||
setUserData((prev) => {
|
||||
// Initialize groups array if it doesn't exist
|
||||
const currentGroups = prev.groups?.groupings || [];
|
||||
|
||||
// Create a new array with all existing groups
|
||||
const newGroups = [...currentGroups];
|
||||
|
||||
// Update the specific group with new values, preserving other fields
|
||||
newGroups[index] = {
|
||||
...newGroups[index],
|
||||
...updates,
|
||||
};
|
||||
|
||||
if (index === 0) {
|
||||
// set condition for first group to true
|
||||
newGroups[index].condition = 'true';
|
||||
}
|
||||
|
||||
return {
|
||||
...prev,
|
||||
groups: {
|
||||
@@ -1240,131 +1237,164 @@ function AddonGroupCard() {
|
||||
});
|
||||
};
|
||||
|
||||
const handleModeChange = (newMode: string) => {
|
||||
setMode(newMode);
|
||||
setUserData((prev) => ({
|
||||
...prev,
|
||||
groups: {
|
||||
...prev.groups,
|
||||
enabled: newMode === 'groups',
|
||||
},
|
||||
dynamicAddonFetching: {
|
||||
...prev.dynamicAddonFetching,
|
||||
enabled: newMode === 'dynamic',
|
||||
},
|
||||
}));
|
||||
};
|
||||
|
||||
const descriptions = {
|
||||
default:
|
||||
'Fetch from all addons simultaneously and wait for all addons to finish fetching before returning results.',
|
||||
groups:
|
||||
'Organize addons into groups. Streams are fetched based on group conditions, either sequentially or in parallel.',
|
||||
dynamic:
|
||||
'Fetch from all addons at once, and exit once a specified condition is met.',
|
||||
};
|
||||
|
||||
return (
|
||||
<SettingsCard
|
||||
title="Groups"
|
||||
// description="Optionally assign your addons to groups. Streams are only fetched from your first group initially,
|
||||
// and only if a certain condition is met, will streams be fetched from the next group, and so on. Leaving this blank will mean streams are
|
||||
// fetched from all addons. For a guide and a reference to the group system,"
|
||||
title="Addon Fetching Strategy"
|
||||
description="Choose how streams are fetched from your addons"
|
||||
>
|
||||
<div className="text-sm text-[--muted] mb-2">
|
||||
Optionally assign your addons to groups. Streams are only fetched from
|
||||
your first group initially, and only if a certain condition is met, will
|
||||
streams be fetched from the next group, and so on. Leaving this blank
|
||||
will mean streams are fetched from all addons. Check the{' '}
|
||||
<a
|
||||
href="https://github.com/Viren070/AIOStreams/wiki/Groups"
|
||||
target="_blank"
|
||||
rel="noopener noreferrer"
|
||||
className="text-[--brand] hover:text-[--brand]/80 hover:underline"
|
||||
>
|
||||
wiki
|
||||
</a>{' '}
|
||||
for a detailed guide to using groups.
|
||||
</div>
|
||||
<Switch
|
||||
label="Enable"
|
||||
value={userData.groups?.enabled ?? false}
|
||||
onValueChange={(value) => {
|
||||
setUserData((prev) => ({
|
||||
...prev,
|
||||
groups: { ...prev.groups, enabled: value },
|
||||
}));
|
||||
}}
|
||||
side="right"
|
||||
/>
|
||||
<Select
|
||||
label="Behaviour"
|
||||
value={userData.groups?.behaviour ?? 'parallel'}
|
||||
onValueChange={(value) => {
|
||||
setUserData((prev) => ({
|
||||
...prev,
|
||||
groups: {
|
||||
...prev.groups,
|
||||
behaviour: value as 'sequential' | 'parallel',
|
||||
},
|
||||
}));
|
||||
}}
|
||||
label="Strategy"
|
||||
value={mode}
|
||||
onValueChange={handleModeChange}
|
||||
options={[
|
||||
{ label: 'Parallel', value: 'parallel' },
|
||||
{ label: 'Sequential', value: 'sequential' },
|
||||
{ label: 'Default', value: 'default' },
|
||||
{ label: 'Groups', value: 'groups' },
|
||||
{ label: 'Dynamic', value: 'dynamic' },
|
||||
]}
|
||||
disabled={userData.groups?.enabled === false}
|
||||
help={
|
||||
userData.groups?.behaviour === 'sequential'
|
||||
? 'Streams are fetched from the first group only to begin with. If the condition for the next group is met, streams are fetched from the next group, and so on.'
|
||||
: 'Streams are fetched from all groups at the same time. When a condition is not met, results from its group onwards are simply not shown.'
|
||||
}
|
||||
/>
|
||||
{(userData.groups?.groupings || []).map((group, index) => (
|
||||
<div key={index} className="flex gap-2">
|
||||
<div className="flex-1 flex gap-2">
|
||||
<div className="flex-1">
|
||||
<Combobox
|
||||
multiple
|
||||
disabled={userData.groups?.enabled === false}
|
||||
value={group.addons}
|
||||
options={getAvailablePresets(index)}
|
||||
emptyMessage="You haven't installed any addons yet or they are already in a group"
|
||||
label="Addons"
|
||||
placeholder="Select addons"
|
||||
onValueChange={(value) => {
|
||||
updateGroup(index, { addons: value });
|
||||
}}
|
||||
/>
|
||||
</div>
|
||||
<div className="flex-1">
|
||||
<TextInput
|
||||
value={index === 0 ? 'true' : group.condition}
|
||||
disabled={index === 0 || userData.groups?.enabled === false}
|
||||
label="Condition"
|
||||
placeholder="Enter condition"
|
||||
onValueChange={(value) => {
|
||||
updateGroup(index, { condition: value });
|
||||
}}
|
||||
/>
|
||||
</div>
|
||||
</div>
|
||||
<IconButton
|
||||
size="sm"
|
||||
rounded
|
||||
disabled={userData.groups?.enabled === false}
|
||||
icon={<FaRegTrashAlt />}
|
||||
intent="alert-subtle"
|
||||
onClick={() => {
|
||||
setUserData((prev) => {
|
||||
const newGroups = [...(prev.groups?.groupings || [])];
|
||||
newGroups.splice(index, 1);
|
||||
return {
|
||||
...prev,
|
||||
groups: { ...prev.groups, groupings: newGroups },
|
||||
};
|
||||
});
|
||||
}}
|
||||
/>
|
||||
</div>
|
||||
))}
|
||||
<div className="mt-2 flex gap-2 items-center">
|
||||
<IconButton
|
||||
rounded
|
||||
size="sm"
|
||||
intent="primary-subtle"
|
||||
icon={<FaPlus />}
|
||||
disabled={userData.groups?.enabled === false}
|
||||
onClick={() => {
|
||||
setUserData((prev) => {
|
||||
const currentGroups = prev.groups?.groupings || [];
|
||||
return {
|
||||
|
||||
<div className="text-sm text-[--muted] mt-2 mb-4">
|
||||
{descriptions[mode as keyof typeof descriptions]}
|
||||
</div>
|
||||
|
||||
{mode === 'groups' && (
|
||||
<>
|
||||
<Select
|
||||
label="Group Behaviour"
|
||||
value={userData.groups?.behaviour ?? 'parallel'}
|
||||
onValueChange={(value) => {
|
||||
setUserData((prev) => ({
|
||||
...prev,
|
||||
groups: {
|
||||
...prev.groups,
|
||||
groupings: [...currentGroups, { addons: [], condition: '' }],
|
||||
behaviour: value as 'sequential' | 'parallel',
|
||||
},
|
||||
};
|
||||
});
|
||||
}));
|
||||
}}
|
||||
options={[
|
||||
{ label: 'Parallel', value: 'parallel' },
|
||||
{ label: 'Sequential', value: 'sequential' },
|
||||
]}
|
||||
help={
|
||||
userData.groups?.behaviour === 'sequential'
|
||||
? 'Streams are fetched from the first group only to begin with. If the condition for the next group is met, streams are fetched from the next group, and so on.'
|
||||
: 'Streams are fetched from all groups at the same time. When a condition is not met, results from its group onwards are simply not shown.'
|
||||
}
|
||||
/>
|
||||
|
||||
{(userData.groups?.groupings || []).map((group, index) => (
|
||||
<div key={index} className="flex gap-2">
|
||||
<div className="flex-1 flex gap-2">
|
||||
<div className="flex-1">
|
||||
<Combobox
|
||||
multiple
|
||||
value={group.addons}
|
||||
options={getAvailablePresets(index)}
|
||||
emptyMessage="You haven't installed any addons yet or they are already in a group"
|
||||
label="Addons"
|
||||
placeholder="Select addons"
|
||||
onValueChange={(value) => {
|
||||
updateGroup(index, { addons: value });
|
||||
}}
|
||||
/>
|
||||
</div>
|
||||
<div className="flex-1">
|
||||
<TextInput
|
||||
value={index === 0 ? 'true' : group.condition}
|
||||
disabled={index === 0}
|
||||
label="Condition"
|
||||
placeholder="Enter condition"
|
||||
onValueChange={(value) => {
|
||||
updateGroup(index, { condition: value });
|
||||
}}
|
||||
/>
|
||||
</div>
|
||||
</div>
|
||||
<IconButton
|
||||
size="sm"
|
||||
rounded
|
||||
icon={<FaRegTrashAlt />}
|
||||
intent="alert-subtle"
|
||||
onClick={() => {
|
||||
setUserData((prev) => {
|
||||
const newGroups = [...(prev.groups?.groupings || [])];
|
||||
newGroups.splice(index, 1);
|
||||
return {
|
||||
...prev,
|
||||
groups: { ...prev.groups, groupings: newGroups },
|
||||
};
|
||||
});
|
||||
}}
|
||||
/>
|
||||
</div>
|
||||
))}
|
||||
|
||||
<div className="mt-2 flex gap-2 items-center">
|
||||
<IconButton
|
||||
rounded
|
||||
size="sm"
|
||||
intent="primary-subtle"
|
||||
icon={<FaPlus />}
|
||||
onClick={() => {
|
||||
setUserData((prev) => {
|
||||
const currentGroups = prev.groups?.groupings || [];
|
||||
return {
|
||||
...prev,
|
||||
groups: {
|
||||
...prev.groups,
|
||||
groupings: [
|
||||
...currentGroups,
|
||||
{ addons: [], condition: '' },
|
||||
],
|
||||
},
|
||||
};
|
||||
});
|
||||
}}
|
||||
/>
|
||||
</div>
|
||||
</>
|
||||
)}
|
||||
|
||||
{mode === 'dynamic' && (
|
||||
<TextInput
|
||||
label="Exit Condition"
|
||||
value={userData.dynamicAddonFetching?.condition ?? ''}
|
||||
onValueChange={(value) => {
|
||||
setUserData((prev) => ({
|
||||
...prev,
|
||||
dynamicAddonFetching: {
|
||||
...prev.dynamicAddonFetching,
|
||||
condition: value,
|
||||
},
|
||||
}));
|
||||
}}
|
||||
help="When this condition is met, no more addons will be fetched from"
|
||||
/>
|
||||
</div>
|
||||
)}
|
||||
</SettingsCard>
|
||||
);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user