commit 4066c8d73730207e002a9944983e199f0a04255d Author: Cristiano Hoshikawa Date: Wed Nov 12 08:53:59 2025 -0300 first commit diff --git a/.idea/.gitignore b/.idea/.gitignore new file mode 100644 index 0000000..7d9a8e5 --- /dev/null +++ b/.idea/.gitignore @@ -0,0 +1,12 @@ +# Default ignored files +/shelf/ +/workspace.xml +# Editor-based HTTP Client requests +/httpRequests/ +# Environment-dependent path to Maven home directory +/mavenHomeManager.xml +# Datasource local storage ignored files +/dataSources/ +/dataSources.local.xml +# Zeppelin ignored files +/ZeppelinRemoteNotebooks/ diff --git a/.idea/ai_logistic_outlook_integration.iml b/.idea/ai_logistic_outlook_integration.iml new file mode 100644 index 0000000..38e06ab --- /dev/null +++ b/.idea/ai_logistic_outlook_integration.iml @@ -0,0 +1,20 @@ + + + + + + + + + + + + + + \ No newline at end of file diff --git a/.idea/codeStyles/Project.xml b/.idea/codeStyles/Project.xml new file mode 100644 index 0000000..919ce1f --- /dev/null +++ b/.idea/codeStyles/Project.xml @@ -0,0 +1,7 @@ + + + + + + \ No newline at end of file diff --git a/.idea/codeStyles/codeStyleConfig.xml b/.idea/codeStyles/codeStyleConfig.xml new file mode 100644 index 0000000..a55e7a1 --- /dev/null +++ b/.idea/codeStyles/codeStyleConfig.xml @@ -0,0 +1,5 @@ + + + + \ No newline at end of file diff --git a/.idea/misc.xml b/.idea/misc.xml new file mode 100644 index 0000000..89ee753 --- /dev/null +++ b/.idea/misc.xml @@ -0,0 +1,6 @@ + + + + + + \ No newline at end of file diff --git a/.idea/modules.xml b/.idea/modules.xml new file mode 100644 index 0000000..9c85304 --- /dev/null +++ b/.idea/modules.xml @@ -0,0 +1,8 @@ + + + + + + + + \ No newline at end of file diff --git a/.idea/vcs.xml b/.idea/vcs.xml new file mode 100644 index 0000000..8306744 --- /dev/null +++ b/.idea/vcs.xml @@ -0,0 +1,7 @@ + + + + + + + \ No newline at end of file diff --git a/.oca/custom_code_review_guidelines.txt b/.oca/custom_code_review_guidelines.txt new file mode 100644 index 0000000..a0a3b63 --- /dev/null +++ b/.oca/custom_code_review_guidelines.txt @@ -0,0 +1,24 @@ +# Sample guideline, please follow similar structure for guideline with code samples +# 1. Suggest using streams instead of simple loops for better readability. +# +# *Comment: +# Category: Minor +# Issue: Use streams instead of a loop for better readability. +# Code Block: +# +# ```java +# // Calculate squares of numbers +# List squares = new ArrayList<>(); +# for (int number : numbers) { +# squares.add(number * number); +# } +# ``` +# Recommendation: +# +# ```java +# // Calculate squares of numbers +# List squares = Arrays.stream(numbers) +# .map(n -> n * n) // Map each number to its square +# .toList(); +# ``` +# diff --git a/README.md b/README.md new file mode 100644 index 0000000..1d588fa --- /dev/null +++ b/README.md @@ -0,0 +1,353 @@ +# 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. + +![img.png](images/img.png) + +--- + +## 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) + +```mermaid +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/`** 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`): + +```sql +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: +```sql +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): + +```json +{ + "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): +```bash +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): +```bash +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 +```bash +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/`. + +--- + +## 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) + diff --git a/files/app_emails.py b/files/app_emails.py new file mode 100644 index 0000000..201ce67 --- /dev/null +++ b/files/app_emails.py @@ -0,0 +1,233 @@ +# -*- coding: utf-8 -*- +from flask import Flask, render_template, request, send_file, redirect, url_for, render_template_string +import oracledb, json +from datetime import datetime +import io +import oci +import urllib.parse +import extract_msg + +# ============================================================ +# CONFIGURATION +# ============================================================ +with open("config", "r") as f: + config_data = json.load(f) + +WALLET_PATH = config_data["WALLET_PATH"] +DB_ALIAS = config_data["DB_ALIAS"] +USERNAME = config_data["USERNAME"] +PASSWORD = config_data["PASSWORD"] + +connection = oracledb.connect( + user=USERNAME, + password=PASSWORD, + dsn=DB_ALIAS, + config_dir=WALLET_PATH, + wallet_location=WALLET_PATH, + wallet_password=PASSWORD, +) + +# OCI client configuration (same standard as process_emails_v2_en.py) +BUCKET_PROFILE = config_data["bucket-profile"] +BUCKET_NAME = config_data["bucket"] +COMPARTMENT_ID = config_data["compartment_id"] +bucket_name = BUCKET_NAME + "/processed" + +# Initialize OCI client and get namespace +config = oci.config.from_file("~/.oci/config", BUCKET_PROFILE) +object_storage = oci.object_storage.ObjectStorageClient(config) +namespace = object_storage.get_namespace().data + +app = Flask(__name__) + +# ============================================================ +# HELPER FUNCTIONS +# ============================================================ +def get_distinct_status(): + """Return all distinct statuses + 'Forwarded Email' option.""" + with connection.cursor() as c: + c.execute("SELECT DISTINCT STATUS FROM PROCESSED_EMAILS ORDER BY STATUS") + values = [r[0] for r in c.fetchall() if r[0] and str(r[0]).strip()] + if "Forwarded Email" not in values: + values.append("Forwarded Email") + return values + +# ============================================================ +# MAIN ROUTE +# ============================================================ +@app.route("/", methods=["GET"]) +def search_emails(): + booking = request.args.get("booking", "").strip().upper() + alert_flag = request.args.get("alert", "") + status = request.args.get("status", "") + start_date = request.args.get("start", "") + end_date = request.args.get("end", "") + include_forwarded = request.args.get("include_forwarded") == "on" + + sql = """ + SELECT DATE_SENT, BOOKING, EMAIL_FROM, EMAIL_TO, SUBJECT, + STATUS, BRIEF_DESCRIPTION, ALERT, FILE_NAME, THREAD_INDEX + FROM PROCESSED_EMAILS_REDACTED + WHERE 1=1 \ + """ + params = {} + + if booking: + sql += " AND UPPER(BOOKING) LIKE :bk" + params["bk"] = f"%{booking}%" + if alert_flag in ("YES", "NO"): + sql += " AND ALERT = :alert" + params["alert"] = alert_flag + if status: + if status == "Forwarded Email": + sql += " AND (STATUS IS NULL OR TRIM(STATUS) = '')" + else: + sql += " AND STATUS = :status" + params["status"] = status + else: + # If the checkbox is NOT checked, exclude forwarded emails (STATUS is null) + if not include_forwarded: + sql += " AND (STATUS IS NOT NULL)" + + if start_date: + sql += " AND DATE_SENT >= TO_TIMESTAMP(:start, 'YYYY-MM-DD')" + params["start"] = start_date + if end_date: + sql += " AND DATE_SENT <= TO_TIMESTAMP(:end, 'YYYY-MM-DD') + INTERVAL '1' DAY" + params["end"] = end_date + + sql += " ORDER BY DATE_SENT ASC" + + with connection.cursor() as c: + c.execute(sql, params) + rows = c.fetchall() + cols = [d[0] for d in c.description] + records = [dict(zip(cols, row)) for row in rows] + + for r in records: + if isinstance(r["DATE_SENT"], datetime): + r["DATE_SENT"] = r["DATE_SENT"].strftime("%Y-%m-%d %H:%M:%S") + if not r.get("STATUS") or not str(r["STATUS"]).strip(): + r["STATUS"] = "Forwarded Email" + + status_options = get_distinct_status() + + return render_template( + "emails_filter.html", + records=records, + filters={ + "booking": booking, + "alert": alert_flag, + "status": status, + "start": start_date, + "end": end_date, + "include_forwarded": include_forwarded + }, + status_options=status_options + ) + +# ============================================================ +# ROUTE: VIEW EMAIL +# ============================================================ +@app.route("/view_email/") +def view_email(file_name): + try: + # Fix URL encoding + file_name = urllib.parse.unquote(file_name).strip() + + # Try multiple possible paths + possible_paths = [ + file_name, + f"processed/{file_name}", + f"inbox/{file_name}", + f"emails/{file_name}", + f"emails/processed/{file_name}", + ] + + msg_obj = None + for path in possible_paths: + try: + obj = object_storage.get_object(namespace, BUCKET_NAME, path) + msg_obj = extract_msg.Message(io.BytesIO(obj.data.content)) + print(f"[INFO] File found: {path}") + break + except oci.exceptions.ServiceError as e: + if e.status == 404: + continue + raise + + if not msg_obj: + return ( + f"

