Manage Atom feeds in a persistent git repository

update to not use metadata.json

Changed files
+129 -143
src
thicket
tests
+6 -14
ARCH.md
···
│ │ ├── commands/ # Subcommands
│ │ │ ├── __init__.py
│ │ │ ├── init.py # Initialize git store
-
│ │ │ ├── add.py # Add feed to config
+
│ │ │ ├── add.py # Add users and feeds
│ │ │ ├── sync.py # Sync feeds
-
│ │ │ ├── list.py # List users/feeds
-
│ │ │ └── search.py # Search entries
+
│ │ │ ├── list_cmd.py # List users/feeds
+
│ │ │ └── duplicates.py # Manage duplicate entries
│ │ └── utils.py # CLI utilities (progress, formatting)
│ ├── core/ # Core business logic
│ │ ├── __init__.py
│ │ ├── feed_parser.py # Feed parsing and normalization
-
│ │ ├── git_store.py # Git repository operations
-
│ │ ├── cache.py # Cache management
-
│ │ └── sanitizer.py # Filename and HTML sanitization
+
│ │ └── git_store.py # Git repository operations
│ ├── models/ # Pydantic data models
│ │ ├── __init__.py
│ │ ├── config.py # Configuration models
│ │ ├── feed.py # Feed/Entry models
│ │ └── user.py # User metadata models
│ └── utils/ # Shared utilities
-
│ ├── __init__.py
-
│ ├── paths.py # Path handling
-
│ └── network.py # HTTP client wrapper
+
│ └── __init__.py
├── tests/
│ ├── __init__.py
│ ├── conftest.py # pytest configuration
···
├── index.json # User directory index
├── duplicates.json # Manual curation of duplicate entries
├── user1/
-
│ ├── metadata.json # User metadata
│ ├── entry_id_1.json # Sanitized entry files
│ ├── entry_id_2.json
│ └── ...
···
thicket list users
thicket list feeds --user alyssa
-
# Search entries
-
thicket search "keyword" --user alyssa --since 2025-01-01
-
# Manage duplicate entries
thicket duplicates list
thicket duplicates add <entry_id_1> <entry_id_2> # Mark as duplicates
···
homepage=self.author_uri or self.link,
icon=self.logo or self.icon or self.image_url
)
-
```
+
```
+29 -41
src/thicket/cli/commands/add.py
···
from ...core.feed_parser import FeedParser
from ...core.git_store import GitStore
-
from ...models import UserConfig
from ..main import app
from ..utils import (
create_progress,
···
print_error,
print_info,
print_success,
-
save_config,
)
···
# Load configuration
config = load_config(config_file)
+
# Initialize Git store
+
git_store = GitStore(config.git_store)
+
# Check if user already exists
-
existing_user = config.find_user(username)
+
existing_user = git_store.get_user(username)
if existing_user:
print_error(f"User '{username}' already exists")
print_error("Use 'thicket add feed' to add additional feeds")
···
if auto_discover:
discovered_metadata = asyncio.run(discover_feed_metadata(validated_feed_url))
-
# Create user config with manual overrides taking precedence
-
user_config = UserConfig(
-
username=username,
-
feeds=[validated_feed_url],
-
email=email or (discovered_metadata.author_email if discovered_metadata else None),
-
homepage=HttpUrl(homepage) if homepage else (discovered_metadata.author_uri or discovered_metadata.link if discovered_metadata else None),
-
icon=HttpUrl(icon) if icon else (discovered_metadata.logo or discovered_metadata.icon or discovered_metadata.image_url if discovered_metadata else None),
-
display_name=display_name or (discovered_metadata.author_name or discovered_metadata.title if discovered_metadata else None),
-
)
-
-
# Add user to configuration
-
config.add_user(user_config)
-
-
# Save configuration
-
save_config(config, config_file)
+
# Prepare user data with manual overrides taking precedence
+
user_display_name = display_name or (discovered_metadata.author_name or discovered_metadata.title if discovered_metadata else None)
+
user_email = email or (discovered_metadata.author_email if discovered_metadata else None)
+
user_homepage = homepage or (str(discovered_metadata.author_uri or discovered_metadata.link) if discovered_metadata else None)
+
user_icon = icon or (str(discovered_metadata.logo or discovered_metadata.icon or discovered_metadata.image_url) if discovered_metadata else None)
# Add user to Git store
-
git_store = GitStore(config.git_store)
git_store.add_user(
username=username,
-
display_name=user_config.display_name,
-
email=user_config.email,
-
homepage=str(user_config.homepage) if user_config.homepage else None,
-
icon=str(user_config.icon) if user_config.icon else None,
-
feeds=[str(f) for f in user_config.feeds],
+
display_name=user_display_name,
+
email=user_email,
+
homepage=user_homepage,
+
icon=user_icon,
+
feeds=[str(validated_feed_url)],
)
# Commit changes
···
if discovered_metadata and auto_discover:
print_info("Auto-discovered metadata:")
-
if user_config.display_name:
-
print_info(f" Display name: {user_config.display_name}")
-
if user_config.email:
-
print_info(f" Email: {user_config.email}")
-
if user_config.homepage:
-
print_info(f" Homepage: {user_config.homepage}")
-
if user_config.icon:
-
print_info(f" Icon: {user_config.icon}")
+
if user_display_name:
+
print_info(f" Display name: {user_display_name}")
+
if user_email:
+
print_info(f" Email: {user_email}")
+
if user_homepage:
+
print_info(f" Homepage: {user_homepage}")
+
if user_icon:
+
print_info(f" Icon: {user_icon}")
def add_feed(username: str, feed_url: Optional[str], config_file: Path) -> None:
···
# Load configuration
config = load_config(config_file)
+
# Initialize Git store
+
git_store = GitStore(config.git_store)
+
# Check if user exists
-
user = config.find_user(username)
+
user = git_store.get_user(username)
if not user:
print_error(f"User '{username}' not found")
print_error("Use 'thicket add user' to add a new user")
raise typer.Exit(1)
# Check if feed already exists
-
if validated_feed_url in user.feeds:
+
if str(validated_feed_url) in user.feeds:
print_error(f"Feed already exists for user '{username}': {feed_url}")
raise typer.Exit(1)
# Add feed to user
-
if config.add_feed_to_user(username, validated_feed_url):
-
save_config(config, config_file)
-
-
# Update Git store
-
git_store = GitStore(config.git_store)
-
git_store.update_user(username, feeds=[str(f) for f in user.feeds])
+
updated_feeds = user.feeds + [str(validated_feed_url)]
+
if git_store.update_user(username, feeds=updated_feeds):
git_store.commit_changes(f"Add feed to user {username}: {feed_url}")
-
print_success(f"Added feed to user '{username}': {feed_url}")
else:
print_error(f"Failed to add feed to user '{username}'")
+21 -15
src/thicket/cli/commands/list_cmd.py
···
load_config,
print_error,
print_feeds_table,
+
print_feeds_table_from_git,
print_info,
print_users_table,
+
print_users_table_from_git,
)
···
# Load configuration
config = load_config(config_file)
+
# Initialize Git store
+
git_store = GitStore(config.git_store)
+
if what == "users":
-
list_users(config)
+
list_users(git_store)
elif what == "feeds":
-
list_feeds(config, user)
+
list_feeds(git_store, user)
elif what == "entries":
-
list_entries(config, user, limit)
+
list_entries(git_store, user, limit)
else:
print_error(f"Unknown list type: {what}")
print_error("Use 'users', 'feeds', or 'entries'")
raise typer.Exit(1)
-
def list_users(config) -> None:
+
def list_users(git_store: GitStore) -> None:
"""List all users."""
-
if not config.users:
+
index = git_store._load_index()
+
users = list(index.users.values())
+
+
if not users:
print_info("No users configured")
return
-
print_users_table(config)
+
print_users_table_from_git(users)
-
def list_feeds(config, username: Optional[str] = None) -> None:
+
def list_feeds(git_store: GitStore, username: Optional[str] = None) -> None:
"""List feeds, optionally filtered by user."""
if username:
-
user = config.find_user(username)
+
user = git_store.get_user(username)
if not user:
print_error(f"User '{username}' not found")
raise typer.Exit(1)
···
print_info(f"No feeds configured for user '{username}'")
return
-
print_feeds_table(config, username)
+
print_feeds_table_from_git(git_store, username)
-
def list_entries(config, username: Optional[str] = None, limit: Optional[int] = None) -> None:
+
def list_entries(git_store: GitStore, username: Optional[str] = None, limit: Optional[int] = None) -> None:
"""List entries, optionally filtered by user."""
-
# Initialize Git store
-
git_store = GitStore(config.git_store)
-
if username:
# List entries for specific user
-
user = config.find_user(username)
+
user = git_store.get_user(username)
if not user:
print_error(f"User '{username}' not found")
raise typer.Exit(1)
···
all_entries = []
all_usernames = []
-
for user in config.users:
+
index = git_store._load_index()
+
for user in index.users.values():
entries = git_store.list_entries(user.username, limit)
if entries:
all_entries.append(entries)
+15 -14
src/thicket/cli/commands/sync.py
···
# Load configuration
config = load_config(config_file)
-
# Determine which users to sync
+
# Initialize Git store
+
git_store = GitStore(config.git_store)
+
+
# Determine which users to sync from git repository
users_to_sync = []
if all_users:
-
users_to_sync = config.users
+
index = git_store._load_index()
+
users_to_sync = list(index.users.values())
elif user:
-
user_config = config.find_user(user)
-
if not user_config:
-
print_error(f"User '{user}' not found")
+
user_metadata = git_store.get_user(user)
+
if not user_metadata:
+
print_error(f"User '{user}' not found in git repository")
raise typer.Exit(1)
-
users_to_sync = [user_config]
+
users_to_sync = [user_metadata]
else:
print_error("Specify --all to sync all users or --user to sync a specific user")
raise typer.Exit(1)
···
if not users_to_sync:
print_info("No users configured to sync")
return
-
-
# Initialize Git store
-
git_store = GitStore(config.git_store)
# Sync each user
total_new_entries = 0
total_updated_entries = 0
-
for user_config in users_to_sync:
-
print_info(f"Syncing user: {user_config.username}")
+
for user_metadata in users_to_sync:
+
print_info(f"Syncing user: {user_metadata.username}")
user_new_entries = 0
user_updated_entries = 0
# Sync each feed for the user
-
for feed_url in track(user_config.feeds, description=f"Syncing {user_config.username}'s feeds"):
+
for feed_url in track(user_metadata.feeds, description=f"Syncing {user_metadata.username}'s feeds"):
try:
new_entries, updated_entries = asyncio.run(
-
sync_feed(git_store, user_config.username, feed_url, dry_run)
+
sync_feed(git_store, user_metadata.username, feed_url, dry_run)
)
user_new_entries += new_entries
user_updated_entries += updated_entries
···
print_error(f"Failed to sync feed {feed_url}: {e}")
continue
-
print_info(f"User {user_config.username}: {user_new_entries} new, {user_updated_entries} updated")
+
print_info(f"User {user_metadata.username}: {user_new_entries} new, {user_updated_entries} updated")
total_new_entries += user_new_entries
total_updated_entries += user_updated_entries
+1
src/thicket/cli/main.py
···
# Import commands to register them
+
from .commands import add, duplicates, init, list_cmd, sync
if __name__ == "__main__":
app()
+49 -1
src/thicket/cli/utils.py
···
from rich.progress import Progress, SpinnerColumn, TextColumn
from rich.table import Table
-
from ..models import ThicketConfig
+
from ..models import ThicketConfig, UserMetadata
+
from ..core.git_store import GitStore
console = Console()
···
def print_info(message: str) -> None:
"""Print an info message."""
console.print(f"[blue]ℹ[/blue] {message}")
+
+
+
def print_users_table_from_git(users: list[UserMetadata]) -> None:
+
"""Print a table of users from git repository."""
+
table = Table(title="Users and Feeds")
+
table.add_column("Username", style="cyan", no_wrap=True)
+
table.add_column("Display Name", style="magenta")
+
table.add_column("Email", style="blue")
+
table.add_column("Homepage", style="green")
+
table.add_column("Feeds", style="yellow")
+
+
for user in users:
+
feeds_str = "\n".join(user.feeds)
+
table.add_row(
+
user.username,
+
user.display_name or "",
+
user.email or "",
+
user.homepage or "",
+
feeds_str,
+
)
+
+
console.print(table)
+
+
+
def print_feeds_table_from_git(git_store: GitStore, username: Optional[str] = None) -> None:
+
"""Print a table of feeds from git repository."""
+
table = Table(title=f"Feeds{f' for {username}' if username else ''}")
+
table.add_column("Username", style="cyan", no_wrap=True)
+
table.add_column("Feed URL", style="blue")
+
table.add_column("Status", style="green")
+
+
if username:
+
user = git_store.get_user(username)
+
users = [user] if user else []
+
else:
+
index = git_store._load_index()
+
users = list(index.users.values())
+
+
for user in users:
+
for feed in user.feeds:
+
table.add_row(
+
user.username,
+
feed,
+
"Active", # TODO: Add actual status checking
+
)
+
+
console.print(table)
+3 -14
src/thicket/core/git_store.py
···
last_updated=datetime.now(),
)
-
# Save user metadata
-
metadata_path = user_dir / "metadata.json"
-
with open(metadata_path, "w") as f:
-
json.dump(user_metadata.model_dump(mode="json"), f, indent=2, default=str)
# Update index
index.add_user(user_metadata)
···
user.update_timestamp()
-
# Save user metadata
-
user_dir = self.repo_path / user.directory
-
metadata_path = user_dir / "metadata.json"
-
with open(metadata_path, "w") as f:
-
json.dump(user.model_dump(mode="json"), f, indent=2, default=str)
# Update index
index.add_user(user)
···
# Update user metadata if new entry
if not entry_exists:
-
user.increment_entry_count()
-
self.update_user(username, entry_count=user.entry_count)
+
index = self._load_index()
+
index.update_entry_count(username, 1)
+
self._save_index(index)
return True
···
entries = []
entry_files = sorted(user_dir.glob("*.json"), key=lambda p: p.stat().st_mtime, reverse=True)
-
# Filter out metadata.json
-
entry_files = [f for f in entry_files if f.name != "metadata.json"]
if limit:
entry_files = entry_files[:limit]
···
continue
entry_files = user_dir.glob("*.json")
-
entry_files = [f for f in entry_files if f.name != "metadata.json"]
for entry_file in entry_files:
try:
-38
src/thicket/models/config.py
···
git_store: Path
cache_dir: Path
users: list[UserConfig] = []
-
-
def find_user(self, username: str) -> Optional[UserConfig]:
-
"""Find a user by username."""
-
for user in self.users:
-
if user.username == username:
-
return user
-
return None
-
-
def add_user(self, user: UserConfig) -> None:
-
"""Add a new user or update existing user."""
-
existing = self.find_user(user.username)
-
if existing:
-
# Update existing user
-
existing.feeds = list(set(existing.feeds + user.feeds))
-
existing.email = user.email or existing.email
-
existing.homepage = user.homepage or existing.homepage
-
existing.icon = user.icon or existing.icon
-
existing.display_name = user.display_name or existing.display_name
-
else:
-
# Add new user
-
self.users.append(user)
-
-
def remove_user(self, username: str) -> bool:
-
"""Remove a user by username. Returns True if user was found and removed."""
-
for i, user in enumerate(self.users):
-
if user.username == username:
-
del self.users[i]
-
return True
-
return False
-
-
def add_feed_to_user(self, username: str, feed_url: HttpUrl) -> bool:
-
"""Add a feed to an existing user. Returns True if user was found."""
-
user = self.find_user(username)
-
if user:
-
if feed_url not in user.feeds:
-
user.feeds.append(feed_url)
-
return True
-
return False
+5 -6
tests/test_git_store.py
···
# Check that user directory was created
user_dir = store.repo_path / "testuser"
assert user_dir.exists()
-
assert (user_dir / "metadata.json").exists()
-
# Check metadata file content
-
with open(user_dir / "metadata.json") as f:
-
metadata = json.load(f)
-
assert metadata["username"] == "testuser"
-
assert metadata["display_name"] == "Test User"
+
# Check user exists in index
+
stored_user = store.get_user("testuser")
+
assert stored_user is not None
+
assert stored_user.username == "testuser"
+
assert stored_user.display_name == "Test User"
def test_get_user(self, temp_dir):
"""Test getting user metadata."""