Spaces:
Sleeping
Sleeping
File size: 3,483 Bytes
00a5d67 84693e0 00a5d67 84693e0 00a5d67 7ea504f 84693e0 7ea504f 00a5d67 7ea504f 00a5d67 7ea504f 00a5d67 84693e0 00a5d67 84693e0 00a5d67 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 |
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}") |