Updates.

Author: simesy

ac-etl/cloud-functions/email-nfe-warehouse/README.md

@@ -42,12 +42,12 @@ Example Cloud deployment
 # From ac-etl/cloud-functions/email-nfe-warehouse
 
 gcloud functions deploy ac-email-processing \
-  --entry-point processEmails
-  --runtime nodejs18
-  --gen2
-  --trigger-topic=your-topic-id
-  --env-vars-file .env.yaml
-  --max-instances=1
+  --entry-point processEmails \
+  --runtime nodejs18 \
+  --gen2 \
+  --trigger-topic=your-topic-id \
+  --env-vars-file .env.yaml \
+  --max-instances=1 \
   --concurrency=1
 ```
 

ac-etl/cloud-functions/email-nfe-warehouse/package-lock.json

@@ -1,4 +1,4 @@
-import { BigQuery } from '@google-cloud/bigquery';
+import { BigQuery, RowsResponse } from '@google-cloud/bigquery';
 import assert from 'assert';
 import * as Ac from './util';
 
@@ -49,6 +49,34 @@ export async function getBigQueryClient(): Promise<BigQuery> {
   return bigQuery;
 }
 
+export async function coundExistingByNfeId(nfeId: string): Promise<number> {
+  const bigQuery = await getBigQueryClient();
+  const datasetId: string = 'ac_ops_data';
+  const invoiceTableId = 'base-nfe-supplier-invoice';
+
+  try {
+    // Build the query to filter rows based on the column and value
+    const query = `
+        SELECT *
+        FROM \`${datasetId}.${invoiceTableId}\`
+        WHERE ${nfeId} = @filterValue
+    `;
+    const options = {
+        query,
+        params: { nfeId },
+    };
+    const [rows] = await bigQuery.query(options);
+    const rowCount = rows.length;
+
+    console.log(`Number of rows retrieved: ${rowCount}`);
+    return rowCount;
+  }
+  catch (error) {
+    console.error('Error retrieving filtered rows from BigQuery:', error);
+    throw error;
+  }
+}
+
 export async function insertInvoiceRecords(invoices: InvoiceRecord[], invoiceLines: InvoiceRecordLine[], logId: string) {
   bigQuery = await getBigQueryClient();
   const datasetId: string = 'ac_ops_data';
@@ -69,6 +97,7 @@ export async function insertInvoiceRecords(invoices: InvoiceRecord[], invoiceLin
     console.log(`${logId}/Inserted ${invoices.length} invoice headers`);
   }
   catch (err: any) {
+    Ac.socialiseIt(`:grimacing: There was a BigQuery error, probably not good. (${logId})`);
     if (err.name === 'PartialFailureError') {
       console.error(`${logId}/BigQuery PartialFailureError` + JSON.stringify(err.errors));
     } else {

ac-etl/cloud-functions/email-nfe-warehouse/src/bigquery.ts

@@ -46,7 +46,7 @@ async function getAuthClient(keyFile: string, gmailUser: string, authScopes: str
     return await auth.getClient();
   }
   catch (error) {
-    console.log([`Error getting Google Auth client for ${gmailUser}:`, error]);
+    console.error([`Error getting Google Auth client for ${gmailUser}:`, error]);
   }
 }
 
@@ -65,7 +65,7 @@ export async function getGmailClient(): Promise<gmail_v1.Gmail> {
     return gmail;
   }
   catch(error) {
-    console.log(['Error getting Gmail client', error]);
+    console.error(['Error getting Gmail client', error]);
     throw new Error('Error getting Gmail client');
   }
 }
@@ -108,7 +108,7 @@ export async function getMessages(query: string = '', label: string): Promise<Gm
   }
   catch (error) {
     if (res.data.resultSizeEstimate == undefined ) {
-      console.log("Not iterable?", res.data);
+      console.error("Not iterable?", res.data);
       throw new Error("Messages from Gmail were not iterable gmail messages. See log.");
     }
     return [];
@@ -147,14 +147,14 @@ export async function getMessages(query: string = '', label: string): Promise<Gm
       });
       // Not quite prepared to assert on this yet.
       if (!isGmailMessage(res.data)) {
-        console.log(["Probably fatal.. there a gmail message in this response?", res]);
+        console.error(["Probably fatal.. there a gmail message in this response?", res]);
       }
       return res.data;
     }) || [];
     return await Promise.all(messagePromises);
   }
   catch (error) {
-    console.log(error);
+    console.error(error);
     throw new Error("Failed to load all gmail messages.")
   }
 }

ac-etl/cloud-functions/email-nfe-warehouse/src/gmail.ts

@@ -6,8 +6,6 @@ import * as St from './storage';
 import * as ff from '@google-cloud/functions-framework';
 import { MessagePublishedData } from '@google/events/cloud/pubsub/v1/MessagePublishedData';
 import NfeDocument, * as nfe from './nfe';
