📄 main_v2.py 20,356 bytes Today 04:39 📋 Raw

"""HoffDesk API - Host-based subdomain routing.

Phase 1: Restructure as sub-applications for domain separation.
- family.hoffdesk.com → Family dashboard + automation
- notes.hoffdesk.com → Content generation + blog admin
- hoffdesk.com → Public blog

Usage: uvicorn main_v2:application --host 0.0.0.0 --port 8000
"""

from fastapi import FastAPI
from starlette.responses import HTMLResponse
from fastapi.middleware.cors import CORSMiddleware
from pathlib import Path
from starlette.applications import Starlette
from starlette.routing import Host, Mount
from starlette.staticfiles import StaticFiles
from contextlib import asynccontextmanager
import sys
import inspect

Load .env file if python-dotenv is available

try:
from dotenv import load_dotenv
env_path = Path(file).parent / ".env"
if env_path.exists():
load_dotenv(env_path)
except ImportError:
pass

from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response
import time

── Analytics ──────────────────────────────────────────────────────────────

from analytics import AnalyticsMiddleware, get_stats, _ensure_db

class CacheControlMiddleware(BaseHTTPMiddleware):
"""Set Cache-Control headers for static assets and HTML responses.

Cloudflare respects CDN-Cache-Control over its default caching behavior.
Static assets: 1 hour cache in browser, 1 hour at CDN edge.
HTML pages + API responses: never cache anywhere.
"""
async def dispatch(self, request: Request, call_next):
    response: Response = await call_next(request)
    path = request.url.path

    # Static assets: short cache so iterative work propagates quickly
    if path.startswith("/blog/static/") or path.startswith("/static/"):
        response.headers["Cache-Control"] = "public, max-age=3600"
        response.headers["CDN-Cache-Control"] = "public, max-age=3600"
    # HTML pages: always fresh from origin
    elif response.headers.get("content-type", "").startswith("text/html"):
        response.headers["Cache-Control"] = "no-cache, no-store, must-revalidate"
        response.headers["CDN-Cache-Control"] = "no-cache"
    # API/admin responses
    elif path.startswith("/admin/") or path.startswith("/api/") or path.startswith("/auth/"):
        response.headers["Cache-Control"] = "no-cache, no-store, must-revalidate"
        response.headers["CDN-Cache-Control"] = "no-cache"

    return response

from shared.session_auth import SessionAuthMiddleware

Import all routers

from blog import blog_router, brief_router
from family import family_router
from family.brain import brain_router as brain_viz_router
from brain import brain_router as brain_query_router
from webhook_public import webhook_router
from auth.router import auth_router
from dashboard.router import router as dashboard_router
from dashboard.ui_router import router as ui_router
from imap_proxy.dashboard_router import router as imap_router, public_router as imap_public_router
from imap_proxy.proxy import IMAPProxy, IMAPConfig, record_email_processed

RTSport mock preview (isolated, no bleed)

from rtsport_mock import setup_rtsport_mounts

=============================================================================

FAMILY APP (family.hoffdesk.com)

Dashboard + family automation

=============================================================================

family_app = FastAPI(
title="Family Dashboard",
description="HoffDesk family dashboard and automation",
version="1.0.0"
)

Middleware

family_app.add_middleware(SessionAuthMiddleware)
family_app.add_middleware(AnalyticsMiddleware)
family_app.add_middleware(
CORSMiddleware,
allow_origins=[""],
allow_credentials=True,
allow_methods=[""],
allow_headers=["*"],
)

Cache-Control for static assets

family_app.add_middleware(CacheControlMiddleware)

Static files

DASHBOARD_STATIC_DIR = Path("/home/hoffmann_admin/.openclaw/workspace-socrates/hoffdesk-api/dashboard/static")
family_app.mount("/static", StaticFiles(directory=str(DASHBOARD_STATIC_DIR)), name="family_static")

Auth router (no prefix - mounted at root)

family_app.include_router(auth_router, tags=["auth"])

Dashboard (main page)

family_app.include_router(dashboard_router, tags=["dashboard"])

UI fragments (HTMX endpoints per CONTRACT.md)

family_app.include_router(ui_router, tags=["ui"])

Family automation endpoints

family_app.include_router(family_router, prefix="/admin", tags=["family-automation"])

