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:

Ingest Outlook emails (.msg) placed in an OCI Object Storage bucket (e.g., inbox/).
Parse each email (including forwarded/replied headers) and split the thread into individual communications.
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.)
Persist the result to Oracle Database (PROCESSED_EMAILS), one row per communication (thread index).
Move processed files to a processed/ prefix to avoid re‑processing.
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)

README.md Unescape Escape

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