ta-outlook-tasks-sync
readme
md
# ta-outlook-tasks-sync
## 概要
Outlook タスクを Markdown と同期して使う運用、のためのツール。
1. Outlook のタスクフォルダを Markdown にエクスポート（フォルダ名＝ファイル名）
2. Markdown 編集、チェック、追加など
3. Markdown からOutlookへの同期
4. DueDate ありタスクは必ずリマインダをつける
## コマンド名
- `ta-outlook-tasks-sync`
## コマンド引数からの入力
- `ACTION`
- 意味
- 操作内容
- 入力制限
- pull と push のみ可で必須
- 指定なしの場合は、上記以外の場合はヘルプを出す
- `--outlook-folder`
- 意味
- 対象フォルダ
- その他条件
- フォルダ名は、デフォルトの olFolderTasks 基準で指定
- 指定フォルダの直下のみで、子孫階層は辿らない
- 指定なし または空文字の場合
- olFolderTasks 直下
- `--tasks-dir`
- 意味
- Markdown 配置ディレクトリ名
- デフォルト値
- `.` (カレントディレクトリ)
- `--tasks-file`
- 意味
- Markdown ファイル名
- デフォルト値
- `<outlook-folder>.md`
- `--outlook-folder` が未指定または "" の場合は `Tasks.md` とする
- `--reminder-default-time`
- 時刻指定なし時の ReminderTime
- デフォルト値 `07:00`
## Outlook 項目
### 必須項目
- 参照のみ
- EntryID : 主キーとする
- 参照/更新
- Subject
- DueDate
- ReminderSet
- ReminderTime
- Body
- 更新のみ
- Complete
### あえて対象外とする項目と理由
- StartDate : 開始の概念がよく分からないから
- CreationTime : API側で自動更新されるため
- LastModificationTime : API側で自動更新されるため
- DateCompleted : API側で自動更新されるため
## Markdown 記載ルール
### Markdown フォーマット
#### Subject 行
- DueDate + ReminderTime あり
- `- [ ] ${Subject} (DueDate: ${yyyy/MM/dd HH:mm}) <!-- EntryID: ${EntryID} -->`
- DueDate あり
- `- [ ] ${Subject} (DueDate: ${yyyy/MM/dd}) <!-- EntryID: ${EntryID} -->`
- DueDate なし
- `- [ ] ${Subject} <!-- EntryID: ${EntryID} -->`
#### DueDate + ReminderTime 記載について
- パース時
- `DueDate` : `yyyy/MM/dd` 部分
- `CalcTime` : `DueDate` + `HH:mm` 部分。指定ない場合は `DueDate` + `reminder-default-time`
- `MM`, `dd`, `HH` は 1文字も許容する（`mm` は 1 文字は許容しない）
- 出力時
- `(DueDate: ${yyyy/MM/dd HH:mm})` で出力
- ただし、`reminder-default-time` と一致する場合は ` HH:mm` 部分を省略
- 共通
- Outlook COM 上では未設定日を None または番兵値として扱う
#### Body
- 共通
- Body については インデントされた fenced code block に記載する
- Markdown 出力時
- fenced code block 内は fence が存在する場合は、backtick を 1 つ増やした fence で囲う
- fence, コンテンツとも、空白 4 つのインデントをする
- trim して empty になる場合は fenced code block 自体を出力しない
- Markdown パース時
- Body 部分のインデントは、1 行目の空白文字数分の空白を、2 行目以降も削る
- 2 行目以降の行頭空白が 1 行目より短い場合も、空白部分のみ削る
````
- [ ] ${Subject} ...
```
Body 1 行目
Body 2 行目
...
```
````
### 記載箇所
`## Tasks` ブロックに記載。
`## Tasks` ブロックとは、`## Tasks\n` で始まり、
次の見出し (`#` 個数に限らず) かファイル末尾、近いほうまで、の範囲を指す。
## Task の取得 ( `pull` 時の動作 )
### 取得条件
- 指定フォルダ内のみ
- Complete = False のみ
### 書き出し
`## Tasks` ブロックに書き出す
- 存在しない場合は末尾に追記
- 存在する場合は ブロック置き換え
- `## Tasks` 行の前後と、ブロックの最後に空行を1つずつ入れる
## Task の更新 ( `push` 時の動作 )
### 基本動作
- `## Tasks` ブロックの `- [ ] `, `- [x] ` 行を読み出す
- `## Tasks` ブロックがない場合はエラーとして処理中断
- Conflict は考慮せず更新
### 更新処理
下記のように更新するが、更新前に更新対象を一覧し、処理続行するかを y/n で問い合わせる。
y/n についてはデフォルト入力は不可とし、y または n で答えることを求めるメッセージを出す。
- 削除 : Outlook 対象フォルダで未完了で、Markdown 側に無い EntryID のレコード
- 更新 : EntryID の一致するレコード
- Subject → Markdown 上で変更されていたら更新
- DueDate, ReminderSet, ReminderTime → Markdown 上で変更されていたら更新
- Complete → Markdown の `- [x] ` なら更新 (COM 的には MarkComplete() 実行)
- Body → Markdown の fenced code block があれば更新
- 追加 : EntryID の無いレコード
- Subject
- DueDate → 指定あれば設定
- 登録完了したら、Markdown 側に `<!-- EntryID: ${EntryID} -->` を書き戻す
DueDate + ReminderTime の更新の詳細
- DueDate が無い場合
- ReminderSet = False とする
- DueDate がある場合
- ReminderSet = True
- ReminderTime は Markdown に時刻指定があればその時刻、なければ --reminder-default-time を使う。
### 入力チェック
更新操作前に Markdown パースを実施し、エラーの場合は処理を中止。
- `## Tasks` ブロック自体が無い場合はエラー
- インデントされたタスクは入力エラー
- `## Tasks` ブロック内で、上記フォーマット以外での記載 (空白ブレを除く) は入力エラー
- `Subject` にキー項目が入ってきた場合もエラー
- つまり `DueDate`, `EntryID` の順番違いや、複数回登場した場合は、エラー
- ただし空行(空白のみの行を含む)は無視
- Outlook 側に存在しない `EntryID` が指定されていた場合はエラー
## 前提
- スクリプトは Python を使用する
- pywin32 が使用可能であること
- `pip install pywin32`
- Outlook がインストール済で、アカウントも設定済であること
- エラーメッセージやヘルプは 英語のみ
- 時刻は 実行環境のロケールに従う
## インストールする場合
初回のみ以下を実行。
```ps1
pip install -e .
```
pyproject.toml に従って wrapper の実行ファイルが作られて、以下のように呼び出し可能となる。
(editable install (-e) ではスクリプトを参照する wrapper が作られるため、都度ビルドしなおしは不要だが、所定ディレクトリに exe は配置される)
```ps1
ta-outlook-tasks-pull
```
インストールしないで使用するには、以下のように直接スクリプトを実行すればよい
```ps
python ta_outlook_tasks_sync.py pull
```
scripts
md
from __future__ import annotations
import argparse
import re
import sys
from dataclasses import dataclass
from datetime import date, datetime, time
from pathlib import Path
from typing import Iterable
OL_FOLDER_TASKS = 13
OL_TASK_ITEM = 3
OL_OBJECT_CLASS_TASK = 48
OUTLOOK_EMPTY_DATE_TEXT = "4501/01/01"
DEFAULT_REMINDER_TIME = "07:00"
class SyncError(Exception):
pass
@dataclass(frozen=True)
class MarkdownTask:
checked: bool
subject: str
entry_id: str | None
due_date: date | None
reminder_time: time | None
body: str | None = None
@dataclass(frozen=True)
class OutlookTaskSnapshot:
entry_id: str
subject: str
due_date: date | None
reminder_set: bool
reminder_time: datetime | None
complete: bool
item: object
body: str = ""
@dataclass(frozen=True)
class TaskBlock:
start_line: int
end_line: int
TASK_LINE_RE = re.compile(
r"""
^-\s\[(?P<mark>[ xX])\]\s+
(?P<body>.*?)
(?:\s+\(DueDate:\s+
(?P<date>\d{4}/\d{1,2}/\d{1,2})
(?:\s+(?P<hour>\d{1,2}):(?P<minute>\d{2}))?
\))?
(?:\s+<!--\s*EntryID:\s*(?P<entry_id>[^<>]+?)\s*-->)?
\s*$
""",
re.VERBOSE,
)
def parse_args(argv: list[str]) -> argparse.Namespace:
parser = argparse.ArgumentParser(
prog="ta-outlook-tasks-sync",
description="Sync Outlook tasks with a Markdown ## Tasks block.",
)
parser.add_argument("action", choices=["pull", "push"])
parser.add_argument("--outlook-folder", help="Direct child folder name under the default Tasks folder.")
parser.add_argument(
"--tasks-dir",
default=".",
help="Directory containing the Markdown tasks file. Default: current directory.",
)
parser.add_argument(
"--tasks-file",
help="Markdown tasks file name. Default: <outlook-folder>.md, or Tasks.md when --outlook-folder is omitted.",
)
parser.add_argument(
"--reminder-default-time",
default=DEFAULT_REMINDER_TIME,
help=f"Default ReminderTime when DueDate has no time. Default: {DEFAULT_REMINDER_TIME}",
)
return parser.parse_args(argv)
def parse_hhmm(value: str) -> time:
match = re.fullmatch(r"(\d{1,2}):(\d{2})", value)
if not match:
raise SyncError(f"Invalid time: {value}")
hour = int(match.group(1))
minute = int(match.group(2))
try:
return time(hour, minute)
except ValueError as exc:
raise SyncError(f"Invalid time: {value}") from exc
def parse_markdown_date(value: str) -> date:
parts = value.split("/")
if len(parts) != 3:
raise SyncError(f"Invalid DueDate: {value}")
year, month, day = (int(part) for part in parts)
try:
return date(year, month, day)
except ValueError as exc:
raise SyncError(f"Invalid DueDate: {value}") from exc
def find_tasks_block(lines: list[str]) -> TaskBlock | None:
start = None
for index, line in enumerate(lines):
if line.rstrip("\r\n") == "## Tasks":
start = index
break
if start is None:
return None
end = len(lines)
for index in range(start + 1, len(lines)):
if re.match(r"^#+\s+", lines[index]):
end = index
break
return TaskBlock(start, end)
def parse_task_line(line: str, line_number: int) -> MarkdownTask:
if line.startswith((" ", "\t")):
raise SyncError(f"Indented task is not allowed at line {line_number}.")
match = TASK_LINE_RE.fullmatch(line.rstrip("\r\n"))
if not match:
raise SyncError(f"Invalid task line at line {line_number}: {line.rstrip()}")
subject = match.group("body").strip()
if not subject:
raise SyncError(f"Subject is empty at line {line_number}.")
if "DueDate:" in subject or "<!--" in subject or "-->" in subject:
raise SyncError(f"Reserved key appears in subject at line {line_number}.")
due_date_value = match.group("date")
due_date = parse_markdown_date(due_date_value) if due_date_value else None
reminder_time = None
if match.group("hour") is not None:
reminder_time = parse_hhmm(f"{match.group('hour')}:{match.group('minute')}")
return MarkdownTask(
checked=match.group("mark").lower() == "x",
subject=subject,
entry_id=match.group("entry_id"),
due_date=due_date,
reminder_time=reminder_time,
)
def count_leading_whitespace(value: str) -> int:
return len(value) - len(value.lstrip(" \t"))
def remove_body_indentation(line: str, indent_width: int) -> str:
content = line.rstrip("\r\n")
leading_width = count_leading_whitespace(content)
return content[min(leading_width, indent_width) :]
def parse_body_block(lines: list[str], start_index: int, block_end: int) -> tuple[str | None, int]:
if start_index >= block_end or lines[start_index].strip() == "":
return None, start_index
opening = re.fullmatch(r"(?P<indent>[ \t]+)(?P<fence>`{3,})\s*", lines[start_index].rstrip("\r\n"))
if not opening:
if lines[start_index].startswith((" ", "\t")):
raise SyncError(f"Invalid body block at line {start_index + 1}: {lines[start_index].rstrip()}")
return None, start_index
indent = opening.group("indent")
fence = opening.group("fence")
content_lines: list[str] = []
index = start_index + 1
while index < block_end:
line = lines[index]
closing = re.fullmatch(
rf"{re.escape(indent)}`{{{len(fence)},}}\s*",
line.rstrip("\r\n"),
)
if closing:
return "\n".join(content_lines), index + 1
content_lines.append(remove_body_indentation(line, len(indent)))
index += 1
raise SyncError(f"Unclosed body block at line {start_index + 1}.")
def parse_tasks_block(markdown: str) -> list[MarkdownTask]:
lines = markdown.splitlines(keepends=True)
block = find_tasks_block(lines)
if block is None:
raise SyncError("Missing ## Tasks block.")
tasks: list[MarkdownTask] = []
seen_entry_ids: set[str] = set()
index = block.start_line + 1
while index < block.end_line:
raw_line = lines[index]
if raw_line.strip() == "":
index += 1
continue
task = parse_task_line(raw_line, index + 1)
body, next_index = parse_body_block(lines, index + 1, block.end_line)
task = MarkdownTask(
checked=task.checked,
subject=task.subject,
entry_id=task.entry_id,
due_date=task.due_date,
reminder_time=task.reminder_time,
body=body,
)
if task.entry_id:
if task.entry_id in seen_entry_ids:
raise SyncError(f"Duplicate EntryID at line {index + 1}: {task.entry_id}")
seen_entry_ids.add(task.entry_id)
tasks.append(task)
index = next_index
return tasks
def replace_tasks_block(markdown: str, task_lines: list[str]) -> str:
lines = markdown.splitlines(keepends=True)
block = find_tasks_block(lines)
replacement = ["\n", "## Tasks\n", "\n", *[line.rstrip("\r\n") + "\n" for line in task_lines], "\n"]
if block is None:
if lines and not lines[-1].endswith(("\n", "\r")):
lines[-1] += "\n"
while lines and lines[-1].strip() == "":
lines.pop()
lines.extend(replacement)
return "".join(lines)
before = lines[: block.start_line]
while before and before[-1].strip() == "":
before.pop()
after = lines[block.end_line :]
while after and after[0].strip() == "":
after.pop(0)
return "".join([*before, *replacement, *after])
def is_empty_outlook_date(value: object) -> bool:
if value is None:
return True
return getattr(value, "year", None) == 4501
def as_date(value: object) -> date | None:
if is_empty_outlook_date(value):
return None
if isinstance(value, datetime):
return value.date()
if isinstance(value, date):
return value
return None
def as_datetime(value: object) -> datetime | None:
if is_empty_outlook_date(value):
return None
if isinstance(value, datetime):
return value
return None
def format_date(value: date) -> str:
return f"{value.year:04d}/{value.month:02d}/{value.day:02d}"
def format_time(value: time) -> str:
return f"{value.hour:02d}:{value.minute:02d}"
def format_outlook_date(value: date) -> str:
return format_date(value)
def format_outlook_datetime(value: datetime) -> str:
return f"{format_date(value.date())} {format_time(value.time())}"
def get_outlook_app() -> object:
try:
import win32com.client
except ImportError as exc:
raise SyncError("pywin32 is required. Install it with: pip install pywin32") from exc
return win32com.client.Dispatch("Outlook.Application")
def get_tasks_folder(outlook: object, outlook_folder_name: str | None) -> object:
namespace = outlook.GetNamespace("MAPI")
root = namespace.GetDefaultFolder(OL_FOLDER_TASKS)
if not outlook_folder_name:
return root
available_names: list[str] = []
for child in root.Folders:
child_name = child.Name
available_names.append(child_name)
if child_name == outlook_folder_name or strip_folder_suffix(child_name) == outlook_folder_name:
return child
available = ", ".join(available_names) if available_names else "(none)"
raise SyncError(f"Task folder not found: {outlook_folder_name}. Available folders: {available}")
def strip_folder_suffix(folder_name: str) -> str:
return re.sub(r"\s+\([^()]+\)$", "", folder_name)
def snapshot_task(item: object) -> OutlookTaskSnapshot:
reminder_set = bool(getattr(item, "ReminderSet", False))
return OutlookTaskSnapshot(
entry_id=item.EntryID,
subject=item.Subject or "",
body=item.Body or "",
due_date=as_date(getattr(item, "DueDate", None)),
reminder_set=reminder_set,
reminder_time=as_datetime(getattr(item, "ReminderTime", None)) if reminder_set else None,
complete=bool(getattr(item, "Complete", False)),
item=item,
)
def collect_snapshots(folder: object) -> list[OutlookTaskSnapshot]:
snapshots: list[OutlookTaskSnapshot] = []
for item in folder.Items:
if getattr(item, "Class", OL_OBJECT_CLASS_TASK) != OL_OBJECT_CLASS_TASK:
continue
snapshots.append(snapshot_task(item))
return snapshots
def default_tasks_file_name(outlook_folder_name: str | None) -> str:
if outlook_folder_name:
return f"{outlook_folder_name}.md"
return "Tasks.md"
def resolve_tasks_path(tasks_dir: str, tasks_file: str | None, outlook_folder_name: str | None) -> Path:
file_name = tasks_file or default_tasks_file_name(outlook_folder_name)
return Path(tasks_dir) / file_name
def render_task(snapshot: OutlookTaskSnapshot, default_reminder_time: time) -> str:
checked = "x" if snapshot.complete else " "
due_part = ""
if snapshot.due_date is not None:
reminder_time = snapshot.reminder_time.time() if snapshot.reminder_time else default_reminder_time
if reminder_time == default_reminder_time:
due_part = f" (DueDate: {format_date(snapshot.due_date)})"
else:
due_part = f" (DueDate: {format_date(snapshot.due_date)} {format_time(reminder_time)})"
line = f"- [{checked}] {snapshot.subject}{due_part} <!-- EntryID: {snapshot.entry_id} -->"
rendered_body = render_body(snapshot.body)
if rendered_body:
return f"{line}\n{rendered_body}"
return line
def body_fence_for(body: str) -> str:
longest = max((len(match.group(0)) for match in re.finditer(r"`+", body)), default=0)
return "`" * max(3, longest + 1)
def render_body(body: str | None) -> str:
if body is None or body.strip() == "":
return ""
fence = body_fence_for(body)
body_lines = body.split("\n")
indented_lines = [f" {line}" for line in body_lines]
return "\n".join([f" {fence}", *indented_lines, f" {fence}"])
def read_text(path: Path) -> str:
if not path.exists():
return ""
return path.read_text(encoding="utf-8")
def read_existing_text(path: Path) -> str:
if not path.exists():
raise SyncError(f"Tasks file not found: {path}")
return path.read_text(encoding="utf-8")
def write_text(path: Path, text: str) -> None:
path.write_text(text, encoding="utf-8", newline="\n")
def sync_pull(folder: object, tasks_path: Path, default_reminder_time: time) -> None:
print(f"Outlook folder: {getattr(folder, 'FolderPath', '(unknown)')}")
snapshots = [snapshot for snapshot in collect_snapshots(folder) if not snapshot.complete]
task_lines = [render_task(snapshot, default_reminder_time) for snapshot in snapshots]
markdown = read_text(tasks_path)
write_text(tasks_path, replace_tasks_block(markdown, task_lines))
print(f"Wrote {len(task_lines)} task(s) to {tasks_path}")
def validate_entry_ids(tasks: list[MarkdownTask], snapshots_by_id: dict[str, OutlookTaskSnapshot]) -> None:
for task in tasks:
if task.entry_id and task.entry_id not in snapshots_by_id:
raise SyncError(f"EntryID does not exist in the target folder: {task.entry_id}")
def describe_changes(
markdown_tasks: list[MarkdownTask],
snapshots: list[OutlookTaskSnapshot],
default_reminder_time: time,
) -> tuple[list[MarkdownTask], list[MarkdownTask], list[OutlookTaskSnapshot], list[str]]:
snapshots_by_id = {snapshot.entry_id: snapshot for snapshot in snapshots}
validate_entry_ids(markdown_tasks, snapshots_by_id)
markdown_ids = {task.entry_id for task in markdown_tasks if task.entry_id}
incomplete_snapshots = [snapshot for snapshot in snapshots if not snapshot.complete]
deletes = [snapshot for snapshot in incomplete_snapshots if snapshot.entry_id not in markdown_ids]
adds = [task for task in markdown_tasks if task.entry_id is None]
updates = [
task
for task in markdown_tasks
if task.entry_id is not None
and changed_fields(task, snapshots_by_id[task.entry_id], default_reminder_time)
]
descriptions: list[str] = []
for snapshot in deletes:
descriptions.append(f"[Delete] {', '.join(delete_field_descriptions(snapshot, default_reminder_time))}")
for task in updates:
snapshot = snapshots_by_id[task.entry_id or ""]
if changed_fields(task, snapshot, default_reminder_time):
changes = changed_field_descriptions(task, snapshot, default_reminder_time)
descriptions.append(f"[Update] {', '.join(changes)}")
for task in adds:
descriptions.append(f"[Add] {', '.join(add_field_descriptions(task, default_reminder_time))}")
return updates, adds, deletes, descriptions
def normalized_reminder_datetime(task: MarkdownTask, default_reminder_time: time) -> datetime | None:
if task.due_date is None:
return None
return datetime.combine(task.due_date, task.reminder_time or default_reminder_time)
def changed_fields(task: MarkdownTask, snapshot: OutlookTaskSnapshot, default_reminder_time: time) -> list[str]:
fields: list[str] = []
if task.subject != snapshot.subject:
fields.append("Subject")
if task.body is not None and task.body != snapshot.body:
fields.append("Body")
if task.due_date != snapshot.due_date:
fields.append("DueDate")
expected_reminder_set = task.due_date is not None
if expected_reminder_set != snapshot.reminder_set:
fields.append("ReminderSet")
expected_reminder_time = normalized_reminder_datetime(task, default_reminder_time)
current_reminder_time = snapshot.reminder_time if snapshot.reminder_set else None
if expected_reminder_time != current_reminder_time:
fields.append("ReminderTime")
if task.checked and not snapshot.complete:
fields.append("Complete")
return fields
def format_optional_date(value: date | None) -> str:
return format_date(value) if value is not None else "None"
def format_optional_datetime(value: datetime | None, default_reminder_time: time) -> str:
if value is None:
return "None"
if value.time() in (time.min, default_reminder_time):
return format_date(value.date())
return f"{format_date(value.date())} {format_time(value.time())}"
def effective_due_datetime(snapshot: OutlookTaskSnapshot) -> datetime | None:
if snapshot.due_date is None:
return None
if snapshot.reminder_set and snapshot.reminder_time is not None:
return snapshot.reminder_time
return datetime.combine(snapshot.due_date, time.min)
def changed_field_descriptions(
task: MarkdownTask,
snapshot: OutlookTaskSnapshot,
default_reminder_time: time,
) -> list[str]:
changes: list[str] = []
if task.subject != snapshot.subject:
changes.append(f"Subject: {snapshot.subject} => {task.subject}")
else:
changes.append(f"Subject: {task.subject}")
if task.body is not None and task.body != snapshot.body:
changes.append("Body: changed")
due_date_changed = task.due_date != snapshot.due_date
if due_date_changed:
before = format_optional_date(snapshot.due_date)
after_datetime = normalized_reminder_datetime(task, default_reminder_time)
after = format_optional_datetime(after_datetime, default_reminder_time)
changes.append(f"DueDate: {before} => {after}")
else:
changes.append(f"DueDate: {format_optional_date(task.due_date)}")
expected_reminder_set = task.due_date is not None
if not due_date_changed and expected_reminder_set != snapshot.reminder_set:
changes.append(f"ReminderSet: {str(snapshot.reminder_set).lower()} => {str(expected_reminder_set).lower()}")
expected_reminder_time = normalized_reminder_datetime(task, default_reminder_time)
current_reminder_time = snapshot.reminder_time if snapshot.reminder_set else None
if not due_date_changed and expected_reminder_time != current_reminder_time:
before = format_optional_datetime(current_reminder_time, default_reminder_time)
after = format_optional_datetime(expected_reminder_time, default_reminder_time)
changes.append(f"ReminderTime: {before} => {after}")
if task.checked and not snapshot.complete:
changes.append("Complete: false => true")
elif snapshot.complete:
changes.append(f"Complete: {str(snapshot.complete).lower()}")
return changes
def add_field_descriptions(task: MarkdownTask, default_reminder_time: time) -> list[str]:
due_datetime = normalized_reminder_datetime(task, default_reminder_time)
fields = [
f"Subject: {task.subject}",
f"DueDate: {format_optional_datetime(due_datetime, default_reminder_time)}",
]
if task.checked:
fields.append("Complete: true")
if task.body is not None and task.body.strip():
fields.append("Body: set")
return fields
def delete_field_descriptions(snapshot: OutlookTaskSnapshot, default_reminder_time: time) -> list[str]:
return [
f"Subject: {snapshot.subject}",
f"DueDate: {format_optional_datetime(effective_due_datetime(snapshot), default_reminder_time)}",
]
def confirm_changes(descriptions: list[str]) -> bool:
if not descriptions:
print("No changes.")
return False
print("Pending changes:")
for description in descriptions:
print(f" - {description}")
while True:
answer = input("Continue? [y/n]: ").strip().lower()
if answer == "y":
return True
if answer == "n":
return False
print("Please enter y or n.")
def apply_due_and_reminder(item: object, task: MarkdownTask, default_reminder_time: time) -> None:
if task.due_date is None:
item.DueDate = OUTLOOK_EMPTY_DATE_TEXT
item.ReminderSet = False
return
item.DueDate = format_outlook_date(task.due_date)
item.ReminderSet = True
reminder_time = normalized_reminder_datetime(task, default_reminder_time)
if reminder_time is not None:
item.ReminderTime = format_outlook_datetime(reminder_time)
def apply_update(item: object, task: MarkdownTask, default_reminder_time: time) -> None:
item.Subject = task.subject
if task.body is not None:
item.Body = task.body
apply_due_and_reminder(item, task, default_reminder_time)
if task.checked and not bool(getattr(item, "Complete", False)):
item.MarkComplete()
item.Save()
def create_task(outlook: object, task: MarkdownTask, folder: object, default_reminder_time: time) -> object:
item = outlook.CreateItem(OL_TASK_ITEM)
item.Subject = task.subject
if task.body is not None:
item.Body = task.body
apply_due_and_reminder(item, task, default_reminder_time)
item.Save()
parent = getattr(item, "Parent", None)
parent_path = getattr(parent, "FolderPath", None)
folder_path = getattr(folder, "FolderPath", None)
if parent_path != folder_path:
item = item.Move(folder)
if task.checked:
item.MarkComplete()
item.Save()
return item
def update_markdown_entry_ids(markdown: str, original_tasks: list[MarkdownTask], created_items: Iterable[object]) -> str:
created_ids = iter([item.EntryID for item in created_items])
lines = markdown.splitlines(keepends=True)
block = find_tasks_block(lines)
if block is None:
raise SyncError("Missing ## Tasks block.")
add_indexes = [index for index, task in enumerate(original_tasks) if task.entry_id is None]
next_add = 0
task_index = 0
line_index = block.start_line + 1
while line_index < block.end_line:
if lines[line_index].strip() == "":
line_index += 1
continue
if next_add < len(add_indexes) and task_index == add_indexes[next_add]:
line = lines[line_index].rstrip("\r\n")
lines[line_index] = f"{line} <!-- EntryID: {next(created_ids)} -->\n"
next_add += 1
task_index += 1
_, line_index = parse_body_block(lines, line_index + 1, block.end_line)
return "".join(lines)
def sync_push(outlook: object, folder: object, tasks_path: Path, default_reminder_time: time) -> None:
print(f"Outlook folder: {getattr(folder, 'FolderPath', '(unknown)')}")
markdown = read_existing_text(tasks_path)
markdown_tasks = parse_tasks_block(markdown)
snapshots = collect_snapshots(folder)
snapshots_by_id = {snapshot.entry_id: snapshot for snapshot in snapshots}
updates, adds, deletes, descriptions = describe_changes(markdown_tasks, snapshots, default_reminder_time)
if not descriptions:
print("No changes.")
return
if not confirm_changes(descriptions):
print("Aborted.")
return
for snapshot in deletes:
snapshot.item.Delete()
for task in updates:
apply_update(snapshots_by_id[task.entry_id or ""].item, task, default_reminder_time)
created_items = [create_task(outlook, task, folder, default_reminder_time) for task in adds]
if created_items:
write_text(tasks_path, update_markdown_entry_ids(markdown, markdown_tasks, created_items))
print("Done.")
def main(argv: list[str] | None = None) -> int:
args = parse_args(sys.argv[1:] if argv is None else argv)
try:
default_reminder_time = parse_hhmm(args.reminder_default_time)
tasks_path = resolve_tasks_path(args.tasks_dir, args.tasks_file, args.outlook_folder)
outlook = get_outlook_app()
folder = get_tasks_folder(outlook, args.outlook_folder)
if args.action == "pull":
sync_pull(folder, tasks_path, default_reminder_time)
else:
sync_push(outlook, folder, tasks_path, default_reminder_time)
except SyncError as exc:
print(f"Error: {exc}", file=sys.stderr)
return 1
return 0
def main_pull() -> int:
return main(["pull", *sys.argv[1:]])
def main_push() -> int:
return main(["push", *sys.argv[1:]])
if __name__ == "__main__":
raise SystemExit(main())