UIND
/
marketing-SQS-RDS-script
4
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

initial commit

main
saif 2025-07-02 15:26:26 +05:00
commit a4ae903b7d
6 changed files with 1808 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
.env Normal file
View File

@ -0,0 +1,30 @@
## for NA - US
SQS_QUEUE_US_URL = https://sqs.us-east-1.amazonaws.com/416818527652/sp-traffic-stream-us-destination-queue
## for NA -CA
SQS_QUEUE_CA_URL = https://sqs.us-east-1.amazonaws.com/416818527652/sp-traffic-stream-ca-destination-queue
## for EU - UK
SQS_QUEUE_UK_URL = https://sqs.eu-west-1.amazonaws.com/416818527652/sp-traffic-stream-uk-destination-queue
## for EU - DE
SQS_QUEUE_DE_URL = https://sqs.eu-west-1.amazonaws.com/416818527652/sp-traffic-stream-de-destination-queue
## FOR EU - FR
SQS_QUEUE_FR_URL = https://sqs.eu-west-1.amazonaws.com/416818527652/sp-traffic-stream-fr-destination-queue
## FOR EU - IT
SQS_QUEUE_IT_URL = https://sqs.eu-west-1.amazonaws.com/416818527652/sp-traffic-stream-it-destination-queue
## FOR EU - ES
SQS_QUEUE_ES_URL = https://sqs.eu-west-1.amazonaws.com/416818527652/sp-traffic-stream-es-destination-queue
## REGIONS
AWS_REGION_US_EAST = us-east-1
AWS_REGION_EU_WEST = eu-west-1
## DB CREDENTIALS
DB_HOST = utopia-2.c5qech8o9lgg.us-east-1.rds.amazonaws.com
DB_USER = stream-user
DB_PASSWORD = Utopia01
DB_NAME = ads_stream_api
## AWS keys
AWS_ACCESS_KEY = AKIAWCDCR7WSP2F3Y4WJ
AWS_SECRET_KEY = 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

46
.gitignore vendored Normal file
View File

@ -0,0 +1,46 @@
# Node.js
node_modules/
npm-debug.log*
yarn-debug.log*
yarn-error.log*
pnpm-debug.log*
# Logs
logs/
*.log
# OS-specific
.DS_Store
Thumbs.db
# Build artifacts
dist/
build/
out/
coverage/
.next/
.vercel/
.cache/
# Testing
coverage/
jest-coverage/
nyc_output/
# IDE/editor
.vscode/
.idea/
*.swp
# TypeScript
*.tsbuildinfo
# PM2 logs
pids/
*.pid
*.seed
# Optional: Ignore local database or secrets
*.sqlite
*.db
*.pem

122
marketplaces.json Normal file
View File

@ -0,0 +1,122 @@
{
"APJ6JRA9NG5V4": {
"profileId": 4253617697009484,
"countryCode": "IT",
"currencyCode": "EUR",
"timezone": "Europe/Paris",
"timezoneUTC": "UTC+02:00",
"marketplace": "AMAZON_IT"
},
"A28R8C7NBKEWEA": {
"profileId": 1293255248900780,
"countryCode": "IE",
"currencyCode": "EUR",
"timezone": "Europe/Dublin",
"timezoneUTC": "UTC+01:00",
"marketplace": "AMAZON_IE"
},
"A2NODRKZP88ZB9": {
"profileId": 3880305188764010,
"countryCode": "SE",
"currencyCode": "SEK",
"timezone": "Europe/Stockholm",
"timezoneUTC": "UTC+02:00",
"marketplace": "AMAZON_SE"
},
"A1C3SOZRARQ6R3": {
"profileId": 3901843410471762,
"countryCode": "PL",
"currencyCode": "PLN",
"timezone": "Europe/Warsaw",
"timezoneUTC": "UTC+02:00",
"marketplace": "AMAZON_PL"
},
"A13V1IB3VIYZZH": {
"profileId": 2247462980733839,
"countryCode": "FR",
"currencyCode": "EUR",
"timezone": "Europe/Paris",
"timezoneUTC": "UTC+02:00",
"marketplace": "AMAZON_FR"
},
"A33AVAJ2PDY3EV": {
"profileId": 3109287371398194,
"countryCode": "TR",
"currencyCode": "TRY",
"timezone": "Europe/Istanbul",
"timezoneUTC": "UTC+03:00",
"marketplace": "AMAZON_TR"
},
"AMEN7PMS3EDWL": {
"profileId": 1633590029473539,
"countryCode": "BE",
"currencyCode": "EUR",
"timezone": "Europe/Brussels",
"timezoneUTC": "UTC+02:00",
"marketplace": "AMAZON_BE"
},
"A1805IZSGTT6HS": {
"profileId": 2173727205472935,
"countryCode": "NL",
"currencyCode": "EUR",
"timezone": "Europe/Amsterdam",
"timezoneUTC": "UTC+02:00",
"marketplace": "AMAZON_NL"
},
"A1RKKUPIHCS9HS": {
"profileId": 1741306081337515,
"countryCode": "ES",
"currencyCode": "EUR",
"timezone": "Europe/Paris",
"timezoneUTC": "UTC+02:00",
"marketplace": "AMAZON_ES"
},
"A1PA6795UKMFR9": {
"profileId": 1717389288563376,
"countryCode": "DE",
"currencyCode": "EUR",
"timezone": "Europe/Paris",
"timezoneUTC": "UTC+02:00",
"marketplace": "AMAZON_DE"
},
"A1F83G8C2ARO7P": {
"profileId": 3391272133480174,
"countryCode": "UK",
"currencyCode": "GBP",
"timezone": "Europe/London",
"timezoneUTC": "UTC+01:00",
"marketplace": "AMAZON_UK"
},
"A2Q3Y263D00KWC": {
"profileId": 4134162108117933,
"countryCode": "BR",
"currencyCode": "BRL",
"timezone": "America/Sao_Paulo",
"timezoneUTC": "UTC-03:00",
"marketplace": "AMAZON_BR"
},
"A2EUQ1WTGCTBG2": {
"profileId": 3047619299309873,
"countryCode": "CA",
"currencyCode": "CAD",
"timezone": "America/Los_Angeles",
"timezoneUTC": "UTC-07:00",
"marketplace": "AMAZON_CA"
},
"A1AM78C64UM0Y8": {
"profileId": "2042812141248173",
"countryCode": "MX",
"currencyCode": "MXN",
"timezone": "America/Los_Angeles",
"timezoneUTC": "UTC-07:00",
"marketplace": "AMAZON_MX"
},
"ATVPDKIKX0DER": {
"profileId": "109435392217298",
"countryCode": "US",
"currencyCode": "USD",
"timezone": "America/Los_Angeles",
"timezoneUTC": "UTC-07:00",
"marketplace": "AMAZON_USA"
}
}

