Spaces:
Sleeping
Sleeping
| import os, requests | |
| import smtplib | |
| from email.mime.text import MIMEText | |
| from email.mime.multipart import MIMEMultipart | |
| from dotenv import load_dotenv | |
| from pyairtable import Table | |
| MAKE_WEBHOOK = os.getenv("MAKE_WEBHOOK") | |
| load_dotenv() | |
| # Airtable settings | |
| AIRTABLE_BASE = os.getenv("AIRTABLE_BASE") | |
| AIRTABLE_TABLE = os.getenv("AIRTABLE_TABLE") | |
| AIRTABLE_KEY = os.getenv("AIRTABLE_KEY") | |
| SMTP_SERVER = os.getenv("SMTP_SERVER", "smtp.gmail.com") | |
| SMTP_PORT = int(os.getenv("SMTP_PORT", 587)) | |
| SMTP_USER = os.getenv("SMTP_USER") | |
| SMTP_PASS = os.getenv("SMTP_PASS") | |
| FROM_EMAIL = os.getenv("FROM_EMAIL", "ops@bookleaf.example") | |
| def update_airtable(isbn, status, confidence, issues, overlay_url, validation_message): | |
| table = Table(AIRTABLE_KEY.strip(), AIRTABLE_BASE.strip(), AIRTABLE_TABLE.strip()) | |
| # Check if record already exists | |
| records = table.all(formula=f"{{Book ID}} = '{isbn}'") | |
| # Define valid issue types as allowed by Airtable | |
| valid_issue_types = { | |
| "Badge Overlap", | |
| "Safe Margin", | |
| "Low Resolution", | |
| "OCR Low Confidence", | |
| "Filename Error" | |
| } | |
| # Normalize issues to match allowed options | |
| filtered_issues = [] | |
| for issue in issues or []: | |
| normalized = issue.strip() | |
| if normalized in valid_issue_types: | |
| filtered_issues.append(normalized) | |
| fields = { | |
| "Book ID": isbn, | |
| "Status": status, | |
| "Confidence Score": round(confidence, 2), | |
| "Issue Type": filtered_issues if filtered_issues else [], | |
| "Visual Annotations URL": overlay_url, | |
| "Correction Instructions": validation_message, | |
| } | |
| if records: | |
| record_id = records[0]["id"] | |
| table.update(record_id, fields) | |
| return {"id": record_id, "fields": fields} | |
| else: | |
| record = table.create(fields) | |
| return record | |
| def send_email(isbn, status, issues, overlay_url, confidence, to_email): | |
| if not MAKE_WEBHOOK: | |
| print("Make webhook not configured.") | |
| return | |
| payload = { | |
| "isbn": isbn, | |
| "status": status, | |
| "issues": issues, | |
| "overlay_url": overlay_url, | |
| "confidence": confidence, | |
| "to_email": to_email, | |
| "subject": f"BookLeaf Cover Validation: {status}", | |
| "body": f""" | |
| <html> | |
| <body style="font-family: Arial, sans-serif; color: #222;"> | |
| <p>Dear Author,</p> | |
| <p>Your book cover has been reviewed by the automated validation system.</p> | |
| <table style="border-collapse: collapse; margin: 15px 0;"> | |
| <tr><td><strong>Status:</strong></td><td>{'✅ PASS' if status == 'PASS' else '❌ REVIEW NEEDED'}</td></tr> | |
| <tr><td><strong>Confidence Score:</strong></td><td>{confidence}%</td></tr> | |
| <tr><td><strong>Detected Issues:</strong></td><td>{', '.join(issues) if isinstance(issues, list) else issues}</td></tr> | |
| </table> | |
| <p> | |
| {f'<a href="{overlay_url}" style="color:#1a73e8;">View annotated cover overlay</a>' if overlay_url else 'No overlay available.'} | |
| </p> | |
| <p> | |
| If your status is <strong>REVIEW NEEDED</strong>, please address the issues listed above and re-upload your revised cover. | |
| </p> | |
| <p>Thank you,<br> | |
| <strong>BookLeaf Validation Team</strong></p> | |
| </body> | |
| </html> | |
| """, | |
| } | |
| try: | |
| r = requests.post(MAKE_WEBHOOK, json=payload, timeout=10) | |
| r.raise_for_status() | |
| print("Make webhook triggered successfully.") | |
| except Exception as e: | |
| print(f"Make webhook send failed: {e}") |