A python script to migrate obsidian notes to logseq, including tasks and daily notes, use at your own risk.

obsidian.py

from datetime import datetime
import re
import os
import shutil
# migrate obsidian note to logseq

# input: - [ ] Rename `dev-triage-china` to `cn-triage` 📅 2023-09-01
# output:
# - TODO aa
#   SCHEDULED: <2023-08-30 Wed>

def migrate_task(obsidian_task: str):
    t = parse_obsidian_task(obsidian_task)
    logseq_task = format_logseq(t['task'], scheduled_time=t.get('scheduled_date'), status=t['status'])
    return logseq_task

def format_logseq(task: str, scheduled_time: str | None = None, status: str | None = None):
    if status == None:
        status = 'TODO'

    status = status.upper()
    if status not in ['TODO', 'DOING', 'DONE']:
        raise ValueError(f'invalid status {status}')

    task_line = f"- {status} {task}"
    if scheduled_time == None:
        return task_line + "\n"

    scheduled_time = datetime.strptime(scheduled_time, '%Y-%m-%d').strftime('%Y-%m-%d %a')
    return task_line + f"\n  SCHEDULED: <{scheduled_time}>\n"

def parse_obsidian_task(s: str) -> dict:
    match = re.search(r"-\s*\[(?P<status>[\w\s\/])\]\s*(?P<task>[^:]*)(?:\s*:date:\s*(?P<scheduled_date>\d{4}-\d{2}-\d{2}))?(?:\s*:white_check_mark:\s*(?P<completed_date>\d{4}-\d{2}-\d{2}))?", s)

    if match:
        status = match.group('status')
        task = match.group('task')
        scheduled_date = match.group('scheduled_date')
        completed_date = match.group('completed_date')

        ret = {
            'task': task.rstrip(),
        }

        if status == 'x':
            ret['status'] = 'DONE'
        else:
            ret['status'] = 'TODO'

        if scheduled_date:
            ret['scheduled_date'] = scheduled_date
        if completed_date:
            ret['completed_date'] = completed_date

        return ret

def traverse_files(folder_path, callback=None):
    for root, _, files in os.walk(folder_path):
        for file in files:
            file_path = os.path.join(root, file)
            if callback is None:
                print(file_path)
            else:
                callback(file_path)

target_journal_folder = '/Users/vincenthuang/Notes/journals'
target_page_foler = '/Users/vincenthuang/Notes/pages'

# copy all the files with "Daily" in the path and end up with .md
def migrate_file(obsidian_file_path: str):
    if '/Daily/' in obsidian_file_path and obsidian_file_path.endswith('.md'):
        target_file_name = os.path.basename(obsidian_file_path).replace('-', '_')
        append_or_copy(target_journal_folder, obsidian_file_path, target_file_name)
        return
    elif obsidian_file_path.endswith('.md'):
        append_or_copy(target_page_foler, obsidian_file_path, os.path.basename(obsidian_file_path))
        return



def append_or_copy(folder_path: str, filepath: str, target_file_name: str):
    target_file = os.path.join(folder_path, target_file_name)

    if os.path.exists(target_file):
        mtime = os.path.getmtime(target_file)

        # if the file is modified today, skip it
        if datetime.fromtimestamp(mtime).date() == datetime.today().date():
            return

        with open(filepath, 'r') as obsidian_file:
            with open(target_file, 'a') as target_file:
                data = obsidian_file.read()
                target_file.write(data)
    else:
        shutil.copy(filepath, folder_path)

def clean_up(filepath: str):
    match = re.search(r'\d{4}_\d{2}_\d{2}.*md$', filepath)

    if match == None:
        # delete file
        os.remove(filepath)


if __name__ == '__main__':
    traverse_files('./program/note/Vincent Notes', callback=migrate_file)
    with open('Inbox.md', 'a') as file:
        with open('/Users/vincenthuang/Library/Mobile Documents/iCloud~md~obsidian/Documents/Vincent Notes/Inbox.md', 'r') as obsidian_file:
            for line in obsidian_file:
                logseq_task = migrate_task(line)
                file.write(logseq_task)
    traverse_files(target_journal_folder, callback=clean_up)



def test_format_logseq_basic():
    assert format_logseq('aa') == '- TODO aa\n'

def test_format_logseq_with_status():
    assert format_logseq('aa', status='DONE') == '- DONE aa\n'

def test_format_logseq_with_due_date():
    assert format_logseq('aa', status='DONE', scheduled_time='2023-08-30') == '- DONE aa\n  SCHEDULED: <2023-08-30 Wed>\n'

def test_parse_obsidian_scheduled_and_completed_task():
    s = "- [x] Airwallex webhook configuration :date: 2023-08-28 :white_check_mark: 2023-08-29"
    t = parse_obsidian_task(s)
    assert t['status'] == 'DONE'
    assert t['task'] == 'Airwallex webhook configuration'
    assert t['scheduled_date'] == '2023-08-28'
    assert t['completed_date'] == '2023-08-29'

def test_parse_obsidian_scheduled_task():
    s = "- [x] Airwallex webhook configuration :date: 2023-08-28"
    t = parse_obsidian_task(s)
    assert t['status'] == 'DONE'
    assert t['task'] == 'Airwallex webhook configuration'
    assert t.get('scheduled_date') == '2023-08-28'
    assert t.get('completed_date') == None

def test_parse_obsidian_completed_task():
    s = "- [x] Airwallex webhook configuration :white_check_mark: 2023-08-29"
    t = parse_obsidian_task(s)
    assert t['status'] == 'DONE'
    assert t['task'] == 'Airwallex webhook configuration'
    assert t.get('scheduled_date') == None
    assert t.get('completed_date') == '2023-08-29'

def test_parse_obsidian_task():
    s = "- [x] Airwallex webhook configuration"
    t = parse_obsidian_task(s)
    assert t['status'] == 'DONE'
    assert t['task'] == 'Airwallex webhook configuration'
    assert t.get('scheduled_date') == None
    assert t.get('completed_date') == None

def test_parse_obsidian_todo_task():
    s = "- [ ] Airwallex webhook configuration"
    t = parse_obsidian_task(s)
    assert t['status'] == 'TODO'
    assert t['task'] == 'Airwallex webhook configuration'
    assert t.get('scheduled_date') == None
    assert t.get('completed_date') == None

def test_parse_obsidian_in_progress_task():
    s = "- [/] Airwallex webhook configuration"
    t = parse_obsidian_task(s)
    assert t['status'] == 'TODO'
    assert t['task'] == 'Airwallex webhook configuration'
    assert t.get('scheduled_date') == None
    assert t.get('completed_date') == None