Public webhooks (also available on family domain)

family_app.include_router(webhook_router, tags=["webhook"])

Family Brain — live network visualization

family_app.include_router(brain_viz_router, tags=["brain-viz"])

Brain Query — natural language search with temporal weighting

family_app.include_router(brain_query_router, tags=["brain-query"])

IMAP Proxy dashboard (private auth required)

family_app.include_router(imap_router, tags=["imap-proxy"])

IMAP Proxy public status/metrics (no auth required)

family_app.include_router(imap_public_router, tags=["imap-proxy"])

Analytics dashboard

@family_app.get("/admin/analytics", response_class=HTMLResponse)
async def analytics_dashboard():
_ensure_db()
stats = get_stats(days=30)
from analytics import _render_dashboard
return _render_dashboard(stats)

IMAP Proxy lifecycle — start on app startup if configured

_imap_proxy: IMAPProxy | None = None

def _on_email_received(email_data: dict):
"""Handle incoming email from IMAP proxy.

Phase 6.2: Route to FamilyPipeline for classification → extraction → calendar.

Note: This runs in the IMAP background thread, so we need to handle async carefully.
"""
import logging
from datetime import datetime

logger = logging.getLogger("imap_proxy")
logger.info(f"New email: {email_data.get('subject', '(no subject)')[:60]} from {email_data.get('from', 'unknown')}")

# Record metrics first (sync)
record_email_processed(email_data)

# Pipeline processing - runs in thread pool to avoid blocking IMAP thread
def _process_sync():
    try:
        import asyncio
        from shared.llm import LLMClient
        from shared.notify import TelegramNotifier
        from family.calendar import CalendarClient
        from family.pipeline import FamilyPipeline

        # Initialize pipeline components
        llm = LLMClient()
        calendar = CalendarClient()
        telegram = TelegramNotifier()
        pipeline = FamilyPipeline(llm, calendar, telegram)

        # Create new event loop for this thread
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)

        try:
            result = loop.run_until_complete(pipeline.process_email(
                subject=email_data.get("subject", ""),
                body=email_data.get("body", ""),
                sender=email_data.get("from", ""),
                received_at=email_data.get("fetched_at", datetime.now().isoformat())
            ))

            logger.info(f"Pipeline result: classification={result.get('classification')}, "
                       f"processed={result.get('processed')}, "
                       f"event_created={result.get('event_created')}")

            if result.get("errors"):
                logger.warning(f"Pipeline errors: {result['errors']}")
        finally:
            loop.close()

    except Exception as e:
        logger.exception(f"Pipeline processing failed: {e}")

# Run processing in background thread to avoid blocking IMAP
import threading
thread = threading.Thread(target=_process_sync, daemon=True)
thread.start()

@family_app.on_event("startup")
async def start_imap_proxy():
global _imap_proxy
import os
imap_user = os.getenv("IMAP_USER", "")
imap_pass = os.getenv("IMAP_PASSWORD", "")
if imap_user and imap_pass:
config = IMAPConfig.from_env()
_imap_proxy = IMAPProxy(config, handler=_on_email_received, metrics_callback=record_email_processed)
_imap_proxy.start()
from imap_proxy.dashboard_router import set_proxy
set_proxy(_imap_proxy)
else:
# No credentials — proxy stays in "not configured" state
from imap_proxy.dashboard_router import set_proxy
set_proxy(None)

@family_app.on_event("shutdown")
async def stop_imap_proxy():
if _imap_proxy:
_imap_proxy.stop()

Root handler for family - dashboard (requires auth, handled by dashboard_router)

The dashboard_router handles / and /dashboard/ already

=============================================================================

NOTES APP (notes.hoffdesk.com)

Content generation + blog admin

=============================================================================

notes_app = FastAPI(
title="Notes & Content",
description="HoffDesk content generation and blog administration",
version="1.0.0"
)

Middleware

notes_app.add_middleware(SessionAuthMiddleware)
notes_app.add_middleware(AnalyticsMiddleware)
notes_app.add_middleware(
CORSMiddleware,
allow_origins=[""],
allow_credentials=True,
allow_methods=[""],
allow_headers=["*"],
)

Static files (shared with blog for now)

