430 lines
17 KiB
Python
430 lines
17 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
|
||
import os
|
||
import shutil
|
||
import time
|
||
import email
|
||
from email.utils import parseaddr
|
||
from typing import List, Dict, Optional, Tuple
|
||
from PIL import Image
|
||
|
||
from config import Config
|
||
from logger import logger
|
||
from temp_manager import TempManager
|
||
from mail_handler import MailHandler
|
||
from manifest_parser import ManifestParser
|
||
from image_processor import ImageProcessor
|
||
from zip_handler import ZipHandler
|
||
from agreement import AgreementManager
|
||
|
||
|
||
class MailConverter:
|
||
def __init__(self):
|
||
self.temp_mgr = TempManager()
|
||
self.mail = MailHandler()
|
||
self.img_proc = ImageProcessor()
|
||
self.zip_proc = ZipHandler()
|
||
self.parser = ManifestParser()
|
||
|
||
# ==================== 规则匹配核心 ====================
|
||
def match_rules_for_file(
|
||
self,
|
||
filename: str,
|
||
file_ext: str,
|
||
tasks: List[Dict],
|
||
current_archive: Optional[str] = None
|
||
) -> List[Dict]:
|
||
base_name = os.path.splitext(filename)[0]
|
||
results = []
|
||
|
||
applicable = []
|
||
for task in tasks:
|
||
task_scope = task.get('scope')
|
||
task_inline = task.get('inline_archive')
|
||
effective_scope = task_inline if task_inline else task_scope
|
||
if current_archive:
|
||
if effective_scope is None or effective_scope == current_archive:
|
||
applicable.append(task)
|
||
else:
|
||
if effective_scope is None:
|
||
applicable.append(task)
|
||
|
||
logger.debug(f"文件 {filename} 在作用域 {current_archive} 下匹配到 {len(applicable)} 条规则")
|
||
|
||
# 1. 重命名
|
||
rename_tasks = [t for t in applicable if t.get('type') == 'rename' and t.get('src_name') == base_name]
|
||
if rename_tasks:
|
||
for task in rename_tasks:
|
||
dst_name = task['dst_name']
|
||
dst_ext = os.path.splitext(dst_name)[1][1:].lower()
|
||
size = task.get('size')
|
||
results.append({
|
||
'output_name': dst_name,
|
||
'format': dst_ext,
|
||
'size': size
|
||
})
|
||
return results
|
||
|
||
# 2. 精准文件名
|
||
name_tasks = [t for t in applicable if t.get('type') == 'by_name' and t.get('src_name') == base_name]
|
||
if name_tasks:
|
||
for task in name_tasks:
|
||
for fmt, size in task.get('targets', []):
|
||
output_name = f"{base_name}.{fmt}"
|
||
results.append({
|
||
'output_name': output_name,
|
||
'format': fmt,
|
||
'size': size
|
||
})
|
||
return results
|
||
|
||
# 3. 按格式批量
|
||
format_tasks = [t for t in applicable if t.get('type') == 'by_format' and t.get('src_format') == file_ext.lower()]
|
||
if format_tasks:
|
||
for task in format_tasks:
|
||
for fmt, size in task.get('targets', []):
|
||
output_name = f"{base_name}.{fmt}"
|
||
results.append({
|
||
'output_name': output_name,
|
||
'format': fmt,
|
||
'size': size
|
||
})
|
||
return results
|
||
|
||
# 4. 全局默认
|
||
global_tasks = [t for t in applicable if t.get('type') == 'global']
|
||
if global_tasks:
|
||
for task in global_tasks:
|
||
for fmt, size in task.get('targets', []):
|
||
output_name = f"{base_name}.{fmt}"
|
||
results.append({
|
||
'output_name': output_name,
|
||
'format': fmt,
|
||
'size': size
|
||
})
|
||
return results
|
||
|
||
# 无规则，原样保留
|
||
results.append({
|
||
'output_name': filename,
|
||
'format': file_ext,
|
||
'size': None
|
||
})
|
||
return results
|
||
|
||
# ==================== 单张图片处理 ====================
|
||
def process_single_image(
|
||
self,
|
||
input_path: str,
|
||
output_dir: str,
|
||
tasks: List[Dict],
|
||
current_archive: Optional[str] = None
|
||
) -> List[str]:
|
||
filename = os.path.basename(input_path)
|
||
file_ext = os.path.splitext(filename)[1][1:].lower()
|
||
output_files = []
|
||
|
||
output_tasks = self.match_rules_for_file(filename, file_ext, tasks, current_archive)
|
||
logger.info(f"图片 {filename} 将生成 {len(output_tasks)} 个输出任务")
|
||
|
||
for task in output_tasks:
|
||
out_name = task['output_name']
|
||
out_fmt = task['format']
|
||
out_size = task['size']
|
||
out_path = os.path.join(output_dir, out_name)
|
||
|
||
try:
|
||
if out_size:
|
||
with Image.open(input_path) as img:
|
||
if out_size[0] == 'ratio':
|
||
img = ImageProcessor.resize_by_ratio(img, out_size[1], out_size[2])
|
||
elif out_size[0] == 'pixel':
|
||
img = ImageProcessor.resize_by_pixel(img, out_size[1], out_size[2])
|
||
pillow_format = ImageProcessor.FORMAT_MAP.get(out_fmt.lower())
|
||
if not pillow_format:
|
||
logger.error(f"不支持的输出格式: {out_fmt}")
|
||
continue
|
||
if pillow_format == 'JPEG' and img.mode in ('RGBA', 'P'):
|
||
rgb_img = Image.new('RGB', img.size, (255, 255, 255))
|
||
rgb_img.paste(img, mask=img.split()[-1] if img.mode == 'RGBA' else None)
|
||
img = rgb_img
|
||
save_kwargs = {}
|
||
if pillow_format in ('JPEG', 'WEBP'):
|
||
save_kwargs['quality'] = Config.DEFAULT_QUALITY
|
||
img.save(out_path, format=pillow_format, **save_kwargs)
|
||
logger.info(f"转换+缩放成功: {input_path} -> {out_path}")
|
||
else:
|
||
success = ImageProcessor.convert_image(input_path, out_path, out_fmt, Config.DEFAULT_QUALITY)
|
||
if not success:
|
||
logger.error(f"转换失败: {input_path} -> {out_path}")
|
||
continue
|
||
output_files.append(out_path)
|
||
except Exception as e:
|
||
logger.error(f"处理图片异常 {input_path}: {e}", exc_info=True)
|
||
continue
|
||
|
||
return output_files
|
||
|
||
# ==================== 压缩包处理 ====================
|
||
def process_archive(
|
||
self,
|
||
archive_path: str,
|
||
output_base_dir: str,
|
||
tasks: List[Dict]
|
||
) -> List[str]:
|
||
archive_name = os.path.basename(archive_path)
|
||
extract_dir = os.path.join(output_base_dir, f"ext_{archive_name}")
|
||
os.makedirs(extract_dir, exist_ok=True)
|
||
|
||
ext = os.path.splitext(archive_path)[1].lower()
|
||
try:
|
||
if ext == '.zip':
|
||
ZipHandler.extract_zip(archive_path, extract_dir)
|
||
elif ext == '.7z':
|
||
ZipHandler.extract_7z(archive_path, extract_dir)
|
||
else:
|
||
logger.warning(f"不支持的压缩包格式: {archive_path}")
|
||
return []
|
||
except Exception as e:
|
||
logger.error(f"解压失败 {archive_path}: {e}")
|
||
return []
|
||
|
||
all_output_files = []
|
||
for root, _, files in os.walk(extract_dir):
|
||
for file in files:
|
||
file_path = os.path.join(root, file)
|
||
file_ext_lower = os.path.splitext(file)[1][1:].lower()
|
||
if file_ext_lower not in Config.SUPPORTED_INPUT_FORMATS:
|
||
continue
|
||
output_files = self.process_single_image(
|
||
file_path,
|
||
output_base_dir,
|
||
tasks,
|
||
current_archive=archive_name
|
||
)
|
||
all_output_files.extend(output_files)
|
||
|
||
if not Config.KEEP_TEMP_FILES:
|
||
shutil.rmtree(extract_dir, ignore_errors=True)
|
||
return all_output_files
|
||
|
||
# ==================== 邮件处理主流程 ====================
|
||
def process_one_email(self, msg_id, raw_email, proto_conn, proto_type) -> bool:
|
||
msg = email.message_from_bytes(raw_email)
|
||
sender = msg.get('From')
|
||
sender_email = parseaddr(sender)[1]
|
||
subject = msg.get('Subject', '无主题')
|
||
logger.info(f"开始处理邮件: {subject} from {sender_email}")
|
||
|
||
if not self.mail.is_domain_allowed(sender_email):
|
||
logger.warning(f"域名不在白名单: {sender_email}，删除原邮件")
|
||
self.mail.delete_email(msg_id, proto_conn, proto_type)
|
||
return False
|
||
|
||
mail_work_dir = os.path.join(Config.TEMP_DIR, f"mail_{msg_id}_{int(time.time())}")
|
||
os.makedirs(mail_work_dir, exist_ok=True)
|
||
|
||
try:
|
||
attachments, body, _ = self.mail.download_attachments(raw_email, mail_work_dir)
|
||
logger.info(f"下载到附件: {[os.path.basename(a) for a in attachments]}")
|
||
|
||
manifest_path = None
|
||
for att in attachments:
|
||
if os.path.basename(att) == 'manifest.txt':
|
||
manifest_path = att
|
||
break
|
||
if manifest_path:
|
||
with open(manifest_path, 'r', encoding='utf-8') as f:
|
||
rule_content = f.read()
|
||
logger.info("使用 manifest.txt 规则")
|
||
else:
|
||
rule_content = body
|
||
logger.info("使用邮件正文规则")
|
||
|
||
logger.info(f"规则原始内容:\n{rule_content}")
|
||
|
||
if not rule_content or not rule_content.strip():
|
||
logger.warning("规则内容为空，删除原邮件")
|
||
self.mail.delete_email(msg_id, proto_conn, proto_type)
|
||
return False
|
||
|
||
tasks = self.parser.parse(rule_content)
|
||
logger.info(f"解析出 {len(tasks)} 条规则指令")
|
||
for idx, t in enumerate(tasks):
|
||
logger.debug(f"规则{idx}: {t}")
|
||
|
||
all_output_files = []
|
||
original_info = []
|
||
|
||
for att in attachments:
|
||
if os.path.basename(att) == 'manifest.txt':
|
||
continue
|
||
att_basename = os.path.basename(att)
|
||
att_ext = os.path.splitext(att)[1][1:].lower()
|
||
|
||
if ZipHandler.is_archive(att):
|
||
archive_output_dir = os.path.join(mail_work_dir, f"out_{att_basename}")
|
||
os.makedirs(archive_output_dir, exist_ok=True)
|
||
output_files = self.process_archive(att, archive_output_dir, tasks)
|
||
all_output_files.extend(output_files)
|
||
original_info.append({
|
||
'original_name': att_basename,
|
||
'output_dir': archive_output_dir,
|
||
'output_files': output_files
|
||
})
|
||
elif att_ext in Config.SUPPORTED_INPUT_FORMATS:
|
||
single_output_dir = os.path.join(mail_work_dir, f"out_{att_basename}")
|
||
os.makedirs(single_output_dir, exist_ok=True)
|
||
output_files = self.process_single_image(att, single_output_dir, tasks, current_archive=None)
|
||
all_output_files.extend(output_files)
|
||
original_info.append({
|
||
'original_name': att_basename,
|
||
'output_dir': single_output_dir,
|
||
'output_files': output_files
|
||
})
|
||
else:
|
||
logger.warning(f"跳过不支持的文件: {att_basename} (扩展名: {att_ext})")
|
||
|
||
logger.info(f"总共生成 {len(all_output_files)} 个输出文件")
|
||
if not all_output_files:
|
||
logger.warning("没有生成任何转换结果，删除原邮件")
|
||
self.mail.delete_email(msg_id, proto_conn, proto_type)
|
||
return False
|
||
|
||
# 保持原附件样式打包
|
||
return_attachments = []
|
||
for info in original_info:
|
||
if len(info['output_files']) == 1:
|
||
return_attachments.append(info['output_files'][0])
|
||
else:
|
||
zip_name = f"{os.path.splitext(info['original_name'])[0]}_result.zip"
|
||
zip_path = os.path.join(mail_work_dir, zip_name)
|
||
ZipHandler.pack_to_zip(info['output_dir'], zip_path)
|
||
return_attachments.append(zip_path)
|
||
|
||
if Config.FLATTEN_OUTPUT:
|
||
return_attachments = all_output_files
|
||
|
||
# ========== 构建详细主题 ==========
|
||
conversion_details = []
|
||
for info in original_info:
|
||
orig_name = info['original_name']
|
||
out_files = info['output_files']
|
||
if len(out_files) == 1:
|
||
out_name = os.path.basename(out_files[0])
|
||
out_ext = os.path.splitext(out_name)[1][1:].upper()
|
||
conversion_details.append(f"{orig_name} → {out_ext}")
|
||
else:
|
||
conversion_details.append(f"{orig_name} → {len(out_files)}个文件")
|
||
|
||
if len(conversion_details) > 2:
|
||
detail_str = f"{', '.join(conversion_details[:2])} 等{len(conversion_details)}项"
|
||
else:
|
||
detail_str = ', '.join(conversion_details)
|
||
|
||
time_str = time.strftime("%Y-%m-%d %H:%M:%S")
|
||
result_subject = f"MailC[{time_str}]：{detail_str}"
|
||
|
||
# 发送结果
|
||
self.mail.send_result(
|
||
recipient=sender_email,
|
||
subject=result_subject,
|
||
attachments=return_attachments,
|
||
split_volume_mb=Config.SPLIT_VOLUME_SIZE_MB
|
||
)
|
||
logger.info(f"结果已发送至 {sender_email}")
|
||
|
||
self.mail.delete_email(msg_id, proto_conn, proto_type)
|
||
return True
|
||
|
||
except Exception as e:
|
||
error_msg = f"处理邮件时异常: {str(e)}"
|
||
logger.error(error_msg, exc_info=True)
|
||
try:
|
||
self.mail.send_error_report(sender_email, error_msg)
|
||
except Exception as send_err:
|
||
logger.error(f"发送错误报告失败: {send_err}")
|
||
self.mail.delete_email(msg_id, proto_conn, proto_type)
|
||
return False
|
||
finally:
|
||
if not Config.KEEP_TEMP_FILES:
|
||
shutil.rmtree(mail_work_dir, ignore_errors=True)
|
||
|
||
# ==================== 循环监听 ====================
|
||
def run_forever(self):
|
||
if not AgreementManager.is_agreed():
|
||
logger.info("首次使用，需要同意协议")
|
||
recipient, subject, body = AgreementManager.request_agreement(Config.ADMIN_EMAIL, self.mail)
|
||
self.mail.send_result(recipient, subject, [])
|
||
logger.info(f"协议请求已发送至 {recipient}，请在 .env 中设置 AGREED_TOS=True")
|
||
return
|
||
|
||
logger.info(f"启动监听循环，轮询间隔 {Config.POLL_INTERVAL_SECONDS} 秒，{'无限循环' if Config.RUN_FOREVER else '单次运行'}")
|
||
|
||
while True:
|
||
TempManager.cleanup_stale()
|
||
conn = None
|
||
proto_type = None
|
||
try:
|
||
conn, proto_type = self.mail.connect_inbox()
|
||
if proto_type == "IMAP":
|
||
typ, data = conn.search(None, 'UNSEEN')
|
||
if typ != 'OK':
|
||
logger.error("IMAP搜索失败")
|
||
continue
|
||
msg_ids = data[0].split()
|
||
else:
|
||
msg_count = len(conn.list()[1])
|
||
msg_ids = list(range(1, msg_count + 1))
|
||
|
||
if not msg_ids:
|
||
logger.debug("无新邮件")
|
||
else:
|
||
logger.info(f"发现 {len(msg_ids)} 封待处理邮件")
|
||
for msg_id in msg_ids[:Config.MAX_EMAILS_PER_RUN]:
|
||
try:
|
||
if proto_type == "IMAP":
|
||
typ, msg_data = conn.fetch(msg_id, '(RFC822)')
|
||
if typ != 'OK':
|
||
continue
|
||
raw_email = msg_data[0][1]
|
||
else:
|
||
raw_email = b'\n'.join(conn.retr(msg_id)[1])
|
||
self.process_one_email(msg_id, raw_email, conn, proto_type)
|
||
except Exception as e:
|
||
logger.error(f"处理邮件 {msg_id} 崩溃: {e}", exc_info=True)
|
||
try:
|
||
self.mail.delete_email(msg_id, conn, proto_type)
|
||
except:
|
||
pass
|
||
if proto_type == "IMAP":
|
||
try:
|
||
conn.expunge()
|
||
except:
|
||
pass
|
||
except Exception as e:
|
||
logger.error(f"主循环异常: {e}", exc_info=True)
|
||
finally:
|
||
if conn:
|
||
try:
|
||
if proto_type == "IMAP":
|
||
conn.close()
|
||
conn.logout()
|
||
else:
|
||
conn.quit()
|
||
except:
|
||
pass
|
||
if not Config.RUN_FOREVER:
|
||
break
|
||
time.sleep(Config.POLL_INTERVAL_SECONDS)
|
||
|
||
|
||
def main():
|
||
converter = MailConverter()
|
||
converter.run_forever()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|