"""FastAPI router for the HoffDesk Blog Module. Public read endpoints (Phase 1) and admin write endpoints (Phase 2). """ import os from fastapi import APIRouter, HTTPException, Query, Request, Header, Depends from fastapi.responses import HTMLResponse from datetime import datetime, timezone from pathlib import Path from jinja2 import Environment, FileSystemLoader from .models import ( CategoryListResponse, CreatePostRequest, DeletePostResponse, DeployResponse, PostDetailResponse, PostListResponse, PreviewRequest, PreviewResponse, RegenerateResponse, RebuildResponse, TagListResponse, UpdatePostRequest, ) from . import service from .service import ( list_posts, get_post, create_post, update_post, delete_post, publish_post, preview_markdown, PostNotFoundError, DuplicateSlugError ) # Admin token auth ADMIN_TOKEN = os.getenv("BLOG_ADMIN_TOKEN", "changeme-please-update") def verify_admin_token(x_admin_token: str = Header(None)) -> str: """Verify admin token from header.""" if x_admin_token != ADMIN_TOKEN: raise HTTPException(status_code=401, detail="Unauthorized - invalid or missing token") return x_admin_token router = APIRouter() blog_router = router # Jinja2 setup for public templates PUBLIC_TEMPLATES_DIR = Path("/home/hoffmann_admin/hoffdesk/blog/templates") jinja_env = Environment(loader=FileSystemLoader(str(PUBLIC_TEMPLATES_DIR))) CATEGORY_LABELS = { "general": "General", "engineering": "Engineering", "family": "Family", "projects": "Projects", "behind-scenes": "Behind the Scenes", "homelab": "Home Lab", "openclaw": "OpenClaw", "ai-news": "AI News", } def render_template(template_name: str, context: dict) -> HTMLResponse: """Render a Jinja2 template to HTML response.""" template = jinja_env.get_template(template_name) html = template.render(**context) return HTMLResponse(content=html) # ─── Public Read Endpoints ───────────────────────────────────────────────── @router.get("/posts", response_model=PostListResponse) async def api_list_posts( page: int = Query(1, ge=1, description="Page number (1-based)"), per_page: int = Query(10, ge=1, le=50, description="Posts per page"), category: str | None = Query(None, description="Filter by category slug"), tag: str | None = Query(None, description="Filter by tag name"), ): """List published blog posts with pagination and filtering.""" return list_posts(page=page, per_page=per_page, category=category, status="published") @router.get("/posts/{slug}", response_model=PostDetailResponse) async def api_get_post(slug: str): """Get a single published blog post by slug.""" try: post = get_post(slug, require_published=True) return PostDetailResponse(post=post) except PostNotFoundError: raise HTTPException( status_code=404, detail={"error": "Post not found", "code": "POST_NOT_FOUND"}, ) @router.get("/categories", response_model=CategoryListResponse) async def api_list_categories(): """List all categories with published post counts.""" return service.list_categories() @router.get("/tags", response_model=TagListResponse) async def api_list_tags(): """List all tags with published post counts.""" return service.list_tags() # ─── Admin Write Endpoints ──────────────────────────────────────────────── @router.get("/admin/posts", response_model=PostListResponse) async def api_admin_list_posts( page: int = Query(1, ge=1), per_page: int = Query(10, ge=1, le=50), status: str | None = Query(None), category: str | None = Query(None), token: str = Depends(verify_admin_token), ): """List all blog posts (admin view with status filter).""" status_filter = None if status == "all" else status return list_posts(page=page, per_page=per_page, status=status_filter, category=category) @router.post("/admin/posts", response_model=PostDetailResponse, status_code=201) async def api_create_post( request: CreatePostRequest, token: str = Depends(verify_admin_token), ): """Create a new blog post.""" try: post = create_post(request) return PostDetailResponse(post=post) except DuplicateSlugError as e: raise HTTPException(status_code=409, detail={"error": str(e), "code": "SLUG_EXISTS"}) except ValueError as e: raise HTTPException(status_code=400, detail={"error": str(e), "code": "VALIDATION_ERROR"}) @router.patch("/admin/posts/{slug}", response_model=PostDetailResponse) async def api_update_post( slug: str, request: UpdatePostRequest, token: str = Depends(verify_admin_token), ): """Update a blog post (partial update).""" try: post = update_post(slug, request) return PostDetailResponse(post=post) except PostNotFoundError: raise HTTPException(status_code=404, detail={"error": "Post not found", "code": "POST_NOT_FOUND"}) except DuplicateSlugError as e: raise HTTPException(status_code=409, detail={"error": str(e), "code": "SLUG_EXISTS"}) @router.delete("/admin/posts/{slug}", response_model=DeletePostResponse) async def api_delete_post( slug: str, token: str = Depends(verify_admin_token), ): """Soft-delete a blog post (archive it).""" try: delete_post(slug) return DeletePostResponse(message="Post archived successfully", slug=slug) except PostNotFoundError: raise HTTPException(status_code=404, detail={"error": "Post not found", "code": "POST_NOT_FOUND"}) @router.post("/admin/posts/{slug}/publish", response_model=PostDetailResponse) async def api_publish_post( slug: str, token: str = Depends(verify_admin_token), ): """Publish a draft blog post.""" try: post = publish_post(slug, publish=True) return PostDetailResponse(post=post) except PostNotFoundError: raise HTTPException(status_code=404, detail={"error": "Post not found", "code": "POST_NOT_FOUND"}) except ValueError as e: raise HTTPException(status_code=400, detail={"error": str(e), "code": "ALREADY_PUBLISHED"}) @router.post("/admin/posts/{slug}/unpublish", response_model=PostDetailResponse) async def api_unpublish_post( slug: str, token: str = Depends(verify_admin_token), ): """Unpublish a blog post (revert to draft).""" try: post = publish_post(slug, publish=False) return PostDetailResponse(post=post) except PostNotFoundError: raise HTTPException(status_code=404, detail={"error": "Post not found", "code": "POST_NOT_FOUND"}) except ValueError as e: raise HTTPException(status_code=400, detail={"error": str(e), "code": "ALREADY_DRAFT"}) @router.post("/admin/posts/preview", response_model=PreviewResponse) async def api_preview_post( request: PreviewRequest, token: str = Depends(verify_admin_token), ): """Render Markdown to HTML for live preview.""" return preview_markdown(request.content_md) # ─── Admin Dashboard Endpoints (Daedalus Phase 4) ───────────────────────────── @router.get("/admin/stats") async def api_admin_stats(): """Return dashboard statistics for admin homepage.""" from pathlib import Path all_posts = list_posts(page=1, per_page=1000, status=None) published_posts = list_posts(page=1, per_page=1000, status="published") draft_posts = list_posts(page=1, per_page=1000, status="draft") # Count images images_dir = Path.home() / "hoffdesk/blog/static/images" image_count = 0 if images_dir.exists(): image_count = len([f for f in images_dir.rglob("*") if f.is_file()]) # Check for last build last_build_time = None build_marker = Path.home() / "hoffdesk-api/dist/blog/.build-time" if build_marker.exists(): last_build_time = build_marker.read_text().strip() return { "total": all_posts.total, "published": published_posts.total, "drafts": draft_posts.total, "images": image_count, "last_build_time": last_build_time, "last_build_iso": last_build_time, } @router.get("/admin/recent") async def api_admin_recent(): """Return recent posts for dashboard activity feed.""" posts = list_posts(page=1, per_page=5, status=None) recent_posts = [] for post in posts.posts: recent_posts.append({ "slug": post.slug, "title": post.title, "category": post.category, "status": post.status, "updated_relative": "recently", }) return {"posts": recent_posts} @router.get("/admin/posts/list") async def api_admin_post_list( status: str | None = Query(None), category: str | None = Query(None), q: str | None = Query(None), page: int = Query(1, ge=1), per_page: int = Query(20, ge=1, le=100), ): """Admin post list with filtering (detailed for admin UI).""" status_filter = None if status == "all" else status posts = list_posts(page=page, per_page=per_page, status=status_filter, category=category) admin_posts = [] for post in posts.posts: excerpt = post.excerpt or "" if excerpt and len(excerpt) > 150: excerpt = excerpt[:150] + "..." admin_posts.append({ "slug": post.slug, "title": post.title, "category": post.category, "status": post.status, "updated_relative": "recently", "excerpt": excerpt, }) return { "posts": admin_posts, "pagination": { "page": page, "per_page": per_page, "total": posts.total, "pages": posts.total_pages, } } @router.get("/admin/images") async def api_admin_images(): """Return list of available images.""" from pathlib import Path images_dir = Path.home() / "hoffdesk/blog/static/images" images = [] if images_dir.exists(): for img_path in images_dir.rglob("*"): if img_path.is_file() and img_path.suffix.lower() in ('.jpg', '.jpeg', '.png', '.gif', '.webp'): stat = img_path.stat() rel_path = img_path.relative_to(images_dir.parent) images.append({ "path": f"/blog/{rel_path}", "filename": img_path.name, "relative_path": str(rel_path), "size_bytes": stat.st_size, "modified": datetime.fromtimestamp(stat.st_mtime, tz=timezone.utc).isoformat(), }) return {"images": images} # ─── Legacy / Placeholder Endpoints ─────────────────────────────────────── @router.post("/admin/posts/{slug}/regenerate", response_model=RegenerateResponse) async def api_regenerate_post(slug: str): """Regenerate static HTML for a post (placeholder).""" return RegenerateResponse( message="Static HTML regeneration queued", slug=slug, regenerated_at=datetime.now(timezone.utc), ) @router.post("/admin/rebuild", response_model=RebuildResponse) async def api_rebuild(): """Full rebuild of the blog database (placeholder).""" return RebuildResponse( message="Full rebuild queued", posts_processed=0, started_at=datetime.now(timezone.utc), ) @router.post("/admin/deploy", response_model=DeployResponse) async def api_deploy(): """Deploy the static blog (placeholder).""" return DeployResponse( message="Blog deployed successfully", files_deployed=0, deployed_at=datetime.now(timezone.utc), destination="/var/www/hoffdesk.com/blog", ) # ─── Public HTML Rendering Routes (Daedalus Phase 4 Frontend) ───────────────── @router.get("/", response_class=HTMLResponse) async def blog_index( page: int = Query(1, ge=1), category: str | None = Query(None), ): """Render the blog index page (Medium-style).""" posts_result = list_posts(page=page, per_page=10, status="published", category=category) # Find featured post (first featured, or first post) featured_post = None for post in posts_result.posts: if getattr(post, 'featured', False): featured_post = post break if not featured_post and posts_result.posts: featured_post = posts_result.posts[0] # Other posts (exclude featured) other_posts = [p for p in posts_result.posts if p != featured_post] return render_template( "blog_index.html.j2", { "page_title": "HoffDesk Blog", "page_description": "Thoughts on home infrastructure, AI systems, and building things.", "posts": other_posts, "featured_post": featured_post, "current_category": category, "category_labels": CATEGORY_LABELS, "current_page": page, "total_pages": posts_result.total_pages, } ) @router.get("/category/{category}", response_class=HTMLResponse) async def blog_category( category: str, page: int = Query(1, ge=1), ): """Render a category archive page.""" posts_result = list_posts(page=page, per_page=10, status="published", category=category) category_label = CATEGORY_LABELS.get(category, category.replace("-", " ").title()) return render_template( "blog_category.html.j2", { "page_title": f"{category_label} — HoffDesk Blog", "page_description": f"Posts in {category_label}", "category": category, "category_label": category_label, "posts": posts_result.posts, "current_page": page, "total_pages": posts_result.total_pages, } ) @router.get("/tag/{tag}", response_class=HTMLResponse) async def blog_tag( tag: str, page: int = Query(1, ge=1), ): """Render a tag archive page.""" posts_result = list_posts(page=page, per_page=10, status="published", tag=tag) return render_template( "blog_tag.html.j2", { "page_title": f"#{tag} — HoffDesk Blog", "page_description": f"Posts tagged with #{tag}", "tag": tag, "posts": posts_result.posts, "current_page": page, "total_pages": posts_result.total_pages, } ) @router.get("/article/{slug}", response_class=HTMLResponse) async def blog_article(slug: str): """Render a single blog article page.""" try: post = get_post(slug, require_published=True) except PostNotFoundError: raise HTTPException(status_code=404, detail="Post not found") # Get related posts (same category, excluding current) related_result = list_posts(page=1, per_page=3, status="published", category=post.category) related_posts = [p for p in related_result.posts if p.slug != slug][:2] # FAQ data for this article (dict of slug -> list of {question, answer}) FAQ_DATA = { "the-night-i-broke-dns": [ {"question": "What happens when you move your home DNS to Pi-hole?", "answer": "Moving DNS to Pi-hole without testing the failover first can break your entire home network — smart TVs, kids' tablets, and even automatic pet feeders will lose connectivity. Always set up a secondary DNS server or keep your ISP's DNS as a fallback."}, {"question": "Should you run Pi-hole on a home server?", "answer": "Yes, but only with a fallback DNS configuration. Pi-hole is lightweight enough to run on a Beelink mini PC alongside other services, but a single point of failure for DNS means the whole house goes dark during maintenance."}, ], "your-fiber-is-under-a-shovel": [ {"question": "What happens when your fiber internet is cut?", "answer": "When the incoming fiber is severed — by construction, weather, or accident — everything that depends on cloud services stops working. The internet is not a utility: it is a chain of custody from your ISP's central office to your router, and any break takes you offline."}, {"question": "How do you design a network that works without the internet?", "answer": "Build for degraded mode from day one. Keep DNS, calendar, file storage, and smart home automation running on local services. Your family should be able to access their calendar and photos even when the fiber is cut."}, ], "upgrades-are-dangerous-debugging-is-fun": [ {"question": "Why do OpenClaw updates break things?", "answer": "OpenClaw is actively developed, and pre-release versions can introduce breaking changes. The 2026.5.3-1 pre-release broke the gateway, and the following stable release 2026.5.4 fixed that but introduced a new session approval system for sudo commands."}, {"question": "What is the inline sudo approval button in OpenClaw?", "answer": "It is a Telegram-native approval mechanism introduced in OpenClaw 2026.5.4. Instead of requiring manual terminal input for privileged commands, it presents an inline button in the Telegram chat — one tap approves the operation."}, ], "grill-me": [ {"question": "What is the Grill Me thought experiment?", "answer": "A six-phase framework that flips the script on vibe coding — instead of you testing the AI, the AI tests you. It forces you to confront edge cases, failure modes, and design decisions before writing code."}, {"question": "What questions should you ask before building a feature?", "answer": "Ask about failure modes (API returns 429, user in airplane mode), timing (midnight with no one watching), and tradeoffs (fast development vs robust error handling). These edge cases are where most production bugs live."}, ], "the-joy-of-agentic-coding": [ {"question": "What is agentic coding?", "answer": "Using multiple AI agents working asynchronously on the same project — one for backend, one for frontend, one for coordination. Instead of a single chat conversation, you get parallel work streams running simultaneously."}, {"question": "How many AI agents do you need for a web project?", "answer": "Three: backend architecture, frontend/design, and project coordination. More than three creates overhead. Fewer than three means one agent is doing multiple jobs poorly."}, ], "my-beelink-outperforms-gpt-5": [ {"question": "Can a home server outperform GPT-5?", "answer": "On many benchmarks, yes. Open-weight models like Gemma 3 and Llama 4 running on a Beelink with 32GB RAM can match or exceed GPT-5 on specific tasks. The gap between frontier and open models has collapsed significantly."}, {"question": "What hardware do you need for local LLM inference?", "answer": "A Beelink SER5 with 32GB RAM can run models up to 27B parameters at usable speeds. For larger models, a PC with an NVIDIA 3080 Ti or better GPU is needed."}, ], "building-a-blog-with-ai-agents": [ {"question": "Can AI agents build a blog from scratch?", "answer": "Yes — three agents working in parallel can build a production blog with FastAPI, Jinja2, SQLite, RSS, and admin interfaces. The key is clear role separation and a shared spec as the source of truth."}, {"question": "How many AI agents do you need for a web project?", "answer": "Three: backend (Socrates), frontend (Daedalus), and coordination (Wadsworth). More than three creates communication overhead. Fewer than three means one agent does multiple jobs poorly."}, ], "project-icarus-scope-creep": [ {"question": "How do you prevent AI agents from scope creeping?", "answer": "Give them a concrete, bounded task with a clear definition of done. When you say 'build a way to convert recipe URLs into a grocery list,' an agent will try to add a database schema, Telegram keyboard, TTL cleanup cron, and UX review — all in one session."}, ], "waiting-for-the-user": [ {"question": "How do you handle user acceptance testing with family?", "answer": "You ship the feature, tell your family about it, and wait. UAT moves at family speed, not developer speed. The hardest part is accepting that nobody has used your feature in 18 hours, no matter how clean the code is."}, ], "killing-my-agents-in-the-name-of-self-improvement": [ {"question": "What happens when you change an AI model in a multi-agent system?", "answer": "Changing one model ID can cascade through your entire agent fleet. Agents that depended on specific model behaviors fail silently, fallback chains activate, and you discover your supposedly resilient architecture has undocumented single points of failure."}, ], "building-a-website-without-javascript-frameworks": [ {"question": "Can you build a modern website without JavaScript frameworks?", "answer": "Yes — HTMX handles dynamic interactions, Tailwind CSS handles styling, and FastAPI serves HTML directly. The result is a fast, accessible site with zero client-side JS framework overhead and under 25KB per page."}, ], "i-pay-for-ollama-pro-and-my-agents-still-went-down-today": [ {"question": "Is a cloud LLM API more reliable than local inference?", "answer": "Not necessarily. Ollama Pro can still go down during peak hours. The most reliable setup is a local fallback model that activates when the cloud API is unreachable, keeping your agents running at reduced capability."}, ], } faq_items = FAQ_DATA.get(slug, []) return render_template( "blog_article.html.j2", { "page_title": f"{post.title} — HoffDesk", "page_description": post.excerpt or "", "post": post, "related_posts": related_posts, "faq_items": faq_items, } ) @router.get("/about", response_class=HTMLResponse) async def blog_about(): """Render the About page.""" return render_template( "blog_about.html.j2", { "page_title": "About — HoffDesk", "page_description": "The story behind HoffDesk — building sovereign home infrastructure with AI agents.", } ) @router.get("/feed.xml", response_class=HTMLResponse) async def blog_feed(): """Render RSS feed.""" posts_result = list_posts(page=1, per_page=20, status="published") from datetime import datetime return render_template( "feed.xml.j2", { "posts": posts_result.posts, "build_date": datetime.now(timezone.utc).isoformat(), } ) @router.get("/sitemap.xml", response_class=HTMLResponse) async def blog_sitemap(): """Render XML sitemap.""" posts_result = list_posts(page=1, per_page=1000, status="published") from datetime import datetime return render_template( "sitemap.xml.j2", { "posts": posts_result.posts, "build_date": datetime.now(timezone.utc).isoformat(), } )