import json import logging import math import traceback import datetime from typing import Optional from fastapi import APIRouter, Depends, Request from pydantic import BaseModel, ValidationError from sqlalchemy import and_, or_, select, func, update from sqlalchemy.ext.asyncio import AsyncSession from app.db.session import get_db from app.models.generated_tattoo_history import GeneratedTattooHistory from app.schemas.generated_tattoo import TattooHistoryQueryParams, GeneratedTattooHistoryRequest, DeleteTattooRequest from app.core.security import validate_token from app.api.v1.category_api import decrypt_data from app.utils.responses import custom_response from app.core.logging import logger from app.models.tattoo_style import TattooStyle from app.models.category import Category router = APIRouter() import time def now_epoch() -> int: return int(time.time()) # ---------- Get Tattoo History (POST) ---------- @router.post("/tattoo/history") async def get_tattoo_history( request: Request, db: AsyncSession = Depends(get_db), token_data: dict = Depends(validate_token) ): """ Get tattoo history for the logged-in user with pagination, optional search, and tag filtering. """ logger.info("Received request for /tattoo/history endpoint.") try: logger.debug(f"Token data: {token_data}") # Step 1: Authorization check if token_data.get('code') != 200: logger.warning( f"Unauthorized access attempt to /tattoo/history. " f"Message: {token_data.get('message', 'N/A')}" ) return custom_response( code=token_data.get('code', 500), message=token_data.get('message', 'Unauthorized') ) # if not token_data.get('user_id'): # logger.warning("User not authenticated or user_id is missing from the token.") # return custom_response(code=404, message="User not found or not authenticated") user_id = token_data.get('user_id') user_device = token_data.get('user_device_id') if user_id: logger.info(f"Fetching tattoo history for user_id: {user_id}") else: print("User ID not found in token for tattoo history request.") # Step 2: Read raw encrypted data try: encrypted_data = await request.body() logger.debug(f"Encrypted request body: {encrypted_data}") except Exception as e: logger.error(f"Failed to read request body: {str(e)}", exc_info=True) return custom_response(code=400, message=f"Failed to read request body: {str(e)}") # Step 3: Decrypt data decrypted_data = decrypt_data(encrypted_data) logger.debug(f"Decrypted data: {decrypted_data}") # Step 4: Parse decrypted JSON and validate schema try: decrypted_json = decrypted_data # already dict after decryption params = TattooHistoryQueryParams.model_validate(decrypted_json) logger.info(f"Successfully parsed and validated payload for /tattoo/history: {params}") except json.JSONDecodeError: logger.warning(f"Invalid JSON after decryption: {decrypted_data}") return custom_response(code=400, message="Invalid JSON after decryption") except ValidationError as ve: logger.warning(f"Validation failed. Errors: {ve.errors()}") return custom_response(code=422, data=ve.errors(), message="Validation failed") # Step 5: Build query conditions dynamically conditions = [ GeneratedTattooHistory.is_delete == 0 ] if user_id: conditions.append(GeneratedTattooHistory.user_id == user_id) logger.info(f"Filtering history for user_id: {user_id}") else: conditions.append(GeneratedTattooHistory.user_id.is_(None)) conditions.append(GeneratedTattooHistory.user_device_id == user_device) logger.info("No user_id provided; fetching history for guest device only.") # Search filter if params.search: search_term = f"%{params.search}%" conditions.append(GeneratedTattooHistory.name.ilike(search_term)) logger.info(f"Applying search filter for term: '{params.search}'") # Tag filter if params.tag and params.tag.lower() != "all": conditions.append(GeneratedTattooHistory.tattoo_type == params.tag.lower()) logger.info(f"Applying tag filter for type: '{params.tag}'") # Category type filter (exact match) if hasattr(params, "category_type") and params.category_type: conditions.append(GeneratedTattooHistory.category_type == params.category_type) logger.info(f"Applying category_type filter: '{params.category_type}'") # ---------- Query 1: Paginated data ---------- data_stmt = ( select( GeneratedTattooHistory, TattooStyle.style_id.label("style_id"), TattooStyle.name.label("style_name"), Category.category_id.label("category_id"), Category.name.label("category_name"), Category.icon_url.label("icon_url"), ) .outerjoin( TattooStyle, GeneratedTattooHistory.tattoo_style_id == TattooStyle.style_id ) .outerjoin( Category, TattooStyle.category_id == Category.category_id ) .where(and_(*conditions)) .order_by(GeneratedTattooHistory.created_at.desc()) .offset((params.page - 1) * params.limit) .limit(params.limit) ) # ---------- Query 2: Total count ---------- count_stmt = ( select(func.count()) .select_from(GeneratedTattooHistory) # .outerjoin( # TattooStyle, # GeneratedTattooHistory.tattoo_style_id == TattooStyle.style_id # ) # .outerjoin( # Category, # TattooStyle.category_id == Category.category_id # ) .where(and_(*conditions)) ) # Execute queries result_data = await db.execute(data_stmt) rows = result_data.mappings().all() result_count = await db.execute(count_stmt) total_count = result_count.scalar() or 0 total_pages = math.ceil(total_count / params.limit) if total_count > 0 else 0 has_next = params.page < total_pages items = [ { "history_id": row["GeneratedTattooHistory"].history_id, "bg_history_id":None, "user_id": row["GeneratedTattooHistory"].user_id, "tattoo_style_id": row["GeneratedTattooHistory"].tattoo_style_id, "tattoo_style": row["GeneratedTattooHistory"].name, "search_name": row["GeneratedTattooHistory"].search_name, "image_url": row["GeneratedTattooHistory"].image_url, "bg_url": row["GeneratedTattooHistory"].bg_url, "placement_name": row["GeneratedTattooHistory"].placement_name, "text_content": row["GeneratedTattooHistory"].text_content, "font_family": row["GeneratedTattooHistory"].font_family, "size": row["GeneratedTattooHistory"].size, "alignment": row["GeneratedTattooHistory"].alignment, "color_code": row["GeneratedTattooHistory"].color, "color_name": row["GeneratedTattooHistory"].color_name, "tattoo_type": row["GeneratedTattooHistory"].tattoo_type, "category_type": row["GeneratedTattooHistory"].category_type, "is_delete": row["GeneratedTattooHistory"].is_delete, "is_history": row["GeneratedTattooHistory"].is_history, "is_template": row["GeneratedTattooHistory"].is_template, "created_at": row["GeneratedTattooHistory"].created_at, "updated_at": row["GeneratedTattooHistory"].updated_at, # "thumbnail_url":row["thumbnail_url"], "category_icon": row["icon_url"], # "name":row["name"], "category_name": row["category_name"] } for row in rows ] tatto_history = { "details": items, "page": params.page, "limit": params.limit, "total_count": total_count, "total_pages": total_pages, "has_next": has_next, } if not items: logger.info(f"No tattoo history found for user {user_id} with the given criteria.") return custom_response(code=404, data=tatto_history, message="No tattoo history found for this user") logger.info(f"Successfully retrieved {len(items)} history items for user {user_id}.") return custom_response(200, data=tatto_history, message="History retrieved successfully") except Exception as e: logger.error(f"Error retrieving tattoo history: {str(e)}", exc_info=True) return custom_response(code=500, message=f"Internal server error: {str(e)}") # ---------- Store Tattoo History (POST) ---------- @router.post("/tattoo/store") async def store_tattoo_history( request: Request, db: AsyncSession = Depends(get_db), token_data: dict = Depends(validate_token) ): """ Store a new tattoo history entry for the logged-in user. """ logger.info("Received request for /tattoo/store endpoint.") try: # Step 1: Authorization check if token_data.get('code') != 200: logger.warning(f"Unauthorized access attempt to /tattoo/store. Message: {token_data.get('message', 'N/A')}") return custom_response(code=token_data.get('code', 500), message=token_data.get('message', 'Unauthorized')) user_id = token_data.get('user_id') if not user_id: logger.warning("User ID not found in token for tattoo store request.") return custom_response(code=401, message="User not authenticated") logger.info(f"Storing tattoo history for user_id: {user_id}") # Step 2: Read raw encrypted data try: encrypted_data = await request.body() logger.debug(f"Encrypted request body: {encrypted_data}") except Exception as e: logger.error(f"Failed to read request body for /tattoo/store: {str(e)}", exc_info=True) return custom_response(code=400, message=f"Failed to read request body: {str(e)}") # Step 3: Decrypt data decrypted_data = decrypt_data(encrypted_data) logger.debug(f"Decrypted data: {decrypted_data}") # Step 4: Parse decrypted JSON and validate schema try: # decrypted_json = json.loads(decrypted_data) decrypted_json = decrypted_data params = GeneratedTattooHistoryRequest.model_validate(decrypted_json) logger.info(f"Successfully parsed and validated payload for /tattoo/store: {params}") except json.JSONDecodeError: logger.warning(f"Invalid JSON after decryption for /tattoo/store. Decrypted data: {decrypted_data}") return custom_response(code=400, message="Invalid JSON after decryption") except ValidationError as ve: logger.warning(f"Validation failed for tattoo history store request. Errors: {ve.errors()}") return custom_response(code=422, data=ve.errors(), message="Validation failed") # Step 5: Create record new_tattoo = GeneratedTattooHistory( user_id=user_id, name=params.name, search_name=params.search_name, image_url=params.image_url, is_template=params.is_template, tattoo_style_id=params.tattoo_style_id, tattoo_type=params.tattoo_type, placement_name=params.placement_name, text_content=params.text_content, font_family=params.font_family, size=params.size, alignment=params.alignment, color=params.color, created_at=now_epoch(), updated_at=now_epoch()) db.add(new_tattoo) await db.commit() await db.refresh(new_tattoo) response_data = { "history_id": new_tattoo.history_id, "user_id": new_tattoo.user_id, "name": new_tattoo.name, "search_name": new_tattoo.search_name, "image_url": new_tattoo.image_url, "tattoo_type": new_tattoo.tattoo_type, "placement_name": new_tattoo.placement_name, "text_content": new_tattoo.text_content, "font_family": new_tattoo.font_family, "size": new_tattoo.size, "alignment": new_tattoo.alignment, "color": new_tattoo.color, "is_template": new_tattoo.is_template, "tattoo_style_id": new_tattoo.tattoo_style_id, "created_at": new_tattoo.created_at, "updated_at": new_tattoo.updated_at } logger.info(f"Tattoo history record created with ID: {new_tattoo.history_id} for user: {user_id}") return custom_response(200, data=response_data, message="Tattoo history stored successfully") except Exception as e: await db.rollback() logger.error(f"Error storing tattoo history: {str(e)}", exc_info=True) return custom_response(code=500, message=f"Internal server error: {str(e)}") # ---------- Delete Tattoo History (POST) ---------- @router.delete("/tattoo/history/delete") async def store_tattoo_history( request: Request, db: AsyncSession = Depends(get_db), token_data: dict = Depends(validate_token) ): """ delete a tattoo history entry for the logged-in user. """ logger.info("Received request for /tattoo/store endpoint.") try: # Step 1: Authorization check if token_data.get('code') != 200: logger.warning(f"Unauthorized access attempt to /tattoo/store. Message: {token_data.get('message', 'N/A')}") return custom_response(code=token_data.get('code', 500), message=token_data.get('message', 'Unauthorized')) user_id = token_data.get('user_id') # if not user_id: # logger.warning("User ID not found in token for tattoo store request.") # return custom_response(code=401, message="User not authenticated") logger.info(f"Storing tattoo history for user_id: {user_id}") # Step 2: Read raw encrypted data try: encrypted_data = await request.body() logger.debug(f"Encrypted request body: {encrypted_data}") except Exception as e: logger.error(f"Failed to read request body for /tattoo/store: {str(e)}", exc_info=True) return custom_response(code=400, message=f"Failed to read request body: {str(e)}") # Step 3: Decrypt data decrypted_data = decrypt_data(encrypted_data) logger.debug(f"Decrypted data: {decrypted_data}") # Step 4: Parse decrypted JSON and validate schema try: # decrypted_json = json.loads(decrypted_data) decrypted_json = decrypted_data params = DeleteTattooRequest.model_validate(decrypted_json) logger.info(f"Successfully parsed and validated payload for /tattoo/store: {params}") history_id = params.history_id except json.JSONDecodeError: logger.warning(f"Invalid JSON after decryption for /tattoo/store. Decrypted data: {decrypted_data}") return custom_response(code=400, message="Invalid JSON after decryption") except ValidationError as ve: logger.warning(f"Validation failed for tattoo history store request. Errors: {ve.errors()}") return custom_response(code=422, data=ve.errors(), message="Validation failed") # Step 5: Update DB (soft delete) try: stmt = ( update(GeneratedTattooHistory) .where( # GeneratedTattooHistory.user_id == user_id, GeneratedTattooHistory.history_id == history_id ) .values(is_delete=1) ) await db.execute(stmt) await db.commit() logger.info(f"Tattoo history {history_id} marked as deleted for user_id {user_id}") return custom_response(code=200, message="Tattoo history deleted successfully") except Exception as e: await db.rollback() logger.error(f"Database error while deleting tattoo history: {str(e)}", exc_info=True) return custom_response(code=500, message="Database error") except Exception as e: await db.rollback() logger.error(f"Error storing tattoo history: {str(e)}", exc_info=True) return custom_response(code=500, message=f"Internal server error: {str(e)}")