mirror of
https://github.com/hoshikawa2/ai_logistic_outlook_integration.git
synced 2026-07-09 16:44:21 +00:00
first commit
This commit is contained in:
458
files/process_emails_v2.py
Normal file
458
files/process_emails_v2.py
Normal file
@@ -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()
|
||||
Reference in New Issue
Block a user