`src/digest.py`

107 lines
"""
Weekly vertical digest builder.

Pulls new assets in a vertical over the last 7 days and renders a plain-text
digest suitable for emailing to dealer subscribers. Prints to stdout; caller
decides how to send (George Gmail agent, Resend, manual, etc.).

Usage:
    python digest.py --vertical restaurant_equipment --days 7
"""
from __future__ import annotations

import argparse
import os
from datetime import date, timedelta

import psycopg

PG_DSN = os.environ.get("BANKRUPT_LEADS_PG", "postgresql:///bankrupt_leads")

VERTICAL_FILTERS = {
    "restaurant_equipment": {
        "categories": ("equipment", "inventory"),
        "keywords": (
            "kitchen", "oven", "fryer", "hood", "walk-in", "walkin", "refrigerator",
            "freezer", "restaurant", "cafe", "bar", "pizza", "grill", "bakery",
            "dish", "prep", "hobart", "vulcan", "true", "traulsen",
        ),
    },
    "machine_tools": {
        "categories": ("equipment",),
        "keywords": (
            "cnc", "lathe", "mill", "milling", "grinder", "press", "haas", "mazak",
            "okuma", "doosan", "fanuc", "bridgeport", "machine shop",
        ),
    },
    "medical_equipment": {
        "categories": ("equipment",),
        "keywords": (
            "mri", "ct scan", "ultrasound", "x-ray", "xray", "surgical", "dental",
            "exam table", "sterilizer", "autoclave", "ge healthcare", "siemens",
        ),
    },
}


def fetch_assets(vertical: str, since: date) -> list[dict]:
    cfg = VERTICAL_FILTERS[vertical]
    like_clause = " OR ".join(["description ILIKE %s"] * len(cfg["keywords"]))
    sql = f"""
        SELECT a.category, a.subcategory, a.description, a.location,
               a.book_value_usd, a.sale_date, a.sale_venue,
               a.contact_name, a.contact_email, a.contact_phone,
               c.debtor_name, c.court, c.docket_number
          FROM bankrupt_leads.assets a
          JOIN bankrupt_leads.cases c ON c.case_id = a.case_id
         WHERE a.extracted_at >= %s
           AND a.category = ANY(%s)
           AND ({like_clause})
         ORDER BY a.sale_date NULLS LAST, a.book_value_usd DESC NULLS LAST
    """
    params = [since, list(cfg["categories"])] + [f"%{kw}%" for kw in cfg["keywords"]]
    with psycopg.connect(PG_DSN) as conn, conn.cursor() as cur:
        cur.execute(sql, params)
        cols = [d[0] for d in cur.description]
        return [dict(zip(cols, row)) for row in cur.fetchall()]


def render(vertical: str, assets: list[dict], since: date) -> str:
    title = vertical.replace("_", " ").title()
    lines = [
        f"{title} — Bankruptcy Asset Digest",
        f"New leads filed {since.isoformat()} → {date.today().isoformat()}",
        f"{len(assets)} asset{'s' if len(assets) != 1 else ''}",
        "=" * 60,
        "",
    ]
    for a in assets:
        val = f"${a['book_value_usd']:,.0f}" if a["book_value_usd"] else "value n/d"
        sale = f" — sale {a['sale_date']}" if a["sale_date"] else ""
        lines.append(f"• {a['debtor_name']} ({a['court']} {a['docket_number']})")
        lines.append(f"  {a['subcategory'] or a['category']}: {a['description'][:200]}")
        lines.append(f"  {a['location'] or 'location n/d'} — {val}{sale}")
        if a["contact_name"]:
            lines.append(
                f"  contact: {a['contact_name']}"
                + (f" <{a['contact_email']}>" if a["contact_email"] else "")
                + (f" {a['contact_phone']}" if a["contact_phone"] else "")
            )
        lines.append("")
    return "\n".join(lines)


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--vertical", required=True, choices=sorted(VERTICAL_FILTERS))
    ap.add_argument("--days", type=int, default=7)
    args = ap.parse_args()

    since = date.today() - timedelta(days=args.days)
    assets = fetch_assets(args.vertical, since)
    print(render(args.vertical, assets, since))


if __name__ == "__main__":
    main()