BLOG_STATIC_DIR = Path("/home/hoffmann_admin/hoffdesk/blog/static")
notes_app.mount("/static", StaticFiles(directory=str(BLOG_STATIC_DIR)), name="notes_static")

Auth router (no prefix)

notes_app.include_router(auth_router, tags=["auth"])

Content briefs

notes_app.include_router(brief_router, prefix="/api/v1/content/briefs", tags=["content-briefs"])

Blog public API - now at root instead of /api/blog

notes_app.include_router(blog_router, tags=["blog"])

Backward compatibility for notes subdomain: /api/blog/ → /

@notes_app.get("/api/blog")
async def notes_api_blog_root_redirect():
from fastapi.responses import RedirectResponse
return RedirectResponse(url="/", status_code=301)

@notes_app.get("/api/blog/{path:path}")
async def notes_api_blog_path_redirect(path: str):
from fastapi.responses import RedirectResponse
return RedirectResponse(url=f"/{path}", status_code=301)

Root handler for notes - simple placeholder (subdomain may be repurposed later)

@notes_app.get("/", response_class=HTMLResponse)
def notes_root():
return """<!DOCTYPE html>

notes.hoffdesk.com

📝 notes.hoffdesk.com

Under Construction

← Back to hoffdesk.com

"""

=============================================================================

BLOG APP (hoffdesk.com, www.hoffdesk.com)

Public blog only

=============================================================================

blog_app = FastAPI(
title="HoffDesk Blog",
description="Public HoffDesk blog",
version="1.0.0"
)

CORS (no session auth for public blog)

blog_app.add_middleware(AnalyticsMiddleware)
blog_app.add_middleware(
CORSMiddleware,
allow_origins=[""],
allow_credentials=True,
allow_methods=[""],
allow_headers=["*"],
)

Cache-Control for static assets and HTML

blog_app.add_middleware(CacheControlMiddleware)

Static files

blog_app.mount("/blog/static", StaticFiles(directory=str(BLOG_STATIC_DIR)), name="blog_static")

Public blog API - now at root instead of /api/blog

blog_app.include_router(blog_router, tags=["blog"])

Public webhooks

blog_app.include_router(webhook_router, tags=["webhook"])

UI fragments (HTMX endpoints) - make available on blog_app for localhost access

blog_app.include_router(ui_router, tags=["ui"])

Health check

@blog_app.get("/health")
def health():
return {"status": "healthy", "app": "blog"}

Backward compatibility: /api/blog/ redirects to / (preserving query params)

@blog_app.get("/api/blog")
async def api_blog_root_redirect(request: Request):
from fastapi.responses import RedirectResponse
qs = request.url.query
url = "/?" + qs if qs else "/"
return RedirectResponse(url=url, status_code=301)

@blog_app.get("/api/blog/{path:path}")
async def api_blog_path_redirect(path: str, request: Request):
from fastapi.responses import RedirectResponse
qs = request.url.query
url = f"/{path}" + ("?" + qs if qs else "")
return RedirectResponse(url=url, status_code=301)

Legacy /blog/ prefix → redirect to new root paths (preserving query params)

@blog_app.get("/blog")
async def blog_root_redirect(request: Request):
from fastapi.responses import RedirectResponse
qs = request.url.query
url = "/?" + qs if qs else "/"
return RedirectResponse(url=url, status_code=301)

@blog_app.get("/blog/{path:path}")
async def blog_path_redirect(path: str, request: Request):
from fastapi.responses import RedirectResponse
qs = request.url.query
# Legacy URL mapping: /blog/{slug} → /article/{slug}
# but keep /blog/category/{cat} and /blog/tag/{tag} intact
if not path or path == "/":
url = "/?" + qs if qs else "/"
return RedirectResponse(url=url, status_code=301)
if path.startswith("category/") or path.startswith("tag/"):
target = f"/{path}"
url = target + ("?" + qs if qs else "")
return RedirectResponse(url=url, status_code=301)
elif path.endswith("feed.xml") or path.endswith("sitemap.xml"):
target = f"/{path}"
url = target + ("?" + qs if qs else "")
return RedirectResponse(url=url, status_code=301)
else:
# Treat as article slug
slug = path.rstrip("/")
target = f"/article/{slug}"
url = target + ("?" + qs if qs else "")
return RedirectResponse(url=url, status_code=301)

