panleicim/appointment_tool
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'master' of git.lpaconsulting.fr:panleicim/appointment_tool

Browse Source
This commit is contained in:
Lei PAN
2025-12-02 09:50:15 +01:00
parent fafdbfc695 eed971779b
commit 1050bb3fc9
4 changed files with 102 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
requirements.txt
+3 -2
View File
@@ -5,7 +5,7 @@ pandas~=1.3.5
playwright==1.28.0
dataclasses~=0.6
SpeechRecognition==3.8.1
pymongo==4.1.1
pymongo==4.12.0
wget==3.2.0
numpy==1.26.4
pocketsphinx==0.1.15
@@ -19,4 +19,5 @@ Mako~=1.2.0
mailjet_rest~=1.3.4
Flask~=2.2.2
flask_httpauth~=4.7.0
APScheduler~=3.9.1
APScheduler~=3.9.1
imapclient
src/mail/mail_address_validator.py
+98 -39
View File
@@ -1,4 +1,11 @@
"""
邮件地址验证器模块
负责验证邮箱地址的有效性并管理无效邮箱的存储
"""
import logging
from concurrent.futures.thread import ThreadPoolExecutor
from typing import List
from imapclient import IMAPClient
@@ -6,66 +13,118 @@ from src.db.mongo_manager import MONGO_STORE_MANAGER
from src.mail.mail_constants import create_imap
from src.pojo.mail.mail_pojo import MailAddress
# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class MailAddressValidator():
def __init__(self, login, password):
class MailAddressValidator:
"""邮件地址验证器类"""
def __init__(self, login: str, password: str):
"""
初始化邮件地址验证器
Args:
login (str): 邮箱地址
password (str): 邮箱密码
"""
self.login = login
self.password = password
def is_valid_email_address(self) -> bool:
# authenticate
imap = create_imap(self.login)
isImapClient = isinstance(imap, IMAPClient)
isValid = True
"""
验证邮箱地址是否有效
Returns:
bool: 如果邮箱地址有效返回True，否则返回False
"""
# 认证
try:
if isImapClient:
imap = create_imap(self.login)
is_imap_client = isinstance(imap, IMAPClient)
if is_imap_client:
dat = imap.login(self.login, str(self.password))
print("dat is " + str(dat) + "for:" + self.login)
logger.info(f"成功登录邮箱: {self.login}, 响应: {dat}")
else:
type, dat = imap.login(self.login, str(self.password))
print("type is " + str(type) + "for:" + self.login)
type_, dat = imap.login(self.login, str(self.password))
logger.info(f"成功登录邮箱: {self.login}, 响应: {type_}, {dat}")
imap.logout()
return True
except Exception as error:
print(error)
isValid = False
return isValid
logger.error(f"验证邮箱地址失败 {self.login}: {error}")
return False
def check_and_save_to_db(self):
"""
检查邮箱地址有效性并保存到数据库
"""
if not self.is_valid_email_address():
MONGO_STORE_MANAGER.insert_invalid_mail(MailAddress(self.login, self.password))
print("{} is not valid".format(self.login))
logger.info(f"{self.login} 不是有效的邮箱地址")
else:
logger.info(f"{self.login} 是有效的邮箱地址")
def remove_invalid_email():
invalid = MONGO_STORE_MANAGER.get_invalid_emails()
for mail in invalid:
MONGO_STORE_MANAGER.remove_email_from_destination_email_list(mail)
"""
从数据库中移除无效邮箱
"""
try:
invalid = MONGO_STORE_MANAGER.get_invalid_emails()
for mail in invalid:
MONGO_STORE_MANAGER.remove_email_from_destination_email_list(mail)
logger.info(f"已从数据库中移除 {len(invalid)} 个无效邮箱")
except Exception as error:
logger.error(f"移除无效邮箱时出错: {error}")
def find_and_update_invalid_emails(mail_list):
# mail_address1 = MailAddress(mail="perrateke1983@onet.pl", password="8EQh#UuyMx8zVO9")
# mail_address4 = MailAddress(mail="ryan_meacham2856@yahoo.com", password="ulgggkodxqbvrpgm")
# mail_list = [mail_address1]
with ThreadPoolExecutor(max_workers=200) as executor:
for mail in mail_list:
valiator = MailAddressValidator(mail.mail, mail.password)
executor.submit(valiator.check_and_save_to_db)
def find_and_update_invalid_emails(mail_list: List[MailAddress]):
"""
并发检查邮箱地址有效性并更新数据库
Args:
mail_list (List[MailAddress]): 邮箱地址列表
"""
try:
with ThreadPoolExecutor(max_workers=200) as executor:
futures = []
for mail in mail_list:
validator = MailAddressValidator(mail.mail, mail.password)
future = executor.submit(validator.check_and_save_to_db)
futures.append(future)
# 等待所有任务完成
for future in futures:
future.result()
logger.info(f"已处理 {len(mail_list)} 个邮箱地址")
except Exception as error:
logger.error(f"处理邮箱列表时出错: {error}")
def check_mails():
mail_list = MONGO_STORE_MANAGER.get_destination_emails()
_mail_list_to_check = []
for _mail in mail_list:
if "outlook.com" not in _mail.mail:
_mail_list_to_check.append(_mail)
# excel_reader = ExcelHelper()
# mail_list = excel_reader.read_mails_and_pwd(file_name="/Users/lpan/Downloads/hotmail_list.xlsx")
# mail_address1 = MailAddress(mail="christinnyua@gmx.net", password="q1J1HHY2sXN")
# mail_list = [mail_address1]
find_and_update_invalid_emails(_mail_list_to_check)
"""
检查目标邮箱地址的有效性
"""
try:
mail_list = MONGO_STORE_MANAGER.get_destination_emails()
_mail_list_to_check = []
for _mail in mail_list:
if "outlook.com" not in _mail.mail:
_mail_list_to_check.append(_mail)
logger.info(f"准备检查 {_mail_list_to_check} 个邮箱地址")
find_and_update_invalid_emails(_mail_list_to_check)
except Exception as error:
logger.error(f"检查邮箱时出错: {error}")
if __name__ == '__main__':
# remove_invalid_email()
check_mails()
check_mails()
src/mail/mail_confirmation.py
-1
View File
@@ -224,7 +224,6 @@ def find_confirmation_contacts_mail_list(mail_list):
def read_mails_and_find_confirmation_contacts(all_mails=False):
# mail_list = MONGO_STORE_MANAGER.get_destination_emails()
if all_mails:
mail_list = MONGO_STORE_MANAGER.get_destination_emails()
else:
src/notification/mailer.py
+1 -1
View File
@@ -27,7 +27,7 @@ class Mailer:
def send_email(self, result: AcceptedResultPojo, to_all: bool = False):
if to_all:
recipients = ['panleicim@gmail.com', 'tangliang0411@gmail.com']
recipients = ['yulin_huang680730@gmx.com', 'tangliang0411@gmail.com']
else:
recipients = ['panleicim@gmail.com']
# result.phone = str(result.phone.replace(".0", "")