1355
package-lock.json generated Normal file
View File

File diff suppressed because it is too large Load Diff

18
package.json Normal file
View File

@ -0,0 +1,18 @@
{
"name": "amz-marketing-script",
"version": "1.0.0",
"description": "",
"main": "index.js",
"scripts": {
"test": "echo \"Error: no test specified\" && exit 1"
},
"keywords": [],
"author": "",
"license": "ISC",
"dependencies": {
"@aws-sdk/client-sqs": "^3.840.0",
"dotenv": "^17.0.0",
"luxon": "^3.6.1",
"mysql2": "^3.14.1"
}
}

237
script.js Normal file
View File

@ -0,0 +1,237 @@
require("dotenv").config();
const fs = require("fs");
const { DateTime } = require("luxon");
const {
SQSClient,
ReceiveMessageCommand,
DeleteMessageBatchCommand,
} = require("@aws-sdk/client-sqs");
const mysql = require("mysql2/promise");
const { off } = require("process");
const raw = fs.readFileSync("./marketplaces.json", "utf8");
const marketplaces = JSON.parse(raw);
const pool = mysql.createPool({
host: process.env.DB_HOST,
user: process.env.DB_USER,
password: process.env.DB_PASSWORD,
database: process.env.DB_NAME,
waitForConnections: true,
connectionLimit: 10,
});
// All 7 queues with their regions
const QUEUES = [
{ url: process.env.SQS_QUEUE_US_URL, region: process.env.AWS_REGION_US_EAST },
{ url: process.env.SQS_QUEUE_CA_URL, region: process.env.AWS_REGION_US_EAST },
{ url: process.env.SQS_QUEUE_UK_URL, region: process.env.AWS_REGION_EU_WEST },
{ url: process.env.SQS_QUEUE_DE_URL, region: process.env.AWS_REGION_EU_WEST },
{ url: process.env.SQS_QUEUE_FR_URL, region: process.env.AWS_REGION_EU_WEST },
{ url: process.env.SQS_QUEUE_IT_URL, region: process.env.AWS_REGION_EU_WEST },
{ url: process.env.SQS_QUEUE_ES_URL, region: process.env.AWS_REGION_EU_WEST },
];
const clients = {};
function getClient(region) {
if (!clients[region]) {
clients[region] = new SQSClient({
region,
accessKeyId: process.env.AWS_ACCESS_KEY,
secretAccessKey: process.env.AWS_SECRET_KEY,
});
}
return clients[region];
}
function parseMessage(jsonStr) {
const data = JSON.parse(jsonStr);
return {
idempotency_id: data.idempotency_id,
dataset_id: data.dataset_id,
marketplace_id: data.marketplace_id,
marketplace: getMarketplaceName(data.marketplace_id),
currency: data.currency,
advertiser_id: data.advertiser_id,
campaign_id: data.campaign_id,
ad_group_id: data.ad_group_id,
ad_id: data.ad_id,
keyword_id: data.keyword_id,
keyword_text: data.keyword_text,
match_type: data.match_type,
placement: data.placement,
time_window_start_original: data.time_window_start,
time_window_start: formatToMysqlDatetime(data.time_window_start),
utc_date_time: getUtcDateTime(data.time_window_start, data.marketplace_id),
clicks: Number(data.clicks || 0),
impressions: Number(data.impressions || 0),
cost: Number(data.cost || 0),
};
}
function getMarketplaceName(marketplaceId) {
const map = marketplaces[marketplaceId];
return map?.marketplace;
}
function offsetToMinutes(offsetStr) {
const match = offsetStr.match(/^UTC([+-])(\d{2}):(\d{2})$/);
if (!match) throw new Error(`Invalid offset format: ${offsetStr}`);
const [, sign, hours, minutes] = match;
const total = parseInt(hours) * 60 + parseInt(minutes);
return sign === "+" ? total : -total;
}
function convertToUtcFromOffset(localIsoString, offsetStr) {
const local = new Date(localIsoString);
const offsetMinutes = offsetToMinutes(offsetStr);
const utcDate = new Date(local.getTime() - offsetMinutes * 60 * 1000);
const pad = (n) => String(n).padStart(2, "0");
return (
`${utcDate.getUTCFullYear()}-${pad(utcDate.getUTCMonth() + 1)}-${pad(
utcDate.getUTCDate()
)} ` +
`${pad(utcDate.getUTCHours())}:${pad(utcDate.getUTCMinutes())}:${pad(
utcDate.getUTCSeconds()
)}`
);
}
function formatToMysqlDatetime(dateTime, marketplace_id) {
const timezone = marketplaces[marketplace_id]?.timezone;
const local = DateTime.fromISO(dateTime, { zone: timezone }).toUTC();
return local.toFormat("yyyy-MM-dd HH:mm:ss");
}
function getUtcDateTime(startDateTime, marketplaceId) {
const offset = marketplaces[marketplaceId]?.timezoneUTC;
const utcTime = convertToUtcFromOffset(startDateTime, offset);
return utcTime;
}
async function saveToDbBulk(traffics) {
if (traffics.length === 0) return true;
const conn = await pool.getConnection();
try {
const sql = `
REPLACE INTO sponsored_product_traffic (
idempotency_id, dataset_id, marketplace_id, marketplace, currency, advertiser_id,
campaign_id, ad_group_id, ad_id, keyword_id, keyword_text, match_type, placement,
time_window_start, utc_date_time, clicks, impressions, cost
) VALUES ?
`;
const values = traffics.map((t) => [
t.idempotency_id,
t.dataset_id,
t.marketplace_id,
t.marketplace,
t.currency,
t.advertiser_id,
t.campaign_id,
t.ad_group_id,
t.ad_id,
t.keyword_id,
t.keyword_text,
t.match_type,
t.placement,
t.time_window_start,
t.utc_date_time,
t.clicks,
t.impressions,
t.cost,
]);
await conn.query(sql, [values]);
return true;
} catch (err) {
console.error("DB Bulk Error:", err);
return false;
} finally {
conn.release();
}
}
async function processQueue(queue) {
const client = getClient(queue.region);
const BATCHES = 50;
const traffics = [];
const deleteEntries = [];
for (let i = 0; i < BATCHES; i++) {
const command = new ReceiveMessageCommand({
QueueUrl: queue.url,
MaxNumberOfMessages: 10,
WaitTimeSeconds: 10,
});
const response = await client.send(command);
const messages = response.Messages || [];
if (messages.length === 0) {
`No Messages in Queue ${queue.url}`;
break;
}
for (const msg of messages) {
try {
const traffic = parseMessage(msg.Body);
traffics.push(traffic);
deleteEntries.push({
Id: msg.MessageId,
ReceiptHandle: msg.ReceiptHandle,
});
} catch (err) {
console.error(`Invalid message in ${queue.url}:`, err.message);
}
}
}
if (traffics.length > 0) {
const saved = await saveToDbBulk(traffics);
if (saved && deleteEntries.length > 0) {
for (let i = 0; i < deleteEntries.length; i += 10) {
const batch = deleteEntries.slice(i, i + 10);
await client.send(
new DeleteMessageBatchCommand({
QueueUrl: queue.url,
Entries: batch,
})
);
}
}
}
console.log(
`[${queue.region}] ${queue.url} -> processed ${traffics.length} messages`
);
}
// async function main() {
// // Parallel processing of all queues
// await Promise.all(QUEUES.map(processQueue));
// }
// main().catch(console.error);
async function main() {
await Promise.all(QUEUES.map(processQueue));
}
async function runLoop() {
try {
await main(); // wait for current execution to finish
} catch (err) {
console.error('Error in main():', err);
}
// Wait 5 seconds before running again
setTimeout(runLoop, 5000);
}
runLoop();