Price Violation Notice

from __future__ import annotations import logging from io import BytesIO from typing import Any, Iterable, List, Optional, Tuple from uuid import UUID import openpyxl from fastapi import HTTPException, status from sqlalchemy import func, select, update, delete from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.sql import Select from app.models.vendor import Vendor from app.models.violation import Violation from app.models.scraping_result import ScrapingResult from app.core.config import settings logger = logging.getLogger(__name__) # Default notification constants DEFAULT_NOTIFICATION_SUBJECT = "Price Violation Notice – Immediate Attention Required" # Template for individual violation details (repeated for each violation) VIOLATION_DETAILS_TEMPLATE = """Product Name: {{PRODUCT_NAME}} Minimum Selling Price (MSP): {{MSP}} Current Selling Price: {{SCRAPED_PRICE}} Product URL: {{PRODUCT_URL}} Detected On: {{VIOLATION_DATE}}""" DEFAULT_NOTIFICATION_MESSAGE = """Dear {{VENDOR_NAME}}, We hope you are doing well. We have identified that one or more of your product listings are currently being sold below the agreed Minimum Selling Price (MSP). This may lead to pricing inconsistencies across the market and impact brand positioning. Below are the details of the identified violation: {{VIOLATIONS}} We kindly request you to review the above listing(s) and take corrective action at the earliest to ensure compliance with the pricing policy. If you believe this has been flagged incorrectly or need further clarification, please feel free to reach out to us. We appreciate your cooperation and prompt action in resolving this matter. Best regards, {{COMPANY_NAME}} {{SUPPORT_EMAIL}}""" # email template (raw string to avoid escaping) EMAIL_TEMPLATE = r""" Price Violation Notice

⚠ Price Violation Notice

Hello {VENDOR_NAME},

We have identified a pricing violation for one of your listed products associated with the brand {BRAND_NAME}.

Violation Details:

Product Name: {PRODUCT_NAME}
Minimum Selling Price (MSP): ₹{MSP}
Detected Selling Price: ₹{SCRAPED_PRICE}
Price Difference: ₹{PRICE_DIFFERENCE}
Violation Detected On: {VIOLATION_DATE}
Product URL: View Listing

As per the agreed pricing policy, selling below the MSP is not permitted. We request you to review and correct the pricing at the earliest.

If you believe this notification was sent in error, Please contact us at {SUPPORT_EMAIL}.

We appreciate your prompt attention to this matter.

Regards,
{COMPANY_NAME}

