ai_logistic_outlook_integration/README.md

Email Export Process Monitor with OCI Generative AI, Outlook Emails and Oracle Database

This project automates the monitoring of export and logistics communications by reading Outlook .msg emails from OCI Object Storage, analyzing them with OCI Generative AI to extract status and alerts, summarizing content, and displaying results in a Flask web application backed by Oracle Database.

---

Email Export Process Monitor with OCI Generative AI, Outlook e-mails and Oracle Database

A production‑grade pipeline that reads Outlook .msg emails from an OCI Object Storage inbox, uses Generative AI to understand the status of an export/logistics process, summarizes each message, flags alerts, and serves a timeline report via a Flask web app backed by Oracle Database.

---

1) What problem this project solves (Use Case)

Export and logistics operations generate long email threads (forwarded/quoted chains, multiple senders, mixed languages). Manually reading and updating spreadsheets or systems is slow and error‑prone.
This project automates that routine:

1. Ingest Outlook emails (.msg) placed in an OCI Object Storage bucket (e.g., inbox/).
2. Parse each email (including forwarded/replied headers) and split the thread into individual communications.
3. Summarize and classify each communication with OCI Generative AI (via LangChain), extracting structured fields like:
   • date_sent (ISO‑8601 when possible)
   • booking
   • email_from, email_to, subject
   • status (e.g., “Send Note to Forwarder”, “Approve Invoice”, “Perform Weighing”, etc.)
   • brief_description (1–2 lines)
   • alert (“YES”/“NO” based on urgency, problems, missing docs, delays, etc.)
4. Persist the result to Oracle Database (PROCESSED_EMAILS), one row per communication (thread index).
5. Move processed files to a processed/ prefix to avoid re‑processing.
6. Visualize a timeline report and filters in a Flask UI (date range, booking, status, alerts). You can also open each original email to inspect content.

Outcome: A near‑real‑time control panel of export process communications with actionable alerts and structured status—no manual triage of huge mail chains.

---

2) Other use cases you can support with the same architecture

• Accounts Payable/Receivable: triage invoice/dispute emails, extract amounts, due dates, and blockers.
• Customer Support: summarize tickets by email, detect intent/severity, and push alerts to a dashboard.
• Procurement: track RFQs/quotes/PO confirmations arriving by email and update ERP status fields.
• Compliance: auto‑detect phrases indicating risk/backlogs or missing mandatory documents.
• Sales Ops: parse quote/booking confirmations and flag risks (price mismatch, dates, missing attachments).

---

3) Architecture (high level)

flowchart LR
    A[Outlook emails .msg\nOCI Object Storage bucket: inbox/] --> B[Processor\nprocess_emails_v2.py]
    B -->|Split thread + chunk| C[LLM OCI Generative AI\nmodel_id: cohere.command-a-03-2025]
    C -->|JSON results| D[Oracle DB 23ai\nEMAILS_PROCESSADOS]
    B -->|Move done| A2[Object Storage: processed/]
    D --> E[Flask App app_emails.py\nTimeline + Filters + View .msg]

Key behaviors:

• Thread splitter detects “From/De, Sent/Sent, To/Para, Subject/Subject…” headers and separators (“Original Message / Original Message / Forwarded message”).
• Chunking limits each LLM call to ~10k chars (≈2.5k tokens) to keep cost/latency predictable.
• Prompting uses a deterministic, JSON‑only instruction in Portuguese tailored to international logistics.
• DB write normalizes dates and strings, stores thread_index per communication.
• UI renders a chronological timeline with filters and redacted emails if desired.

---

4) Technologies

Layer	Technology	Why
Email parsing	extract_msg	Robust reader for Outlook .msg files (headers, body)
Object storage	OCI Object Storage	Durable inbox/processed area, cheap and scalable
AI runtime	OCI Generative AI via langchain_community.chat_models.oci_generative_ai.ChatOCIGenAI	Secure, low‑latency, enterprise LLM serving
Orchestration	LangChain (ChatPromptTemplate)	Structured prompting & message assembly
Database	Oracle Database 23ai (oracledb)	Reliable transactional store, SQL analytics, wallet auth
API/UI	Flask	Simple server‑side UI for timeline + filters
Config/Auth	oci SDK + ~/.oci/config profile	Standardized credential management

---

5) Project structure (key files)

• process_emails_v2.py – batch processor:
  • Lists .msg files in BUCKET_NAME inbox
  • Downloads to temp, extracts top headers + body
  • Splits thread into messages and chunks text to fit LLM limits
  • Builds prompt and calls OCI GenAI for each chunk
  • Parses JSON, normalizes fields, and inserts into PROCESSED_EMAILS
  • On success, moves the .msg object to processed/
  • Propagates last detected booking to blank items of the same file
• app_emails.py – Flask web app:
  • Connects to Oracle using wallet (WALLET_PATH) and DB_ALIAS
  • Reads from PROCESSED_EMAILS_REDACTED view (recommended) or the base table
  • Provides filters: booking, alert (YES/NO), status, date range, and option to include forwarded-only messages
  • Shows timeline sorted by DATE_SENT ASC
  • Route /view_email/<file_name> renders the original .msg (downloaded from OCI) for inspection

