from enum import Enum import time from datetime import datetime from typing import Optional from pydantic import BaseModel from fastapi import FastAPI, Query, UploadFile, File from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware import json import urllib.request import dataset,json import dataset from datetime import datetime, timedelta from email.mime.text import MIMEText from email.mime.image import MIMEImage from email.mime.multipart import MIMEMultipart from email.mime.base import MIMEBase from email.mime.application import MIMEApplication import os import boto3 from botocore.exceptions import ClientError app = FastAPI() origins = [ "http://139.162.121.30", "http://139.162.121.30:8002", ] app.add_middleware( CORSMiddleware, # allow_origins=origins, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # 寄送逾期執行表單Email通知 def hhh_send_mail(email, subject, html): SENDER = "Gorgeous Space " RECIPIENT = email AWS_REGION = "us-east-1" CHARSET = "UTF-8" client = boto3.client('ses',region_name=AWS_REGION) try: msg = MIMEMultipart() # msg["Subject"] = "逾期執行表單通知_" + str(datetime.date.today()) msg["Subject"] = subject msg["From"] = "noreply@hhh.com.tw" msg["To"] = email # Set message body body = MIMEText(html, "html") msg.attach(body) response = client.send_raw_email( Source=msg["From"], Destinations=[msg["To"]], RawMessage={"Data": msg.as_string()} ) print(response) # Display an error if something goes wrong. except ClientError as e: print(e.response['Error']['Message']) else: print("Email sent! Message ID:"), print(response['MessageId']) # Choozmo DB # db = dataset.connect("mysql://choozmo:pAssw0rd@db.ptt.cx:3306/hhh?charset=utf8mb4") # 測試機 DB db = dataset.connect('mysql://hhh7796hhh:lYmWsu^ujcA1@hhh-v57-cluster.cluster-cmab1ctkglka.ap-northeast-2.rds.amazonaws.com:3306/stage?charset=utf8mb4') # 正式機 DB # db = dataset.connect('mysql://hhh7796hhh:lYmWsu^ujcA1@hhh-v57-cluster.cluster-cmab1ctkglka.ap-northeast-2.rds.amazonaws.com:3306/xoops?charset=utf8mb4') # class Execute(BaseModel): # exf_id: str # num: str # 合約編號 # company: str #設計公司 # mobile: str # 手機 # telete: str # 電話 # designer: str #設計師 # sdate: str # 上架日期 # edate: str # 下架日期 # note: str # 備註說明 # contract_time: str # 到期日 # contract_person: str # 聯絡人 # sales_dept: str # 部門別 # sales_man: str # 接案業務 # is_close: str # 是否結案 (N: 未結案，Y: 已結案，T: 未上線，D: 續約，E: 不續約延期，F: 不續約需下線，G: 已到期未續約) # price: str @app.get("/is_close_changed_notice") async def is_close_changed_notice(exf_id: str = '', is_close: str = ''): subject = "" # 取得合約資料 q = "SELECT * \ FROM execute_form \ WHERE exf_id = " + exf_id q_result_count = len(list(db.query(q))) if q_result_count == 0: # 無合約資料 return for r in db.query(q): subject = '[合約狀態修改]：' + r['company'] + ' (合約編號：' + r['num'] + ')_' + datetime.now().strftime("%Y-%m-%d %H:%M") html = """ """ if is_close == 'D': html += '

本合約狀態已修改為續約

'; elif is_close == 'E': html += '

本合約狀態已修改為不續約延期

'; elif is_close == 'F': html += '

本合約狀態已修改為不續約需下線

'; elif is_close == 'G': html += '

本合約狀態已修改為已到期未續約

'; html += """ """ # Email清單 email_list = ['mike@choozmo.com'] # 寄送逾期執行表單Email通知 for email in email_list: hhh_send_mail(email, subject, html)