from typing import List, Dict import pandas as pd from datetime import datetime from pathlib import Path from jinja2 import Environment, FileSystemLoader from weasyprint import HTML import logging import asyncio from app.schemas.file import FileInDB from app.services.base_service import BaseService from sqlalchemy.orm import Session logger = logging.getLogger(__name__) class PullSheetService(BaseService): def __init__(self): super().__init__(None) self.template_dir = Path("app/data/assets/templates") self.env = Environment(loader=FileSystemLoader(str(self.template_dir))) self.template = self.env.get_template("pull_sheet.html") async def get_or_create_rendered_pull_sheet(self, db: Session, order_ids: list[str]) -> FileInDB: # get file service file_service = self.get_service('file') # check if rendered pull sheet exists rendered_pull_sheet = await file_service.get_file_by_metadata(db, "order_ids", order_ids, "rendered_pull_sheet", "application/pdf") if rendered_pull_sheet: return rendered_pull_sheet # check if pull sheet data file exists pull_sheet_data_file = await file_service.get_file_by_metadata(db, "order_ids", order_ids, "pull_sheet", "text/csv") if pull_sheet_data_file: # generate pdf from pull sheet data file return await self.generate_pull_sheet_pdf(db, pull_sheet_data_file) # if no pull sheet data file exists, get it from order management service order_service = self.get_service('order_management') pull_sheet_data_file = await order_service.get_pull_sheet(db, order_ids) return await self.generate_pull_sheet_pdf(db, pull_sheet_data_file) async def generate_pull_sheet_pdf(self, db: Session, file: FileInDB) -> FileInDB: """Generate a PDF pull sheet from a CSV file. Args: file: FileInDB object containing the pull sheet data Returns: Path to the generated PDF file """ try: # Read and process CSV data items = await self._read_and_process_csv(file.path) # Prepare template data template_data = { 'items': items, 'generation_date': datetime.now().strftime("%Y-%m-%d %H:%M:%S") } # Render HTML html_content = self.template.render(**template_data) # Ensure metadata is properly formatted metadata = file.file_metadata.copy() if file.file_metadata else {} if 'order_ids' in metadata: metadata['order_ids'] = sorted(metadata['order_ids']) file_service = self.get_service('file') return await file_service.save_file( db=db, file_data=html_content, filename=f"rendered_pull_sheet_{datetime.now().strftime('%Y%m%d_%H%M%S')}.pdf", subdir="tcgplayer/pull_sheets/rendered", file_type="rendered_pull_sheet", content_type="application/pdf", metadata=metadata, html_content=True # This tells FileService to convert HTML to PDF ) except Exception as e: logger.error(f"Error generating pull sheet PDF: {str(e)}") raise async def _read_and_process_csv(self, csv_path: str) -> List[Dict]: """Read and process CSV data using pandas. Args: csv_path: Path to the CSV file Returns: List of processed items """ # Read CSV into pandas DataFrame in a separate thread to avoid blocking df = await asyncio.get_event_loop().run_in_executor( None, lambda: pd.read_csv(csv_path) ) # Filter out the "Orders Contained in Pull Sheet" row df = df[df['Product Line'] != 'Orders Contained in Pull Sheet:'].copy() # Convert Set Release Date to datetime df['Set Release Date'] = pd.to_datetime(df['Set Release Date'], format='%m/%d/%Y %H:%M:%S') # Sort by Set Release Date (descending) and then Product Name (ascending) df = df.sort_values(['Set Release Date', 'Set', 'Product Name'], ascending=[False, True, True]) # Convert to list of dictionaries items = [] for _, row in df.iterrows(): items.append({ 'product_name': row['Product Name'], 'condition': row['Condition'], 'quantity': str(int(row['Quantity'])), # Convert to string for template 'set': row['Set'], 'rarity': row['Rarity'], 'card_number': str(int(row['Number'])) if 'Number' in row else '' }) return items