feat: Scaffold Novi-Mailman Bridge integration

- app/methods/e_novi_mailman_methods.py: full sync engine, per-member
  sync helper, webhook handler, and Mailman 3 REST subscribe/unsubscribe
- app/routers/api_v3_actions_e_novi_mailman.py: test_connection, list
  inspection, full sync trigger, and Novi webhook receiver endpoints
- registry.py: registered at /v3/action/e_novi_mailman
- TODO: marked as scaffolded, pending Novi field verification + data_store setup

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

app/methods/e_novi_mailman_methods.py

@@ -0,0 +1,377 @@
+import json, requests
+from typing import Dict, List, Optional
+from app.db_sql import sql_select, sql_update, sql_insert
+from app.lib_general import log, logging, logger_reset
+
+# ---------------------------------------------------------------------------
+# Novi-Mailman Bridge
+# Synchronizes Novi AMS membership data with Mailman 3 mailing lists.
+#
+# Credential Storage (data_store table):
+#   code='novi_api_config'     → JSON: { "api_key": "...", "base_url": "https://..." }
+#   code='mailman_api_config'  → JSON: { "base_url": "http://...:8001", "username": "...", "password": "..." }
+#
+# Sync Logic:
+#   - Active Novi members → subscribed in target Mailman list(s)
+#   - Lapsed/expired members → unsubscribed (or held, depending on policy)
+#   - Driven by webhooks (Novi membership events) + optional full-sync
+# ---------------------------------------------------------------------------
+
+# ── Credential Helpers ────────────────────────────────────────────────────
+
+@logger_reset
+def _load_novi_config() -> Optional[Dict]:
+    """Load Novi AMS API credentials from data_store (code='novi_api_config')."""
+    rec = sql_select(table_name='data_store', field_name='code', field_value='novi_api_config')
+    if not rec:
+        log.error("Novi API config not found in data_store (code='novi_api_config').")
+        return None
+    try:
+        return json.loads(rec['text'])
+    except Exception as e:
+        log.error(f"Failed to parse Novi config: {e}")
+        return None
+
+
+@logger_reset
+def _load_mailman_config() -> Optional[Dict]:
+    """Load Mailman 3 REST API credentials from data_store (code='mailman_api_config')."""
+    rec = sql_select(table_name='data_store', field_name='code', field_value='mailman_api_config')
+    if not rec:
+        log.error("Mailman API config not found in data_store (code='mailman_api_config').")
+        return None
+    try:
+        return json.loads(rec['text'])
+    except Exception as e:
+        log.error(f"Failed to parse Mailman config: {e}")
+        return None
+
+
+# ── Novi AMS Methods ──────────────────────────────────────────────────────
+
+@logger_reset
+def test_novi_connection() -> Dict:
+    """
+    Verify Novi AMS API credentials are valid.
+    Returns a dict with 'ok' bool and optional error message.
+    """
+    config = _load_novi_config()
+    if not config:
+        return {"ok": False, "error": "Credentials not configured."}
+
+    # TODO: Confirm the actual test endpoint for the target Novi instance.
+    # Common pattern: GET /api/v1/members?$top=1 with ApiKey header.
+    base_url = config.get('base_url', '').rstrip('/')
+    api_key  = config.get('api_key', '')
+    headers  = {"ApiKey": api_key, "Accept": "application/json"}
+
+    try:
+        resp = requests.get(f"{base_url}/api/v1/members", headers=headers, params={"$top": 1}, timeout=10)
+        if resp.status_code == 200:
+            return {"ok": True}
+        return {"ok": False, "error": f"HTTP {resp.status_code}: {resp.text[:200]}"}
+    except Exception as e:
+        log.exception(f"Novi connection test failed: {e}")
+        return {"ok": False, "error": str(e)}
+
+
+@logger_reset
+def get_novi_members(status_filter: Optional[str] = None, page_size: int = 500, offset: int = 0) -> Optional[List[Dict]]:
+    """
+    Fetch member records from Novi AMS.
+
+    Args:
+        status_filter: Optional Novi membership status (e.g. 'Active', 'Lapsed').
+                       None returns all members.
+        page_size:     Records per page (Novi uses OData $top/$skip).
+        offset:        Pagination offset ($skip).
+
+    Returns:
+        List of member dicts, or None on failure.
+
+    TODO: Confirm OData filter field names against your Novi instance schema.
+    """
+    config = _load_novi_config()
+    if not config:
+        return None
+
+    base_url = config.get('base_url', '').rstrip('/')
+    api_key  = config.get('api_key', '')
+    headers  = {"ApiKey": api_key, "Accept": "application/json"}
+    params   = {"$top": page_size, "$skip": offset}
+
+    if status_filter:
+        params["$filter"] = f"MembershipStatus eq '{status_filter}'"
+
+    try:
+        resp = requests.get(f"{base_url}/api/v1/members", headers=headers, params=params, timeout=30)
+        if resp.status_code != 200:
+            log.error(f"Novi API error: {resp.status_code} - {resp.text[:200]}")
+            return None
+        data = resp.json()
+        # Novi typically wraps results in a 'value' key (OData convention)
+        return data.get('value', data) if isinstance(data, dict) else data
+    except Exception as e:
+        log.exception(f"Failed to fetch Novi members: {e}")
+        return None
+
+
+# ── Mailman 3 Methods ─────────────────────────────────────────────────────
+
+@logger_reset
+def test_mailman_connection() -> Dict:
+    """
+    Verify Mailman 3 REST API credentials are valid.
+    Returns a dict with 'ok' bool and optional error message.
+    """
+    config = _load_mailman_config()
+    if not config:
+        return {"ok": False, "error": "Credentials not configured."}
+
+    base_url = config.get('base_url', '').rstrip('/')
+    auth     = (config.get('username', 'restadmin'), config.get('password', ''))
+
+    try:
+        resp = requests.get(f"{base_url}/3.1/system/versions", auth=auth, timeout=10)
+        if resp.status_code == 200:
+            return {"ok": True, "version": resp.json().get('mailman_version')}
+        return {"ok": False, "error": f"HTTP {resp.status_code}: {resp.text[:200]}"}
+    except Exception as e:
+        log.exception(f"Mailman connection test failed: {e}")
+        return {"ok": False, "error": str(e)}
+
+
+@logger_reset
+def get_mailman_lists() -> Optional[List[Dict]]:
+    """
+    Return all mailing lists from this Mailman 3 instance.
+    """
+    config = _load_mailman_config()
+    if not config:
+        return None
+
+    base_url = config.get('base_url', '').rstrip('/')
+    auth     = (config.get('username', 'restadmin'), config.get('password', ''))
+
+    try:
+        resp = requests.get(f"{base_url}/3.1/lists", auth=auth, timeout=10)
+        if resp.status_code != 200:
+            log.error(f"Mailman list fetch failed: {resp.status_code}")
+            return None
+        return resp.json().get('entries', [])
+    except Exception as e:
+        log.exception(f"Failed to fetch Mailman lists: {e}")
+        return None
+
+
+@logger_reset
+def subscribe_member_to_list(list_id: str, email: str, display_name: str = '') -> bool:
+    """
+    Subscribe an email address to a Mailman 3 list.
+    Uses pre-confirmed subscription (no confirmation email sent).
+
+    Args:
+        list_id:      Mailman list ID, e.g. 'members@yourdomain.org'
+        email:        Member email address
+        display_name: Optional display name
+
+    Returns:
+        True on success or already subscribed, False on error.
+    """
+    config = _load_mailman_config()
+    if not config:
+        return False
+
+    base_url = config.get('base_url', '').rstrip('/')
+    auth     = (config.get('username', 'restadmin'), config.get('password', ''))
+
+    payload = {
+        "list_id":            list_id.replace('@', '.'),  # Mailman uses dot notation for list IDs
+        "subscriber":         email,
+        "display_name":       display_name,
+        "pre_verified":       True,
+        "pre_confirmed":      True,
+        "pre_approved":       True,
+        "send_welcome_message": False,
+    }
+
+    try:
+        resp = requests.post(f"{base_url}/3.1/members", auth=auth, json=payload, timeout=10)
+        if resp.status_code in (200, 201):
+            log.info(f"Subscribed {email} to {list_id}")
+            return True
+        if resp.status_code == 409:
+            log.debug(f"{email} already subscribed to {list_id} — skipping.")
+            return True  # Already a member, treat as success
+        log.error(f"Subscribe failed for {email}: {resp.status_code} - {resp.text[:200]}")
+        return False
+    except Exception as e:
+        log.exception(f"Error subscribing {email} to {list_id}: {e}")
+        return False
+
+
+@logger_reset
+def unsubscribe_member_from_list(list_id: str, email: str) -> bool:
+    """
+    Unsubscribe an email address from a Mailman 3 list.
+
+    Returns:
+        True on success or not found (already unsubscribed), False on error.
+    """
+    config = _load_mailman_config()
+    if not config:
+        return False
+
+    base_url = config.get('base_url', '').rstrip('/')
+    auth     = (config.get('username', 'restadmin'), config.get('password', ''))
+    list_id_dot = list_id.replace('@', '.')
+
+    try:
+        # Mailman member ID is {email}_{list_id} (base64 encoded in REST path)
+        resp = requests.delete(
+            f"{base_url}/3.1/lists/{list_id_dot}/member/{email}",
+            auth=auth,
+            timeout=10
+        )
+        if resp.status_code in (200, 204):
+            log.info(f"Unsubscribed {email} from {list_id}")
+            return True
+        if resp.status_code == 404:
+            log.debug(f"{email} not found in {list_id} — skipping.")
+            return True  # Not a member, treat as success
+        log.error(f"Unsubscribe failed for {email}: {resp.status_code} - {resp.text[:200]}")
+        return False
+    except Exception as e:
+        log.exception(f"Error unsubscribing {email} from {list_id}: {e}")
+        return False
+
+
+# ── Sync Engine ───────────────────────────────────────────────────────────
+
+@logger_reset
+def sync_single_member(email: str, display_name: str, list_id: str, is_active: bool) -> str:
+    """
+    Sync one member's subscription state for a given Mailman list.
+
+    Args:
+        email:        Member email
+        display_name: Member display name
+        list_id:      Target Mailman list ID
+        is_active:    True = subscribe, False = unsubscribe
+
+    Returns:
+        'subscribed', 'unsubscribed', or 'error'
+    """
+    if is_active:
+        ok = subscribe_member_to_list(list_id, email, display_name)
+        return 'subscribed' if ok else 'error'
+    else:
+        ok = unsubscribe_member_from_list(list_id, email)
+        return 'unsubscribed' if ok else 'error'
+
+
+@logger_reset
+def sync_novi_to_mailman(list_id: str, active_status: str = 'Active') -> Optional[Dict]:
+    """
+    Full sync: pull all Novi members and reconcile their Mailman subscription state.
+
+    - Novi members with `active_status` → subscribed to `list_id`
+    - All other statuses → unsubscribed from `list_id`
+
+    Args:
+        list_id:       Target Mailman list ID (e.g. 'members@yourdomain.org')
+        active_status: Novi membership status string that counts as active.
+
+    Returns:
+        Result dict with counts, or None on fatal error.
+
+    TODO: Adjust the Novi member field names (Email, FirstName, LastName,
+          MembershipStatus) to match your actual Novi instance schema.
+    """
+    log.info(f"Starting full Novi → Mailman sync for list: {list_id}")
+
+    members = get_novi_members()
+    if members is None:
+        log.error("Novi member fetch failed — aborting sync.")
+        return None
+
+    log.info(f"Fetched {len(members)} members from Novi.")
+
+    results = {"total": len(members), "subscribed": 0, "unsubscribed": 0, "error": 0, "skipped": 0}
+
+    for member in members:
+        # TODO: Confirm exact field names from your Novi instance
+        email  = (member.get('Email') or member.get('email') or '').strip()
+        fname  = member.get('FirstName') or member.get('first_name') or ''
+        lname  = member.get('LastName')  or member.get('last_name')  or ''
+        status = member.get('MembershipStatus') or member.get('membership_status') or ''
+
+        if not email:
+            results['skipped'] += 1
+            continue
+
+        display_name = f"{fname} {lname}".strip()
+        is_active    = (status == active_status)
+
+        outcome = sync_single_member(email, display_name, list_id, is_active)
+        results[outcome] = results.get(outcome, 0) + 1
+
+    log.info(f"Novi → Mailman sync complete: {results}")
+    return results
+
+
+@logger_reset
+def handle_novi_webhook(payload: Dict) -> Optional[Dict]:
+    """
+    Process a Novi membership webhook event and update the appropriate Mailman list.
+
+    Expected payload shape (Novi webhook format — confirm against Novi docs):
+        {
+            "EventType": "MembershipActivated" | "MembershipLapsed" | "MembershipExpired" | ...,
+            "Member": {
+                "Email": "...",
+                "FirstName": "...",
+                "LastName": "...",
+                "MembershipStatus": "Active" | "Lapsed" | ...
+            }
+        }
+
+    TODO: Confirm Novi webhook payload format and EventType values.
+    TODO: Pull target list_id from data_store config or per-account settings.
+    """
+    event_type = payload.get('EventType', '')
+    member     = payload.get('Member', {})
+    email      = (member.get('Email') or '').strip()
+    fname      = member.get('FirstName', '')
+    lname      = member.get('LastName', '')
+    status     = member.get('MembershipStatus', '')
+
+    if not email:
+        log.warning(f"Novi webhook received with no email — skipping. Payload: {payload}")
+        return None
+
+    # Load target list_id from config
+    # TODO: Support per-account list routing (e.g. multiple orgs, each with their own list)
+    novi_config = _load_novi_config()
+    if not novi_config:
+        return None
+    list_id = novi_config.get('mailman_list_id', '')
+    if not list_id:
+        log.error("'mailman_list_id' not set in novi_api_config — cannot route webhook.")
+        return None
+
+    ACTIVE_EVENTS = {'MembershipActivated', 'MembershipRenewed', 'MembershipCreated'}
+    INACTIVE_EVENTS = {'MembershipLapsed', 'MembershipExpired', 'MembershipTerminated', 'MembershipCancelled'}
+
+    if event_type in ACTIVE_EVENTS:
+        is_active = True
+    elif event_type in INACTIVE_EVENTS:
+        is_active = False
+    else:
+        log.info(f"Novi webhook: unhandled EventType '{event_type}' for {email} — ignoring.")
+        return {"action": "ignored", "reason": f"Unhandled EventType: {event_type}"}
+
+    display_name = f"{fname} {lname}".strip()
+    outcome = sync_single_member(email, display_name, list_id, is_active)
+    log.info(f"Novi webhook processed: {email} → {outcome} (EventType: {event_type})")
+    return {"action": outcome, "email": email, "event_type": event_type}