""" async def create_vendor(db: AsyncSession, *, name: str, brand_name: str, email: str, website_url: Optional[str], is_active: bool) -> Vendor: # uniqueness of email enforced at database level but we check early existing = await db.execute(select(Vendor).where(Vendor.email == email)) if existing.scalar_one_or_none(): raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Email already registered") # any HttpUrl passed from pydantic must be coerced to string if website_url is not None: website_url = str(website_url) vendor = Vendor(name=name, brand_name=brand_name, email=email, website_url=website_url, is_active=is_active) db.add(vendor) await db.commit() await db.refresh(vendor) return vendor async def get_vendor(db: AsyncSession, *, vendor_id: UUID) -> Vendor: result = await db.execute(select(Vendor).where(Vendor.id == vendor_id)) vendor = result.scalar_one_or_none() if not vendor: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Vendor not found") return vendor async def get_vendor_with_violations(db: AsyncSession, *, vendor_id: UUID) -> tuple[Vendor, List[Violation]]: """Return the vendor and a list of all its violations. Raises HTTPException(404) if the vendor does not exist. """ vendor = await get_vendor(db, vendor_id=vendor_id) result = await db.execute(select(Violation).where(Violation.vendor_id == vendor_id)) violations = result.scalars().all() return vendor, violations async def update_vendor(db: AsyncSession, *, vendor_id: UUID, data: dict[str, Any], changed_by: Optional[str] = None) -> Vendor: vendor = await get_vendor(db, vendor_id=vendor_id) # audit: log the changes updates = {} # convert potential HttpUrl in update data if "website_url" in data and data["website_url"] is not None: data["website_url"] = str(data["website_url"]) for key, value in data.items(): if hasattr(vendor, key): old = getattr(vendor, key) if old != value: updates[key] = (old, value) setattr(vendor, key, value) if updates: logger.info("vendor updated", extra={"vendor_id": str(vendor_id), "changes": updates, "changed_by": changed_by}) await db.commit() await db.refresh(vendor) return vendor async def delete_vendor(db: AsyncSession, *, vendor_id: UUID) -> None: """Delete a vendor and cascade-delete related scraping results and violations.""" from app.models.scraping_result import ScrapingResult from app.models.violation import Violation try: # Delete related scraping results first (to avoid foreign key violation) await db.execute(delete(ScrapingResult).where(ScrapingResult.vendor_id == vendor_id)) logging.info(f"[Vendor Delete] Deleted scraping results for vendor {vendor_id}") # Delete related violations await db.execute(delete(Violation).where(Violation.vendor_id == vendor_id)) logging.info(f"[Vendor Delete] Deleted violations for vendor {vendor_id}") # Now delete the vendor await db.execute(delete(Vendor).where(Vendor.id == vendor_id)) await db.commit() logging.info(f"[Vendor Delete] ✓ Vendor {vendor_id} and related data deleted successfully") except Exception as e: await db.rollback() logging.error(f"[Vendor Delete] ✗ Failed to delete vendor: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to delete vendor: {str(e)}" ) async def list_vendors( db: AsyncSession, page: int = 1, page_size: int = 20, sort_by: str | None = None, search: Optional[str] = None, ) -> Tuple[List[Vendor], int]: stmt: Select = select(Vendor) if search: stmt = stmt.where(Vendor.name.ilike(f"%{search}%")) # count for pagination (reuse same filter but without sorting/offset) count_stmt = select(func.count()).select_from(Vendor) if search: count_stmt = count_stmt.where(Vendor.name.ilike(f"%{search}%")) total_count = (await db.execute(count_stmt)).scalar_one() # handle sorting if sort_by == "name": stmt = stmt.order_by(Vendor.name) elif sort_by == "violation_count": # join violations and order by count stmt = ( stmt.outerjoin(Violation) .group_by(Vendor.id) .order_by(func.count(Violation.id).desc()) ) elif sort_by == "average_discount": # compute average avg_expr = func.coalesce(func.avg((Violation.msp - Violation.scraped_price) / Violation.msp), 0) stmt = ( stmt.outerjoin(Violation) .group_by(Vendor.id) .order_by(avg_expr.desc()) ) offset = (page - 1) * page_size stmt = stmt.offset(offset).limit(page_size) result = await db.execute(stmt) vendors = result.scalars().unique().all() return vendors, total_count async def compute_metrics(db: AsyncSession, vendor: Vendor) -> Tuple[int, float]: # returns (count, average_discount_percent) result = await db.execute( select( func.count(Violation.id), func.coalesce(func.avg((Violation.msp - Violation.scraped_price) / Violation.msp * 100), 0), ).where(Violation.vendor_id == vendor.id) ) count, avg = result.one() return int(count or 0), float(avg or 0.0) async def export_violations( db: AsyncSession, *, vendor_ids: Optional[List[UUID]] = None ) -> bytes: stmt = select(Violation, Vendor).join(Vendor) # Handle multiple vendor_ids (or None to get all) if vendor_ids and len(vendor_ids) > 0: stmt = stmt.where(Violation.vendor_id.in_(vendor_ids)) result = await db.execute(stmt) rows = result.all() wb = openpyxl.Workbook() ws = wb.active ws.title = "Violations" headers = [ "Vendor Name", "Brand Name", "Product Name", "Barcode Number", "Reference ID", "MSP", "Scraped Price", "Price Difference", "Violation Date", "URL", ] ws.append(headers) for violation, vendor in rows: ws.append([ vendor.name, vendor.brand_name, violation.product_name, violation.barcode_number or "", violation.reference_id or "", float(violation.msp), float(violation.scraped_price), float(violation.msp - violation.scraped_price), violation.violation_date.isoformat(), violation.url, ]) stream = BytesIO() wb.save(stream) return stream.getvalue() async def send_violation_notifications(db: AsyncSession, *, violation_ids: Iterable[UUID]) -> None: from datetime import datetime # fetch violations with vendor stmt = select(Violation, Vendor).join(Vendor).where(Violation.id.in_(list(violation_ids))) result = await db.execute(stmt) records = result.all() if not records: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="No violations found for given IDs") # Group violations by vendor violations_by_vendor: dict[UUID, tuple[Vendor, List[Violation]]] = {} for violation, vendor in records: if vendor.id not in violations_by_vendor: violations_by_vendor[vendor.id] = (vendor, []) violations_by_vendor[vendor.id][1].append(violation) # send one email per vendor with all their violations for vendor_id, (vendor, violations) in violations_by_vendor.items(): _send_violation_email(vendor, violations) # Mark violations as notified for violation in violations: violation.notification_sent_at = datetime.utcnow() # Commit the changes to mark violations as notified await db.commit() def _send_violation_email(vendor: Vendor, violations: List[Violation]) -> None: host = settings.SMTP_HOST port = settings.SMTP_PORT smtp_user = settings.SMTP_USER smtp_pass = settings.SMTP_PASS sender = settings.SMTP_FROM if not host or not port or not sender: raise RuntimeError("SMTP configuration incomplete") # Build violation details for each violation violation_items_html = "" for idx, violation in enumerate(violations, 1): # Format prices to 3 decimal places msp_formatted = f"{float(violation.msp):.3f}" scraped_price_formatted = f"{float(violation.scraped_price):.3f}" if violation.scraped_price else "N/A" price_difference = float(violation.msp) - float(violation.scraped_price) if violation.scraped_price else 0 price_difference_formatted = f"{price_difference:.3f}" violation_items_html += f"""

