vodorod/parcer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: use integrations from updated robot model

Browse Source
This commit is contained in:
RohitR311
2024-11-09 18:37:47 +05:30
parent d5c56afd0b
commit ac29a15ea8
1 changed files with 78 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

120
server/src/workflow-management/integrations/gsheet.ts
View File

@@ -6,13 +6,14 @@ import Robot from "../../models/Robot";
interface GoogleSheetUpdateTask { interface GoogleSheetUpdateTask {
robotId: string; robotId: string;
runId: string; runId: string;
status: 'pending' | 'completed' | 'failed'; status: "pending" | "completed" | "failed";
retries: number; retries: number;
} }
const MAX_RETRIES = 5; const MAX_RETRIES = 5;
export let googleSheetUpdateTasks: { [runId: string]: GoogleSheetUpdateTask } = {}; export let googleSheetUpdateTasks: { [runId: string]: GoogleSheetUpdateTask } =
{};
export async function updateGoogleSheet(robotId: string, runId: string) { export async function updateGoogleSheet(robotId: string, runId: string) {
try { try {
@@ -24,51 +25,74 @@ export async function updateGoogleSheet(robotId: string, runId: string) {
const plainRun = run.toJSON(); const plainRun = run.toJSON();
if (plainRun.status === 'success') { if (plainRun.status === "success") {
let data: { [key: string]: any }[] = []; let data: { [key: string]: any }[] = [];
if (plainRun.serializableOutput && Object.keys(plainRun.serializableOutput).length > 0) { if (
data = plainRun.serializableOutput['item-0'] as { [key: string]: any }[]; plainRun.serializableOutput &&
Object.keys(plainRun.serializableOutput).length > 0
} else if (plainRun.binaryOutput && plainRun.binaryOutput['item-0']) { ) {
data = plainRun.serializableOutput["item-0"] as {
[key: string]: any;
}[];
} else if (plainRun.binaryOutput && plainRun.binaryOutput["item-0"]) {
// Handle binaryOutput by setting the URL as a data entry // Handle binaryOutput by setting the URL as a data entry
const binaryUrl = plainRun.binaryOutput['item-0'] as string; const binaryUrl = plainRun.binaryOutput["item-0"] as string;
// Create a placeholder object with the binary URL // Create a placeholder object with the binary URL
data = [{ "Screenshot URL": binaryUrl }]; data = [{ "Screenshot URL": binaryUrl }];
} }
const robot = await Robot.findOne({ where: { 'recording_meta.id': robotId } }); const robot = await Robot.findOne({
where: { "recording_meta.id": robotId },
});
if (!robot) { if (!robot) {
throw new Error(`Robot not found for robotId: ${robotId}`); throw new Error(`Robot not found for robotId: ${robotId}`);
} }
const plainRobot = robot.toJSON(); const plainRobot = robot.toJSON();
const spreadsheetId = plainRobot.integrations?.google_sheets?.sheet_id;
const spreadsheetId = plainRobot.google_sheet_id; if (plainRobot.integrations?.google_sheets?.email && spreadsheetId) {
if (plainRobot.google_sheet_email && spreadsheetId) { console.log(
console.log(`Preparing to write data to Google Sheet for robot: ${robotId}, spreadsheetId: ${spreadsheetId}`); `Preparing to write data to Google Sheet for robot: ${robotId}, spreadsheetId: ${spreadsheetId}`
);
const headers = Object.keys(data[0]); const headers = Object.keys(data[0]);
const rows = data.map((row: { [key: string]: any }) => Object.values(row)); const rows = data.map((row: { [key: string]: any }) =>
Object.values(row)
);
const outputData = [headers, ...rows]; const outputData = [headers, ...rows];
await writeDataToSheet(robotId, spreadsheetId, outputData); await writeDataToSheet(robotId, spreadsheetId, outputData);
console.log(`Data written to Google Sheet successfully for Robot: ${robotId} and Run: ${runId}`); console.log(
`Data written to Google Sheet successfully for Robot: ${robotId} and Run: ${runId}`
);
} else { } else {
console.log('Google Sheets integration not configured.'); console.log("Google Sheets integration not configured.");
} }
} else { } else {
console.log('Run status is not success or serializableOutput is missing.'); console.log(
"Run status is not success or serializableOutput is missing."
);
} }
} catch (error: any) { } catch (error: any) {
console.error(`Failed to write data to Google Sheet for Robot: ${robotId} and Run: ${runId}: ${error.message}`); console.error(
`Failed to write data to Google Sheet for Robot: ${robotId} and Run: ${runId}: ${error.message}`
);
} }
}; }
export async function writeDataToSheet(robotId: string, spreadsheetId: string, data: any[]) { export async function writeDataToSheet(
robotId: string,
spreadsheetId: string,
data: any[]
) {
try { try {
const robot = await Robot.findOne({ where: { 'recording_meta.id': robotId } }); const robot = await Robot.findOne({
where: { "recording_meta.id": robotId },
});
if (!robot) { if (!robot) {
throw new Error(`Robot not found for robotId: ${robotId}`); throw new Error(`Robot not found for robotId: ${robotId}`);
@@ -76,8 +100,11 @@ export async function writeDataToSheet(robotId: string, spreadsheetId: string, d
const plainRobot = robot.toJSON(); const plainRobot = robot.toJSON();
if (!plainRobot.google_access_token || !plainRobot.google_refresh_token) { const access_token = plainRobot.integrations?.google_sheets?.access_token;
throw new Error('Google Sheets access not configured for user'); const refresh_token = plainRobot.integrations?.google_sheets?.refresh_token;
if (!access_token || !refresh_token) {
throw new Error("Google Sheets access not configured for user");
} }
const oauth2Client = new google.auth.OAuth2( const oauth2Client = new google.auth.OAuth2(
@@ -87,11 +114,11 @@ export async function writeDataToSheet(robotId: string, spreadsheetId: string, d
); );
oauth2Client.setCredentials({ oauth2Client.setCredentials({
access_token: plainRobot.google_access_token, access_token: access_token,
refresh_token: plainRobot.google_refresh_token, refresh_token: refresh_token,
}); });
oauth2Client.on('tokens', async (tokens) => { oauth2Client.on("tokens", async (tokens) => {
if (tokens.refresh_token) { if (tokens.refresh_token) {
await robot.update({ google_refresh_token: tokens.refresh_token }); await robot.update({ google_refresh_token: tokens.refresh_token });
} }
@@ -100,22 +127,22 @@ export async function writeDataToSheet(robotId: string, spreadsheetId: string, d
} }
}); });
const sheets = google.sheets({ version: 'v4', auth: oauth2Client }); const sheets = google.sheets({ version: "v4", auth: oauth2Client });
const resource = { values: data }; const resource = { values: data };
console.log('Attempting to write to spreadsheet:', spreadsheetId); console.log("Attempting to write to spreadsheet:", spreadsheetId);
const response = await sheets.spreadsheets.values.append({ const response = await sheets.spreadsheets.values.append({
spreadsheetId, spreadsheetId,
range: 'Sheet1!A1', range: "Sheet1!A1",
valueInputOption: 'USER_ENTERED', valueInputOption: "USER_ENTERED",
requestBody: resource, requestBody: resource,
}); });
if (response.status === 200) { if (response.status === 200) {
console.log('Data successfully appended to Google Sheet.'); console.log("Data successfully appended to Google Sheet.");
} else { } else {
console.error('Google Sheets append failed:', response); console.error("Google Sheets append failed:", response);
} }
logger.log(`info`, `Data written to Google Sheet: ${spreadsheetId}`); logger.log(`info`, `Data written to Google Sheet: ${spreadsheetId}`);
@@ -130,33 +157,42 @@ export const processGoogleSheetUpdates = async () => {
let hasPendingTasks = false; let hasPendingTasks = false;
for (const runId in googleSheetUpdateTasks) { for (const runId in googleSheetUpdateTasks) {
const task = googleSheetUpdateTasks[runId]; const task = googleSheetUpdateTasks[runId];
console.log(`Processing task for runId: ${runId}, status: ${task.status}`); console.log(
`Processing task for runId: ${runId}, status: ${task.status}`
);
if (task.status === 'pending') { if (task.status === "pending") {
hasPendingTasks = true; hasPendingTasks = true;
try { try {
await updateGoogleSheet(task.robotId, task.runId); await updateGoogleSheet(task.robotId, task.runId);
console.log(`Successfully updated Google Sheet for runId: ${runId}`); console.log(`Successfully updated Google Sheet for runId: ${runId}`);
delete googleSheetUpdateTasks[runId]; delete googleSheetUpdateTasks[runId];
} catch (error: any) { } catch (error: any) {
console.error(`Failed to update Google Sheets for run ${task.runId}:`, error); console.error(
`Failed to update Google Sheets for run ${task.runId}:`,
error
);
if (task.retries < MAX_RETRIES) { if (task.retries < MAX_RETRIES) {
googleSheetUpdateTasks[runId].retries += 1; googleSheetUpdateTasks[runId].retries += 1;
console.log(`Retrying task for runId: ${runId}, attempt: ${task.retries}`); console.log(
`Retrying task for runId: ${runId}, attempt: ${task.retries}`
);
} else { } else {
googleSheetUpdateTasks[runId].status = 'failed'; googleSheetUpdateTasks[runId].status = "failed";
console.log(`Max retries reached for runId: ${runId}. Marking task as failed.`); console.log(
`Max retries reached for runId: ${runId}. Marking task as failed.`
);
} }
} }
} }
} }
if (!hasPendingTasks) { if (!hasPendingTasks) {
console.log('No pending tasks. Exiting loop.'); console.log("No pending tasks. Exiting loop.");
break; break;
} }
console.log('Waiting for 5 seconds before checking again...'); console.log("Waiting for 5 seconds before checking again...");
await new Promise(resolve => setTimeout(resolve, 5000)); await new Promise((resolve) => setTimeout(resolve, 5000));
} }
}; };