Nguồn gốc ý tưởng

Tôi muốn thêm chức năng Newsletter cho blog của mình và tốt nhất là hỗ trợ chuyển đổi RSS sang Newsletter. Như vậy, mỗi khi blog được cập nhật, bản tin cũng sẽ tự động cập nhật.

Có rất nhiều dịch vụ như vậy trên thị trường, nhưng phần lớn đều tính phí và không rẻ chút nào, ví dụ như Mailchimp nổi tiếng. Cũng có các giải pháp miễn phí, chẳng hạn như Mailbrew mà tôi đã sử dụng trong một thời gian. Tuy nhiên, sản phẩm này hiện không còn được duy trì, thậm chí gần đây tôi không thể đăng nhập, vì vậy tôi buộc phải từ bỏ.

“Tự tay làm tỷ lệ bóng đá hôm nay lấy thì mới đủ ăn”, tôi bắt đầu thử nghiệm cách tự keo bd xây dựng hệ thống.

Ý tưởng

Để chuyển đổi RSS thành email, chỉ cần hai chức năng chính:

1. Công cụ biểu mẫu để thu thập địa chỉ email của độc giả, cần hỗ trợ hủy đăng ký.
2. Dịch vụ gửi email, phân tích nội dung RSS và gửi qua email.

Về điểm thứ hai, chỉ cần tìm một máy chủ gửi email, sau đó sử dụng script Python, rất dễ dàng để xử lý. Tôi đã chọn dịch vụ AWS SES của Amazon, tính phí theo số lượng email được gửi, với giá 0,1 đô la cho mỗi 1000 email. Với tần suất cập nhật blog và số lượng người đăng ký của tôi, chi phí hầu như có thể coi là không đáng kể.

Về điểm thứ nhất, tôi không muốn (thực tế là cũng không biết) viết mã nguồn phía trước, càng không muốn phát triển phía máy chủ hay quản lý cơ sở dữ liệu. Vì vậy, tôi đã cân nhắc việc sử dụng một cơ sở dữ liệu bên thứ ba hỗ trợ API. Danh sách email của độc giả là dữ liệu khá quan trọng, nên cần một cơ sở dữ liệu đáng tin cậy và dễ quản lý. Notion Database và Google Sheet là những lựa chọn tốt.

Cuối cùng, tôi đã chọn Notion, kết hợp với NotionForms. NotionForms cung cấp trang/giao diện biểu mẫu để thu thập dữ liệu và lưu trữ vào cơ sở dữ liệu của Notion. Phiên bản miễn phí không giới hạn số lượng dữ liệu được thu thập, hoàn toàn phù hợp với nhu cầu của tôi.

OK, giờ đến thời gian quảng cáo rồi! Mời bạn điền biểu mẫu dưới đây để đăng ký nhận bản tin. Hiện tại, nội dung chủ yếu là bản tin tuần, được cập nhật vào mỗi sáng thứ Hai.

(Toàn bộ quy trình tôi đã kiểm tra kỹ lưỡng, nhưng đây là lần đầu tiên gửi vào thứ Hai tới. Nếu nhận được nội dung kỳ lạ, xin vui lòng thông cảm.)

Một nhược điểm duy nhất của phiên bản miễn phí NotionForms là không hỗ trợ cập nhật dữ liệu, nghĩa là không thể trực tiếp hỗ trợ hủy đăng ký. Giải pháp tạm thời là tạo một biểu mẫu riêng biệt và cơ sở dữ liệu Notion tương ứng cho yêu cầu hủy đăng ký. Sau khi nhận yêu cầu, tôi sẽ xóa thủ công danh sách người đăng ký. Nếu có quá nhiều yêu cầu hủy, tôi có thể viết script dựa trên API Notion để xử lý, sẽ cân nhắc thực hiện sau này.

Thực hiện kỹ thuật

Dựa trên ý tưởng trên, chỉ cần script Python để hoàn thành tất cả công việc, gồm ba bước:

1. Đọc danh sách email từ cơ sở dữ liệu Notion.
2. Thu thập dữ liệu RSS để lấy nội dung cập nhật của blog.
3. Gọi API AWS SES để gửi email.

Cuối cùng, cấu hình crontab để chạy nhiệm vụ định kỳ mỗi ngày.

Chuẩn bị

Tất nhiên, cần một số bước chuẩn bị, bao gồm:

1. Chuẩn bị cơ sở dữ liệu
2. Chuẩn bị biểu mẫu
3. Chuẩn bị Token Notion và quyền truy cập
   • Hệ thống sẽ tạo ra một Token, hãy lưu giữ kỹ lưỡng vì sẽ cần dùng trong script Python.
   • Trở lại cơ sở dữ liệu “Danh sách đăng ký” trong Notion, nhấn vào menu Database, chọn “Add connections” và thêm quyền cho “NewsletterAPP” vừa tạo.