app/routers/api_v3_actions_e_novi_mailman.py

@@ -0,0 +1,116 @@
+from fastapi import APIRouter, Body, Depends, Query
+from typing import Optional
+import asyncio
+
+from app.lib_general import log
+from app.lib_general_v3 import AccountContext, get_account_context, DelayParams
+from app.models.response_models import Resp_Body_Base, mk_resp
+from app.methods.e_novi_mailman_methods import (
+    test_novi_connection,
+    test_mailman_connection,
+    get_mailman_lists,
+    get_novi_members,
+    sync_novi_to_mailman,
+    handle_novi_webhook,
+)
+
+router = APIRouter()
+
+
+# ── Connection Tests ──────────────────────────────────────────────────────
+
+@router.get('/test_connection/novi', response_model=Resp_Body_Base)
+async def test_novi(
+        account: AccountContext = Depends(get_account_context),
+        delay: DelayParams = Depends(),
+        ):
+    """Verify Novi AMS API credentials stored in data_store."""
+    if delay.sleep_time_s > 0: await asyncio.sleep(delay.sleep_time_s)
+    result = test_novi_connection()
+    if result.get('ok'):
+        return mk_resp(data=result)
+    return mk_resp(data=result, status_code=401, status_message="Novi connection failed.")
+
+
+@router.get('/test_connection/mailman', response_model=Resp_Body_Base)
+async def test_mailman(
+        account: AccountContext = Depends(get_account_context),
+        delay: DelayParams = Depends(),
+        ):
+    """Verify Mailman 3 REST API credentials stored in data_store."""
+    if delay.sleep_time_s > 0: await asyncio.sleep(delay.sleep_time_s)
+    result = test_mailman_connection()
+    if result.get('ok'):
+        return mk_resp(data=result)
+    return mk_resp(data=result, status_code=401, status_message="Mailman connection failed.")
+
+
+# ── Inspection / Preview ──────────────────────────────────────────────────
+
+@router.get('/mailman/lists', response_model=Resp_Body_Base)
+async def list_mailman_lists(
+        account: AccountContext = Depends(get_account_context),
+        delay: DelayParams = Depends(),
+        ):
+    """Return all mailing lists from this Mailman 3 instance."""
+    if delay.sleep_time_s > 0: await asyncio.sleep(delay.sleep_time_s)
+    data = get_mailman_lists()
+    if data is not None:
+        return mk_resp(data=data)
+    return mk_resp(data=False, status_code=500, status_message="Failed to fetch Mailman lists.")
+
+
+@router.get('/novi/members', response_model=Resp_Body_Base)
+async def list_novi_members(
+        status_filter: Optional[str] = Query(None, description="Novi membership status filter (e.g. 'Active', 'Lapsed')"),
+        page_size: int = Query(100, ge=1, le=500),
+        offset: int = Query(0, ge=0),
+        account: AccountContext = Depends(get_account_context),
+        delay: DelayParams = Depends(),
+        ):
+    """Fetch a page of Novi AMS members. Useful for inspecting data before a full sync."""
+    if delay.sleep_time_s > 0: await asyncio.sleep(delay.sleep_time_s)
+    data = get_novi_members(status_filter=status_filter, page_size=page_size, offset=offset)
+    if data is not None:
+        return mk_resp(data={"count": len(data), "members": data})
+    return mk_resp(data=False, status_code=500, status_message="Failed to fetch members from Novi.")
+
+
+# ── Sync ──────────────────────────────────────────────────────────────────
+
+@router.post('/sync', response_model=Resp_Body_Base)
+async def sync_full(
+        list_id: str = Query(..., description="Target Mailman list ID, e.g. 'members@yourdomain.org'"),
+        active_status: str = Query('Active', description="Novi membership status that maps to 'subscribed'"),
+        account: AccountContext = Depends(get_account_context),
+        delay: DelayParams = Depends(),
+        ):
+    """
+    Full sync: pull all Novi members and reconcile Mailman subscription state.
+    Active members are subscribed; all others are unsubscribed.
+    This is the manual / scheduled trigger — the webhook handles real-time updates.
+    """
+    if delay.sleep_time_s > 0: await asyncio.sleep(delay.sleep_time_s)
+    result = sync_novi_to_mailman(list_id=list_id, active_status=active_status)
+    if result:
+        return mk_resp(data=result, status_message="Novi → Mailman sync complete.")
+    return mk_resp(data=False, status_code=500, status_message="Sync failed.")
+
+
+# ── Webhook ───────────────────────────────────────────────────────────────
+
+@router.post('/webhook/novi', response_model=Resp_Body_Base, include_in_schema=True)
+async def novi_membership_webhook(
+        payload: dict = Body(...),
+        ):
+    """
+    Receives Novi AMS membership webhook events and immediately updates
+    the corresponding Mailman subscription — no auth required (Novi pushes to this endpoint).
+
+    TODO: Add HMAC signature verification once Novi webhook secret is configured.
+    """
+    log.info(f"Novi webhook received: EventType={payload.get('EventType')} Email={payload.get('Member', {}).get('Email')}")
+    result = handle_novi_webhook(payload)
+    if result:
+        return mk_resp(data=result)
+    return mk_resp(data=False, status_code=400, status_message="Webhook payload could not be processed.")

