procyon/db.py

# Procyon — Job Hunt Workflow Engine
# Multi-step pipeline: discover -> score -> tailor -> audit -> approve -> send -> track

import sqlite3
import os
from datetime import datetime, timezone

DB_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'data')
DB = os.path.join(DB_DIR, 'procyon.db')


def now():
    return datetime.now(timezone.utc).isoformat()


def get_conn():
    os.makedirs(DB_DIR, exist_ok=True)
    conn = sqlite3.connect(DB, timeout=15)
    conn.row_factory = sqlite3.Row
    return conn


SCHEMA = """
CREATE TABLE IF NOT EXISTS jobs (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  title TEXT NOT NULL,
  company TEXT,
  location TEXT,
  url TEXT UNIQUE,
  source TEXT,
  description TEXT,
  posted_at TEXT,
  discovered_at TEXT,
  fit_score REAL,
  opinion TEXT,
  contact_email TEXT,
  status TEXT DEFAULT 'new'
);
CREATE TABLE IF NOT EXISTS applications (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  job_id INTEGER NOT NULL,
  resume_path TEXT,
  tailored_pdf TEXT,
  email_subject TEXT,
  email_body TEXT,
  audit_json TEXT,
  status TEXT DEFAULT 'draft',
  outcome TEXT,
  sent_at TEXT,
  created_at TEXT,
  notes TEXT
);
CREATE TABLE IF NOT EXISTS outcomes (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  application_id INTEGER NOT NULL,
  stage TEXT,
  happened_at TEXT,
  notes TEXT
);
CREATE TABLE IF NOT EXISTS audit_log (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  application_id INTEGER,
  job_id INTEGER,
  item TEXT,
  result TEXT,
  detail TEXT,
  checked_at TEXT
);
CREATE TABLE IF NOT EXISTS settings (
  key TEXT PRIMARY KEY,
  value TEXT
);
CREATE INDEX IF NOT EXISTS idx_jobs_status ON jobs(status);
CREATE INDEX IF NOT EXISTS idx_apps_job ON applications(job_id);
"""


DEFAULTS = {
    'llm_base_url': 'http://10.30.20.69:11434',
    'llm_model': 'qwen3.5:9b',
    'llm_timeout': '120',
    'smtp_host': 'smtp.mail.me.com',
    'smtp_port': '587',
    'smtp_user': '',
    'smtp_pass': '',
    'from_email': 'indianaholmes1@icloud.com',
    'from_name': 'Indiana Holmes',
    'resume_phone': '(425) 280-0023',
    'resume_email': 'indianaholmes1@icloud.com',
    'website': 'https://thetempleofdoom.com',
    'proxy_url': '',
    'master_resume': '',
    'rss_feeds': '',
    'headless_browser': '0',
    'firecrawl_url': 'http://10.30.20.182:3002',
    'firecrawl_enabled': '1',
    'firecrawl_queries': 'remote software engineer\nremote devops engineer\nremote platform engineer\nremote linux administrator\nwork from home software engineer seattle\nremote infrastructure engineer',
    'autopilot_enabled': '1',
    'autopilot_interval_hours': '6',
    'autopilot_max_prepare': '3',
    'auto_send': '0',
}


def init_db():
    conn = get_conn()
    try:
        conn.executescript(SCHEMA)
        for k, v in DEFAULTS.items():
            conn.execute('INSERT OR IGNORE INTO settings(key, value) VALUES (?, ?)', (k, v))
        conn.commit()
    finally:
        conn.close()


def get_settings():
    conn = get_conn()
    try:
        rows = conn.execute('SELECT key, value FROM settings').fetchall()
        return {r['key']: r['value'] for r in rows}
    finally:
        conn.close()


def get_setting(key, default=None):
    conn = get_conn()
    try:
        r = conn.execute('SELECT value FROM settings WHERE key = ?', (key,)).fetchone()
        return r['value'] if r else default
    finally:
        conn.close()


def set_setting(key, value):
    conn = get_conn()
    try:
        conn.execute(
            'INSERT INTO settings(key, value) VALUES (?, ?) '
            'ON CONFLICT(key) DO UPDATE SET value = excluded.value',
            (key, str(value)),
        )
        conn.commit()
    finally:
        conn.close()