4. Chuẩn bị AWS SES và quyền truy cập
   • Tài khoản mới tạo sẽ ở chế độ sandbox, không thể gửi email ra ngoài. Cần ty le chau a gửi yêu cầu hỗ trợ để thoát khỏi sandbox. Trong yêu cầu cần nêu rõ mục đích sử dụng, số lượng email dự kiến, cách xử lý email bị trả về, v.v.
   • Tạo IAM role cho tài khoản, sau đó lấy aws_access_key_id và aws_secret_access_key, lưu vào file ~/.aws/credentials.
5. Thư viện Python cần thiết
   • requests: Không cần giải thích thêm.
   • feedparser: Thư viện phân tích RSS, rất dễ sử dụng.
   • boto3: Thư viện chính thức của AWS SES, giúp gửi email một cách thuận tiện.

Mã nguồn đầy đủ

1. Chạy mã nguồn, cần đặt thông tin xác thực AWS SES vào file ~/.aws/credentials. Các thông tin khác có thể bổ sung trực tiếp trong mã nguồn.
2. Mã nguồn đầy đủ:

#!/usr/local/bin/python3
# -*- coding: UTF-8 -*-
import boto3,feedparser,requests,re,datetime
from botocore.exceptions import ClientError
from time import mktime

# ++++++++++
# Chuẩn bị các tham số
# ++++++++++
# Tham số Notion
EMAIL_DATABASE_ID = '' # ID database danh sách đăng ký, nằm trong URL của database
NOTION_API_TOKEN = '' # Xem trong trang integration của Notion
EMAIL_DATABASE_URL = ''
HEADERS = {
  'accept': 'application/json',
  'Authorization': 'Bearer {token}'.format(token=NOTION_API_TOKEN),
  'Notion-Version': '2022-06-28',
  'content-type': 'application/json'
}
PAGE_SIZE = 100

# Tham số AWS SES
REGION_NAME = 'ap-northeast-2' # Khu vực AWS SES, xem trong tài khoản AWS
SOURCE = 'Tên <địa chỉ_email>' # Địa chỉ gửi email đã được xác thực trong AWS SES
client = boto3.client('ses',region_name=REGION_NAME)

# Email cá nhân và RSS
NOTI_MYSELF_EMAIL = 'địa_chỉ_email' # Mỗi lần gửi xong bản tin, sẽ gửi thông báo đến email này
BLOG_URL = 'URL_blog' # Sử dụng trong phần cuối của email
RSS_URL = 'URL_RSS' # Địa chỉ RSS cần chuyển đổi
AUTHOR_URL = 'URL_tác_giả' # Liên kết tác giả trong email
UNSCRIBE_URL = '' # Đường dẫn hủy đăng ký

# ++++++++++
# Bắt đầu triển khai chức năng
# ++++++++++
# Lấy danh sách email từ API Notion
def get_emails():
  payload = {"page_size": PAGE_SIZE}
  emails = []
  response = requests.post(EMAIL_DATABASE_URL, json=payload, headers=HEADERS)
  for result in response.json()['results']:
    if result['properties']['Email']['email']:
      emails.append(result['properties']['Email']['email'])
  next_cursor = response.json()['next_cursor']
  while next_cursor:
    payload = {"page_size": PAGE_SIZE, 'start_cursor': next_cursor}
    response = requests.post(EMAIL_DATABASE_URL, json=payload, headers=HEADERS)
    for result in response.json()['results']:
      if result['properties']['Email']['email']:
        emails.append(result['properties']['Email']['email'])
    next_cursor = response.json()['next_cursor']
  emails = list(set(filter_email(emails)))
  return emails

# Hàm lọc địa chỉ email không đúng định dạng
def filter_email(emails):
  pattern = r'(^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$)'
  result = []
  for e in emails:
    if re.match(pattern, e.strip()):
      result.append(e.strip())
  return result

