File size: 9,804 Bytes
d712cef | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 | import sqlite3
import json
import os
import time
from datetime import datetime, timedelta
from EmailGenerator import generate_cold_email
# --- CONSTANTS & PATHS ---
BASE_PATH = os.environ.get('WORKSPACE_ROOT', '.')
# Source: companies imported from .xlsx / .numbers files
DB_EXCEL_DATA = os.path.join(BASE_PATH, 'Database/outreach_data/excel_data.db')
# Destination: review queue shown in the frontend dashboard
DB_TRACKING = os.path.join(BASE_PATH, 'Database/EmailsUnderReview/emailsUnderReview.db')
# Ensure both directories exist before any DB connection is attempted
os.makedirs(os.path.dirname(DB_EXCEL_DATA), exist_ok=True)
os.makedirs(os.path.dirname(DB_TRACKING), exist_ok=True)
# ─────────────────────────────────────────────────────────────────────────────
# DATABASE SETUP
# ─────────────────────────────────────────────────────────────────────────────
def init_tracking_db():
"""Create the tracking table if it doesn't exist yet."""
os.makedirs(os.path.dirname(DB_TRACKING), exist_ok=True)
with sqlite3.connect(DB_TRACKING) as conn:
conn.execute('''
CREATE TABLE IF NOT EXISTS tracking (
id INTEGER PRIMARY KEY AUTOINCREMENT,
body_json TEXT,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
followup_date DATETIME,
status TEXT DEFAULT 'under review',
-- core identifiers
company_name TEXT,
company_email TEXT,
-- rich outreach fields (carried from excel_data.db)
website TEXT,
address TEXT,
total_shipments TEXT,
top_suppliers TEXT,
hs_codes TEXT,
company_description TEXT,
key_executives TEXT,
deep_research_pdf TEXT,
-- email meta
generated_subject TEXT,
date_added DATETIME,
Unique_application_id TEXT
)
''')
conn.commit()
def dict_factory(cursor, row):
return {col[0]: row[idx] for idx, col in enumerate(cursor.description)}
# ─────────────────────────────────────────────────────────────────────────────
# DUPLICATE CHECK
# ─────────────────────────────────────────────────────────────────────────────
def already_tracked(company_name: str, company_email: str) -> bool:
"""Return True if this company is already in the tracking DB."""
with sqlite3.connect(DB_TRACKING) as conn:
row = conn.execute(
"SELECT id FROM tracking WHERE company_name = ? AND company_email = ?",
(company_name, company_email)
).fetchone()
return row is not None
# ─────────────────────────────────────────────────────────────────────────────
# CORE AGENT LOGIC
# ─────────────────────────────────────────────────────────────────────────────
def process_next_company():
"""
Picks the next unprocessed company from outreach_companies,
generates a cold email, saves everything to emailsUnderReview.db,
then marks the source row as 'processed'.
"""
# 1. Fetch next pending company
try:
with sqlite3.connect(DB_EXCEL_DATA) as conn:
conn.row_factory = dict_factory
company = conn.execute(
"SELECT * FROM outreach_companies WHERE status = 'under_review' LIMIT 1"
).fetchone()
except sqlite3.OperationalError:
# Table doesn't exist yet — Excel_Processor.py hasn't been run
print(f"[{datetime.now():%H:%M:%S}] Source table not initialised yet. Run Excel_Processor.py first.")
return False
if not company:
print(f"[{datetime.now():%H:%M:%S}] No pending companies found. Sleeping...")
return False
company_name = company.get("company_name", "Unknown")
company_email = company.get("email", "not updated")
# 2. Duplicate guard
if already_tracked(company_name, company_email):
print(f"[{datetime.now():%H:%M:%S}] Duplicate: '{company_name}' already tracked — marking processed.")
with sqlite3.connect(DB_EXCEL_DATA) as conn:
conn.execute(
"UPDATE outreach_companies SET status = 'processed' WHERE id = ?",
(company["id"],)
)
return True
# 3. Generate cold email
print(f"[{datetime.now():%H:%M:%S}] Generating email for '{company_name}'...")
try:
raw_email = generate_cold_email(company) # returns "Subject: ...\n\n<body>"
except Exception as e:
print(f" ❌ Email generation failed: {e}")
return False
# Split subject from body
subject, body = "", raw_email
if raw_email.startswith("Subject:"):
lines = raw_email.split("\n\n", 1)
subject = lines[0].replace("Subject:", "").strip()
body = lines[1].strip() if len(lines) > 1 else raw_email
# 4. Build the JSON blob stored in body_json (keeps backward compat with the frontend)
# 5. Generate a persistent Unique Application ID immediately
import random, string
unique_app_id = ''.join(random.choices(string.digits, k=20))
final_payload = {
"body": {
"generated_content": body,
"subject": subject,
"Unique_application_id": unique_app_id,
"outreach_data": {
"company_name": company_name,
"company_email": company_email,
"website": company.get("website", ""),
"address": company.get("address", ""),
"total_shipments": company.get("total_shipments", ""),
"top_suppliers": company.get("top_suppliers", ""),
"hs_codes": company.get("hs_codes", ""),
"company_description": company.get("company_description", ""),
"key_executives": company.get("key_executives", ""),
"deep_research_pdf": company.get("deep_research_pdf", ""),
}
}
}
body_json_str = json.dumps(final_payload)
# 5. Save to tracking DB
now = datetime.now()
followup_dt = now + timedelta(days=4)
with sqlite3.connect(DB_TRACKING) as conn:
conn.execute('''
INSERT INTO tracking (
body_json, timestamp, followup_date, status,
company_name, company_email,
website, address, total_shipments, top_suppliers,
hs_codes, company_description, key_executives, deep_research_pdf,
generated_subject, date_added, Unique_application_id
) VALUES (
?, ?, ?, ?,
?, ?,
?, ?, ?, ?,
?, ?, ?, ?,
?, ?, ?
)
''', (
body_json_str,
now.strftime("%Y-%m-%d %H:%M:%S"),
followup_dt.strftime("%Y-%m-%d %H:%M:%S"),
'under review',
company_name,
company_email,
company.get("website", ""),
company.get("address", ""),
company.get("total_shipments", ""),
company.get("top_suppliers", ""),
company.get("hs_codes", ""),
company.get("company_description", ""),
company.get("key_executives", ""),
company.get("deep_research_pdf", ""),
subject,
now.strftime("%Y-%m-%d %H:%M:%S"),
unique_app_id
))
conn.commit()
# 6. Mark source row as processed
with sqlite3.connect(DB_EXCEL_DATA) as conn:
conn.execute(
"UPDATE outreach_companies SET status = 'processed' WHERE id = ?",
(company["id"],)
)
conn.commit()
print(f" ✅ Saved to tracking DB — '{company_name}' | email: {company_email}")
print(f" Subject: {subject}")
return True
# ─────────────────────────────────────────────────────────────────────────────
# AGENT DAEMON LOOP
# ─────────────────────────────────────────────────────────────────────────────
if __name__ == "__main__":
print("=" * 60)
print(" Textile Outreach Agent — starting single run")
print("=" * 60)
init_tracking_db()
try:
result = process_next_company()
if not result:
print("Nothing to process.")
except Exception as e:
print(f"[{datetime.now():%H:%M:%S}] Agent error: {e}")
raise |