Press n or j to go to the next uncovered block, b, p or k for the previous block.
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 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 93 94 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 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 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 184 | 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x | /**
* Crawler API - Shared Helpers
* Common functions used by both org-scoped and single-resource routes
*/
const axios = require('axios');
const { v4: uuidv4 } = require('uuid');
const Redis = require('ioredis');
const config = require('../config');
const { signedAxios } = require('../../shared/signedAxios');
let redisClient = null;
const CONFIG_PREFIX = 'config:';
const JOB_POOL_PREFIX = 'pool:job:';
const JOB_POOL_SUFFIX = ':params';
const CONFIG_TTL = 86400;
const JOB_TTL = 86400;
function getRedisClient() {
if (!redisClient) {
const redisHost = process.env.REDIS_HOST || 'redis';
const redisPort = parseInt(process.env.REDIS_PORT || '6379');
const redisPassword = process.env.REDIS_PASSWORD || undefined;
redisClient = new Redis({
host: redisHost,
port: redisPort,
password: redisPassword,
retryStrategy: (times) => Math.min(times * 50, 2000),
});
redisClient.on('error', (err) => {
console.error('[CrawlerAPI/Redis] Error:', err.message);
});
}
return redisClient;
}
async function storeConfigById(configId, configData) {
const redis = getRedisClient();
const key = `${CONFIG_PREFIX}${configId}`;
await redis.setex(key, CONFIG_TTL, JSON.stringify(configData));
return configId;
}
async function storeJobById(jobId, jobData) {
const redis = getRedisClient();
const key = `${JOB_POOL_PREFIX}${jobId}${JOB_POOL_SUFFIX}`;
await redis.setex(key, JOB_TTL, JSON.stringify(jobData));
return jobId;
}
const JOB_MANAGER_URL = process.env.JOB_MANAGER_URL || 'http://job-manager:8000';
async function launchCrawlingWorkflow(jobId, startUrls, crawlerConfig, userId, authToken, organizationId) {
const cfg = crawlerConfig.config || crawlerConfig;
const configId = uuidv4();
let llmApiKey = null;
try {
const configResponse = await signedAxios.get(
`/internal/${organizationId}/config/openrouter`,
{ timeout: 5000 }
);
if (configResponse.data?.success) {
llmApiKey = configResponse.data.data.value;
}
} catch (e) {
console.warn('[CrawlerAPI] Could not fetch LLM API key:', e.message);
}
let markerApiKey = null;
try {
const markerResponse = await signedAxios.get(
`/internal/${organizationId}/config/marker`,
{ timeout: 5000 }
);
if (markerResponse.data?.success) {
markerApiKey = markerResponse.data.data.value;
}
} catch (e) {
console.warn('[CrawlerAPI] Could not fetch Marker API key:', e.message);
}
const llmModel = cfg.llmModel || 'deepseek/deepseek-v4-flash';
const configData = {
llmApiKey,
markerApiKey,
authToken,
userId,
llmModel,
storedAt: new Date().toISOString(),
};
await storeConfigById(configId, configData);
// Le workflow config contient DEUX niveaux de prompt :
// - prompt → utilisé pour l'évaluation de pertinence + synthèse (si extractPrompt absent)
// - extractPrompt → utilisé pour la synthèse UNIQUEMENT (remplace prompt si présent)
// - config.synthetiser.spec → inutilisé dans le flux crawler (réservé workflows API)
// Voir triggerSynthesisWorkflow dans crawler-activity-worker (ligne 1042) pour le mapping
const workflowConfig = {
jobId,
configId,
startUrls,
prompt: crawlerConfig.prompt || '',
extractPrompt: crawlerConfig.extractPrompt || '',
config: {
maxPages: cfg.maxPages ?? 10,
maxDepth: cfg.maxDepth ?? 3,
pageThreshold: cfg.pageThreshold ?? 60,
linkThreshold: cfg.linkThreshold ?? 40,
crawlDelay: cfg.crawlDelay ?? 100,
sameDomain: cfg.sameDomain === true,
enableSynthesis: cfg.enableSynthesis !== false,
synthetiser: cfg.synthetiser || { mode: 'fact', spec: {} },
llmModel: cfg.llmModel || 'deepseek/deepseek-v4-flash',
},
router: cfg.router || [],
userId,
organizationId,
};
await storeJobById(jobId, workflowConfig);
const response = await axios.post(
`${JOB_MANAGER_URL}/pool/add`,
{
jobId,
configId,
organizationId,
workflowType: 'CrawlerOrchestrationWorkflow',
input: workflowConfig,
},
{ timeout: 5000 }
);
console.log(`[CrawlerAPI] Job ${jobId} queued via job-manager, position: ${response.data.position}, model: ${llmModel}`);
return jobId;
}
async function createKnowledgeBase(authToken, organizationId, name, description) {
const response = await axios.post(
`${config.knowledgeBaseUrl}/knowledge-bases`,
{ name, description, organizationId },
{ headers: { 'Authorization': `Bearer ${authToken}` }, timeout: 10000 }
);
return response.data.data.knowledge_base;
}
async function createKnowledgeRouter(authToken, organizationId, name, description, targetKbIds) {
const response = await axios.post(
`${config.knowledgeRouterUrl}/knowledge-routers`,
{ name, description, targetKnowledgeBases: targetKbIds, organizationId },
{ headers: { 'Authorization': `Bearer ${authToken}` }, timeout: 10000 }
);
return response.data.data;
}
async function deleteKnowledgeBase(authToken, kbId) {
await axios.delete(
`${config.knowledgeBaseUrl}/knowledge-bases/${kbId}`,
{ headers: { 'Authorization': `Bearer ${authToken}` }, timeout: 5000 }
);
}
async function deleteKnowledgeRouter(authToken, routerId) {
await axios.delete(
`${config.knowledgeRouterUrl}/knowledge-routers/${routerId}`,
{ headers: { 'Authorization': `Bearer ${authToken}` }, timeout: 5000 }
);
}
module.exports = {
getRedisClient,
storeConfigById,
storeJobById,
launchCrawlingWorkflow,
createKnowledgeBase,
createKnowledgeRouter,
deleteKnowledgeBase,
deleteKnowledgeRouter,
};
|