-// import BigQuery from '@google-cloud/bigquery';
-// import { isMemberName } from 'typescript';
 
 async function processEmailXml(cloudEvent: any): Promise<void> {
 
@@ -41,23 +39,33 @@ async function processEmailXml(cloudEvent: any): Promise<void> {
     Ac.getEnv('GMAIL_LABEL_IN_QUEUE'),
   );
 
-  if (message.length == 0) {
+  if (messages.length == 0) {
     console.log(`${logId}: Nothing found process, perhaps it recently ran.`);
   }
+  else {
+    Ac.socialiseIt(`:man-running: Started processing ${messages.length} tagged email(s) with batch sequence ${batchSequence}.`);
+  }
+
+  let messagesProcessed = 0;
+  let timePassed = 0;
 
   // Keeping this basic and procedural while working out the kinks.
   for (const message of messages) {
     console.log(`${logId}: ${message.snippet}`);
 
     const currentTime = new Date();
-    if ((startTime.getTime() - currentTime.getTime()) > 3e4) {
+    timePassed = currentTime.getTime() - startTime.getTime();
+    console.log(`${logId}: Passed: ${timePassed}, processed: ${messagesProcessed}`);
+    if (timePassed > 3e4) {
       // Only process for 30 seconds or so.
+      console.log(`${logId}: Passed: ${timePassed}, processed: ${messagesProcessed}`);
       continue;
     }
 
     if (message?.payload?.parts) {
       for (const part of message.payload.parts) {
         if (part?.body?.attachmentId && message.id) {
+          const messageLink = `https://mail.google.com/mail/u/1/#inbox/${message.id}`;
           const attachment = await gmail.users.messages.attachments.get({
             id: part.body.attachmentId,
             messageId: message.id,
@@ -67,13 +75,20 @@ async function processEmailXml(cloudEvent: any): Promise<void> {
           if (attachment?.data?.data && part?.filename) {
             try {
               if (!part.filename.endsWith('.xml')) {
-                console.log(`${logId}/IGNORE: ${part.filename}`);
+                // console.log(`${logId}/IGNORE: ${part.filename}`);
                 continue;
               }
               const xml = Buffer.from(attachment.data.data, 'base64').toString();
               const nfeDocument = new NfeDocument(xml, part?.filename || 'a file');
+              const existingCount = await Bq.coundExistingByNfeId(nfeDocument.getDocumentId());
 
-              if (nfeDocument.isNfeValid()) {
+              if (existingCount > 0) {
+                console.log(`${logId}/SKIP: stubbornly refusing to re-add ${nfeDocument.getDocumentId()} for ${nfeDocument.getSupplierDisplayName()}`);
+                Ac.socialiseIt(`:sweat_smile: oops looks like ${nfeDocument.getDocumentId()} for ${nfeDocument.getSupplierDisplayName()} was already processed!`);
+                continue;
+              }
+
+              if (nfeDocument.isNfeValid() && existingCount < 1) {
                 console.log(`${logId}/VALID: ${part.filename} is a/an ${nfeDocument.getType()} record for ${nfeDocument.getSupplierDisplayName()}`);
                 if (nfeDocument.getType() == 'invoice') {
                   invoiceRows.push({
@@ -88,6 +103,7 @@ async function processEmailXml(cloudEvent: any): Promise<void> {
                     verboseDescription: nfeDocument.getInvoiceDescription(),
                     batchSequence: batchSequence
                   });
+
                 }
                 else {
                   invoiceRows.push({
@@ -116,14 +132,17 @@ async function processEmailXml(cloudEvent: any): Promise<void> {
               }
             }
             catch (error) {
-              console.log([`${logId}/Error processing ${part.filename}`, error]);
+              console.error([`${logId}/Error processing`, error]);
+              Ac.socialiseIt(`:thinking_face: Error processing NFe data for <${messageLink}|${part.filename}> (${logId})}`);
             }
             // console.log(`${logId}/Finished: ${part.filename}`);
             // Ac.writeAttachmentToBucket(bucket, part.filename, 'application/xml', attachment.data.data);
           }
         }
       }
     }
+    messagesProcessed++;
+    console.log(`${logId}/Error processing`)
   }
 
   if (invoiceRows.length > 0 || invoiceLinesRows.length > 0) {
@@ -135,26 +154,28 @@ async function processEmailXml(cloudEvent: any): Promise<void> {
     }
     catch (error) {
       console.log("There was an error inserting records.", error);
-      Ac.socialiseIt(`There was an error inserting invoices to the data warehouse, data my be inaccurate for these suppliers: ${suppliers.join(", ")}`);
+      Ac.socialiseIt(`:thinking_face: There was an error inserting invoices to the data warehouse, data my be inaccurate for these suppliers: ${suppliers.join(", ")}`);
       return;
     }
 
     try {
       await Gm.updateMessageLabels(messages, Ac.getEnv('GMAIL_LABEL_IN_QUEUE'), Ac.getEnv('GMAIL_LABEL_DONE'));
     }
     catch (error) {
-      console.log("There was an error updating labels.", error);
-      Ac.socialiseIt(`There was an error updating \`ops/in-queue\` and :white_check_mark labels, which may cause period duplicate records for these suppliers: ${suppliers.join(", ")}`);
+      console.log(`${logId}: There was an error updating labels.`, error);
+      Ac.socialiseIt(`:thinking_face: There was an error updating \`ops/in-queue\` and \`ops/done\` labels, which may cause duplicate records for these suppliers: ${suppliers.join(", ")}. (${batchSequence})`);
     }
 
-    console.log(`${logId}: Completed processing ${messages.length} email(s).`);
-    Ac.socialiseIt(`Processed ${messages.length} tagged email(s) and imported ${invoiceRows.length} NFe documents for these suppliers: ${suppliers.join(", ")}`);
+    const currentTime = new Date();
+    timePassed = Math.round((currentTime.getTime() - startTime.getTime()) / 1000);
+    console.log(`${logId}: Completed processing ${messagesProcessed} email(s).`);
+    Ac.socialiseIt(`:white_check_mark: Processed ${messagesProcessed} tagged email(s) and imported ${invoiceRows.length} NFe documents for these suppliers: ${suppliers.join(", ")}. The cloud function ran for about ${timePassed} seconds, with batch sequence ${batchSequence}.`);
   }
   else {
     console.log(`${logId}: Nothing found process.`);
+    Ac.socialiseIt(`:white_check_mark: No documents found to process.`);
   }
 }
 
 ff.cloudEvent('processEmailXml', processEmailXml);
-
 export { processEmailXml };

ac-etl/cloud-functions/email-nfe-warehouse/src/index.ts

@@ -1,5 +1,5 @@
 import assert from 'assert';
-import { XMLParser } from 'fast-xml-parser';
+import { XMLParser, X2jOptions } from 'fast-xml-parser';
 import { InvoiceRecordLine } from './bigquery';
 
 /**
@@ -13,15 +13,28 @@ class NfeDocument {
 
   constructor(xmlString: string, debugContext: string) {
     try {
-      const options = {
+      const options: X2jOptions = {
         ignoreAttributes: false,
-        attributeNamePrefix : "_"
+        attributeNamePrefix: "_",
+        parseAttributeValue: true,
+        trimValues: true,
+        // Disable entity expansion to prevent XXE attacks
+        processEntities: false,
+        // Ignore DOCTYPE declarations
+        ignoreDeclaration: true,
+        ignorePiTags: true,
+        // Set a reasonable limit for string lengths
+        tagValueProcessor: (tagName: string, tagValue: string) =>
+          tagValue.length > 1000 ? tagValue.slice(0, 1000) : tagValue,
       };
+
       const parser = new XMLParser(options);
       this.nfeJson = parser.parse(xmlString);
     }
-    catch {
-      console.log(`XML parsing error for ${debugContext}`);
+    catch (error: any) {
+      console.error(`XML parsing error for ${debugContext}:`, error);
+      // Consider throwing a custom error or handling this case appropriately
+      throw new Error(`Failed to parse XML: ${error.message}`);
     }
   }
 
@@ -169,13 +182,16 @@ class NfeDocument {
     try {
       switch (this.getType()) {
         case 'invoice':
-          dateStr = this.nfeJson?.nfeProc?.NFe?.infNFe?.cobr?.dup?.dVenc;
+          // Sometimes no due date.
+          dateStr = this.nfeJson?.nfeProc?.NFe?.infNFe?.cobr?.dup?.dVenc || this.getDateTime();
           break;
         case 'service':
           dateStr = this.nfeJson?.CompNfse?.Nfse?.InfNfse?.DataEmissao;
           break;
       }
-    } catch (err) {}
+    } catch (err) {
+      console.error(`Date fail for ${dateStr}:`, err);
+    }
     let date = new Date(dateStr);
     dateStr = date.toISOString().slice(0, 19).replace("T", " ");
     return dateStr;
@@ -195,7 +211,9 @@ class NfeDocument {
           dateStr = this.nfeJson?.CompNfse?.Nfse?.InfNfse?.DataEmissao;
           break;
       }
-    } catch (err) {}
+    } catch (err) {
+      console.error(`Date fail for ${dateStr}:`, err);
+    }
     let date = new Date(dateStr);
     dateStr = date.toISOString().slice(0, 19).replace("T", " ");
     return dateStr;