import os import json import boto3 from datetime import datetime from pathlib import Path from typing import Optional, List from fastapi import FastAPI, HTTPException, Header, Depends from fastapi.middleware.cors import CORSMiddleware from fastapi.staticfiles import StaticFiles from fastapi.responses import FileResponse from pydantic import BaseModel from botocore.exceptions import ClientError # --- CONFIG & AWS SETUP --- ADMIN_PASSWORD = os.getenv("ADMIN_PASSWORD", "secret") S3_BUCKET = os.getenv("S3_BUCKET_NAME") AWS_REGION = os.getenv("AWS_REGION", "us-east-1") # Initialize S3 Client s3_client = boto3.client( "s3", aws_access_key_id=os.getenv("AWS_ACCESS_KEY_ID"), aws_secret_access_key=os.getenv("AWS_SECRET_ACCESS_KEY"), region_name=AWS_REGION ) def read_s3_json(file_key: str): """Helper to read JSON directly from S3.""" try: response = s3_client.get_object(Bucket=S3_BUCKET, Key=file_key) return json.loads(response['Body'].read().decode('utf-8')) except ClientError as e: if e.response['Error']['Code'] == "NoSuchKey": return [] raise HTTPException(status_code=500, detail="S3 Read Error") def write_s3_json(file_key: str, data: list): """Helper to write JSON directly to S3.""" try: s3_client.put_object( Bucket=S3_BUCKET, Key=file_key, Body=json.dumps(data, indent=2, default=str), ContentType='application/json' ) except Exception: raise HTTPException(status_code=500, detail="S3 Write Error") async def verify_admin(x_admin_password: Optional[str] = Header(None)): if x_admin_password != ADMIN_PASSWORD: raise HTTPException(status_code=401, detail="Unauthorized Access") return True # --- MODELS --- class ArticleCreate(BaseModel): title: str slug: str content: str excerpt: Optional[str] = "" tags: List[str] = [] published: bool = True class ArticleResponse(BaseModel): id: int title: str slug: str content: str excerpt: str tags: List[str] published: bool created_at: str class LinkCreate(BaseModel): title: str url: str category: str description: Optional[str] = "" tags: List[str] = [] class LinkResponse(LinkCreate): id: int created_at: str # --- APP INIT --- app = FastAPI(title="Aditya writes here") app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], ) # Serve UI app.mount("/static", StaticFiles(directory="static"), name="static") @app.get("/") async def read_root(): return FileResponse("static/index.html") @app.get("/admin") async def read_admin(): return FileResponse("static/admin.html") # --- API ROUTES (ARTICLES) --- @app.get("/api/auth/verify") async def check_auth(auth: bool = Depends(verify_admin)): return {"status": "Authorized"} @app.get("/api/articles", response_model=List[ArticleResponse]) async def get_articles(published_only: bool = True): articles = read_s3_json("articles.json") if published_only: articles = [a for a in articles if a.get('published', True)] for a in articles: if "tags" not in a or a["tags"] is None: a["tags"] = [] if "excerpt" not in a or a["excerpt"] is None: a["excerpt"] = "" return sorted(articles, key=lambda x: x['created_at'], reverse=True) @app.get("/api/articles/{slug}", response_model=ArticleResponse) async def get_article(slug: str): articles = read_s3_json("articles.json") article = next((a for a in articles if a['slug'].lower() == slug.lower()), None) if not article: raise HTTPException(status_code=404, detail="Article not found") return article @app.post("/api/articles", response_model=ArticleResponse) async def create_article(article: ArticleCreate, auth: bool = Depends(verify_admin)): articles = read_s3_json("articles.json") if any(a['slug'] == article.slug for a in articles): raise HTTPException(status_code=400, detail="Slug already exists") new_id = max([a['id'] for a in articles], default=0) + 1 new_article = { "id": new_id, **article.dict(), "created_at": datetime.utcnow().isoformat() } articles.append(new_article) write_s3_json("articles.json", articles) return new_article @app.put("/api/articles/{article_id}", response_model=ArticleResponse) async def update_article(article_id: int, article_update: ArticleCreate, auth: bool = Depends(verify_admin)): articles = read_s3_json("articles.json") for i, a in enumerate(articles): if a['id'] == article_id: if article_update.slug != a['slug'] and any(other['slug'] == article_update.slug for other in articles): raise HTTPException(status_code=400, detail="Slug already exists") articles[i].update(article_update.dict()) write_s3_json("articles.json", articles) return articles[i] raise HTTPException(status_code=404, detail="Article not found") @app.delete("/api/articles/{article_id}") async def delete_article(article_id: int, auth: bool = Depends(verify_admin)): articles = read_s3_json("articles.json") initial_len = len(articles) articles = [a for a in articles if a['id'] != article_id] if len(articles) == initial_len: raise HTTPException(status_code=404, detail="Article not found") write_s3_json("articles.json", articles) return {"message": "Data Purged"} # --- API ROUTES (SAVED LINKS) --- @app.get("/api/links", response_model=List[LinkResponse]) async def get_links(category: Optional[str] = None): links = read_s3_json("links.json") if category: links = [l for l in links if l.get('category', '').lower() == category.lower()] return sorted(links, key=lambda x: x.get('created_at', ''), reverse=True) @app.post("/api/links", response_model=LinkResponse) async def create_link(link: LinkCreate, auth: bool = Depends(verify_admin)): links = read_s3_json("links.json") new_id = max([l.get('id', 0) for l in links], default=0) + 1 new_link = { "id": new_id, **link.dict(), "created_at": datetime.utcnow().isoformat() } links.append(new_link) write_s3_json("links.json", links) return new_link @app.put("/api/links/{link_id}", response_model=LinkResponse) async def update_link(link_id: int, link_update: LinkCreate, auth: bool = Depends(verify_admin)): links = read_s3_json("links.json") for i, l in enumerate(links): if l.get('id') == link_id: # Preserve the original creation date created_at = l.get("created_at") links[i].update(link_update.dict()) links[i]["created_at"] = created_at write_s3_json("links.json", links) return links[i] raise HTTPException(status_code=404, detail="Link not found") @app.delete("/api/links/{link_id}") async def delete_link(link_id: int, auth: bool = Depends(verify_admin)): links = read_s3_json("links.json") initial_len = len(links) links = [l for l in links if l.get('id') != link_id] if len(links) == initial_len: raise HTTPException(status_code=404, detail="Link not found") write_s3_json("links.json", links) return {"message": "Link Purged"} if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=7860)