100 lines
3.9 KiB
Python
100 lines
3.9 KiB
Python
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
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class AddressLabelService:
|
|
def __init__(self):
|
|
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.output_dir = "app/data/cache/tcgplayer/address_labels/"
|
|
os.makedirs(self.output_dir, exist_ok=True)
|
|
self.executor = ThreadPoolExecutor()
|
|
|
|
async def generate_labels_from_csv(self, csv_path: str, label_type: Literal["dk1201", "dk1241"]) -> List[str]:
|
|
"""Generate address labels from a CSV file and save them as PDFs.
|
|
|
|
Args:
|
|
csv_path: Path to the CSV file containing address data
|
|
label_type: Type of label to generate ("6x4" or "dk1201")
|
|
|
|
Returns:
|
|
List of paths to 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
|
|
pdf_path = await self._generate_single_label(row, label_type)
|
|
if pdf_path:
|
|
generated_files.append(str(pdf_path))
|
|
|
|
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, row: Dict[str, str], label_type: Literal["dk1201", "dk1241"]) -> Optional[str]:
|
|
"""Generate a single address label PDF.
|
|
|
|
Args:
|
|
row: Dictionary containing address data
|
|
label_type: Type of label to generate ("6x4" or "dk1201")
|
|
|
|
Returns:
|
|
Path to 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_path = self.output_dir + f"{row['Order #']}_{label_type}.pdf"
|
|
await loop.run_in_executor(
|
|
self.executor,
|
|
lambda: HTML(string=html_content).write_pdf(str(pdf_path))
|
|
)
|
|
return pdf_path
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error generating label for order {row.get('Order #', 'unknown')}: {str(e)}")
|
|
return None |