Initial commit: Email Assistant project
This commit is contained in:
@@ -0,0 +1,201 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
migrate_to_email_md.py — конвертация существующего архива из meta.json + body.md в email.md.
|
||||
|
||||
Сканирует /mnt/yandex-disk/hermes/email/<folder>/YYYY/MM/UID/,
|
||||
где лежат meta.json + body.md, и объединяет их в один email.md
|
||||
с YAML-frontmatter. После успешной записи удаляет meta.json и body.md.
|
||||
|
||||
Запуск:
|
||||
python3 migrate_to_email_md.py # все папки
|
||||
python3 migrate_to_email_md.py --dry-run # только показать, что будет
|
||||
python3 migrate_to_email_md.py --folder INBOX # только одну папку
|
||||
"""
|
||||
|
||||
import json
|
||||
import sys
|
||||
import argparse
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
|
||||
ARCHIVE_ROOT = Path("/mnt/yandex-disk/hermes/email")
|
||||
|
||||
|
||||
def make_frontmatter(meta: dict, body: str) -> str:
|
||||
"""Собрать YAML frontmatter + body в один email.md."""
|
||||
# Извлекаем поля из meta.json
|
||||
msg_id = meta.get("id", "0")
|
||||
flags = meta.get("flags", [])
|
||||
subject = meta.get("subject", "")
|
||||
from_ = meta.get("from", {})
|
||||
to_ = meta.get("to", {})
|
||||
date = meta.get("date", "")
|
||||
has_attachment = meta.get("has_attachment", False)
|
||||
|
||||
# Форматируем from/to как "Name <addr>"
|
||||
def fmt_contact(c):
|
||||
if not c:
|
||||
return ""
|
||||
name = c.get("name") or ""
|
||||
addr = c.get("addr") or ""
|
||||
if name and addr:
|
||||
return f"{name} <{addr}>"
|
||||
return addr or name
|
||||
|
||||
from_str = fmt_contact(from_)
|
||||
to_str = fmt_contact(to_)
|
||||
|
||||
# Собираем YAML-поля вручную, чтобы сохранить порядок и читаемость
|
||||
lines = []
|
||||
lines.append("---")
|
||||
lines.append(f"id: {msg_id}")
|
||||
lines.append(f"folder: {meta.get('_folder', '')}")
|
||||
lines.append(f"subject: {json.dumps(subject, ensure_ascii=False)}")
|
||||
lines.append(f"from: {json.dumps(from_str, ensure_ascii=False)}")
|
||||
lines.append(f"to: {json.dumps(to_str, ensure_ascii=False)}")
|
||||
lines.append(f"date: {json.dumps(date, ensure_ascii=False)}")
|
||||
lines.append(f"flags: {json.dumps(flags)}")
|
||||
lines.append(f"has_attachment: {str(has_attachment).lower()}")
|
||||
# Дополнительные поля, если есть (Message-ID и т.д.)
|
||||
for extra_field in ["message_id", "in_reply_to", "references", "cc", "content_type"]:
|
||||
val = meta.get(extra_field)
|
||||
if val:
|
||||
lines.append(f"{extra_field}: {json.dumps(val, ensure_ascii=False)}")
|
||||
lines.append("---")
|
||||
lines.append("")
|
||||
|
||||
# Тело письма — уже в body.md
|
||||
lines.append(body.rstrip("\n"))
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def migrate_email(msg_dir: Path, dry_run: bool = False) -> bool:
|
||||
"""
|
||||
Конвертировать один email из meta.json + body.md в email.md.
|
||||
Возвращает True, если было что конвертировать.
|
||||
"""
|
||||
meta_path = msg_dir / "meta.json"
|
||||
body_path = msg_dir / "body.md"
|
||||
email_path = msg_dir / "email.md"
|
||||
|
||||
if not meta_path.exists() and not body_path.exists():
|
||||
return False
|
||||
|
||||
if email_path.exists() and not meta_path.exists() and not body_path.exists():
|
||||
# Уже сконвертировано
|
||||
return False
|
||||
|
||||
# Читаем meta.json (если есть, иначе пустой словарь)
|
||||
meta = {}
|
||||
if meta_path.exists():
|
||||
try:
|
||||
with meta_path.open("r", encoding="utf-8") as f:
|
||||
meta = json.load(f)
|
||||
except Exception as e:
|
||||
print(f" [WARN] {msg_dir}/meta.json: {e}", file=sys.stderr)
|
||||
|
||||
# Читаем body.md (если есть)
|
||||
body = ""
|
||||
if body_path.exists():
|
||||
try:
|
||||
body = body_path.read_text(encoding="utf-8")
|
||||
except Exception as e:
|
||||
print(f" [WARN] {msg_dir}/body.md: {e}", file=sys.stderr)
|
||||
|
||||
# Добавляем имя папки в meta
|
||||
# Извлекаем относительный путь от ARCHIVE_ROOT
|
||||
try:
|
||||
rel = msg_dir.relative_to(ARCHIVE_ROOT)
|
||||
folder = str(rel.parent.parent.parent) # folder/YYYY/MM/UID -> folder
|
||||
except ValueError:
|
||||
folder = "unknown"
|
||||
meta["_folder"] = folder
|
||||
|
||||
content = make_frontmatter(meta, body)
|
||||
|
||||
if dry_run:
|
||||
uid = meta.get("id", "?")
|
||||
subj = meta.get("subject", "(no subject)")[:50]
|
||||
print(f" [DRY] {folder}/{uid} ({subj}) → email.md")
|
||||
return True
|
||||
|
||||
# Пишем email.md
|
||||
email_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
email_path.write_text(content, encoding="utf-8")
|
||||
|
||||
# Удаляем старые файлы
|
||||
if meta_path.exists():
|
||||
meta_path.unlink()
|
||||
if body_path.exists():
|
||||
body_path.unlink()
|
||||
|
||||
uid = meta.get("id", "?")
|
||||
subj = meta.get("subject", "")[:50]
|
||||
print(f" ✓ {folder}/{uid} ({subj})")
|
||||
return True
|
||||
|
||||
|
||||
def scan_folders(root: Path) -> list[Path]:
|
||||
"""Найти все директории вида <folder>/YYYY/MM/UID/."""
|
||||
msg_dirs = []
|
||||
for folder_dir in root.iterdir():
|
||||
if not folder_dir.is_dir():
|
||||
continue
|
||||
for year_dir in folder_dir.iterdir():
|
||||
if not year_dir.is_dir() or not year_dir.name.isdigit():
|
||||
continue
|
||||
for month_dir in year_dir.iterdir():
|
||||
if not month_dir.is_dir() or not month_dir.name.isdigit():
|
||||
continue
|
||||
for uid_dir in month_dir.iterdir():
|
||||
if not uid_dir.is_dir():
|
||||
continue
|
||||
msg_dirs.append(uid_dir)
|
||||
return msg_dirs
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Миграция meta.json + body.md → email.md с YAML-frontmatter"
|
||||
)
|
||||
parser.add_argument("--dry-run", action="store_true",
|
||||
help="Показать, что будет сделано, без изменений")
|
||||
parser.add_argument("--folder", default=None,
|
||||
help="Конвертировать только одну папку (например, INBOX)")
|
||||
args = parser.parse_args()
|
||||
|
||||
if not ARCHIVE_ROOT.exists():
|
||||
print(f"Ошибка: {ARCHIVE_ROOT} не существует", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
msg_dirs = scan_folders(ARCHIVE_ROOT)
|
||||
|
||||
if args.folder:
|
||||
msg_dirs = [d for d in msg_dirs if str(d.relative_to(ARCHIVE_ROOT)).startswith(args.folder + "/")]
|
||||
|
||||
if not msg_dirs:
|
||||
print("Нет директорий с письмами для конвертации.")
|
||||
return 0
|
||||
|
||||
total = 0
|
||||
skipped = 0
|
||||
for msg_dir in sorted(msg_dirs):
|
||||
try:
|
||||
if migrate_email(msg_dir, dry_run=args.dry_run):
|
||||
total += 1
|
||||
else:
|
||||
skipped += 1
|
||||
except Exception as e:
|
||||
print(f" [ERROR] {msg_dir}: {e}", file=sys.stderr)
|
||||
|
||||
if args.dry_run:
|
||||
print(f"\nБудет сконвертировано: {total}, пропущено (уже email.md): {skipped}")
|
||||
else:
|
||||
print(f"\nСконвертировано: {total}, пропущено: {skipped}")
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user