from fastapi import APIRouter, Depends, File, UploadFile, HTTPException, Form from sqlalchemy.orm import Session from src.marketing.apps.post import controller, schema from src.utils.db import get_db from fastapi import status from typing import List, Optional import os import uuid import shutil from pathlib import Path import traceback from src.marketing.apps.post.utils import save_social_config_to_db from src.marketing.utils.social_media_post_config import social_media_config from fastapi import Query from src.marketing.apps.post.model import PostTypeConfig from src.marketing.apps.post.utils import format_post_type_configs from src.utils.response import APIResponse from typing import Union from datetime import date from src.marketing.apps.post.model import PostImage from src.apps.auth.controller import get_current_user router = APIRouter() post_type_router = APIRouter() post_type_config = APIRouter() generate_content_router = APIRouter() get_url_information = APIRouter() BASE_URL = os.getenv("BASE_URL", "http://localhost:8000") # BASE_URL = "http://localhost:8000" # Adjust path BASE_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "..")) STATIC_PATH = os.path.join(BASE_DIR, "menu_design", "designer", "static", "marketing", "images") os.makedirs(STATIC_PATH, exist_ok=True) @router.post("/posts", response_model=APIResponse[schema.CalendarPostOut],dependencies=[Depends(get_current_user)]) def create_post(post_data: schema.CalendarPostCreate, db: Session = Depends(get_db)): try: result = controller.create_calendar_post_controller(post_data, db) return { "success": True, "message": "Post created successfully", "data": result } except HTTPException as e: raise e except Exception as e: print("ERROR:", e) traceback.print_exc() raise HTTPException(status_code=500, detail="Internal Server Error") # Get a calendar post by post_id router @router.get("/posts/{post_id}", response_model=schema.CalendarPostOut, status_code=status.HTTP_200_OK, dependencies=[Depends(get_current_user)]) def get_post(post_id: int, db: Session = Depends(get_db)) -> schema.CalendarPostOut: return controller.get_calendar_post_by_id_controller(post_id, db) @router.get("/posts", response_model=APIResponse[List[schema.CalendarPostOut]], status_code=status.HTTP_200_OK, dependencies=[Depends(get_current_user)]) def get_calendar_posts_by_branch_id_and_date_range( calendar_id: int, start_date: date, # <-- Optional: change if using date end_date: date, db: Session = Depends(get_db) ) -> APIResponse[List[schema.CalendarPostOut]]: """ Get calendar posts by calendar_id and a schedule_date between start_date and end_date. """ result = controller.get_calendar_posts_by_calendar_id_and_date_range_controller( calendar_id, start_date, end_date, db ) return APIResponse( success=True, message="Calendar posts retrieved successfully", data=result ) # Update a calendar post router @router.put("/posts/{post_id}", response_model=schema.CalendarPostOut, status_code=status.HTTP_200_OK, dependencies=[Depends(get_current_user)]) def update_post(post_id: int, post_data: schema.CalendarPostUpdate, db: Session = Depends(get_db)) -> schema.CalendarPostOut: return controller.update_calendar_post_controller(post_id, post_data, db) #Partially update router @router.patch("/posts/{post_id}", response_model=schema.CalendarPostOut, status_code=status.HTTP_200_OK, dependencies=[Depends(get_current_user)]) def partial_update_post( post_id: int, post_data: schema.CalendarPostUpdate, db: Session = Depends(get_db) ) -> schema.CalendarPostOut: """ Partially update a calendar post. Only the provided fields will be updated. """ return controller.partial_update_calendar_post_controller(post_id, post_data, db) # Delete a calendar post router @router.delete("/posts/{post_id}", status_code=status.HTTP_200_OK, dependencies=[Depends(get_current_user)]) def delete_post(post_id: int, db: Session = Depends(get_db)) -> dict: controller.delete_calendar_post_controller(post_id, db) return {"detail": "Calendar post deleted successfully"} # --------------------------------------------------| # Calendar Post Type Routers # --------------------------------------------------| # Upload post image router @router.post("/upload-post-image", response_model=schema.PostImageOut, status_code=status.HTTP_201_CREATED, dependencies=[Depends(get_current_user)]) def upload_post_image( image: Optional[UploadFile] = File(None), image_url: Optional[str] = Form(None), post_id: Optional[int] = Form(None), post_type_id: Optional[int] = Form(None), db: Session = Depends(get_db) ): # Ensure either image or image_url is provided if not image and not image_url: raise HTTPException(status_code=400, detail="Either image file or image URL is required.") if image: if image.content_type not in ["image/png", "image/jpeg", "image/jpg", "image/webp"]: raise HTTPException(status_code=400, detail="Invalid image type") filename = f"{uuid.uuid4().hex}_{image.filename}" file_path = os.path.join(STATIC_PATH, filename) with open(file_path, "wb") as buffer: shutil.copyfileobj(image.file, buffer) image_url = f"{BASE_URL}/static/marketing/images/{filename}" image_name = filename elif image_url: # Extract image name from URL image_name = image_url.split("/")[-1] post_image = PostImage(post_id=post_id, post_type_id=post_type_id, image_url=image_url, image_name=image_name) db.add(post_image) db.commit() db.refresh(post_image) return post_image ##Delete Photo @router.delete("/delte-upload-image") def delete_photo_router(image_id:Optional[int]=Form(None), post_type_id: Optional[int] = Form(None), db:Session=Depends(get_db)): return controller.delete_uplod_photo_controller(db=db,image_id=image_id,post_type_id=post_type_id) # Create a calendar post type router # @post_type_router.post("/post-types", response_model=schema.CalendarPostTypeOut, status_code=status.HTTP_201_CREATED, dependencies=[Depends(get_current_user)]) # def create_calendar_post_type( # post_type_data: schema.CalendarPostTypeCreate, # db: Session = Depends(get_db) # ): # try: # # if not post_type_data.images: # # raise HTTPException(status_code=400, detail="At least one image ID is required") # return controller.create_calendar_post_type_controller(post_type_data, db) # except Exception as e: # raise HTTPException(status_code=500, detail=str(e)) @post_type_router.post( "/post-types", response_model=APIResponse[schema.CalendarPostTypeOut], status_code=status.HTTP_201_CREATED, dependencies=[Depends(get_current_user)] ) def create_calendar_post_type( post_type_data: schema.CalendarPostTypeCreate, db: Session = Depends(get_db) ): try: created_post_type = controller.create_calendar_post_type_controller(post_type_data, db) return APIResponse( success=True, message="Calendar post type created successfully", data=created_post_type ) except Exception as e: raise HTTPException(status_code=500, detail=str(e)) # Get a calendar post type by ID router @post_type_router.get("/post-types/{id}", response_model=APIResponse[schema.CalendarPostTypeOut], status_code=status.HTTP_200_OK, dependencies=[Depends(get_current_user)]) def get_calendar_post_type_by_id( id: int, db: Session = Depends(get_db) ) -> schema.CalendarPostTypeOut: """ Get a calendar post type by ID. """ result = controller.get_calendar_post_type_by_id_controller(id, db) return APIResponse( success=True, message="Calendar post type retrieved successfully", data=result ) # Get post types by post_id (query param) @post_type_router.get("/post-types", response_model=APIResponse[List[schema.CalendarPostTypeOut]], status_code=status.HTTP_200_OK, dependencies=[Depends(get_current_user)]) def get_post_types_by_post_id( post_id: int, db: Session = Depends(get_db) ): result = controller.get_calendar_post_types_by_post_id_controller(post_id, db) return APIResponse( success=True, message="Post types retrieved successfully", data=result ) # Update a calendar post type router @post_type_router.patch("/post-type/{id}", response_model=schema.CalendarPostTypeOut, status_code=status.HTTP_200_OK, dependencies=[Depends(get_current_user)]) def update_calendar_post_type( id: int, post_type_data: schema.CalendarPostTypeUpdate, db: Session = Depends(get_db) ) -> schema.CalendarPostTypeOut: """ Update a calendar post type. """ return controller.update_calendar_post_type_controller(id, post_type_data,db) # Delete a calendar post type router @post_type_router.delete("/post-type/{id}", status_code=status.HTTP_204_NO_CONTENT, dependencies=[Depends(get_current_user)]) def delete_calendar_post_type( id: int, db: Session = Depends(get_db) ) -> None: """ Delete a calendar post type. """ controller.delete_calendar_post_type_controller(id, db) return {"detail": "Calendar post type deleted successfully"} # Load social media post configuration and save to database @post_type_config.post("/load-config", status_code=status.HTTP_201_CREATED, dependencies=[Depends(get_current_user)]) def load_social_config(db: Session = Depends(get_db)): config_data = social_media_config() if not config_data: raise ValueError("Social media post configuration data is not available.") save_social_config_to_db(db, config_data) return {"message": "Social media post config saved successfully"} # Get post type configurations by master account IDs @post_type_config.get("/post-type-configs", dependencies=[Depends(get_current_user)]) def get_post_type_configs_alt( master_ids: str = Query(..., description="Comma-separated list of master account IDs"), db: Session = Depends(get_db) ): try: # Parse comma-separated string to list of integers master_ids_list = [int(id.strip()) for id in master_ids.split(',')] except ValueError: raise HTTPException(status_code=400, detail="Invalid master_ids format. Use comma-separated integers.") if not master_ids_list: raise HTTPException(status_code=400, detail="master_account_id list cannot be empty") configs = db.query(PostTypeConfig).filter( PostTypeConfig.master_account_id.in_(master_ids_list) ).all() return format_post_type_configs(configs) #---------------------------------------------------| # Generate social media post content #---------------------------------------------------| # Generate text @generate_content_router.post( "/generate-text", response_model=APIResponse, dependencies=[Depends(get_current_user)] ) def generate_text(request_data: schema.GenerateTextContent): try: result = controller.generate_text_controller(request_data) return APIResponse( success=True, message="Text content generated successfully", data=result ) except HTTPException as e: raise e except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e) ) #Generate Image content @generate_content_router.post( "/image-generate", response_model=APIResponse, dependencies=[Depends(get_current_user)] ) def generate_image(request_data: schema.GenerateImageContent, db: Session = Depends(get_db)): try: result = controller.generate_image_controller(request_data, db=db) return APIResponse( success=True, message="Image content generated successfully", data=result ) except HTTPException as e: raise e except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e) ) #-----------------------------------------------------| # Get URL information #------------------------------------------------------| @get_url_information.post( "/check-url", response_model=APIResponse, status_code=status.HTTP_200_OK) def check_url_endpoint(url_data: schema.URLInfoRequest): result = controller.check_url(url_data) return APIResponse( success=True, message="URL information fetched successfully", data=result )