#!/usr/bin/env python3 """ # Script dedup-readwise ## Metadata #obsidian/ob-script #python **Parent**:: [[♯ Readwise]], [[Obsidian Chore Scripts]], [[Executable Markdown File]] ## Synopsis Repair concatenated Readwise exports, retaining the complete last appended export, including its frontmatter. Older-only highlights and annotations are not merged. Single-export files and separately named duplicates are not rewritten. Both legacy `Full Title` and canonical `full-title` are recognized; an export containing both names requires review. Retained frontmatter is never reserialized. Requires PyYAML (`python -m pip install PyYAML`). - `python shore/bin/dedup-readwise.md -n`: scan the library without writing (also the default). - `python shore/bin/dedup-readwise.md -n "path/to/note.md"`: inspect one file. - `python shore/bin/dedup-readwise.md --write --backup-dir "path/to/backups" "path/to/note.md"`: apply to explicit files or directories. Writes preserve UTF-8/LF bytes and exact modification times. Originals are backed up by SHA-256 before writing; without `--backup-dir`, backups use the system temp folder's `readwise-dedup-backups` directory. Keep a persistent backup directory if needed. Backups must be outside write-target directories. Reports include older-only lines, removed block IDs, and changed metadata. Malformed exports, inconsistent titles/URLs, reversed Created dates, non-LF input, and a newer legacy copy lacking an older copy's frontmatter require review. Only structurally complete matching exports are selected, not arbitrary covers. Windows writes deny other writers/deletion while locked; POSIX locks are advisory, so stop other editors/sync writers before applying. No reset or resync is performed. ## Script ```python # """ import argparse from contextlib import contextmanager from datetime import date import hashlib import os from pathlib import Path import re import stat import sys import tempfile from urllib.parse import urlsplit import yaml class ReviewRequired(ValueError): pass class PropertiesLoader(yaml.SafeLoader): def construct_mapping(self, node, deep=False): result = {} for key_node, value_node in node.value: key = self.construct_object(key_node, deep=deep) if not isinstance(key, str) or key == '<<' or key in result: raise ReviewRequired('Invalid or duplicate YAML property key') result[key] = self.construct_object(value_node, deep=deep) return result def properties(text): try: result = yaml.load(text, Loader=PropertiesLoader) except yaml.YAMLError as error: raise ReviewRequired('Malformed YAML') from error if not isinstance(result, dict): raise ReviewRequired('Frontmatter must be a mapping') return result def visible_lines(lines, masked): fence = None comment = None html = None visible = set() for i, line in enumerate(lines): if i in masked: continue if comment: if comment in line: comment = None continue if html: if re.search(r'</' + html + r'\s*>', line, re.I): html = None continue marker = re.match(r'^ {0,3}(`{3,}|~{3,})(.*), line.rstrip('\n')) if fence: if marker and marker[1][0] == fence[0] and len(marker[1]) >= len(fence) and not marker[2].strip(): fence = None continue if marker: fence = marker[1] continue for start, end in [('<!--', '-->'), ('%%', '%%')]: if start in line: if end not in line.split(start, 1)[1]: comment = end break else: tag = re.match(r'^ {0,3}<([a-z][a-z0-9-]*)(?:\s|>)', line, re.I) if tag and tag[1].lower() not in {'br', 'hr', 'img', 'input', 'link', 'meta', 'source'}: if not re.search(r'</' + tag[1] + r'\s*>', line, re.I) and not line.rstrip().endswith('/>'): html = tag[1] continue visible.add(i) return visible def field_key(name): key = name.lower() return 'full-title' if key == 'full title' else key def field(record, name): values = [v for k, v in record.items() if field_key(k) == field_key(name)] if len(values) > 1: raise ReviewRequired('Ambiguous property names: ' + name) return values[0] if values else None def target(value): if not isinstance(value, str): return None link = re.fullmatch(r'\[[^\]]*\]\((https?://.*)\)', value) return link[1] if link else value def created(record): value = field(record, 'Created') if value is None: return None if type(value) is date: return value.isoformat() if not isinstance(value, str): raise ReviewRequired('Unrecognized Created date') if value.startswith('[[') and value.endswith(']]'): value = value[2:-2] if not re.fullmatch(r'\d{4}-\d{2}-\d{2}', value): raise ReviewRequired('Unrecognized Created date') try: return date.fromisoformat(value).isoformat() except ValueError as error: raise ReviewRequired('Invalid Created date') from error def plan(text): lines = text.splitlines(keepends=True) offsets = [] offset = 0 for line in lines: offsets.append(offset) offset += len(line) headers = [i for i, line in enumerate(lines) if re.fullmatch(r'# .+ \(Highlights\)\s*', line)] if len(headers) < 2: return None wrappers = {} masked = set() for h in headers: end = h - 1 while end >= 0 and not lines[end].strip(): end -= 1 if end < 0 or lines[end].strip() != '---': continue start = end - 1 while start >= 0 and not lines[start].rstrip('\n').endswith('---'): start -= 1 try: if start < 0: raise ReviewRequired('Missing frontmatter opening delimiter') data = properties(''.join(lines[start + 1:end])) wrappers[h] = (offsets[start] + len(lines[start].rstrip('\n')) - 3, data) masked.update(range(start, end + 1)) except ReviewRequired as error: wrappers[h] = error visible = visible_lines(lines, masked) headers = [h for h in headers if h in visible] if len(headers) < 2: return None if '\r' in text or text.startswith('\ufeff'): raise ReviewRequired('Only UTF-8 without BOM and LF line endings can be rewritten') records = [] for n, h in enumerate(headers): wrapper = wrappers.get(h) if isinstance(wrapper, ReviewRequired): raise wrapper start, data = wrapper if wrapper else (offsets[h], {}) data = dict(data) end = headers[n + 1] if n + 1 < len(headers) else len(lines) highlights = [i for i in range(h + 1, end) if i in visible and lines[i].strip() == '## Highlights'] if len(highlights) != 1: raise ReviewRequired('Missing or repeated Highlights section') if n == len(headers) - 1 and not ''.join(lines[highlights[0] + 1:]).strip(): raise ReviewRequired('Newer export has an empty Highlights section') for i in range(h + 1, highlights[0]): if i not in visible: continue match = re.fullmatch(r'\*\*([^*]+)\*\*:: (.*)\n?', lines[i]) if match: if any(field_key(k) == field_key(match[1]) for k in data): raise ReviewRequired('Metadata appears in both frontmatter and body') data[match[1]] = match[2] review = target(field(data, 'Review')) title = field(data, 'Full Title') if not isinstance(review, str) or not re.fullmatch(r'https://readwise\.io/bookreview/\d+', review) or not isinstance(title, str) or not title: raise ReviewRequired('Incomplete Readwise metadata') records.append(dict(start=start, heading=lines[h].strip(), data=data, yaml=bool(wrapper))) if records[0]['start'] != 0: raise ReviewRequired('Unowned content precedes the first export') last = records[-1] if not last['yaml'] and any(r['yaml'] for r in records[:-1]): raise ReviewRequired('Newer legacy copy lacks frontmatter; ownership is ambiguous') dates = [] urls = set() for record in records: if record['heading'] != last['heading'] or field(record['data'], 'Full Title') != field(last['data'], 'Full Title'): raise ReviewRequired('Exports have different titles') value = created(record['data']) if value: dates.append(value) url = target(field(record['data'], 'URL')) if url: parsed = urlsplit(url) if parsed.scheme not in {'http', 'https'} or not parsed.netloc: raise ReviewRequired('Unrecognized source URL') urls.add((parsed.netloc.lower(), parsed.path.rstrip('/'))) if dates != sorted(dates) or len(urls) > 1: raise ReviewRequired('Conflicting dates or source URLs') output = text[last['start']:] old = text[:last['start']] block_ids = lambda value: set(re.findall(r'(?m)(?<!\w)\^([\w-]+)[ \t]*, value)) changed = sorted({k for r in records[:-1] for k in r['data'] if r['data'][k] != last['data'].get(k)}) return dict(content=output, exports=len(records), removed_ids=sorted(block_ids(old) - block_ids(output)), older_only_lines=len(set(old.splitlines()) - set(output.splitlines()) - {''}), changed_metadata=changed) def identity(info): return info.st_dev, info.st_ino, info.st_size, info.st_mtime_ns def regular(path, info): if not stat.S_ISREG(info.st_mode) or info.st_nlink != 1 or path.is_symlink(): raise ReviewRequired('Only regular, non-linked files can be rewritten') @contextmanager def locked_file(path): if os.name == 'nt': import ctypes from ctypes import wintypes import msvcrt kernel = ctypes.WinDLL('kernel32', use_last_error=True) create = kernel.CreateFileW create.argtypes = [wintypes.LPCWSTR, wintypes.DWORD, wintypes.DWORD, wintypes.LPVOID, wintypes.DWORD, wintypes.DWORD, wintypes.HANDLE] create.restype = wintypes.HANDLE handle = create(str(path.absolute()), 0xC0000000, 1, None, 3, 0x00200080, None) if handle == ctypes.c_void_p(-1).value: raise ctypes.WinError(ctypes.get_last_error()) try: fd = msvcrt.open_osfhandle(handle, os.O_RDWR | os.O_BINARY) except BaseException: close = kernel.CloseHandle close.argtypes = [wintypes.HANDLE] close(handle) raise with os.fdopen(fd, 'r+b') as stream: yield stream else: import fcntl with path.open('r+b') as stream: fcntl.flock(stream, fcntl.LOCK_EX | fcntl.LOCK_NB) yield stream def restore_times(fd, info): if os.name != 'nt': os.utime(fd, ns=(info.st_atime_ns, info.st_mtime_ns)) return import ctypes from ctypes import wintypes import msvcrt def filetime(ns): ticks = ns // 100 + 116444736000000000 return wintypes.FILETIME(ticks & 0xFFFFFFFF, ticks >> 32) access, modified = filetime(info.st_atime_ns), filetime(info.st_mtime_ns) set_time = ctypes.WinDLL('kernel32', use_last_error=True).SetFileTime set_time.argtypes = [wintypes.HANDLE, ctypes.POINTER(wintypes.FILETIME), ctypes.POINTER(wintypes.FILETIME), ctypes.POINTER(wintypes.FILETIME)] set_time.restype = wintypes.BOOL if not set_time(msvcrt.get_osfhandle(fd), None, ctypes.byref(access), ctypes.byref(modified)): raise ctypes.WinError(ctypes.get_last_error()) def apply(path, before, info, after, backup_dir): if path.resolve().is_relative_to(backup_dir.resolve()): raise ReviewRequired('A backup file cannot also be a write target') with locked_file(path) as stream: current = os.fstat(stream.fileno()) regular(path, current) if identity(current) != identity(info) or identity(path.stat()) != identity(info) or stream.read() != before: raise ReviewRequired('File changed since inspection') backup_dir.mkdir(parents=True, exist_ok=True) backup = backup_dir / (hashlib.sha256(before).hexdigest() + '.md') try: with backup.open('xb') as saved: saved.write(before) saved.flush() os.fsync(saved.fileno()) except FileExistsError: pass regular(backup, backup.lstat()) if backup.read_bytes() != before: raise ReviewRequired('Recovery backup does not match original bytes') stream.seek(0) if stream.write(after) != len(after): raise OSError('Incomplete write; recover original from ' + str(backup)) stream.truncate() stream.flush() os.fsync(stream.fileno()) stream.seek(0) if stream.read() != after or os.fstat(stream.fileno()).st_ino != path.stat().st_ino: raise ReviewRequired('Output changed; timestamp not restored. Backup: ' + str(backup)) restore_times(stream.fileno(), info) if os.fstat(stream.fileno()).st_mtime_ns != info.st_mtime_ns: raise OSError('Exact modification time was not preserved') return backup def fix(file, dry_run, *, backup_dir=None, report=None): path = Path(file) regular(path, path.lstat()) with path.open('rb') as stream: info = os.fstat(stream.fileno()) before = stream.read() if identity(info) != identity(os.fstat(stream.fileno())): raise ReviewRequired('File changed while reading') result = plan(before.decode('utf-8')) if result is None: return None if not dry_run: directory = Path(backup_dir) if backup_dir else Path(tempfile.gettempdir()) / 'readwise-dedup-backups' result['backup'] = str(apply(path, before, info, result['content'].encode('utf-8'), directory)) if report: report(result) return True def linked(path): return path.is_symlink() or getattr(path, 'is_junction', lambda: False)() def files_at(targets): seen = set() for target_path in targets: path = Path(target_path) if linked(path): raise ReviewRequired('Linked targets are not supported: ' + str(path)) if path.is_dir(): for root, dirs, files in os.walk(path): dirs[:] = sorted(d for d in dirs if d not in {'Readwise Syncs', 'Readwise Syncs History'} and not linked(Path(root, d))) for name in sorted(files): file = Path(root, name) key = os.path.normcase(str(file.absolute())) if key not in seen: seen.add(key) yield file elif path.is_file(): key = os.path.normcase(str(path.absolute())) if key not in seen: seen.add(key) yield path else: raise ReviewRequired('Target does not exist: ' + str(path)) def main(argv=None): parser = argparse.ArgumentParser(description='Keep newer complete Readwise exports; default: dry-run.') mode = parser.add_mutually_exclusive_group() mode.add_argument('-n', '--dry-run', action='store_true') mode.add_argument('--write', action='store_true') parser.add_argument('--backup-dir', type=Path) parser.add_argument('paths', nargs='*') args = parser.parse_args(argv) if args.write and not args.paths: parser.error('--write requires explicit file or directory targets') backup_dir = args.backup_dir or Path(tempfile.gettempdir()) / 'readwise-dedup-backups' if args.write: for path in map(Path, args.paths): if path.is_dir() and backup_dir.resolve().is_relative_to(path.resolve()): parser.error('Backup directory must be outside directory write targets') failed = False def report(result): print(f"{'Fix' if args.write else 'Would fix'} {file}: retain export {result['exports']}/{result['exports']}; " f"{result['older_only_lines']} older-only lines; removed IDs {result['removed_ids']}; " f"changed metadata {result['changed_metadata']}" + (f"; backup {result['backup']}" if args.write else '')) try: for file in files_at(args.paths or ['robot/Readwise Library']): if '-2.' in file.name and not file.name.startswith('Readwise Sync '): print(f'Dup {file}') if file.suffix.lower() != '.md': continue try: fix(file, not args.write, backup_dir=backup_dir, report=report) except (OSError, ValueError) as error: failed = True print(f'Review {file}: {error}', file=sys.stderr) except (OSError, ValueError) as error: failed = True print(f'Review: {error}', file=sys.stderr) return int(failed) if __name__ == '__main__': sys.exit(main()) """ # vim: ft=python ``` """