app/routers/registry.py

@@ -7,7 +7,7 @@ from app.routers import (
     event_device, event_exhibit, event_exhibit_tracking, event_file, event_importing,
     event_location, event_person,
     event_presentation, event_presenter, event_session,
-    flask_cfg, hosted_file, api_v3_actions_hosted_file, api_v3_actions_event_file, api_v3_actions_e_zoom, lookup, lookup_v3,
+    flask_cfg, hosted_file, api_v3_actions_hosted_file, api_v3_actions_event_file, api_v3_actions_e_zoom, api_v3_actions_e_novi_mailman, lookup, lookup_v3,
     organization, page, person,
     person_user, qr, site, site_domain, user,
     util_email, websockets, websockets_redis, websockets_v3, e_confex, e_cvent, e_impexium, e_stripe
@@ -51,6 +51,7 @@ def setup_routers(app: FastAPI):
     app.include_router(api_v3_actions_hosted_file.router, prefix='/v3/action/hosted_file', tags=['Hosted File (V3 Actions)'])
     app.include_router(api_v3_actions_event_file.router, prefix='/v3/action/event_file', tags=['Event File (V3 Actions)'])
     app.include_router(api_v3_actions_e_zoom.router, prefix='/v3/action/e_zoom', tags=['Zoom Events (V3 Actions)'])
+    app.include_router(api_v3_actions_e_novi_mailman.router, prefix='/v3/action/e_novi_mailman', tags=['Novi-Mailman Bridge (V3 Actions)'])
     app.include_router(lookup.router, prefix='/lu', tags=['Lookup'])
     app.include_router(lookup_v3.router, prefix='/v3/lookup', tags=['Lookup V3'])
 

documentation/TODO__Agents.md

@@ -33,7 +33,7 @@
 
 ## 🚧 Strategic Goals (V3.5+)
 - [ ] **Pydantic V2 / SQLAlchemy 2.0:** Major framework upgrade for performance and type safety.
-- [ ] **Novi-Mailman Bridge:** Synchronization between Novi AMS and Mailman 3.
+- [~] **Novi-Mailman Bridge:** Synchronization between Novi AMS and Mailman 3. (Scaffolded — needs Novi field name verification + credential setup in data_store)
 - [ ] **Lookup System Batch 2:** Migration of `post_topic`, `user_status`, `file_purpose`.
 - [ ] **Zoom Events Integration:** Implement cron synchronization for OAuth2 ticket retrieval.