devplacepy/devplacepy/main.py

import asyncio
import logging
import os
import time
from collections import defaultdict
from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse, RedirectResponse
from fastapi.staticfiles import StaticFiles
from devplacepy.config import STATIC_DIR, PORT
from devplacepy.database import init_db, get_table, db, get_users_by_uids, get_comment_counts_by_post_uids, get_vote_counts
from devplacepy.templating import templates
from devplacepy.utils import get_current_user, time_ago
from devplacepy.seo import base_seo_context, site_url, website_schema, breadcrumb_schema, combine
from devplacepy.routers import auth, feed, posts, comments, projects, profile, messages, notifications, votes, avatar, follow, admin, seo, bugs, news, gists, services as services_router, uploads
from devplacepy.services.manager import service_manager
from devplacepy.services.news import NewsService

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)
logger = logging.getLogger(__name__)

_rate_limit_store = defaultdict(list)
RATE_LIMIT = 60
RATE_WINDOW = 60

app = FastAPI(title="DevPlace")
app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static")


@app.exception_handler(404)
async def not_found(request: Request, exc):
    seo_ctx = base_seo_context(request, title="Not Found — DevPlace", description="The page you requested does not exist.", robots="noindex")
    return templates.TemplateResponse(request, "error.html", {**seo_ctx, "request": request, "error_code": 404, "error_message": "Page not found"}, status_code=404)


@app.exception_handler(500)
async def server_error(request: Request, exc):
    logger.exception("500 error on %s %s", request.method, request.url.path)
    seo_ctx = base_seo_context(request, title="Server Error — DevPlace", description="Something went wrong.", robots="noindex")
    return templates.TemplateResponse(request, "error.html", {**seo_ctx, "request": request, "error_code": 500, "error_message": "Internal server error"}, status_code=500)

app.include_router(auth.router, prefix="/auth")
app.include_router(feed.router, prefix="/feed")
app.include_router(posts.router, prefix="/posts")
app.include_router(comments.router, prefix="/comments")
app.include_router(projects.router, prefix="/projects")
app.include_router(profile.router, prefix="/profile")
app.include_router(messages.router, prefix="/messages")
app.include_router(notifications.router, prefix="/notifications")
app.include_router(votes.router, prefix="/votes")
app.include_router(avatar.router, prefix="/avatar")
app.include_router(follow.router, prefix="/follow")
app.include_router(admin.router, prefix="/admin")
app.include_router(seo.router)
app.include_router(bugs.router, prefix="/bugs")
app.include_router(gists.router, prefix="/gists")
app.include_router(news.router, prefix="/news")
app.include_router(services_router.router, prefix="/admin/services")
app.include_router(uploads.router, prefix="/uploads")


@app.middleware("http")
async def add_security_headers(request: Request, call_next):
    response = await call_next(request)
    if not response.headers.get("X-Robots-Tag"):
        response.headers["X-Robots-Tag"] = "index, follow"
    response.headers["X-Content-Type-Options"] = "nosniff"
    return response


@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
    if request.method in ("POST", "PUT", "DELETE", "PATCH"):
        ip = request.client.host if request.client else "unknown"
        now = time.time()
        window_start = now - RATE_WINDOW
        _rate_limit_store[ip] = [t for t in _rate_limit_store[ip] if t > window_start]
        if len(_rate_limit_store[ip]) >= RATE_LIMIT:
            return HTMLResponse("Rate limit exceeded. Try again later.", status_code=429)
        _rate_limit_store[ip].append(now)
    return await call_next(request)


@app.on_event("startup")
async def startup():
    init_db()
    if not os.environ.get("DEVPLACE_DISABLE_SERVICES"):
        news_service = NewsService()
        service_manager.register(news_service)
        asyncio.create_task(service_manager.start_all())
    logger.info(f"DevPlace started on port {PORT}")


@app.on_event("shutdown")
async def shutdown():
    logger.info("Shutting down services...")
    await service_manager.stop_all()


@app.get("/")
async def landing(request: Request):
    user = get_current_user(request)
    if user:
        return RedirectResponse(url="/feed", status_code=302)

    landing_articles = []
    if "news" in db.tables:
        news_table = get_table("news")
        raw = list(news_table.find(show_on_landing=1, order_by=["-synced_at"], _limit=6))
        images_table = get_table("news_images") if "news_images" in db.tables else None
        for a in raw:
            image_url = ""
            if images_table:
                img = images_table.find_one(news_uid=a["uid"])
                if img:
                    image_url = img["url"]
            landing_articles.append({
                "uid": a["uid"],
                "slug": a.get("slug", ""),
                "title": a.get("title", ""),
                "description": (a.get("description", "") or "")[:250],
                "url": a.get("url", ""),
                "source_name": a.get("source_name", ""),
                "grade": a.get("grade", 0),
                "synced_at": a.get("synced_at", "") or "",
                "time_ago": time_ago(a["synced_at"]),
                "image_url": image_url,
            })

    landing_posts = []
    if "posts" in db.tables:
        posts_table = get_table("posts")
        raw_posts = list(posts_table.find(order_by=["-created_at"], _limit=6))
        if raw_posts:
            post_uids = [p["uid"] for p in raw_posts]
            author_uids = [p["user_uid"] for p in raw_posts]
            authors = get_users_by_uids(author_uids)
            comment_counts = get_comment_counts_by_post_uids(post_uids)
            upvotes, downvotes = get_vote_counts(post_uids)
            for p in raw_posts:
                landing_posts.append({
                    "post": p,
                    "author": authors.get(p["user_uid"]),
                    "time_ago": time_ago(p["created_at"]),
                    "comment_count": comment_counts.get(p["uid"], 0),
                    "stars": upvotes.get(p["uid"], 0) - downvotes.get(p["uid"], 0),
                    "slug": p.get("slug", "") or p["uid"],
                })

    base = site_url(request)
    seo_ctx = base_seo_context(
        request,
        title="DevPlace — The Developer Social Network",
        description="Track industry shifts. Discover bold releases. Share what you're building in an open, uncensored environment.",
        breadcrumbs=[],
        schemas=[website_schema(base)],
    )
    return templates.TemplateResponse(request, "landing.html", {
        **seo_ctx, "request": request,
        "landing_articles": landing_articles,
        "landing_posts": landing_posts,
    })