Changes include:
- Roboto Flex loaded from Google Fonts for compact UI text - Reduced decorative borders while retaining useful input and divider borders - Fixed blank journal detail pages - Password-protected journal publisher with four-hour signed sessions - New articles saved directly to posts.json - Prominent Spotify, Instagram, and Steam buttons on Contact - Desktop and mobile layouts visually verified - Production build and authorization tests passed
This commit is contained in:
+143
-3
@@ -1,16 +1,21 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import hashlib
|
||||
import hmac
|
||||
import os
|
||||
import re
|
||||
import smtplib
|
||||
import ssl
|
||||
import threading
|
||||
import time
|
||||
from datetime import date
|
||||
from email.message import EmailMessage
|
||||
from functools import lru_cache
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
from typing import Any, Literal
|
||||
|
||||
from fastapi import FastAPI, HTTPException, Query
|
||||
from fastapi import Depends, FastAPI, Header, HTTPException, Query, status
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
from fastapi.responses import FileResponse
|
||||
from pydantic import BaseModel, EmailStr, Field
|
||||
@@ -22,6 +27,9 @@ PROJECT_DIR = BASE_DIR.parent
|
||||
DATA_DIR = BASE_DIR / "data"
|
||||
FRONTEND_DIST = PROJECT_DIR / "frontend" / "dist"
|
||||
RESUME_PATH = PROJECT_DIR / "Alexander Herlan Resume 2024.pdf"
|
||||
POSTS_PATH = DATA_DIR / "posts.json"
|
||||
POSTS_LOCK = threading.Lock()
|
||||
ADMIN_TOKEN_TTL = 60 * 60 * 4
|
||||
|
||||
|
||||
app = FastAPI(
|
||||
@@ -52,6 +60,25 @@ class ContactPayload(BaseModel):
|
||||
company: str = Field(default="", max_length=200)
|
||||
|
||||
|
||||
class LoginPayload(BaseModel):
|
||||
password: str = Field(min_length=1, max_length=200)
|
||||
|
||||
|
||||
class ArticleSection(BaseModel):
|
||||
heading: str = Field(min_length=3, max_length=160)
|
||||
paragraphs: list[str] = Field(min_length=1, max_length=12)
|
||||
|
||||
|
||||
class NewPostPayload(BaseModel):
|
||||
title: str = Field(min_length=5, max_length=160)
|
||||
excerpt: str = Field(min_length=20, max_length=360)
|
||||
published_at: date
|
||||
read_time: int = Field(ge=1, le=60)
|
||||
tags: list[str] = Field(min_length=1, max_length=6)
|
||||
accent: Literal["blue", "lavender", "peach", "yellow", "mint"] = "mint"
|
||||
content: list[ArticleSection] = Field(min_length=1, max_length=8)
|
||||
|
||||
|
||||
@lru_cache(maxsize=8)
|
||||
def read_data(filename: str) -> Any:
|
||||
path = DATA_DIR / filename
|
||||
@@ -60,6 +87,64 @@ def read_data(filename: str) -> Any:
|
||||
return json.loads(path.read_text(encoding="utf-8"))
|
||||
|
||||
|
||||
def journal_credentials() -> tuple[str, str]:
|
||||
password = os.getenv("JOURNAL_ADMIN_PASSWORD", "")
|
||||
token_secret = os.getenv("JOURNAL_TOKEN_SECRET", "")
|
||||
if not password or not token_secret:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
||||
detail="Journal publishing is not configured.",
|
||||
)
|
||||
return password, token_secret
|
||||
|
||||
|
||||
def issue_admin_token() -> tuple[str, int]:
|
||||
_, token_secret = journal_credentials()
|
||||
expires_at = int(time.time()) + ADMIN_TOKEN_TTL
|
||||
payload = str(expires_at)
|
||||
signature = hmac.new(
|
||||
token_secret.encode("utf-8"), payload.encode("utf-8"), hashlib.sha256
|
||||
).hexdigest()
|
||||
return f"{payload}.{signature}", expires_at
|
||||
|
||||
|
||||
def require_admin(authorization: str | None = Header(default=None)) -> None:
|
||||
if not authorization or not authorization.startswith("Bearer "):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Admin sign-in required.",
|
||||
)
|
||||
|
||||
token = authorization.removeprefix("Bearer ").strip()
|
||||
try:
|
||||
expires, supplied_signature = token.split(".", maxsplit=1)
|
||||
expires_at = int(expires)
|
||||
except (TypeError, ValueError):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Invalid admin session.",
|
||||
)
|
||||
|
||||
_, token_secret = journal_credentials()
|
||||
expected_signature = hmac.new(
|
||||
token_secret.encode("utf-8"), expires.encode("utf-8"), hashlib.sha256
|
||||
).hexdigest()
|
||||
if expires_at <= int(time.time()) or not hmac.compare_digest(
|
||||
supplied_signature, expected_signature
|
||||
):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Your admin session has expired. Please sign in again.",
|
||||
)
|
||||
|
||||
|
||||
def post_slug(title: str) -> str:
|
||||
slug = re.sub(r"[^a-z0-9]+", "-", title.casefold()).strip("-")
|
||||
if not slug:
|
||||
raise HTTPException(status_code=422, detail="The title needs letters or numbers.")
|
||||
return slug
|
||||
|
||||
|
||||
@app.get("/api/health")
|
||||
def health() -> dict[str, str]:
|
||||
return {"status": "ok"}
|
||||
@@ -80,6 +165,28 @@ def skills() -> dict[str, Any]:
|
||||
return read_data("skills.json")
|
||||
|
||||
|
||||
@app.post("/api/auth/login")
|
||||
def admin_login(payload: LoginPayload) -> dict[str, Any]:
|
||||
password, _ = journal_credentials()
|
||||
if not hmac.compare_digest(payload.password, password):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="That password is not correct.",
|
||||
)
|
||||
token, expires_at = issue_admin_token()
|
||||
return {
|
||||
"access_token": token,
|
||||
"token_type": "bearer",
|
||||
"expires_at": expires_at,
|
||||
"expires_in": ADMIN_TOKEN_TTL,
|
||||
}
|
||||
|
||||
|
||||
@app.get("/api/auth/session")
|
||||
def admin_session(_: None = Depends(require_admin)) -> dict[str, bool]:
|
||||
return {"authenticated": True}
|
||||
|
||||
|
||||
@app.get("/api/posts")
|
||||
def posts(
|
||||
q: str | None = Query(default=None, max_length=100),
|
||||
@@ -105,6 +212,40 @@ def posts(
|
||||
return sorted(filtered, key=lambda item: item["published_at"], reverse=True)
|
||||
|
||||
|
||||
@app.post("/api/posts", status_code=status.HTTP_201_CREATED)
|
||||
def create_post(
|
||||
payload: NewPostPayload, _: None = Depends(require_admin)
|
||||
) -> dict[str, Any]:
|
||||
slug = post_slug(payload.title)
|
||||
clean_tags = list(dict.fromkeys(tag.strip() for tag in payload.tags if tag.strip()))
|
||||
if not clean_tags:
|
||||
raise HTTPException(status_code=422, detail="Add at least one topic tag.")
|
||||
if any(len(tag) > 40 for tag in clean_tags):
|
||||
raise HTTPException(status_code=422, detail="Topic tags must be 40 characters or fewer.")
|
||||
|
||||
article = payload.model_dump(mode="json")
|
||||
article["slug"] = slug
|
||||
article["tags"] = clean_tags
|
||||
|
||||
with POSTS_LOCK:
|
||||
current_posts = json.loads(POSTS_PATH.read_text(encoding="utf-8"))
|
||||
if any(post["slug"] == slug for post in current_posts):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_409_CONFLICT,
|
||||
detail="A journal post with this title already exists.",
|
||||
)
|
||||
current_posts.append(article)
|
||||
temporary_path = POSTS_PATH.with_suffix(".json.tmp")
|
||||
temporary_path.write_text(
|
||||
json.dumps(current_posts, indent=2, ensure_ascii=False) + "\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
temporary_path.replace(POSTS_PATH)
|
||||
read_data.cache_clear()
|
||||
|
||||
return article
|
||||
|
||||
|
||||
@app.get("/api/posts/{slug}")
|
||||
def post_by_slug(slug: str) -> dict[str, Any]:
|
||||
if not re.fullmatch(r"[a-z0-9-]+", slug):
|
||||
@@ -193,4 +334,3 @@ if FRONTEND_DIST.is_dir():
|
||||
if requested_file.is_file():
|
||||
return FileResponse(requested_file)
|
||||
return FileResponse(FRONTEND_DIST / "index.html")
|
||||
|
||||
|
||||
Reference in New Issue
Block a user