109 lines
4.3 KiB
Python
109 lines
4.3 KiB
Python
import couchdb
|
|
import os
|
|
import base64
|
|
import json
|
|
from urllib.parse import quote
|
|
|
|
class CouchDBSync:
|
|
def __init__(self, url, username, password, db_name, passphrase=""):
|
|
self.url = url.rstrip("/")
|
|
self.username = username.strip()
|
|
self.password = password.strip()
|
|
self.db_name = db_name
|
|
self.passphrase = passphrase
|
|
|
|
def connect(self):
|
|
"""Establishes connection to CouchDB."""
|
|
try:
|
|
if self.username and self.password:
|
|
if "://" in self.url:
|
|
protocol, rest = self.url.split("://", 1)
|
|
else:
|
|
protocol, rest = "http", self.url
|
|
safe_user = quote(self.username, safe="")
|
|
safe_pass = quote(self.password, safe="")
|
|
full_url = f"{protocol}://{safe_user}:{safe_pass}@{rest}"
|
|
else:
|
|
full_url = self.url
|
|
|
|
server = couchdb.Server(full_url)
|
|
_ = server.version()
|
|
return server
|
|
except Exception as e:
|
|
raise Exception(f"Connection Failed: {str(e)}")
|
|
|
|
def fetch_notes(self, target_folder: str) -> list[str]:
|
|
logs = []
|
|
try:
|
|
server = self.connect()
|
|
if self.db_name not in server:
|
|
return [f"Error: Database '{self.db_name}' not found."]
|
|
|
|
db = server[self.db_name]
|
|
logs.append(f"Connected to {self.db_name}.")
|
|
|
|
if not os.path.exists(target_folder):
|
|
os.makedirs(target_folder, exist_ok=True)
|
|
|
|
# 1. Fetch all docs with content in ONE request
|
|
logs.append("Fetching all documents...")
|
|
all_docs = {}
|
|
# include_docs=True is much faster than 278 individual requests
|
|
for row in db.view('_all_docs', include_docs=True):
|
|
doc = row.doc
|
|
if not doc['_id'].startswith("_design"):
|
|
all_docs[doc['_id']] = doc
|
|
|
|
logs.append(f"Retrieved {len(all_docs)} documents. Reconstructing notes...")
|
|
|
|
note_count = 0
|
|
|
|
# 2. Iterate to find "Metadata" documents
|
|
for doc_id, doc in all_docs.items():
|
|
# Self-hosted LiveSync uses 'children' for files
|
|
if "children" in doc and isinstance(doc["children"], list):
|
|
|
|
relative_path = doc.get("path", doc_id)
|
|
|
|
# Construct full local path
|
|
safe_path = relative_path.replace(":", "-").replace("|", "-")
|
|
full_local_path = os.path.join(target_folder, safe_path)
|
|
|
|
# 3. Reconstruct Content
|
|
full_content = []
|
|
is_encrypted = False
|
|
|
|
for chunk_id in doc["children"]:
|
|
chunk = all_docs.get(chunk_id)
|
|
if not chunk:
|
|
continue
|
|
|
|
chunk_data = chunk.get("data") or chunk.get("content") or ""
|
|
|
|
if str(chunk_data).startswith("%") or chunk.get("e_"):
|
|
is_encrypted = True
|
|
break
|
|
|
|
full_content.append(str(chunk_data))
|
|
|
|
if is_encrypted:
|
|
continue
|
|
|
|
note_text = "".join(full_content)
|
|
|
|
# Only save if there is content or it's a known empty file
|
|
if note_text.strip() or len(doc["children"]) > 0:
|
|
os.makedirs(os.path.dirname(full_local_path), exist_ok=True)
|
|
with open(full_local_path, "w", encoding="utf-8") as f:
|
|
f.write(note_text)
|
|
note_count += 1
|
|
if note_count % 10 == 0:
|
|
logs.append(f"Reconstructed {note_count} notes...")
|
|
|
|
logs.append(f"Sync Complete. Reconstructed {note_count} total notes.")
|
|
|
|
except Exception as e:
|
|
logs.append(f"Sync Error: {str(e)}")
|
|
|
|
return logs
|