❌ Email not found in any tested path.

" + f"

Bucket: {BUCKET_NAME}
File: {file_name}

", + 404, + ) + + # Extract main fields + from_field = getattr(msg_obj, "sender", "") + to_field = getattr(msg_obj, "to", "") + subject = getattr(msg_obj, "subject", "") + date = getattr(msg_obj, "date", "") + body = getattr(msg_obj, "body", "") + + # Basic HTML rendering for readability + html = f""" + + + + {subject or "Viewed Email"} + + + + + + + """ + return render_template_string(html) + + except Exception as e: + return f"

Error reading the email:

{e}
", 500 + + +# ============================================================ +# MAIN ENTRY POINT +# ============================================================ +if __name__ == "__main__": + app.run(debug=True, host="0.0.0.0", port=5015) \ No newline at end of file diff --git a/files/config b/files/config new file mode 100644 index 0000000..79973b5 --- /dev/null +++ b/files/config @@ -0,0 +1,11 @@ +{ + "oci_profile": "DEFAULT", + "llm_endpoint": "https://inference.generativeai.us-phoenix-1.oci.oraclecloud.com", + "compartment_id": "ocid1.compartment.oc1..example", + "bucket-profile": "DEFAULT", + "bucket": "emails", + "WALLET_PATH": "/path/to/db/wallet", + "DB_ALIAS": "adb_high", + "USERNAME": "APPUSER", + "PASSWORD": "****" +} \ No newline at end of file diff --git a/files/process_emails_v2.py b/files/process_emails_v2.py new file mode 100644 index 0000000..479d3ec --- /dev/null +++ b/files/process_emails_v2.py @@ -0,0 +1,458 @@ +# -*- coding: utf-8 -*- +import oci +import json +import tempfile +import os +import re +from datetime import datetime +from typing import List, Dict, Any + +import oracledb +from extract_msg import Message as MsgReader +from oci.object_storage import ObjectStorageClient + +from langchain_core.prompts import ChatPromptTemplate +from langchain_community.chat_models.oci_generative_ai import ChatOCIGenAI + +# ============================================================ +# DATABASE TABLE DEFINITION +# ============================================================ +# CREATE TABLE PROCESSED_EMAILS ( +# ID NUMBER GENERATED ALWAYS AS IDENTITY, +# DATE_SENT TIMESTAMP, +# BOOKING VARCHAR2(100), +# EMAIL_FROM VARCHAR2(500), +# EMAIL_TO VARCHAR2(1000), +# SUBJECT VARCHAR2(1000), +# STATUS VARCHAR2(100), +# INSERT_DATE TIMESTAMP DEFAULT CURRENT_TIMESTAMP, +# PRIMARY KEY (ID) +# ); +# +# ALTER TABLE PROCESSED_EMAILS ADD ( +# BRIEF_DESCRIPTION VARCHAR2(1000), +# ALERT VARCHAR2(10), +# THREAD_INDEX NUMERIC(10), +# FILE_NAME VARCHAR(400) +# ); + +# ============================================================ +# CONFIGURATION +# ============================================================ +with open("config", "r") as f: + config_data = json.load(f) + +MODEL_ID = "cohere.command-a-03-2025" +SERVICE_ENDPOINT = config_data["llm_endpoint"] +COMPARTMENT_ID = config_data["compartment_id"] +AUTH_PROFILE = config_data["oci_profile"] + +WALLET_PATH = config_data["WALLET_PATH"] +DB_ALIAS = config_data["DB_ALIAS"] +USERNAME = config_data["USERNAME"] +PASSWORD = config_data["PASSWORD"] + +BUCKET_NAME = config_data["bucket"] +PROCESSED_PREFIX = "processed/" +CONFIG_PROFILE = AUTH_PROFILE + +MAX_CHARS_PER_CALL = 10000 # ~2500 tokens per block + +# ============================================================ +# OCI CLIENTS & DATABASE CONNECTION +# ============================================================ +config = oci.config.from_file("~/.oci/config", CONFIG_PROFILE) +namespace = oci.object_storage.ObjectStorageClient(config).get_namespace().data +object_client = ObjectStorageClient(config) + +connection = oracledb.connect( + user=USERNAME, + password=PASSWORD, + dsn=DB_ALIAS, + config_dir=WALLET_PATH, + wallet_location=WALLET_PATH, + wallet_password=PASSWORD, +) + +# ============================================================ +# UTILS: LIST AND MANAGE OBJECTS IN BUCKET +# ============================================================ +def list_files() -> List[str]: + """List all .msg files in the configured Object Storage bucket.""" + objs = object_client.list_objects(namespace, BUCKET_NAME, fields="name") + return [o.name for o in objs.data.objects if o.name.lower().endswith(".msg")] + + +def download_file(name: str) -> str: + """Download a .msg file from OCI Object Storage to a temporary folder.""" + temp_dir = tempfile.mkdtemp() + path = os.path.join(temp_dir, os.path.basename(name)) + with open(path, "wb") as f: + data = object_client.get_object(namespace, BUCKET_NAME, name).data.content + f.write(data) + return path + + +def move_to_processed(name: str): + """Move a processed file to the 'processed/' prefix in the same bucket.""" + destination = PROCESSED_PREFIX + os.path.basename(name) + obj = object_client.get_object(namespace, BUCKET_NAME, name) + object_client.put_object(namespace, BUCKET_NAME, destination, obj.data.content) + object_client.delete_object(namespace, BUCKET_NAME, name) + +# ============================================================ +# .MSG EXTRACTION +# ============================================================ +def extract_msg_text(path): + """ + Extract only the first (most recent) message from the .msg file. + Returns: + { + "top": {headers}, + "body": "message body" + } + """ + msg = MsgReader(path) + + top = { + "from": getattr(msg, "sender", ""), + "to": getattr(msg, "to", ""), + "subject": getattr(msg, "subject", ""), + "date": str(getattr(msg, "date", "")), + } + + body = getattr(msg, "body", "") or "" + + # Remove previous messages (older thread content) + thread_pattern = re.compile( + r"(?i)(^-{2,}\s*(original message|forwarded message|mensagem original)\s*-{2,}$|^from:|^sent:|^to:|^de:|^enviado:|^para:)", + re.MULTILINE + ) + + parts = thread_pattern.split(body) + cleaned_body = parts[0].strip() if parts else body.strip() + + return {"top": top, "body": cleaned_body} + +# ============================================================ +# THREAD SPLITTING +# ============================================================ +HEADER_PATTERNS = [ + r"^-{2,}\s*Original Message\s*-{2,}$", + r"^-{2,}\s*Forwarded message\s*-{2,}$", + r"^From:\s", r"^Sent:\s", r"^To:\s", r"^Subject:\s" +] +BOUNDARY_REGEX = re.compile("|".join(HEADER_PATTERNS), re.IGNORECASE | re.MULTILINE) + + +def split_email_thread_refined(text): + """ + Split an email body into message blocks within a thread, + considering standard header sets. + """ + text = text.replace('\r\n', '\n').strip() + + regex_start_email = re.compile( + r"(?=^(From|De):.*\n(Sent|Enviado):.*\n(To|Para):.*)", + re.MULTILINE | re.IGNORECASE + ) + + parts = regex_start_email.split(text) + blocks = [] + + if len(parts) <= 1: + return [text.strip()] if text.strip() else [] + + current = "" + for line in parts: + line = line.strip() + if not line: + continue + + if re.match(r"^(From|De):", line, re.IGNORECASE): + if current: + blocks.append(current.strip()) + current = line + else: + current += "\n" + line + + if current.strip(): + blocks.append(current.strip()) + + cleaned_blocks = [] + for b in blocks: + b = b.strip() + if b and b not in cleaned_blocks: + cleaned_blocks.append(b) + + return cleaned_blocks + + +def split_thread_by_headers(body: str) -> List[str]: + lines = body.splitlines() + blocks, curr = [], [] + + def flush(): + if curr: + text = "\n".join(curr).strip() + if text: + blocks.append(text) + + for line in lines: + if BOUNDARY_REGEX.match(line.strip()) and curr: + flush() + curr = [line] + else: + curr.append(line) + flush() + + if not blocks: + return [body.strip()] if body.strip() else [] + return blocks + + +def chunk_by_size(text: str, max_chars: int) -> List[str]: + """Split text into chunks within a character limit.""" + text = text.strip() + if len(text) <= max_chars: + return [text] + chunks, curr = [], "" + for part in re.split(r"(\n\s*\n)", text): + if len(curr) + len(part) <= max_chars: + curr += part + else: + if curr.strip(): + chunks.append(curr.strip()) + if len(part) > max_chars: + for i in range(0, len(part), max_chars): + chunks.append(part[i:i + max_chars].strip()) + curr = "" + else: + curr = part + if curr.strip(): + chunks.append(curr.strip()) + return chunks + + +def prepare_blocks_for_llm(body: str, top_headers: Dict[str, Any]) -> List[str]: + """Prepare text blocks for LLM processing.""" + body = (body or "").strip() + if not body: + hint = f"From: {top_headers.get('from','')}\nTo: {top_headers.get('to','')}\nSubject: {top_headers.get('subject','')}\nDate: {top_headers.get('date','')}" + return [hint] + by_headers = split_thread_by_headers(body) + result = [] + for b in by_headers: + for c in chunk_by_size(b, MAX_CHARS_PER_CALL): + if c.strip(): + result.append(c.strip()) + if not result: + hint = f"From: {top_headers.get('from','')}\nTo: {top_headers.get('to','')}\nSubject: {top_headers.get('subject','')}\nDate: {top_headers.get('date','')}\n\n{body[:8000]}" + result = [hint] + return result + +# ============================================================ +# PROMPT (ENGLISH VERSION) +# ============================================================ +def build_prompt(block_text: str, top_headers: Dict[str, Any]) -> ChatPromptTemplate: + sys_text = """ +You are a data extractor for corporate logistics emails. +Each email may contain multiple communications (replies, forwards, or quoted messages) within the same thread. +Your task is to identify each communication separately and extract the structured fields below, returning everything as a JSON list. + +Extraction rules: + +1. **Communication identification** + - Each individual message starts when a new header with "From:", "Sent:", or similar appears. + - Also consider blocks separated by "Original Message" or "Forwarded Message". + - Each identified block must generate a separate JSON object. + +2. **Fields to extract (per message)** + - "date_sent": Date and time the message was sent (ISO 8601). + - "booking": Booking or shipment code (“Booking”, “Bkg”, “BK”, etc.). + - "email_from": Sender (“From:”). + - "email_to": Main recipients (“To:”). + - "subject": Message subject (“Subject:”). + - "status": Operational status inferred from text using logistics keywords (shipment, invoice, forwarding, approval, etc.). + - "brief_description": Short (1–2 lines) natural language summary. + - "alert": "YES" if the message contains urgency, problems, delays, or missing documents; "NO" otherwise. + +3. **Output format** + - Return ONLY JSON (no extra text). + - Example: + [ + { + "date_sent": "", + "booking": "", + "email_from": "", + "email_to": "", + "subject": "", + "status": "", + "brief_description": "", + "alert": "" + } + ] + +Context: +From: {from_hint} +To: {to_hint} +Subject: {subject_hint} +Date: {date_hint} + +Text to analyze: +""" + prompt = ChatPromptTemplate.from_messages([ + ("system", sys_text), + ("human", block_text) + ]) + prompt = prompt.partial( + from_hint=top_headers.get("from", ""), + to_hint=top_headers.get("to", ""), + subject_hint=top_headers.get("subject", ""), + date_hint=top_headers.get("date", ""), + ) + return prompt + +# ============================================================ +# LLM AND DATABASE OPERATIONS +# ============================================================ +def clean_markdown_fences(text: str) -> str: + text = text.strip() + text = re.sub(r"^```[a-zA-Z]*\s*", "", text) + text = re.sub(r"\s*```$", "", text) + return text.strip() + + +def call_llm_extract(block_text: str, top_headers: Dict[str, Any]) -> Dict[str, Any]: + llm = ChatOCIGenAI( + model_id=MODEL_ID, + service_endpoint=SERVICE_ENDPOINT, + compartment_id=COMPARTMENT_ID, + auth_profile=AUTH_PROFILE, + model_kwargs={"temperature": 0.0, "max_tokens": 1800, "top_p": 0.9}, + ) + prompt = build_prompt(block_text, top_headers) + resp = llm.invoke(prompt.format_messages()) + content = clean_markdown_fences(resp.content or "") + try: + data = json.loads(content) + if isinstance(data, list) and data: + return data[0] + if isinstance(data, dict): + return data + return {"error": "Unexpected format", "raw": content[:500]} + except Exception: + return {"error": "Invalid JSON", "raw": content[:500]} + + +def normalize_datetime(value): + if value is None: + return None + if isinstance(value, datetime): + return value + s = str(value).strip() + if not s: + return None + try: + if s.endswith("Z"): + s = s[:-1] + "+00:00" + return datetime.fromisoformat(s).replace(tzinfo=None) + except Exception: + for fmt in ("%Y-%m-%dT%H:%M:%S", "%Y-%m-%d"): + try: + return datetime.strptime(s, fmt) + except Exception: + continue + return None + + +def sanitize_value(v): + if v is None: + return None + if isinstance(v, (list, tuple, set)): + return ", ".join(str(x) for x in v) + if isinstance(v, dict): + return json.dumps(v, ensure_ascii=False) + return str(v) + + +def insert_into_database(item, file_name, thread_index): + with connection.cursor() as cursor: + binds = { + "date_sent": normalize_datetime(item.get("date_sent")), + "booking": sanitize_value(item.get("booking")), + "email_from": sanitize_value(item.get("email_from")), + "email_to": sanitize_value(item.get("email_to")), + "subject": sanitize_value(item.get("subject")), + "status": sanitize_value(item.get("status")), + "brief_description": sanitize_value(item.get("brief_description")), + "alert": sanitize_value(item.get("alert")), + "file_name": file_name, + "thread_index": thread_index + } + cursor.execute(""" + INSERT INTO PROCESSED_EMAILS + (DATE_SENT, BOOKING, EMAIL_FROM, EMAIL_TO, SUBJECT, STATUS, BRIEF_DESCRIPTION, ALERT, FILE_NAME, THREAD_INDEX) + VALUES + (:date_sent, :booking, :email_from, :email_to, :subject, :status, :brief_description, :alert, :file_name, :thread_index) + """, binds) + connection.commit() + +# ============================================================ +# PIPELINE +# ============================================================ +def process_emails(): + files = list_files() + for name in files: + print(f"\n=== Processing: {name} ===") + path = download_file(name) + data = extract_msg_text(path) + top_headers = data["top"] + body = data["body"] + blocks = split_email_thread_refined(body) + + last_booking_detected = None + thread_counter = 0 + + for block in blocks: + print("-------------------------------------") + print("headers: ", top_headers) + print("block: ", block) + + thread_counter += 1 + result = call_llm_extract(block, top_headers) + + if isinstance(result, dict) and "error" not in result: + current_booking = (result.get("booking") or "").strip() + + if current_booking and current_booking != last_booking_detected: + print(f"[INFO] New booking detected: {current_booking}") + last_booking_detected = current_booking + + elif not current_booking and last_booking_detected: + result["booking"] = last_booking_detected + + insert_into_database(result, + file_name=os.path.basename(name), + thread_index=thread_counter) + print(f"[OK] Thread {thread_counter}: {result}") + + else: + print(f"[WARNING] Error in block {thread_counter}: {result}") + + if last_booking_detected: + with connection.cursor() as cur: + cur.execute(""" + UPDATE PROCESSED_EMAILS + SET BOOKING = :booking + WHERE FILE_NAME = :file_name + AND (BOOKING IS NULL OR TRIM(BOOKING) = '') + """, {"booking": last_booking_detected, "file_name": os.path.basename(name)}) + connection.commit() + print(f"[UPDATE] Booking '{last_booking_detected}' propagated in {os.path.basename(name)}") + + move_to_processed(name) + + +if __name__ == "__main__": + process_emails() \ No newline at end of file diff --git a/files/redaction_23ai.sql b/files/redaction_23ai.sql new file mode 100644 index 0000000..cefb238 --- /dev/null +++ b/files/redaction_23ai.sql @@ -0,0 +1,77 @@ + CREATE TABLE PROCESSED_EMAILS ( + ID NUMBER GENERATED ALWAYS AS IDENTITY, + DATE_SENT TIMESTAMP, + BOOKING VARCHAR2(100), + EMAIL_FROM VARCHAR2(500), + EMAIL_TO VARCHAR2(1000), + SUBJECT VARCHAR2(1000), + STATUS VARCHAR2(100), + INSERT_DATE TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY (ID) + ); + + ALTER TABLE PROCESSED_EMAILS ADD ( + BRIEF_DESCRIPTION VARCHAR2(1000), + ALERT VARCHAR2(10), + THREAD_INDEX NUMERIC(10), + FILE_NAME VARCHAR(400) + ); + +CREATE OR REPLACE VIEW PROCESSED_EMAILS_REDACTED AS +SELECT +---------------------------------------------------------------------- +-- EMAIL_TO: remove names, parentheses, and mask all email addresses +---------------------------------------------------------------------- +REGEXP_REPLACE( + REGEXP_REPLACE( + REGEXP_REPLACE( + -- 1️⃣ Remove names and content inside parentheses before the + EMAIL_TO, +'([[:alpha:]][[:alpha:] .''()_-]*)<([^>]+)>', +'<\2>' +), +-- 2️⃣ Mask email addresses +'([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' +), +-- 3️⃣ Normalize ";" separators +'\s*;\s*', '; ' +) AS EMAIL_TO, + +---------------------------------------------------------------------- +-- EMAIL_FROM: same treatment (remove names, parentheses, mask) +---------------------------------------------------------------------- +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, + +---------------------------------------------------------------------- +-- Mask BOOKING (for example, a code or sensitive text) +---------------------------------------------------------------------- +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; \ No newline at end of file diff --git a/files/requirements.txt b/files/requirements.txt new file mode 100644 index 0000000..4095aa2 --- /dev/null +++ b/files/requirements.txt @@ -0,0 +1,3 @@ +oci +extract-msg +oracledb diff --git a/files/templates/emails_filter.html b/files/templates/emails_filter.html new file mode 100644 index 0000000..8fdb772 --- /dev/null +++ b/files/templates/emails_filter.html @@ -0,0 +1,208 @@ + + + + + + Email Query - International Logistics + + + +
+

Email Search

+
+
+ + +
+
+ + +
+
+ + +
+
+ + +
+
+ + +
+
+ +
+
+ +
+
+ + {% if records|length == 0 %} +

No results found.

+ {% endif %} + + {% for e in records %} + + {% endfor %} +
+ + \ No newline at end of file diff --git a/images/img.png b/images/img.png new file mode 100644 index 0000000..a38b232 Binary files /dev/null and b/images/img.png differ