201 lines
7.1 KiB
Python
201 lines
7.1 KiB
Python
#!/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()) |