Violation #{idx}

Product Name:	{violation.product_name}
Minimum Selling Price (MSP):	₹{msp_formatted}
Detected Selling Price:	₹{scraped_price_formatted}
Price Difference:	₹{price_difference_formatted}
Violation Detected On:	{violation.violation_date.isoformat()}
Product URL:	View Listing

""" # Build html with multiple violations html = f""" Price Violation Notice

⚠ Price Violation Notice

Hello {vendor.name},

We have identified {'a pricing violation' if len(violations) == 1 else f'{len(violations)} pricing violations'} for {'one of your listed products' if len(violations) == 1 else 'your listed products'} associated with the brand {vendor.brand_name}.

Total Violations: {len(violations)}

{violation_items_html}

As per the agreed pricing policy, selling below the MSP is not permitted. We request you to review and correct the pricing at the earliest.

If you believe {'this notification' if len(violations) == 1 else 'any of these notifications'} was sent in error, please contact us at {smtp_user or 'support@company.com'}.

We appreciate your prompt attention to this matter.

Regards,
Your Company

""" from email.message import EmailMessage msg = EmailMessage() msg["Subject"] = f"Price Violation Notice - {len(violations)} {'Violation' if len(violations) == 1 else 'Violations'}" msg["From"] = sender msg["To"] = vendor.email # Set HTML ONLY (not plain text + HTML alternative) to ensure formatting is preserved # msg.set_content(html, subtype="html", charset="utf-8") msg.set_content("This email requires an HTML capable client.") msg.add_alternative(html, subtype="html") import smtplib with smtplib.SMTP(host, port) as smtp: smtp.ehlo() smtp.starttls() smtp.ehlo() if smtp_user and smtp_pass: smtp.login(smtp_user.strip(), smtp_pass.replace(" ", "").strip()) smtp.send_message(msg) async def export_scraping_results( db: AsyncSession, *, vendor_id: Optional[UUID] = None, product_id: Optional[int] = None, status_filter: Optional[str] = None # 'compliant', 'violation', 'complaining', 'unknown' ) -> bytes: """Export scraping results to XLSX.""" stmt = select(ScrapingResult) if vendor_id: stmt = stmt.where(ScrapingResult.vendor_id == vendor_id) if product_id: stmt = stmt.where(ScrapingResult.product_id == product_id) if status_filter: stmt = stmt.where(ScrapingResult.compliance_status == status_filter) stmt = stmt.order_by(ScrapingResult.scraped_at.desc()) result = await db.execute(stmt) rows = result.scalars().all() wb = openpyxl.Workbook() ws = wb.active ws.title = "Scraping Results" headers = [ "Product Name", "Product Reference", "Barcode", "Vendor Name", "Vendor URL", "MSP", "Scraped Price", "Price Difference", "Percentage Difference", "Status", "Source Type", "Scraped At", ] ws.append(headers) for scraping_result in rows: ws.append([ scraping_result.product_name, scraping_result.reference_id or "", scraping_result.barcode or "", scraping_result.vendor_name, scraping_result.vendor_url, float(scraping_result.msp), float(scraping_result.scraped_price) if scraping_result.scraped_price else "N/A", float(scraping_result.price_difference) if scraping_result.price_difference else "N/A", float(scraping_result.percentage_difference) if scraping_result.percentage_difference else "N/A", scraping_result.compliance_status, scraping_result.source_type, scraping_result.scraped_at.isoformat(), ]) stream = BytesIO() wb.save(stream) return stream.getvalue() async def export_combined_report( db: AsyncSession, *, vendor_id: Optional[UUID] = None, product_id: Optional[int] = None ) -> bytes: """Export combined report with both violations and scraping results.""" # Fetch violations violation_stmt = select(Violation, Vendor).join(Vendor) if vendor_id: violation_stmt = violation_stmt.where(Violation.vendor_id == vendor_id) if product_id: # Note: Violation doesn't have product_id, but we can filter by reference_id pass violation_result = await db.execute(violation_stmt) violations = violation_result.all() # Fetch scraping results scraping_stmt = select(ScrapingResult) if vendor_id: scraping_stmt = scraping_stmt.where(ScrapingResult.vendor_id == vendor_id) if product_id: scraping_stmt = scraping_stmt.where(ScrapingResult.product_id == product_id) scraping_result = await db.execute(scraping_stmt) scraping_results = scraping_result.scalars().all() wb = openpyxl.Workbook() # ===== VIOLATIONS SHEET ===== ws_violations = wb.active ws_violations.title = "Violations" violation_headers = [ "Vendor Name", "Brand Name", "Product Name", "Barcode Number", "Reference ID", "MSP", "Scraped Price", "Price Difference", "Confirmation Count", "Source Type", "Violation Date", "Last Confirmed", "URL", ] ws_violations.append(violation_headers) for violation, vendor in violations: ws_violations.append([ vendor.name, vendor.brand_name, violation.product_name, violation.barcode_number or "", violation.reference_id or "", float(violation.msp), float(violation.scraped_price), float(violation.msp - violation.scraped_price), violation.confirmation_count, violation.source_type, violation.violation_date.isoformat(), violation.last_confirmed_date.isoformat() if violation.last_confirmed_date else "", violation.url, ]) # ===== SCRAPING RESULTS SHEET ===== ws_results = wb.create_sheet("Scraping Results") result_headers = [ "Product Name", "Product Reference", "Barcode", "Vendor Name", "Vendor URL", "MSP", "Scraped Price", "Price Difference", "Percentage Difference", "Status", "Source Type", "Scraped At", ] ws_results.append(result_headers) for sr in scraping_results: ws_results.append([ sr.product_name, sr.reference_id or "", sr.barcode or "", sr.vendor_name, sr.vendor_url, float(sr.msp), float(sr.scraped_price) if sr.scraped_price else "N/A", float(sr.price_difference) if sr.price_difference else "N/A", float(sr.percentage_difference) if sr.percentage_difference else "N/A", sr.compliance_status, sr.source_type, sr.scraped_at.isoformat(), ]) stream = BytesIO() wb.save(stream) return stream.getvalue() async def get_violations_prefill_data( db: AsyncSession, *, violation_ids: Iterable[UUID], ) -> dict: """ Get pre-fill data for the notification modal. Returns: - recipient_email: Auto-fetched from vendor (None if multiple vendors) - default_subject: Default subject template - default_message_body: Default message template - violations_count: Number of violations - vendor_names: List of vendor names involved """ # Fetch violations with vendor info stmt = select(Violation, Vendor).join(Vendor).where(Violation.id.in_(list(violation_ids))) result = await db.execute(stmt) records = result.all() if not records: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="No violations found for given IDs") # Extract unique vendors vendors_set = {} for violation, vendor in records: if vendor.id not in vendors_set: vendors_set[vendor.id] = vendor # Determine recipient email recipient_email = None if len(vendors_set) == 1: # Single vendor - pre-fill with their email recipient_email = list(vendors_set.values())[0].email return { "recipient_email": recipient_email, "default_subject": DEFAULT_NOTIFICATION_SUBJECT, "default_message_body": DEFAULT_NOTIFICATION_MESSAGE, "violations_count": len(records), "vendor_names": [v.name for v in vendors_set.values()], "note": "Multiple vendors detected" if len(vendors_set) > 1 else None, } def _substitute_template_variables( message_body: str, vendor_name: str, product_name: str, msp: float, scraped_price: float, product_url: str, violation_date: str, company_name: str = "Your Company", support_email: str = "support@company.com", ) -> str: """ Substitute template variables in message body. Handles both plain text and HTML content. Supported variables: - {{VENDOR_NAME}} - {{PRODUCT_NAME}} - {{MSP}} - {{SCRAPED_PRICE}} - {{PRODUCT_URL}} - {{VIOLATION_DATE}} - {{COMPANY_NAME}} - {{SUPPORT_EMAIL}} """ # Format prices msp_formatted = f"₹{float(msp):.3f}" scraped_price_formatted = f"₹{float(scraped_price):.3f}" if scraped_price else "N/A" # Create substitution mapping substitutions = { "{{VENDOR_NAME}}": vendor_name, "{{PRODUCT_NAME}}": product_name, "{{MSP}}": msp_formatted, "{{SCRAPED_PRICE}}": scraped_price_formatted, "{{PRODUCT_URL}}": product_url, "{{VIOLATION_DATE}}": violation_date, "{{COMPANY_NAME}}": company_name, "{{SUPPORT_EMAIL}}": support_email, } result = message_body for placeholder, value in substitutions.items(): # Replace placeholder with value (preserves all HTML tags) result = result.replace(placeholder, str(value)) return result async def send_custom_violation_notification( db: AsyncSession, *, violation_ids: Iterable[UUID], recipient_email: Optional[str], subject: str, message_body: str, ) -> None: """ Send custom violation notification email to specified recipient. If recipient_email is not provided, automatically fetches vendor email. If multiple violations from different vendors without recipient_email, raises error. The message_body supports template variables: - {{VENDOR_NAME}}, {{PRODUCT_NAME}}, {{MSP}}, {{SCRAPED_PRICE}} - {{PRODUCT_URL}}, {{VIOLATION_DATE}}, {{COMPANY_NAME}}, {{SUPPORT_EMAIL}} If multiple violations are provided, each violation will be sent with individually substituted variables for that violation. """ from datetime import datetime from email.message import EmailMessage import smtplib # Log what we received logger.info(f"[send_custom_violation_notification] Function called with:") logger.info(f" - violation_ids: {list(violation_ids)}") logger.info(f" - recipient_email: {recipient_email}") logger.info(f" - subject: {subject[:100] if subject else 'EMPTY'}") logger.info(f" - message_body length: {len(message_body) if message_body else 0} chars") if message_body: logger.info(f" - message_body preview (first 300 chars): {message_body[:300]}") logger.info(f" - message_body contains HTML tags: {'<' in message_body and '>' in message_body}") # Validate inputs if not message_body or message_body.strip() == "": raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Message body cannot be empty" ) if not subject or subject.strip() == "": raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Subject cannot be empty" ) # Fetch violations with vendor info stmt = select(Violation, Vendor).join(Vendor).where(Violation.id.in_(list(violation_ids))) result = await db.execute(stmt) records = result.all() if not records: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="No violations found for given IDs") logger.info(f"[send_custom] recipient_email received: {recipient_email}") # Validate recipient email is provided if not recipient_email or (isinstance(recipient_email, str) and recipient_email.strip() == ""): raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="recipient_email is required. Please provide the email address where you want to send this notification." ) logger.info(f"[send_custom] Email will be sent to: {recipient_email}") # Get SMTP configuration host = settings.SMTP_HOST port = settings.SMTP_PORT smtp_user = settings.SMTP_USER smtp_pass = settings.SMTP_PASS sender = settings.SMTP_FROM if not host or not port or not sender: raise RuntimeError("SMTP configuration incomplete") # Build email content with all violations violations_list = [] for violation, vendor in records: violations_list.append(violation) # Simple approach: Substitute template variables ONCE and send as-is # (Frontend may have already formatted all violations in the message) # Get first vendor for substitution first_vendor = records[0][1] if records else None combined_html_body = _substitute_template_variables( message_body=message_body, vendor_name=first_vendor.name if first_vendor else "Vendor", product_name="", # Not used in full message msp=0, # Not used in full message scraped_price=0, # Not used in full message product_url="", # Not used in full message violation_date="", # Not used in full message company_name=getattr(settings, "COMPANY_NAME", "Your Company"), support_email=smtp_user or "support@company.com", ) # Wrap in minimal HTML structure (preserve formatting from rich text editor) html_content = f""" {subject} {combined_html_body} """ # Create and send email msg = EmailMessage() msg["Subject"] = subject msg["From"] = sender msg["To"] = recipient_email # Set HTML content (EmailMessage.set_content with subtype="html" automatically sets Content-Type) # This ensures rich text formatting from editor is preserved msg.set_content(html_content, subtype="html", charset="utf-8") # Debug logging logger.info(f"[Email Debug] Sending to: {recipient_email}") logger.info(f"[Email Debug] Subject: {subject}") logger.info(f"[Email Debug] HTML length: {len(html_content)} characters") logger.info(f"[Email Debug] Total violations: {len(violations_list)}") logger.info(f"[Email Debug] Message body preview: {message_body[:200]}...") logger.info(f"[Email Debug] Content-Type header: {msg.get_content_type()}") logger.info(f"[Email Debug] Full headers: {dict(msg.items())}") # Send via SMTP try: with smtplib.SMTP(host, port) as smtp: smtp.ehlo() smtp.starttls() smtp.ehlo() if smtp_user and smtp_pass: smtp.login(smtp_user.strip(), smtp_pass.replace(" ", "").strip()) smtp.send_message(msg) logger.info(f"✓ Custom notification sent to {recipient_email} for {len(violations_list)} violation(s)") # Mark violations as notified for violation in violations_list: violation.notification_sent_at = datetime.utcnow() await db.commit() except smtplib.SMTPAuthenticationError as e: logger.error(f"✗ SMTP Authentication failed: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"SMTP Authentication failed. Please check your email credentials." ) except smtplib.SMTPException as e: logger.error(f"✗ SMTP Error: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"SMTP error: {str(e)}" ) except Exception as e: logger.error(f"✗ Failed to send custom notification: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to send notification: {str(e)}" )