# ---- jobs ----
def upsert_job(title, company, location, url, source, description, posted_at=None):
    conn = get_conn()
    try:
        if url:
            existing = conn.execute('SELECT id FROM jobs WHERE url = ?', (url,)).fetchone()
            if existing:
                return existing['id']
        cur = conn.execute(
            'INSERT INTO jobs(title, company, location, url, source, description, posted_at, discovered_at) '
            'VALUES (?, ?, ?, ?, ?, ?, ?, ?)',
            (title, company, location, url, source, description, posted_at, now()),
        )
        conn.commit()
        return cur.lastrowid
    finally:
        conn.close()


def list_jobs(status=None, limit=100, offset=0):
    conn = get_conn()
    try:
        q = 'SELECT * FROM jobs'
        args = []
        if status:
            q += ' WHERE status = ?'
            args.append(status)
        q += ' ORDER BY COALESCE(fit_score, -1) DESC, discovered_at DESC LIMIT ? OFFSET ?'
        args += [limit, offset]
        return [dict(r) for r in conn.execute(q, args).fetchall()]
    finally:
        conn.close()


def get_job(job_id):
    conn = get_conn()
    try:
        r = conn.execute('SELECT * FROM jobs WHERE id = ?', (job_id,)).fetchone()
        return dict(r) if r else None
    finally:
        conn.close()


def set_job_status(job_id, status):
    conn = get_conn()
    try:
        conn.execute('UPDATE jobs SET status = ? WHERE id = ?', (status, job_id))
        conn.commit()
    finally:
        conn.close()


def update_job_score(job_id, score, opinion):
    conn = get_conn()
    try:
        conn.execute('UPDATE jobs SET fit_score = ?, opinion = ?, status = ? WHERE id = ?',
                     (score, opinion, 'scored', job_id))
        conn.commit()
    finally:
        conn.close()


def set_job_contact_email(job_id, email):
    conn = get_conn()
    try:
        conn.execute('UPDATE jobs SET contact_email = ? WHERE id = ?', (email or '', job_id))
        conn.commit()
    finally:
        conn.close()


# ---- applications ----
def create_application(job_id, subject='', body=''):
    conn = get_conn()
    try:
        cur = conn.execute(
            'INSERT INTO applications(job_id, email_subject, email_body, created_at) VALUES (?, ?, ?, ?)',
            (job_id, subject, body, now()),
        )
        conn.commit()
        return cur.lastrowid
    finally:
        conn.close()


def update_application(app_id, **fields):
    conn = get_conn()
    try:
        cols = ', '.join(f'{k} = ?' for k in fields)
        conn.execute(f'UPDATE applications SET {cols} WHERE id = ?', (*fields.values(), app_id))
        conn.commit()
    finally:
        conn.close()


def get_application(app_id):
    conn = get_conn()
    try:
        r = conn.execute('SELECT * FROM applications WHERE id = ?', (app_id,)).fetchone()
        return dict(r) if r else None
    finally:
        conn.close()


def list_applications(limit=100):
    conn = get_conn()
    try:
        q = ('SELECT a.*, j.title, j.company, j.url FROM applications a '
             'LEFT JOIN jobs j ON j.id = a.job_id ORDER BY a.id DESC LIMIT ?')
        return [dict(r) for r in conn.execute(q, (limit,)).fetchall()]
    finally:
        conn.close()


def add_outcome(app_id, stage, notes=''):
    conn = get_conn()
    try:
        conn.execute('INSERT INTO outcomes(application_id, stage, happened_at, notes) VALUES (?, ?, ?, ?)',
                     (app_id, stage, now(), notes))
        conn.execute('UPDATE applications SET outcome = ? WHERE id = ?', (stage, app_id))
        conn.commit()
    finally:
        conn.close()


def add_audit(app_id, job_id, item, result, detail):
    conn = get_conn()
    try:
        conn.execute('INSERT INTO audit_log(application_id, job_id, item, result, detail, checked_at) '
                     'VALUES (?, ?, ?, ?, ?, ?)', (app_id, job_id, item, result, detail, now()))
        conn.commit()
    finally:
        conn.close()


def get_audit(app_id):
    conn = get_conn()
    try:
        return [dict(r) for r in conn.execute(
            'SELECT * FROM audit_log WHERE application_id = ? ORDER BY id', (app_id,)).fetchall()]
    finally:
        conn.close()