---

6) Data model

6.1 Table

PROCESSED_EMAILS (from the inline DDL in process_emails_v2.py):

CREATE TABLE PROCESSED_EMAILS (
    ID               NUMBER GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    DATE_SENT       TIMESTAMP,
    BOOKING          VARCHAR2(100),
    EMAIL_FROM       VARCHAR2(500),
    EMAIL_TO         VARCHAR2(1000),
    SUBJECT          VARCHAR2(1000),
    STATUS           VARCHAR2(100),
    BRIEF_DESCRIPTION VARCHAR2(1000),
    ALERT            VARCHAR2(10),
    THREAD_INDEX     NUMBER(10),
    FILE_NAME        VARCHAR2(400),
    INSERT_DATE    TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

Tip: For the UI, create a sanitized view that redacts email addresses:

CREATE OR REPLACE VIEW PROCESSED_EMAILS_REDACTED AS
SELECT
    REGEXP_REPLACE(
            REGEXP_REPLACE(
                    REGEXP_REPLACE(
                            EMAIL_TO,
                            '([[:alpha:]][[:alpha:] .''()_-]*)<([^>]+)>',
                            '<\2>'
                    ),
                    '([A-Za-z])([A-Za-z0-9._%+-]*)@([A-Za-z])([A-Za-z0-9.-]*)(\.[A-Za-z]{2,}(?:\.[A-Za-z]{2,})*)',
                    '\1******@\3******\5',
                    1, 0, 'i'
            ),
            '\s*;\s*', '; '
    ) AS EMAIL_TO,
    REGEXP_REPLACE(
            REGEXP_REPLACE(
                    REGEXP_REPLACE(
                            EMAIL_FROM,
                            '([[:alpha:]][[:alpha:] .''()_-]*)<([^>]+)>',
                            '<\2>'
                    ),
                    '([A-Za-z])([A-Za-z0-9._%+-]*)@([A-Za-z])([A-Za-z0-9.-]*)(\.[A-Za-z]{2,}(?:\.[A-Za-z]{2,})*)',
                    '\1******@\3******\5',
                    1, 0, 'i'
            ),
            '\s*;\s*', '; '
    ) AS EMAIL_FROM,
    SUBJECT,
    REGEXP_REPLACE(
            BOOKING,
            '([A-Za-z0-9])([A-Za-z0-9 _.-]*)',
            '***'
    ) AS BOOKING,
    DATE_SENT,
    STATUS,
    FILE_NAME,
    THREAD_INDEX, INSERT_DATE, BRIEF_DESCRIPTION, ALERT
FROM PROCESSED_EMAILS;
/

6.2 Key normalization logic

• normalize_datetime() parses ISO‑8601 and common patterns, removes TZ for DB TIMESTAMP.
• sanitize_value() stringifies lists/dicts and trims empty values.

---

7) Configuration

Create a JSON file named ./config used by both scripts.

Minimum keys (adapt to your tenancy and DB):

{
  "oci_profile": "DEFAULT",
  "llm_endpoint": "https://inference.generativeai.us-phoenix-1.oci.oraclecloud.com",
  "namespace": "xxxxxxxxxxxxxxxx",
  "compartment_id": "ocid1.compartment.oc1..example",
  "bucket-profile": "DEFAULT",
  "bucket": "emails",
  "WALLET_PATH": "/path/to/db/wallet",
  "DB_ALIAS": "adb_high",
  "USERNAME": "APPUSER",
  "PASSWORD": "****"
}

In process_emails_v2.py the following constants are read/used:

• MODEL_ID="cohere.command-a-03-2025" (you can change to another model available in your tenancy)
• SERVICE_ENDPOINT, COMPARTMENT_ID, AUTH_PROFILE
• BUCKET_NAME, PROCESSED_PREFIX="processed/"
• DB_ALIAS, USERNAME, PASSWORD, WALLET_PATH

In app_emails.py you’ll also see:

• bucket-profile, bucket, compartment_id for Object Storage access
• The Flask app runs by default on 0.0.0.0:5015

OCI credentials come from ~/.oci/config under the chosen profile(s).

---

8) How the processing works (deep dive)

8.1 Extract the latest message from a .msg

• extract_msg_text(path) uses extract_msg.Message to read sender, to, subject, date, and body.
• Returns a dict with top (headers) and body.

8.2 Split threads reliably

• split_refined_email_thread(text) detects boundaries using a robust regex that looks for header groups (From/De, Sent/Sent, To/Para, Subject/Subject) and common separators like “Original Message”, “Original Message”, “Forwarded message”.
• Fallback to split_thread_by_headers() if needed.

8.3 Chunking & prompts

