from typing import List, Dict, Optional, Literal import csv import os from pathlib import Path from jinja2 import Environment, FileSystemLoader from weasyprint import HTML import logging import asyncio from concurrent.futures import ThreadPoolExecutor from sqlalchemy.orm import Session from app.schemas.file import FileInDB from app.services.base_service import BaseService logger = logging.getLogger(__name__) class AddressLabelService(BaseService): def __init__(self): super().__init__(None) # BaseService doesn't need a model for this service self.template_dir = Path("app/data/assets/templates") self.env = Environment(loader=FileSystemLoader(str(self.template_dir))) self.templates = { "dk1241": self.env.get_template("address_label_dk1241.html"), "dk1201": self.env.get_template("address_label_dk1201.html") } self.return_address_path = "file://" + os.path.abspath("app/data/assets/images/ccrcardsaddress.png") self.executor = ThreadPoolExecutor() async def get_or_create_address_labels(self, db: Session, order_ids: list[str], label_type: Literal["dk1201", "dk1241"]) -> List[FileInDB]: """Get or create address labels for the specified orders. Args: db: Database session order_ids: List of TCGPlayer order numbers label_type: Type of label to generate ("dk1201" or "dk1241") Returns: List of FileInDB objects for generated PDF files """ # check if address labels exist for the order ids file_service = self.get_service('file') # honestly i just dont feel like caching the address labels bc its hard shipping_csv = await file_service.get_file_by_metadata(db, "order_ids", order_ids, "shipping_csv", "text/csv") if shipping_csv: return await self.generate_labels_from_csv(db, shipping_csv.path, label_type) else: order_management = self.get_service('order_management') shipping_csv = await order_management.get_shipping_csv(db, order_ids) return await self.generate_labels_from_csv(db, shipping_csv.path, label_type) async def generate_labels_from_csv(self, db: Session, csv_path: str, label_type: Literal["dk1201", "dk1241"]) -> List[FileInDB]: """Generate address labels from a CSV file and save them as PDFs. Args: db: Database session csv_path: Path to the CSV file containing address data label_type: Type of label to generate ("6x4" or "dk1201") Returns: List of FileInDB objects for generated PDF files """ generated_files = [] # Read CSV file in a thread pool loop = asyncio.get_event_loop() rows = await loop.run_in_executor(self.executor, self._read_csv, csv_path) for row in rows: # if value of Value Of Products is greater than 50, skip if row.get('Value Of Products') and float(row['Value Of Products']) > 50: logger.info(f"Skipping order {row.get('Order #')} because value of products is greater than 50") continue # Generate label for each row file_record = await self._generate_single_label(db, row, label_type) if file_record: generated_files.append(file_record) return generated_files def _read_csv(self, csv_path: str) -> List[Dict[str, str]]: """Read CSV file and return list of rows.""" with open(csv_path, 'r') as csvfile: reader = csv.DictReader(csvfile) return list(reader) async def _generate_single_label(self, db: Session, row: Dict[str, str], label_type: Literal["dk1201", "dk1241"]) -> Optional[FileInDB]: """Generate a single address label PDF. Args: db: Database session row: Dictionary containing address data label_type: Type of label to generate ("6x4" or "dk1201") Returns: FileInDB object for the generated PDF file or None if generation failed """ try: # Prepare template data template_data = { "recipient_name": f"{row['FirstName']} {row['LastName']}", "address_line1": row['Address1'], "address_line2": row['Address2'], "city": row['City'], "state": row['State'], "zip_code": row['PostalCode'] } # Add return address path only for 6x4 labels if label_type == "dk1241": template_data["return_address_path"] = self.return_address_path # Render HTML html_content = self.templates[label_type].render(**template_data) # Generate PDF in a thread pool loop = asyncio.get_event_loop() pdf_content = await loop.run_in_executor( self.executor, lambda: HTML(string=html_content).write_pdf() ) # Prepare metadata metadata = { "order_number": row.get('Order #'), "label_type": label_type } # Save using FileService filename = f"{row['Order #']}_{label_type}.pdf" file_record = await self.file_service.save_file( db=db, file_data=pdf_content, filename=filename, subdir="address_labels", file_type="address_label", content_type="application/pdf", metadata=metadata ) return file_record except Exception as e: logger.error(f"Error generating label for order {row.get('Order #', 'unknown')}: {str(e)}") return None