def _protect_fragments(text, patterns):
"""Replace regex matches with opaque tokens and return text + token map."""
saved={}
for pattern in patterns:
def repl(m):
token=f"\x00SAFE{len(saved)}\x00"
saved[token]=m.group(0)
return token
text=re.sub(pattern,repl,text,flags=re.S)
return text,saved
def _escape_hashes(s):
"""Escape literal hashes without double-escaping an existing backslash escape."""
return re.sub(r'(?<!\\)#', r'\\#', s)
def _escape_source_page_refs(s):
"""
Literal [[...]] in Dynalist is source text, not Dynalist's native internal
link representation. Escape it so File->DB import does not synthesize pages.
"""
s=re.sub(r'(?<!\\)\[\[', r'\\[\\[', s)
s=re.sub(r'(?<!\\)\]\]', r'\\]\\]', s)
return s
def neutralize_logseq_source(text,stats,tag_mode="neutralize"):
"""
Preserve source text visually while preventing the Logseq Markdown/DB
importer from inventing semantics that were not present in Dynalist.
tag_mode="neutralize" (recommended for archives with no intentional tags):
#foo -> \\#foo
# heading -> \\# heading
[[literal]] -> \\[\\[literal\\]\\]
tag_mode="preserve":
source #tags are retained, but literal [[...]] references and hashes
inside URLs/HTML numeric entities are still protected from accidental
page/tag creation.
Inline/fenced code is protected unchanged. Genuine Dynalist internal links
have already been replaced by opaque placeholders and are restored later.
"""
if not text:
return text
if tag_mode not in {"neutralize","preserve"}:
raise ValueError(f"Unknown tag mode: {tag_mode}")
# Code spans/fences: backslashes are literal inside code, so do not modify.
text,saved=_protect_fragments(
text,
[
r'```.*?```',
r'~~~.*?~~~',
r'``[^`\n]*``',
r'`[^`\n]*`',
]
)
# Existing Markdown links/images. Protect # in destinations because Logseq's
# importer has been observed to treat URL fragments as hashtags/pages.
md_saved={}
md_pat=re.compile(r'(!?\[[^\]\n]*\]\()([^\)\n]*)(\))')
def md_link_repl(m):
prefix,dest,suffix=m.groups()
open_paren=prefix.rfind("](")
label_part=prefix[:open_paren]
tail=prefix[open_paren:]
if tag_mode=="neutralize":
label_part=_escape_hashes(label_part)
label_part=_escape_source_page_refs(label_part)
n_hash=len(re.findall(r'(?<!\\)#',dest))
if n_hash:
stats["markdown_link_destination_hashes_protected"]+=n_hash
dest=_escape_hashes(dest)
rebuilt=label_part+tail+dest+suffix
token=f"\x00MDLINK{len(md_saved)}\x00"
md_saved[token]=rebuilt
return token
text=md_pat.sub(md_link_repl,text)
# Raw URLs containing # are made explicit Markdown links. This preserves the
# visible URL and target while keeping the DB importer out of hashtag mode.
url_saved={}
raw_url_re=re.compile(r'https?://[^\s<>"\']+')
def raw_url_repl(m):
url=m.group(0)
token=f"\x00URL{len(url_saved)}\x00"
if "#" in url:
n_hash=url.count("#")
escaped=_escape_hashes(url)
url_saved[token]=f"[{escaped}]({escaped})"
stats["raw_urls_with_hash_protected"]+=1
stats["raw_url_hash_characters_protected"]+=n_hash
else:
url_saved[token]=url
return token
text=raw_url_re.sub(raw_url_repl,text)
# HTML/XML numeric character entities such as ’ are data, not tags.
# This matters especially in imported RSS/feed/XML material.
entity_count=0
def entity_repl(m):
nonlocal entity_count
entity_count+=1
return m.group(0).replace("#",r"\#",1)
text=re.sub(r'&#(?:[0-9]+|[xX][0-9A-Fa-f]+);?',entity_repl,text)
stats["html_numeric_entity_hashes_protected"]+=entity_count
# Dynalist did not use [[...]] as its native page-link syntax. Preserve such
# source literally so Logseq does not create pages from it.
before_ref=len(re.findall(r'(?<!\\)\[\[',text))
text=_escape_source_page_refs(text)
stats["source_page_refs_neutralized"]+=before_ref
if tag_mode=="neutralize":
before_hash=len(re.findall(r'(?<!\\)#',text))
text=_escape_hashes(text)
stats["source_hashes_neutralized"]+=before_hash
else:
stats["source_hashes_preserved"]+=len(re.findall(r'(?<!\\)#',text))
# Restore protected content.
for mapping in (url_saved,md_saved,saved):
for k,v in mapping.items():
text=text.replace(k,v)
return text
def prepare_source_text(raw,exact,byid,docs,stats,link_prov,tag_mode="neutralize"):
"""
Convert genuine Dynalist internal links, neutralize only the remaining
source syntax, then restore generated native Logseq references.
"""
generated={}
text=convert_links(raw,exact,byid,docs,stats,link_prov,generated)
text=neutralize_logseq_source(text,stats,tag_mode=tag_mode)
for k,v in generated.items():
text=text.replace(k,v)
return text
def quote_note(s,indent):
s=s.replace("\r\n","\n").replace("\r","\n"); return [f"{indent}> {line}" if line else f"{indent}>" for line in s.split("\n")]
def escape_logseq_literal_heading(text):
"""
Dynalist heading level is metadata, not inferred from a literal '# ' prefix.
In Logseq Markdown, however, '# text', '## text', etc. at the beginning of a
block are rendered as headings. Escape only hashes followed by whitespace,
leaving Dynalist-style tags such as '#compress' completely unchanged.
"""
return re.sub(r'^(#{1,6})(?=\s)', r'\\\1', text)
def render_doc(doc,title,exact,byid,docs,literal,provenance,stats,tag_mode="neutralize"):
"""
Render a clean Logseq page.
User-visible/source metadata is NOT emitted as dynalist-* properties.
Full raw metadata remains in provenance.json and canonical.json.
Built-in id:: is retained for stable block references; title:: is retained
so long/encoded physical filenames can still display the intended page name.
"""
ns=doc.get("nodes",[]) or []
by={n.get("id"):n for n in ns if n.get("id")}
root=by.get("root")
seen=set()
# Logseq built-ins only. 'title' is hidden in normal rendered view.
lines=[
f"id:: {page_uuid(doc['file_id'])}",
f"title:: {title}",
""
]
if root is not None:
stats["synthetic_document_roots"]+=1
provenance.setdefault("document_roots",{})[doc["file_id"]]=root
def visit(n,depth,path):
nid=n.get("id")
if not nid:
stats["nodes_missing_id"]+=1
return
if nid in seen:
stats["node_cycle_or_duplicate_reference"]+=1
return
seen.add(nid)
rawc=n.get("content","") or ""
rawn=n.get("note","") or ""
lprov=[]
content=convert_fmt(
prepare_source_text(rawc,exact,byid,docs,stats,lprov,tag_mode=tag_mode),literal
).replace("\r\n","\n").replace("\r","\n").replace("\n","<br>")
content=escape_logseq_literal_heading(content)
cb=bool(n.get("checkbox",False))
checked=bool(n.get("checked",False))
task=("DONE " if checked else "TODO ") if cb else ""
try:
h=max(0,min(3,int(n.get("heading",0) or 0)))
except:
h=0
visible=task + (("#"*h+" ") if h else "") + content
if not visible.strip():
visible="\u200b"
stats["empty_content_nodes"]+=1
indent=" "*depth
pi=" "*(depth+1)
lines.append(f"{indent}- {visible}")
# Required for native Logseq block references; built-in identity property.
lines.append(f"{pi}id:: {block_uuid(nid)}")
# Preserve Logseq semantics, but keep all Dynalist-specific metadata out
# of the page body.
if n.get("collapsed") is True:
lines.append(f"{pi}collapsed:: true")
stats["collapsed"]+=1
if h:
stats["headings"]+=1
try:
color=int(n.get("color",0) or 0)
except:
color=n.get("color",0)
if color:
stats["colored"]+=1
if cb:
stats["done_tasks" if checked else "todo_tasks"]+=1
if rawn:
readable=convert_fmt(
prepare_source_text(rawn,exact,byid,docs,stats,lprov,tag_mode=tag_mode),literal
)
readable="\n".join(escape_logseq_literal_heading(line) for line in readable.splitlines())
lines.extend(quote_note(readable,pi))
stats["notes"]+=1
# This is the lossless metadata layer.
provenance["nodes"][block_uuid(nid)]={
"document_id":doc["file_id"],
"node_id":nid,
"tree_path":list(path),
"raw":n,
"link_conversions":lprov
}
stats["nodes"]+=1
for i,cid in enumerate(n.get("children",[]) or []):
if cid in by:
visit(by[cid],depth+1,path+(i,))
else:
stats["missing_child_ids"]+=1
provenance["missing_children"].append({
"document_id":doc["file_id"],
"parent_node_id":nid,
"missing_child_id":cid
})
if root is not None:
top_ids=root.get("children",[]) or []
for i,cid in enumerate(top_ids):
if cid in by:
visit(by[cid],0,(i,))
else:
stats["missing_child_ids"]+=1
provenance["missing_children"].append({
"document_id":doc["file_id"],
"parent_node_id":"root",
"missing_child_id":cid
})
j=len(top_ids)
else:
child={c for n in ns for c in (n.get("children",[]) or [])}
roots=[n for n in ns if n.get("id") not in child]
for i,n in enumerate(roots):
visit(n,0,(i,))
j=len(roots)
for n in ns:
nid=n.get("id")
if nid not in {None,"root"} and nid not in seen:
stats["unreachable_nodes_recovered"]+=1
visit(n,0,(j,))
j+=1
if not any(n.get("id") not in {None,"root"} for n in ns):
lines.append("- \u200b")
stats["empty_documents"]+=1
return "\n".join(lines).rstrip()+"\n"
def _dir_nonempty(path: Path) -> bool:
return path.exists() and any(path.iterdir())
def _safe_replace_output(out: Path, api: Path) -> None:
"""Delete only an explicitly requested output graph after conservative guards."""
out=out.resolve()
protected={Path("/").resolve(), Path.home().resolve(), api.resolve(), api.parent.resolve()}
try:
protected.add(Path.cwd().resolve())
except Exception:
pass
if out in protected or out.parent==out:
raise ValueError(f"Refusing to replace unsafe output path: {out}")
if len(out.parts) < 3:
raise ValueError(f"Refusing to replace suspiciously short path: {out}")
if out.exists():
shutil.rmtree(out)
def opml_status(cmp):
exact_no=sum(
1 for m in cmp["matches"]
if m["match_method"] in {"title+exact-structure-content-note","unique-signature-fallback"}
and not m["differences"]
)
diffdocs=sum(1 for m in cmp["matches"] if m["differences"])
clean=(
diffdocs==0
and not cmp["unmatched_api"]
and not cmp["unmatched_opml"]
and not cmp["opml_parse_failures"]
and exact_no==len(cmp["matches"])
)
return {
"exact_no_differences":exact_no,
"matched_with_differences":diffdocs,
"clean":clean,
}
def verify_generated_graph(api: Path, out: Path, manifest: dict, tag_mode: str):
"""
Structural verification of the generated legacy Logseq graph. This does not
replace the independent API↔OPML comparison; it verifies the generated files
and manifest are internally consistent.
"""
issues=[]
pages_dir=out/"pages"
docs=manifest.get("documents",[])
expected_files={d["markdown_file"] for d in docs}
actual_files={
f"pages/{p.name}" for p in pages_dir.glob("*.md") if p.is_file()
}
for missing in sorted(expected_files-actual_files):
issues.append(f"missing generated page: {missing}")
for extra in sorted(actual_files-expected_files):
issues.append(f"unexpected generated page: {extra}")
block_ids=0
quoted_titles=0
for d in docs:
p=out/d["markdown_file"]
if not p.exists():
continue
raw=p.read_text(encoding="utf-8",errors="replace")
lines=raw.splitlines()
expected_title=f"title:: {d['page_title']}"
if len(lines)<2 or lines[1] != expected_title:
issues.append(
f"title property mismatch in {d['markdown_file']}: "
f"expected {expected_title!r}, got {(lines[1] if len(lines)>1 else None)!r}"
)
if len(lines)>1 and re.match(r'^title::\s+".*"\s*$',lines[1]):
quoted_titles+=1
block_ids += sum(
1 for line in lines[2:]
if re.match(r'^\s+id::\s+[0-9a-fA-F-]{36}\s*$',line)
)
expected_sha=d.get("markdown_sha256")
if expected_sha and sha256_file(p) != expected_sha:
issues.append(f"checksum mismatch: {d['markdown_file']}")
if quoted_titles:
issues.append(f"{quoted_titles} page title properties are still JSON-quoted")
expected_blocks=manifest.get("stats",{}).get("nodes")
if expected_blocks is not None and block_ids != expected_blocks:
issues.append(
f"block id count mismatch: generated={block_ids}, expected={expected_blocks}"
)
return {
"ok":not issues,
"issues":issues,
"page_files_expected":len(expected_files),
"page_files_actual":len(actual_files),
"block_ids":block_ids,
"tag_mode":tag_mode,
}
def _importer_guidance(tag_mode):
if tag_mode=="neutralize":
return [
"Recommended Logseq File -> DB import options:",
" Extract inline code snippets as child blocks: OFF",
" Import all tags: OFF",
" Import specific tags: blank",
" Remove inline tags: OFF",
" Import additional tags from properties: blank",
" Import tag parents from properties: blank",
]
return [
"Recommended Logseq File -> DB import options:",
" Extract inline code snippets as child blocks: OFF",
" Import all tags: ON",
" Remove inline tags: OFF (unless intentionally desired)",
" Other tag-property options: leave blank unless intentionally used",
]