Family Brain — live SVG endpoint (public, accessible from blog domain)

from family.brain import brain_router as brain_viz_router_blog
blog_app.include_router(brain_viz_router_blog, tags=["brain-viz"])

Brain Query — natural language search (also available on blog domain)

from brain import brain_router as brain_query_router_blog
blog_app.include_router(brain_query_router_blog, tags=["brain-query"])

=============================================================================

MAIN HOST ROUTER

Routes by Host header to appropriate sub-application

=============================================================================

routes = [
Host("family.hoffdesk.com", app=family_app, name="family"),
Host("notes.hoffdesk.com", app=notes_app, name="notes"),
Host("hoffdesk.com", app=blog_app, name="blog"),
Host("www.hoffdesk.com", app=blog_app, name="blog_www"),
Host("hook.hoffdesk.com", app=family_app, name="hook"),
# Fallback: any other host goes to blog
Mount("/", app=blog_app),
]

Lifespan context manager to propagate startup/shutdown to sub-apps

@asynccontextmanager
async def lifespan(app):
"""Propagate lifespan events to all mounted sub-applications."""
# Collect unique sub-apps (blog_app appears multiple times in routes)
sub_apps = [family_app, notes_app, blog_app]

# Trigger startup for each sub-app
for sub_app in sub_apps:
    for event_handler in sub_app.router.on_startup:
        if callable(event_handler):
            if inspect.iscoroutinefunction(event_handler):
                await event_handler()
            else:
                event_handler()

yield

# Trigger shutdown for each sub-app
for sub_app in sub_apps:
    for event_handler in sub_app.router.on_shutdown:
        if callable(event_handler):
            if inspect.iscoroutinefunction(event_handler):
                await event_handler()
            else:
                event_handler()

Create the host-routing application with lifespan

application = Starlette(routes=routes, lifespan=lifespan)

=============================================================================

DEVELOPMENT MODE

For local testing without domain names

=============================================================================

dev_app = FastAPI(
title="HoffDesk API (Dev Mode)",
description="All routes mounted at paths for local development",
version="1.0.0-dev"
)

Session auth

dev_app.add_middleware(SessionAuthMiddleware)
dev_app.add_middleware(
CORSMiddleware,
allow_origins=[""],
allow_credentials=True,
allow_methods=[""],
allow_headers=["*"],
)

Mount all static files

dev_app.mount("/blog/static", StaticFiles(directory=str(BLOG_STATIC_DIR)), name="blog_static_dev")
dev_app.mount("/static", StaticFiles(directory=str(DASHBOARD_STATIC_DIR)), name="dashboard_static_dev")

Mount all routers for local testing

dev_app.include_router(auth_router, prefix="/auth", tags=["auth"])
dev_app.include_router(dashboard_router, prefix="/family", tags=["dashboard"])
dev_app.include_router(family_router, prefix="/admin", tags=["family-automation"])
dev_app.include_router(brief_router, prefix="/api/v1/content/briefs", tags=["content-briefs"])

Blog routes at root for dev mode

dev_app.include_router(blog_router, tags=["blog"])
dev_app.include_router(webhook_router, tags=["webhook"])
dev_app.include_router(ui_router, tags=["ui"])

Family Brain visualization — add after auth middleware so it's public

dev_app.include_router(brain_viz_router, tags=["brain-viz"])

Brain Query — natural language search

dev_app.include_router(brain_query_router, tags=["brain-query"])

Also add brain routes to blog_app in dev mode for testing

blog_app.include_router(brain_viz_router, tags=["brain-viz"])
blog_app.include_router(brain_query_router, tags=["brain-query"])

RTSport mock preview (isolated, no bleed with any project)

setup_rtsport_mounts(dev_app)
setup_rtsport_mounts(blog_app)

from login_router import router as login_router
blog_app.include_router(login_router)

RTSport API proxy (temporary until unified API)

from rtsport_api_proxy import router as rtsport_proxy_router

Coach and parent mock APIs disabled — now routed via proxy to port 8001

from rtsport_coach_api import router as coach_api_router

from rtsport_parent_api import router as parent_api_router

for _app in [family_app, dev_app, blog_app]:

_app.include_router(coach_api_router, prefix="/api/v1", tags=["rtsport-coach"])

