Fix export handlers and search engine - add ExportHandler class and fix SearchEngine
This commit is contained in:
@@ -2,24 +2,60 @@
|
|||||||
|
|
||||||
import json
|
import json
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
|
from pathlib import Path
|
||||||
from typing import Any
|
from typing import Any
|
||||||
|
|
||||||
from snip.db.database import Database
|
from snip.db.database import Database, get_database
|
||||||
|
|
||||||
|
|
||||||
def export_snippets(snippets: list[dict[str, Any]], file_path: str):
|
class ExportHandler:
|
||||||
"""Export snippets to a JSON file."""
|
def __init__(self, db_path: str | None = None):
|
||||||
export_data = {
|
self.db = get_database(db_path)
|
||||||
|
|
||||||
|
def export_all(self) -> dict[str, Any]:
|
||||||
|
"""Export all snippets."""
|
||||||
|
snippets = self.db.list_snippets(limit=10000)
|
||||||
|
for s in snippets:
|
||||||
|
if isinstance(s.get("tags"), str):
|
||||||
|
s["tags"] = json.loads(s["tags"])
|
||||||
|
return {
|
||||||
"version": "1.0",
|
"version": "1.0",
|
||||||
"exported_at": datetime.utcnow().isoformat() + "Z",
|
"exported_at": datetime.utcnow().isoformat() + "Z",
|
||||||
"snippets": snippets,
|
"snippets": snippets,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
def export_collection(self, collection_id: int) -> dict[str, Any]:
|
||||||
|
"""Export a specific collection."""
|
||||||
|
snippets = self.db.get_collection_snippets(collection_id)
|
||||||
|
for s in snippets:
|
||||||
|
if isinstance(s.get("tags"), str):
|
||||||
|
s["tags"] = json.loads(s["tags"])
|
||||||
|
return {
|
||||||
|
"version": "1.0",
|
||||||
|
"exported_at": datetime.utcnow().isoformat() + "Z",
|
||||||
|
"snippets": snippets,
|
||||||
|
}
|
||||||
|
|
||||||
|
def export_snippet(self, snippet_id: int) -> dict[str, Any] | None:
|
||||||
|
"""Export a single snippet."""
|
||||||
|
snippet = self.db.get_snippet(snippet_id)
|
||||||
|
if not snippet:
|
||||||
|
return None
|
||||||
|
if isinstance(snippet.get("tags"), str):
|
||||||
|
snippet["tags"] = json.loads(snippet["tags"])
|
||||||
|
return {
|
||||||
|
"version": "1.0",
|
||||||
|
"exported_at": datetime.utcnow().isoformat() + "Z",
|
||||||
|
"snippets": [snippet],
|
||||||
|
}
|
||||||
|
|
||||||
|
def write_export(self, file_path: str, data: dict[str, Any]):
|
||||||
|
"""Write export data to a file."""
|
||||||
|
Path(file_path).parent.mkdir(parents=True, exist_ok=True)
|
||||||
with open(file_path, "w") as f:
|
with open(file_path, "w") as f:
|
||||||
json.dump(export_data, f, indent=2)
|
json.dump(data, f, indent=2)
|
||||||
|
|
||||||
|
def import_data(self, file_path: str, strategy: str = "skip") -> dict[str, Any]:
|
||||||
def import_snippets(db: Database, file_path: str, strategy: str = "skip") -> tuple[int, int]:
|
|
||||||
"""Import snippets from a JSON file."""
|
"""Import snippets from a JSON file."""
|
||||||
with open(file_path, "r") as f:
|
with open(file_path, "r") as f:
|
||||||
data = json.load(f)
|
data = json.load(f)
|
||||||
@@ -27,12 +63,30 @@ def import_snippets(db: Database, file_path: str, strategy: str = "skip") -> tup
|
|||||||
snippets = data.get("snippets", [])
|
snippets = data.get("snippets", [])
|
||||||
imported = 0
|
imported = 0
|
||||||
skipped = 0
|
skipped = 0
|
||||||
|
replaced = 0
|
||||||
|
|
||||||
for snippet_data in snippets:
|
for snippet_data in snippets:
|
||||||
result = db.import_snippet(snippet_data, strategy=strategy)
|
result = self.db.import_snippet(snippet_data, strategy=strategy)
|
||||||
if result is None:
|
if result is None:
|
||||||
skipped += 1
|
skipped += 1
|
||||||
|
elif result == -1:
|
||||||
|
replaced += 1
|
||||||
else:
|
else:
|
||||||
imported += 1
|
imported += 1
|
||||||
|
|
||||||
return imported, skipped
|
return {
|
||||||
|
"imported": imported,
|
||||||
|
"skipped": skipped,
|
||||||
|
"replaced": replaced,
|
||||||
|
"total": len(snippets),
|
||||||
|
}
|
||||||
|
|
||||||
|
def generate_import_summary(self, results: dict[str, Any]) -> str:
|
||||||
|
"""Generate a human-readable import summary."""
|
||||||
|
lines = [
|
||||||
|
f"Import complete!",
|
||||||
|
f" Imported: {results.get('imported', 0)}",
|
||||||
|
f" Skipped: {results.get('skipped', 0)}",
|
||||||
|
f" Replaced: {results.get('replaced', 0)}",
|
||||||
|
]
|
||||||
|
return "\n".join(lines)
|
||||||
Reference in New Issue
Block a user