Extends the pipeline with infosec classification, sender profile tracking, and configurable purge rules. Adds a web dashboard for managing rules and monitoring email processing. Includes new migrations and seed script. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
170 lines
5.9 KiB
Python
170 lines
5.9 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Seed howl database from pulse_autotask (Autotask contacts/companies).
|
|
|
|
Sources:
|
|
customers + customer_emails <- company_type=1 (Customer) active contacts w/ emails
|
|
vendors + vendor_emails <- company_type=7 (Partner) active contacts w/ emails
|
|
whitelist <- unique domains from all client company email addresses
|
|
"""
|
|
|
|
import psycopg2
|
|
import re
|
|
|
|
PULSE_PARAMS = dict(host='localhost', port=5432, user='pulse_user',
|
|
password='9KuYTjjGEB7NsJc_togj6R9wYLRRrhudZiR4@i@N',
|
|
dbname='pulse_autotask')
|
|
HOWL_PARAMS = dict(host='localhost', port=5434, user='howl',
|
|
password='ztPEGUexRYkfWqUtd15WMdJF98N1jsui',
|
|
dbname='howl')
|
|
|
|
EMAIL_RE = re.compile(r'^[^@\s]+@[^@\s]+\.[^@\s]+$')
|
|
|
|
def valid_email(e):
|
|
return e and EMAIL_RE.match(e.strip())
|
|
|
|
def domain_of(e):
|
|
return e.strip().lower().split('@')[1] if e and '@' in e else None
|
|
|
|
def seed_contacts(pulse_cur, howl_cur, company_type, parent_table, email_table, fk_col):
|
|
"""Generic: pull contacts of given company_type, insert parent rows then email rows."""
|
|
pulse_cur.execute("""
|
|
SELECT
|
|
trim(concat(c.first_name, ' ', c.last_name)) AS name,
|
|
co.company_name AS company,
|
|
COALESCE(NULLIF(c.phone,''), NULLIF(co.phone,'')) AS phone,
|
|
c.email_address,
|
|
c.email_address2,
|
|
c.email_address3,
|
|
c.primary_contact
|
|
FROM contacts c
|
|
JOIN companies co ON c.company_id = co.id
|
|
WHERE co.company_type = %s
|
|
AND co.is_active = true AND co.is_deleted = false
|
|
AND c.is_active = true AND c.is_deleted = false
|
|
AND c.email_address IS NOT NULL AND c.email_address != ''
|
|
ORDER BY co.company_name, c.primary_contact DESC, c.last_name, c.first_name
|
|
""", (company_type,))
|
|
rows = pulse_cur.fetchall()
|
|
|
|
inserted_parents = 0
|
|
inserted_emails = 0
|
|
skipped_emails = 0
|
|
seen_emails = set() # guard against dupes within this batch
|
|
|
|
for row in rows:
|
|
name, company, phone, e1, e2, e3, is_primary = row
|
|
|
|
# Collect valid emails for this contact
|
|
emails = []
|
|
for addr in [e1, e2, e3]:
|
|
if valid_email(addr):
|
|
emails.append(addr.strip().lower())
|
|
if not emails:
|
|
continue
|
|
|
|
# Insert parent row
|
|
howl_cur.execute(
|
|
f"INSERT INTO {parent_table} (name, company, phone) VALUES (%s,%s,%s) RETURNING id",
|
|
(name or 'Unknown', company, phone)
|
|
)
|
|
parent_id = howl_cur.fetchone()[0]
|
|
inserted_parents += 1
|
|
|
|
# Insert email rows
|
|
for i, addr in enumerate(emails):
|
|
if addr in seen_emails:
|
|
skipped_emails += 1
|
|
continue
|
|
try:
|
|
howl_cur.execute(
|
|
f"INSERT INTO {email_table} ({fk_col}, email_address, label, is_primary) VALUES (%s,%s,%s,%s)",
|
|
(parent_id, addr, 'work', i == 0 and is_primary)
|
|
)
|
|
seen_emails.add(addr)
|
|
inserted_emails += 1
|
|
except psycopg2.errors.UniqueViolation:
|
|
howl_cur.connection.rollback()
|
|
skipped_emails += 1
|
|
except Exception as e:
|
|
howl_cur.connection.rollback()
|
|
print(f" WARN email {addr}: {e}")
|
|
skipped_emails += 1
|
|
|
|
return inserted_parents, inserted_emails, skipped_emails
|
|
|
|
|
|
def seed_whitelist(pulse_cur, howl_cur):
|
|
"""Insert one whitelist row per unique domain used by customer company contacts."""
|
|
pulse_cur.execute("""
|
|
SELECT DISTINCT lower(split_part(c.email_address, '@', 2)) AS domain
|
|
FROM contacts c
|
|
JOIN companies co ON c.company_id = co.id
|
|
WHERE co.company_type = 1
|
|
AND co.is_active = true AND co.is_deleted = false
|
|
AND c.is_active = true AND c.is_deleted = false
|
|
AND c.email_address LIKE '%@%'
|
|
ORDER BY 1
|
|
""")
|
|
domains = [r[0] for r in pulse_cur.fetchall() if r[0] and '.' in r[0]]
|
|
|
|
inserted = 0
|
|
for domain in domains:
|
|
try:
|
|
howl_cur.execute(
|
|
"INSERT INTO whitelist (domain, description, added_by) VALUES (%s,%s,%s)",
|
|
(domain, 'Auto-seeded from Autotask customer contacts', 'seed_from_pulse.py')
|
|
)
|
|
inserted += 1
|
|
except psycopg2.errors.UniqueViolation:
|
|
howl_cur.connection.rollback()
|
|
except Exception as e:
|
|
howl_cur.connection.rollback()
|
|
print(f" WARN domain {domain}: {e}")
|
|
|
|
return inserted
|
|
|
|
|
|
def main():
|
|
print("Connecting to pulse_autotask …")
|
|
pulse = psycopg2.connect(**PULSE_PARAMS)
|
|
pulse.autocommit = False
|
|
pc = pulse.cursor()
|
|
|
|
print("Connecting to howl …")
|
|
howl = psycopg2.connect(**HOWL_PARAMS)
|
|
howl.autocommit = False
|
|
hc = howl.cursor()
|
|
|
|
try:
|
|
# --- Customers (company_type=1) ---
|
|
print("\nSeeding customers …")
|
|
cp, ce, cs = seed_contacts(pc, hc, 1, 'customers', 'customer_emails', 'customer_id')
|
|
print(f" customers: {cp:4d} rows")
|
|
print(f" customer_emails: {ce:4d} rows ({cs} skipped/dupes)")
|
|
|
|
# --- Vendors / Partners (company_type=7) ---
|
|
print("\nSeeding vendors …")
|
|
vp, ve, vs = seed_contacts(pc, hc, 7, 'vendors', 'vendor_emails', 'vendor_id')
|
|
print(f" vendors: {vp:4d} rows")
|
|
print(f" vendor_emails: {ve:4d} rows ({vs} skipped/dupes)")
|
|
|
|
# --- Whitelist domains ---
|
|
print("\nSeeding whitelist …")
|
|
wl = seed_whitelist(pc, hc)
|
|
print(f" whitelist: {wl:4d} domain rows")
|
|
|
|
howl.commit()
|
|
print("\nDone — all changes committed.")
|
|
|
|
except Exception as e:
|
|
howl.rollback()
|
|
print(f"\nERROR — rolled back: {e}")
|
|
raise
|
|
finally:
|
|
pc.close(); pulse.close()
|
|
hc.close(); howl.close()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|