# Lấy nội dung bài viết từ RSS và xử lý HTML email
def get_article():
  rss_weekly = feedparser.parse(RSS_URL)
  title = rss_weekly['entries'][0].title
  link = rss_weekly['entries'][0].link
  content = rss_weekly['entries'][0].content[0].value.replace('<a href="', '<a style="color:#3354AA" href="')
  published = datetime.datetime.fromtimestamp(mktime(rss_weekly['entries'][0].published_parsed)) + datetime.timedelta(hours=8)
  
  content_html = '''
  <!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.0 Transitional//EN" "
  <html xmlns="
  <head>
    <meta http-equiv="Content-Type" content="text/html; charset=UTF-8" />
    <title>
      {title}
    </title>
    <meta name="viewport" content="width=device-width, initial-scale=1.0"/>
  </head>
  <body style="margin: 0; padding: 0; font-size: 1.2em; color:#111; text-decoration:none; ">
    <table cellpadding="0" cellspacing="0" width="100%">
      <tr>
        <td> 
          <table align="center" border="0" cellpadding="5" cellspacing="0" width="600" style="border-collapse: collapse; border-color:lightgray; ">
            <tr>
              <td>
                <h1>
                  <a style="color:black;text-decoration:none;" href="{link}">{title}</a>
                </h1>
                <p>
                  <i>by <a style="color:black" href="{author_url}">Tác giả</a></i>
                </p>
                <p>
                  {content}
                </p>
              </td>
            </tr>
            <tr>
              <td>-- HẾT --</td>
            </tr>
            <tr>
              <td>
                <p>
                  Liên kết:<a style="color:black;" href="{blog_url}">Xem các bản tin trước</a> | 
                  <a style="color:black;" href="{unsubscribe_url}">Hủy đăng ký</a>
                </p>
              </td>
            </tr>  
          </table>
        </td>
      </tr>
    </table>
  </body>
  </html>
  '''.format(
    title=title, 
    link=link, 
    content=content,
    blog_url = BLOG_URL,
    unsubscribe_url = UNSCRIBE_URL,
    author_url = AUTHOR_URL
    )  
  
  return {
    'title': title,
    'link': link,
    'published': published,
    'content_html': content_html
  }

# Ghi log
def write_log(text):
  with open('email_send_log.txt', 'a') as f:
    f.write(text)

# Gửi email
def send_email(email_type, to_email, title, body):
  try:
    response = client.send_email(
      Destination={
        'ToAddresses': [
          to_email,
        ],
      },
      Message={
        'Body': {
          'Html': {
            'Charset': "UTF-8",
            'Data': body
          },
        },
        'Subject': {
          'Charset': "UTF-8",
          'Data': title,
        },
      },
      Source=SOURCE,
    )
  except ClientError as e:
    log = '{dt} {email_type} to {email} error info: {error}\n'.format(
      dt=datetime.datetime.now(),
      error=e.response['Error']['Message'],
      email = to_email,
      email_type = email_type
      )
    write_log(log)
    return False
  else:
    log = "{dt} {email_type} to {email} success message_id: {messageid}\n".format(
      dt = datetime.datetime.now(), 
      messageid=response['MessageId'], 
      email = to_email,
      email_type = email_type
      )
    write_log(log)
    return True

# Gửi hàng loạt bản tin
def send_newsletter():
  success_cnt = 0
  failure_cnt = 0
  emails = get_emails()
  article = get_article()
  have_new_post = 0
  if article['published'].strftime('%Y-%m-%d') == datetime.datetime.now().strftime('%Y-%m-%d'):
    have_new_post = 1
    for email_addr in emails:
      status = send_email('send_newsletter_email' ,email_addr, article['title'], article['content_html'])
      if status == True:
        success_cnt = success_cnt + 1
      else:
        failure_cnt = failure_cnt + 1
  return {
    'success_cnt': success_cnt,
    'failure_cnt': failure_cnt,
    'have_new_post': have_new_post
  }
  
if __name__ == '__main__':
  try:
    result = send_newsletter()
    write_log('{dt} run_script success: success_cnt={success_cnt}, failure_cnt={failure_cnt}, new_post={have_new_post}\n'.format(
      dt=datetime.datetime.now(),
      success_cnt=result['success_cnt'],
      failure_cnt=result['failure_cnt'],
      have_new_post=result['have_new_post']
    ))
    send_email('send_noti_myself' ,NOTI_MYSELF_EMAIL, 
          'Thông báo chạy script bản tin',
          '<html>Gửi thành công: {success_cnt}, Gửi thất bại: {failure_cnt}, Số bài viết mới: {have_new_post}</html>'.format(
            success_cnt=result['success_cnt'],
            failure_cnt=result['failure_cnt'],
            have_new_post=result['have_new_post']
    ))
  except Exception as e:
    write_log('{dt} run_script error: {e}\n'.format(dt=datetime.datetime.now(),e=e))
    send_email('send_noti_myself' ,NOTI_MYSELF_EMAIL, 
          'Thông báo chạy script bản tin',
          '<html>Script gặp lỗi: {e}</html>'.format(e=e))

3. Cấu hình nhiệm vụ định kỳ Giả sử tôi lưu mã nguồn ở /sky/job/newsletter/newsletter.py, sau đó thiết lập crontab để chạy vào lúc 8 giờ sáng mỗi thứ Hai:

0 8 * * 1 /usr/bin/python3 /sky/job/newsletter/newsletter.py

Sửa đổi lần cuối vào 2025-03-30