""" Landing page content API. Public GET /landing (no auth); admin CRUD for hero, stats, services block, service cards, contact. Uses get_db_config() for MySQL connection. """ from typing import Any, List, Optional from fastapi import APIRouter, Depends, HTTPException, status from pydantic import BaseModel from backend.app.config import get_db_config from backend.auth.dependencies import require_admin from backend.auth.schemas import UserInfo def _get_connection(): """MySQL connection using app config. Caller must close.""" cfg = get_db_config() if cfg.get("type") != "mysql": raise HTTPException(status_code=501, detail="Landing API requires MySQL") try: import pymysql return pymysql.connect( host=cfg.get("host", "localhost"), port=int(cfg.get("port", 3306)), user=cfg.get("username", "root"), password=cfg.get("password", ""), database=cfg.get("database"), charset=cfg.get("charset", "utf8mb4"), cursorclass=pymysql.cursors.DictCursor, ) except ImportError: import mysql.connector conn = mysql.connector.connect( host=cfg.get("host", "localhost"), port=int(cfg.get("port", 3306)), user=cfg.get("username", "root"), password=cfg.get("password", ""), database=cfg.get("database"), ) conn._landing_dict_cursor = lambda: conn.cursor(dictionary=True) return conn def _cursor(conn): """Return a cursor that returns dict-like rows.""" return (getattr(conn, "_landing_dict_cursor", conn.cursor))() # ---------- Pydantic models ---------- class LandingHero(BaseModel): id: int title: str subtitle: str banner_text: Optional[str] = None background_image_url: Optional[str] = None updated_at: Optional[str] = None class LandingHeroUpdate(BaseModel): title: Optional[str] = None subtitle: Optional[str] = None banner_text: Optional[str] = None background_image_url: Optional[str] = None class LandingStat(BaseModel): id: int label: str value: str sort_order: int created_at: Optional[str] = None class LandingStatCreate(BaseModel): label: str value: str sort_order: int = 0 class LandingStatUpdate(BaseModel): label: Optional[str] = None value: Optional[str] = None sort_order: Optional[int] = None class LandingServicesBlock(BaseModel): id: int title: str body_text: Optional[str] = None image_url: Optional[str] = None button_1_label: Optional[str] = None button_1_url: Optional[str] = None button_2_label: Optional[str] = None button_2_url: Optional[str] = None button_3_label: Optional[str] = None button_3_url: Optional[str] = None updated_at: Optional[str] = None class LandingServicesBlockUpdate(BaseModel): title: Optional[str] = None body_text: Optional[str] = None image_url: Optional[str] = None button_1_label: Optional[str] = None button_1_url: Optional[str] = None button_2_label: Optional[str] = None button_2_url: Optional[str] = None button_3_label: Optional[str] = None button_3_url: Optional[str] = None class LandingServiceCard(BaseModel): id: int title: str description: Optional[str] = None link_url: Optional[str] = None sort_order: int created_at: Optional[str] = None class LandingServiceCardCreate(BaseModel): title: str description: Optional[str] = None link_url: Optional[str] = None sort_order: int = 0 class LandingServiceCardUpdate(BaseModel): title: Optional[str] = None description: Optional[str] = None link_url: Optional[str] = None sort_order: Optional[int] = None class LandingContact(BaseModel): id: int phone: Optional[str] = None email: Optional[str] = None address: Optional[str] = None updated_at: Optional[str] = None class LandingContactUpdate(BaseModel): phone: Optional[str] = None email: Optional[str] = None address: Optional[str] = None class LandingResponse(BaseModel): hero: LandingHero stats: List[LandingStat] services_block: LandingServicesBlock service_cards: List[LandingServiceCard] contact: LandingContact # ---------- Router ---------- router = APIRouter(prefix="/landing", tags=["Landing"]) def _row_hero(r: dict) -> LandingHero: return LandingHero( id=r["id"], title=r.get("title") or "", subtitle=r.get("subtitle") or "", banner_text=r.get("banner_text"), background_image_url=r.get("background_image_url"), updated_at=str(r["updated_at"]) if r.get("updated_at") else None, ) def _row_stat(r: dict) -> LandingStat: return LandingStat( id=r["id"], label=r.get("label") or "", value=r.get("value") or "", sort_order=int(r.get("sort_order", 0)), created_at=str(r["created_at"]) if r.get("created_at") else None, ) def _row_services_block(r: dict) -> LandingServicesBlock: return LandingServicesBlock( id=r["id"], title=r.get("title") or "", body_text=r.get("body_text"), image_url=r.get("image_url"), button_1_label=r.get("button_1_label"), button_1_url=r.get("button_1_url"), button_2_label=r.get("button_2_label"), button_2_url=r.get("button_2_url"), button_3_label=r.get("button_3_label"), button_3_url=r.get("button_3_url"), updated_at=str(r["updated_at"]) if r.get("updated_at") else None, ) def _row_service_card(r: dict) -> LandingServiceCard: return LandingServiceCard( id=r["id"], title=r.get("title") or "", description=r.get("description"), link_url=r.get("link_url"), sort_order=int(r.get("sort_order", 0)), created_at=str(r["created_at"]) if r.get("created_at") else None, ) def _row_contact(r: dict) -> LandingContact: return LandingContact( id=r["id"], phone=r.get("phone"), email=r.get("email"), address=r.get("address"), updated_at=str(r["updated_at"]) if r.get("updated_at") else None, ) def _fetch_landing(cursor: Any) -> LandingResponse: cursor.execute("SELECT * FROM landing_hero WHERE id = 1") hero_row = cursor.fetchone() if not hero_row: raise HTTPException(status_code=404, detail="Landing hero not found. Run landing_content_tables.sql.") hero = _row_hero(hero_row) cursor.execute("SELECT * FROM landing_stats ORDER BY sort_order, id") stats = [_row_stat(r) for r in cursor.fetchall()] cursor.execute("SELECT * FROM landing_services_block WHERE id = 1") sb_row = cursor.fetchone() if not sb_row: raise HTTPException(status_code=404, detail="Landing services block not found.") services_block = _row_services_block(sb_row) cursor.execute("SELECT * FROM landing_service_cards ORDER BY sort_order, id") service_cards = [_row_service_card(r) for r in cursor.fetchall()] cursor.execute("SELECT * FROM landing_contact WHERE id = 1") contact_row = cursor.fetchone() if not contact_row: raise HTTPException(status_code=404, detail="Landing contact not found.") contact = _row_contact(contact_row) return LandingResponse(hero=hero, stats=stats, services_block=services_block, service_cards=service_cards, contact=contact) @router.get("", response_model=LandingResponse) async def get_landing(): """Public: return full landing content (no auth).""" conn = _get_connection() try: cursor = _cursor(conn) return _fetch_landing(cursor) finally: conn.close() @router.get("/admin", response_model=LandingResponse) async def get_landing_admin(_: UserInfo = Depends(require_admin)): """Admin: same as public GET, for prefilling forms.""" conn = _get_connection() try: cursor = _cursor(conn) return _fetch_landing(cursor) finally: conn.close() # ---------- Hero (single row) ---------- @router.put("/hero", response_model=LandingHero) async def put_landing_hero(body: LandingHeroUpdate, _: UserInfo = Depends(require_admin)): conn = _get_connection() try: cursor = _cursor(conn) cursor.execute( """INSERT INTO landing_hero (id, title, subtitle, banner_text, background_image_url) VALUES (1, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE title = COALESCE(VALUES(title), title), subtitle = COALESCE(VALUES(subtitle), subtitle), banner_text = COALESCE(VALUES(banner_text), banner_text), background_image_url = COALESCE(VALUES(background_image_url), background_image_url)""", (body.title or "", body.subtitle or "", body.banner_text, body.background_image_url), ) conn.commit() cursor.execute("SELECT * FROM landing_hero WHERE id = 1") return _row_hero(cursor.fetchone()) finally: conn.close() # ---------- Stats (CRUD) ---------- @router.get("/stats", response_model=List[LandingStat]) async def list_stats(_: UserInfo = Depends(require_admin)): conn = _get_connection() try: cursor = _cursor(conn) cursor.execute("SELECT * FROM landing_stats ORDER BY sort_order, id") return [_row_stat(r) for r in cursor.fetchall()] finally: conn.close() @router.post("/stats", response_model=LandingStat, status_code=status.HTTP_201_CREATED) async def create_stat(body: LandingStatCreate, _: UserInfo = Depends(require_admin)): conn = _get_connection() try: cursor = _cursor(conn) cursor.execute( "INSERT INTO landing_stats (label, value, sort_order) VALUES (%s, %s, %s)", (body.label, body.value, body.sort_order), ) conn.commit() cursor.execute("SELECT * FROM landing_stats WHERE id = %s", (cursor.lastrowid,)) return _row_stat(cursor.fetchone()) finally: conn.close() @router.put("/stats/{id}", response_model=LandingStat) async def update_stat(id: int, body: LandingStatUpdate, _: UserInfo = Depends(require_admin)): conn = _get_connection() try: cursor = _cursor(conn) updates, vals = [], [] if body.label is not None: updates.append("label = %s") vals.append(body.label) if body.value is not None: updates.append("value = %s") vals.append(body.value) if body.sort_order is not None: updates.append("sort_order = %s") vals.append(body.sort_order) if not updates: cursor.execute("SELECT * FROM landing_stats WHERE id = %s", (id,)) row = cursor.fetchone() if not row: raise HTTPException(status_code=404, detail="Stat not found") return _row_stat(row) vals.append(id) cursor.execute("UPDATE landing_stats SET " + ", ".join(updates) + " WHERE id = %s", vals) conn.commit() cursor.execute("SELECT * FROM landing_stats WHERE id = %s", (id,)) row = cursor.fetchone() if not row: raise HTTPException(status_code=404, detail="Stat not found") return _row_stat(row) finally: conn.close() @router.delete("/stats/{id}", status_code=status.HTTP_204_NO_CONTENT) async def delete_stat(id: int, _: UserInfo = Depends(require_admin)): conn = _get_connection() try: cursor = _cursor(conn) cursor.execute("DELETE FROM landing_stats WHERE id = %s", (id,)) conn.commit() if cursor.rowcount == 0: raise HTTPException(status_code=404, detail="Stat not found") finally: conn.close() # ---------- Services block (single row) ---------- @router.put("/services-block", response_model=LandingServicesBlock) async def put_landing_services_block(body: LandingServicesBlockUpdate, _: UserInfo = Depends(require_admin)): conn = _get_connection() try: cursor = _cursor(conn) cursor.execute( """INSERT INTO landing_services_block (id, title, body_text, image_url, button_1_label, button_1_url, button_2_label, button_2_url, button_3_label, button_3_url) VALUES (1, %s, %s, %s, %s, %s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE title = COALESCE(VALUES(title), title), body_text = COALESCE(VALUES(body_text), body_text), image_url = COALESCE(VALUES(image_url), image_url), button_1_label = COALESCE(VALUES(button_1_label), button_1_label), button_1_url = COALESCE(VALUES(button_1_url), button_1_url), button_2_label = COALESCE(VALUES(button_2_label), button_2_label), button_2_url = COALESCE(VALUES(button_2_url), button_2_url), button_3_label = COALESCE(VALUES(button_3_label), button_3_label), button_3_url = COALESCE(VALUES(button_3_url), button_3_url)""", ( body.title or "", body.body_text, body.image_url, body.button_1_label, body.button_1_url, body.button_2_label, body.button_2_url, body.button_3_label, body.button_3_url, ), ) conn.commit() cursor.execute("SELECT * FROM landing_services_block WHERE id = 1") return _row_services_block(cursor.fetchone()) finally: conn.close() # ---------- Service cards (CRUD) ---------- @router.get("/service-cards", response_model=List[LandingServiceCard]) async def list_service_cards(_: UserInfo = Depends(require_admin)): conn = _get_connection() try: cursor = _cursor(conn) cursor.execute("SELECT * FROM landing_service_cards ORDER BY sort_order, id") return [_row_service_card(r) for r in cursor.fetchall()] finally: conn.close() @router.post("/service-cards", response_model=LandingServiceCard, status_code=status.HTTP_201_CREATED) async def create_service_card(body: LandingServiceCardCreate, _: UserInfo = Depends(require_admin)): conn = _get_connection() try: cursor = _cursor(conn) cursor.execute( "INSERT INTO landing_service_cards (title, description, link_url, sort_order) VALUES (%s, %s, %s, %s)", (body.title, body.description, body.link_url, body.sort_order), ) conn.commit() cursor.execute("SELECT * FROM landing_service_cards WHERE id = %s", (cursor.lastrowid,)) return _row_service_card(cursor.fetchone()) finally: conn.close() @router.put("/service-cards/{id}", response_model=LandingServiceCard) async def update_service_card(id: int, body: LandingServiceCardUpdate, _: UserInfo = Depends(require_admin)): conn = _get_connection() try: cursor = _cursor(conn) updates, vals = [], [] if body.title is not None: updates.append("title = %s") vals.append(body.title) if body.description is not None: updates.append("description = %s") vals.append(body.description) if body.link_url is not None: updates.append("link_url = %s") vals.append(body.link_url) if body.sort_order is not None: updates.append("sort_order = %s") vals.append(body.sort_order) if not updates: cursor.execute("SELECT * FROM landing_service_cards WHERE id = %s", (id,)) row = cursor.fetchone() if not row: raise HTTPException(status_code=404, detail="Service card not found") return _row_service_card(row) vals.append(id) cursor.execute("UPDATE landing_service_cards SET " + ", ".join(updates) + " WHERE id = %s", vals) conn.commit() cursor.execute("SELECT * FROM landing_service_cards WHERE id = %s", (id,)) row = cursor.fetchone() if not row: raise HTTPException(status_code=404, detail="Service card not found") return _row_service_card(row) finally: conn.close() @router.delete("/service-cards/{id}", status_code=status.HTTP_204_NO_CONTENT) async def delete_service_card(id: int, _: UserInfo = Depends(require_admin)): conn = _get_connection() try: cursor = _cursor(conn) cursor.execute("DELETE FROM landing_service_cards WHERE id = %s", (id,)) conn.commit() if cursor.rowcount == 0: raise HTTPException(status_code=404, detail="Service card not found") finally: conn.close() # ---------- Contact (single row) ---------- @router.put("/contact", response_model=LandingContact) async def put_landing_contact(body: LandingContactUpdate, _: UserInfo = Depends(require_admin)): conn = _get_connection() try: cursor = _cursor(conn) cursor.execute( """INSERT INTO landing_contact (id, phone, email, address) VALUES (1, %s, %s, %s) ON DUPLICATE KEY UPDATE phone = COALESCE(VALUES(phone), phone), email = COALESCE(VALUES(email), email), address = COALESCE(VALUES(address), address)""", (body.phone, body.email, body.address), ) conn.commit() cursor.execute("SELECT * FROM landing_contact WHERE id = 1") return _row_contact(cursor.fetchone()) finally: conn.close()