_app.include_router(parent_api_router, prefix="/api/v1", tags=["rtsport-parent"])

family_app.include_router(rtsport_proxy_router, prefix="/api/v1", tags=["rtsport-api"])
dev_app.include_router(rtsport_proxy_router, prefix="/api/v1", tags=["rtsport-api"])

Also mount on blog_app since hoffdesk.com routes there

blog_app.include_router(rtsport_proxy_router, prefix="/api/v1", tags=["rtsport-api"])

Dev mode: backward compatibility redirects

@dev_app.get("/api/blog")
async def dev_api_blog_root_redirect():
from fastapi.responses import RedirectResponse
return RedirectResponse(url="/", status_code=301)

@dev_app.get("/api/blog/{path:path}")
async def dev_api_blog_path_redirect(path: str):
from fastapi.responses import RedirectResponse
return RedirectResponse(url=f"/{path}", status_code=301)

@dev_app.get("/health")
def dev_health():
return {"status": "healthy", "mode": "development"}

Export based on environment

import os
if os.getenv("HOST_ROUTING", "true").lower() == "true":
# Production: use host-based routing
app = application
else:
# Development: use path-based routing
app = dev_app

if name == "main":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)

← Back

Load .env file if python-dotenv is available

── Analytics ──────────────────────────────────────────────────────────────

Import all routers

RTSport mock preview (isolated, no bleed)

=============================================================================

FAMILY APP (family.hoffdesk.com)

Dashboard + family automation

=============================================================================

Middleware

Cache-Control for static assets

Static files

Auth router (no prefix - mounted at root)

Dashboard (main page)

UI fragments (HTMX endpoints per CONTRACT.md)

Family automation endpoints

Public webhooks (also available on family domain)

Family Brain — live network visualization

Brain Query — natural language search with temporal weighting

IMAP Proxy dashboard (private auth required)

IMAP Proxy public status/metrics (no auth required)

Analytics dashboard

IMAP Proxy lifecycle — start on app startup if configured

Root handler for family - dashboard (requires auth, handled by dashboard_router)

The dashboard_router handles / and /dashboard/ already

=============================================================================

NOTES APP (notes.hoffdesk.com)

Content generation + blog admin

=============================================================================

Middleware

Static files (shared with blog for now)

Auth router (no prefix)

Content briefs

Blog public API - now at root instead of /api/blog

Backward compatibility for notes subdomain: /api/blog/ → /

Root handler for notes - simple placeholder (subdomain may be repurposed later)

📝 notes.hoffdesk.com

=============================================================================

BLOG APP (hoffdesk.com, www.hoffdesk.com)

Public blog only

=============================================================================

CORS (no session auth for public blog)

Cache-Control for static assets and HTML

Static files

Public blog API - now at root instead of /api/blog

Public webhooks

UI fragments (HTMX endpoints) - make available on blog_app for localhost access

Health check

Backward compatibility: /api/blog/ redirects to / (preserving query params)

Legacy /blog/ prefix → redirect to new root paths (preserving query params)

Family Brain — live SVG endpoint (public, accessible from blog domain)

Brain Query — natural language search (also available on blog domain)

=============================================================================

MAIN HOST ROUTER

Routes by Host header to appropriate sub-application

=============================================================================

Lifespan context manager to propagate startup/shutdown to sub-apps

Create the host-routing application with lifespan

=============================================================================

DEVELOPMENT MODE

For local testing without domain names

=============================================================================

Session auth

Mount all static files

Mount all routers for local testing

Blog routes at root for dev mode

Family Brain visualization — add after auth middleware so it's public

Brain Query — natural language search

Also add brain routes to blog_app in dev mode for testing

RTSport mock preview (isolated, no bleed with any project)

RTSport Login Page

RTSport API proxy (temporary until unified API)

Coach and parent mock APIs disabled — now routed via proxy to port 8001

from rtsport_coach_api import router as coach_api_router

from rtsport_parent_api import router as parent_api_router

for _app in [family_app, dev_app, blog_app]:

_app.include_router(coach_api_router, prefix="/api/v1", tags=["rtsport-coach"])

_app.include_router(parent_api_router, prefix="/api/v1", tags=["rtsport-parent"])

Also mount on blog_app since hoffdesk.com routes there

Dev mode: backward compatibility redirects

Export based on environment