• chunk_by_size(text, MAX_CHARS_PER_CALL=10000) ensures each LLM call is bounded.
• prepare_blocks_for_llm(headers_topo, body) packs the first chunk with top headers if needed.
• build_prompt(bloco_texto, headers_topo) (Portuguese) directs the model to return JSON ONLY with fields:
  data_envio, booking, email_from, email_to, subject, status, brief_description, alert
  and maps domain‑specific phrases to standardized status values (e.g., “Send Note to Forwarder”, “Realizar Estufagem”, “Validar Peso”, “Approve Invoice”, etc.).

8.4 LLM call

• call_llm_extract() instantiates ChatOCIGenAI with:
  • model_id=MODEL_ID (default: cohere.command-a-03-2025)
  • temperature=0.0 for determinism, max_tokens≈1800, top_p=0.9
• Cleans fenced code blocks and parses JSON; takes first dict if a list is returned.

8.5 Persist results

• insert_into_database(item, file_name, thread_index) writes each communication to PROCESSED_EMAILS.
• After processing a file, the last detected booking is propagated to rows where it’s empty (same FILE_NAME).

8.6 Idempotence

• Objects are moved to processed/ after success to prevent reprocessing.

---

9) Running locally (end‑to‑end)

9.1 Prerequisites

• Python 3.10+
• Oracle Instant Client (for thick mode) and DB wallet (place path in WALLET_PATH)
• Access to OCI Tenancy with permissions to read the chosen Object Storage bucket
• A table PROCESSED_EMAILS created in your schema

Install dependencies (example):

python -m venv .venv
source .venv/bin/activate  # Windows: .venv\Scripts\activate
pip install flask oracledb oci extract_msg langchain langchain-core langchain-community

9.2 Prepare Object Storage

• Create bucket (e.g., emails).
• Upload Outlook .msg files into emails/inbox/ (or your chosen prefix).
  Ensure the paths in process_emails_v2.py match your structure.

9.3 Configure

• Create ./config (see section 7).
• Ensure ~/.oci/config has the profile named in "oci_profile" / "bucket-profile".

9.4 Process emails

Run the batch once (or schedule via cron):

python process_emails_v2.py

Expected logs:

• Files listed and downloaded
• Thread chunks sent to LLM
• Rows inserted in PROCESSED_EMAILS
• File moved to processed/

9.5 Start the web app

python app_emails.py

Open: http://localhost:5015/

Use filters:

• Booking: case‑insensitive contains
• Alert: YES or NO
• Status: dropdown from SELECT DISTINCT STATUS …
• Date range: inicio / fim (inclusive)
• Include forwarded‑only: checkbox to include messages without a classified status

Click on entries to view the original email via /view_email/<file_name>.

---

10) Testing tips

• Add a small test .msg with a short thread containing an English and a Portuguese part referencing export steps (e.g., DU‑e, invoice approval, warehouse dates).
• Verify:
  • status gets standardized per prompt rules
  • alert hits “YES” when there’s a delay/missing document/urgent note
  • Multiple communications in the same thread create multiple rows with incrementing THREAD_INDEX
  • Timeline sorts by DATE_SENT ASC
• Create the REDACTED view and point the app to it if sharing the UI broadly.

---

11) Operations & scaling

• Throughput: adjust MAX_CHARS_PER_CALL and chunking to balance latency/cost.
• Retries: wrap call_llm_extract() with exponential backoff if needed.
• Observability: log file_name, thread_index, LLM latency, and DB writes.
• Security: keep wallets and OCI config out of source control; use per‑environment profiles.
• Cost: batch larger emails off‑peak; cache results for unchanged .msg object hashes.

---

12) Appendix – notable functions (by file)

process_emails_v2.py

• list_files() – list .msg in bucket
• download_file(name) – temp‑download object
• move_to_processed(name) – move to processed/
• extract_msg_text(path) – read .msg headers/body
• split_refined_email_thread(texto) / split_thread_by_headers(texto) – reliable thread splitter
• chunk_by_size(text) / prepare_blocks_for_llm(headers, body) – chunk & pack
• build_prompt(bloco, headers_topo) – robust JSON‑only, domain‑specific prompt
• call_llm_extract(bloco, headers_topo) – OCI GenAI call + JSON parse
• insert_into_database(item, file_name, thread_index) – DB insert
• process_emails() – the pipeline entrypoint

app_emails.py

• search_emails() – main route “/”: filters + timeline using PROCESSED_EMAILS_REDACTED
• get_distinct_status() – populates status dropdown
• view_email(file_name) – renders original .msg body for audit

---

13) FAQ

Can I change the LLM?
Yes. Edit MODEL_ID and SERVICE_ENDPOINT. Keep the prompt constraints (JSON‑only) for reliable parsing.

Do I need the redacted view?
Recommended for sharing the UI. The processor writes raw values; the view masks addresses for privacy.

Where do processed .msg go?
To processed/ (prefix configurable by PROCESSED_PREFIX).

How do I handle attachments?
Extend extract_msg_text() to enumerate attachments from extract_msg and store metadata in a separate table.

---

Acknowledgments

• Author - Cristiano Hoshikawa (Oracle LAD A-Team Solution Engineer)