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"""
Dear Author,
Your book cover has been reviewed by the automated validation system.
| Status: | {'✅ PASS' if status == 'PASS' else '❌ REVIEW NEEDED'} |
| Confidence Score: | {confidence}% |
| Detected Issues: | {', '.join(issues) if isinstance(issues, list) else issues} |
{f'View annotated cover overlay' if overlay_url else 'No overlay available.'}
If your status is REVIEW NEEDED, please address the issues listed above and re-upload your revised cover.
Thank you,
BookLeaf Validation Team
""",
}
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}")