initial working state

main.py

@@ -0,0 +1,6 @@
+def main():
+    print("Hello from synclean!")
+
+
+if __name__ == "__main__":
+    main()

src/synclean/__main__.py

@@ -0,0 +1,17 @@
+import sys
+from synclean.cli.commands import cli
+
+
+def main():
+    try:
+        cli()
+    except KeyboardInterrupt:
+        print("\nGoodbye!")
+        sys.exit(0)
+    except Exception as e:
+        print(f"An error occurred: {e}")
+        sys.exit(1)
+
+
+if __name__ == '__main__':
+    main()

src/synclean/api/rooms.py

@@ -0,0 +1,42 @@
+from typing import List
+from urllib.parse import urlencode
+from synclean.api.synapse import SynapseApiClient
+from synclean.models.pagination import PaginationParams, RoomPaginationParams
+from synclean.models.room import Room, RoomListResponse
+from synclean.models.media import MediaURIs
+
+
+class RoomAPI:
+    def __init__(self, client: SynapseApiClient):
+        self.client = client
+
+    def get_rooms_list(self, pagination: RoomPaginationParams = None) -> RoomListResponse:
+        if pagination is None:
+            pagination = RoomPaginationParams()
+
+        params = pagination.to_query_params()
+
+        query_string = urlencode(params)
+        response = self.client.request("GET", "/v1/rooms", params)
+        return RoomListResponse.from_api_response(response)
+
+    def get_room_details(self, room_id: str):
+        response = self.client.request("GET", f"/v1/rooms/{room_id}")
+        return Room.from_api_response(response)
+
+    def get_room_avatar(self, room_id: str):
+        room_details = self.get_room_details(room_id)
+        room_avatar = room_details.avatar
+        if room_avatar:
+            return room_avatar
+
+        return None
+
+    def get_media_from_room(self, room_id):
+        response = self.client.request("GET", f"/v1/room/{room_id}/media")
+        return MediaURIs.from_json(response)
+
+    def delete_room_media_by_id(self, server_name: str,media_id: str):
+        response = self.client.request("DELETE", f"/v1/media/{server_name}/{media_id}")
+        return
+

src/synclean/api/synapse.py

@@ -0,0 +1,28 @@
+from logging import Logger
+from typing import Dict, Any
+
+import requests
+from synclean.models.config import SynapseConfig
+from synclean.utils.exceptions import AuthenticationError, APIError
+
+
+class SynapseApiClient:
+    """Synapse API endpoints."""
+
+    def __init__(self, config: SynapseConfig) -> None:
+        self.config = config
+        self.logger = Logger(__name__)
+        self.session = requests.Session()
+        self.session.headers.update(config.headers)
+
+    def request(self, method: str, endpoint: str, params = None) -> Dict[str, Any]:
+        """Send a request to the Synapse API."""
+        url = f"{self.config.base_url}{endpoint}"
+        try:
+            response = self.session.request(method, url, params)
+            response.raise_for_status()
+            return response.json()
+        except requests.exceptions.HTTPError as e:
+            if e.response.status_code == 401:
+                raise AuthenticationError(f"Authentication failed: {e}") from e
+            raise APIError(f"Network error: {e}") from e

src/synclean/api/users.py

@@ -0,0 +1,19 @@
+from synclean.api.synapse import SynapseApiClient
+from synclean.models.pagination import UserPaginationParams
+from synclean.models.user import UserMediaList
+
+
+class UserAPI:
+    def __init__(self, client: SynapseApiClient):
+        self.client = client
+
+    def get_users_list_by_media(self, pagination: UserPaginationParams = None) -> UserMediaList:
+        if pagination is None:
+            pagination = UserPaginationParams()
+
+        params = pagination.to_query_params()
+
+        response = self.client.request("GET", "/v1/statistics/users/media", params)
+        return UserMediaList.from_api_response(response)
+
+

src/synclean/cli/commands.py

@@ -0,0 +1,180 @@
+from typing import Optional
+import click
+from synclean.api.rooms import RoomAPI
+from synclean.api.synapse import SynapseApiClient
+from synclean.api.users import UserAPI
+from synclean.config.settings import SettingsManager
+from synclean.models.enums import RoomOrderBy, Direction, UserOrderBy
+from synclean.models.pagination import PaginationParams, RoomPaginationParams, UserPaginationParams
+
+pass_api = click.make_pass_decorator(SynapseApiClient)
+
+@click.group()
+@click.pass_context
+def cli(ctx):
+    """Synclean CLI."""
+    settings = SettingsManager()
+    config = settings.load_config()
+    ctx.obj = SynapseApiClient(config)
+
+
+@cli.group()
+def room():
+    """Room commands."""
+    pass
+
+@cli.group()
+def user():
+    """User commands."""
+    pass
+
+@room.command("ls")
+@click.option("--limit", "-l", default=100, help="Number of rooms to display per page.")
+@click.option("--offset", "-o", default=0, help="Number of the room to start from.")
+@click.option("--search", "-s", help="Search term to filter rooms by.")
+@click.option("--order-by", "-O",
+              type=click.Choice(["name", "joined_members", "canonical_alias", "creator", "version",]),
+              default="name", help="Field to order result by.")
+@click.option("--direction", "-d",
+              type=click.Choice(["f", "b"]),
+              default="b", help="Direction to order result by. (f=forward, b=backward)")
+@pass_api
+def room_ls(
+        api: SynapseApiClient,
+        limit: int,
+        offset: int,
+        search: Optional[str],
+        order_by: str,
+        direction: str
+):
+    order_by_enum = RoomOrderBy(order_by)
+    direction_enum = Direction(direction)
+
+    rooms_api = RoomAPI(api)
+
+    while True:
+        pagination_params = RoomPaginationParams(limit, offset, order_by_enum, direction_enum, search)
+
+        rooms = rooms_api.get_rooms_list(pagination_params)
+        if rooms:
+            for r in rooms.rooms:
+                click.echo(f"Name: {r.name}")
+                click.echo(f"ID: {r.room_id}")
+                click.echo(f"Joined Members: {r.joined_members}")
+                click.echo("=" * 50)
+        else:
+            click.echo("No rooms found.")
+
+        offset += limit
+
+        choice = input("Next page? (y/n): ")
+        if choice.lower() != "y":
+            break
+
+@room.command("show")
+@click.argument("room_id")
+@click.option("--media", is_flag=True, help="List media in the room")
+@click.option("--rm-all", is_flag=True, help="Remove all media in the room")
+@click.option("--rm", "remove_media_id", help="Remove media by ID")
+@click.option("--avatar", is_flag=True, help="Show avatar URI")
+@pass_api
+def show_room(
+        api: SynapseApiClient,
+        room_id: str,
+        media: bool,
+        rm_all: bool,
+        remove_media_id: Optional[str],
+        avatar: bool
+):
+    if room_id is None:
+        click.echo("You must specify a room ID.", err=True)
+        return
+
+
+
+    if rm_all or remove_media_id:
+        if not media:
+            click.echo("You must specify --media to remove media.")
+            return
+
+    room_api = RoomAPI(api)
+    room_details = room_api.get_room_details(room_id)
+
+    if not room_details:
+        click.echo(f"Error: Room {room_id} not found.", err=True)
+        return
+
+    if avatar:
+        if not room_details.avatar:
+            click.echo("Room has no avatar.")
+            return
+        click.echo(room_details.avatar)
+        # TODO: implement adding avatar to blacklist
+
+    if rm_all:
+        if click.confirm("Are you sure you want to remove all media in this room?"):
+            # TODO: implement removing all media from the room
+            pass
+
+    if remove_media_id:
+        if click.confirm(f"Are you sure you want to remove media {remove_media_id} from the room?"):
+            # TODO: implement removing media by ID
+            pass
+
+    if not media:
+        click.echo(f"Room details: {room_id}")
+        click.echo("=" * 40)
+        click.echo(f"Name: {room_details.name}")
+        click.echo(f"Avatar: {room_details.avatar}")
+        click.echo(f"ID: {room_details.room_id}")
+        click.echo(f"Members: {room_details.joined_members} (Local: {room_details.joined_local_members})")
+        click.echo(f"Creator: {room_details.creator}")
+        click.echo(f"Encryption: {room_details.encryption}")
+        click.echo(f"Public: {room_details.public}")
+        click.echo(f"Guest Access: {room_details.guest_access}")
+        click.echo(f"History Visibility: {room_details.history_visibility}")
+        click.echo(f"State Events: {room_details.state_events}")
+        click.echo(f"Room Type: {room_details.room_type}")
+    else:
+        room_media = room_api.get_media_from_room(room_id)
+        if room_media:
+            for uri in room_media.all_uris:
+                click.echo(f"URI: {uri}")
+
+@user.command("ls")
+@click.option("--limit", "-l", default=10, help="Number of users to list.")
+@click.option("--offset", "-o", default=0, help="Number of the room to start from.")
+@click.option("--search", "-s", help="Search term to filter by.")
+@click.option("--order-by", "-O",
+              type=click.Choice(["user_id", "display_name", "media_length", "media_count"]),
+              default="media_length", help="Field to order result by.")
+@click.option("--direction", "-d",
+              type=click.Choice(["f", "b"]),
+              default="b", help="Direction to order result by. (f=forward, b=backward)")
+@pass_api
+def user_ls(
+        api: SynapseApiClient,
+        limit: int,
+        offset: int,
+        search: Optional[str],
+        order_by: str,
+        direction: str
+):
+
+    order_by_enum = UserOrderBy(order_by)
+    direction_enum = Direction(direction)
+
+    users_api = UserAPI(api)
+
+    while True:
+        pagination_params = UserPaginationParams(limit, offset, order_by_enum, direction_enum, search)
+
+        users = users_api.get_users_list_by_media(pagination_params)
+
+        if users:
+            for u in users.users:
+                click.echo(f"Display name: {u.display_name}")
+                click.echo(f"Media count: {u.media_count}")
+                click.echo(f"Media length: {u.media_length_mb()}MB")
+                click.echo(f"User ID: {u.user_id}")
+                click.echo("=" * 40)

src/synclean/config/paths.py

@@ -0,0 +1,24 @@
+from pathlib import Path
+import platformdirs
+
+
+APP_NAME = "synclean"
+
+
+def get_app_dir() -> Path:
+    """Get platform-specific application directory"""
+    app_dir = Path(platformdirs.user_data_dir(APP_NAME))
+    app_dir.mkdir(parents=True, exist_ok=True)
+    return app_dir
+
+
+def get_config_path() -> Path:
+    return get_app_dir() / "config.yaml"
+
+
+def get_user_avatars_path() -> Path:
+    return get_app_dir() / "users.json"
+
+
+def get_room_avatars_path() -> Path:
+    return get_app_dir() / "rooms.json"

src/synclean/config/settings.py

@@ -0,0 +1,57 @@
+from pathlib import Path
+from typing import List
+import yaml
+from .paths import get_config_path
+from synclean.models.config import SynapseConfig
+
+
+class SettingsManager:
+    """Manage Synapse configuration persistence in YAML format."""
+    DEFAULT_PATH = get_config_path()
+
+    def __init__(self, config_path: Path = DEFAULT_PATH) -> None:
+        self.config_path = config_path
+
+    def load_config(self) -> SynapseConfig:
+        """Load Synapse configuration from YAML file with validation."""
+        try:
+            with self.config_path.open('r', encoding="utf-8") as f:
+                config_data = yaml.safe_load(f)
+        except FileNotFoundError:
+            raise FileNotFoundError(f"Config file not found: {self.config_path}") from None
+        except yaml.YAMLError as e:
+            raise ValueError(f"Invalid YAML syntax: {e}") from e
+
+        if not config_data:
+            raise ValueError("Config file is empty")
+
+        return SynapseConfig.from_dict(config_data)
+
+    def save_config(self, config: SynapseConfig) -> None:
+        """Save configuration to YAML file with safety checks."""
+        config_dict = config.to_dict()
+
+        self.config_path.parent.mkdir(parents=True, exist_ok=True)
+
+        with self.config_path.open('w', encoding='utf-8') as f:
+            yaml.safe_dump(
+                config_dict,
+                f,
+                sort_keys=False,
+                default_flow_style=None,
+                allow_unicode=True
+            )
+
+        self.config_path.chmod(0o600)
+
+    def update_blacklist(self, add_usernames: List[str] = None,
+                         remove_usernames: List[str] = None) -> None:
+        """Update blacklist in the config file."""
+        config = self.load_config()
+        if add_usernames:
+            config.add_to_blacklist(*add_usernames)
+
+        if remove_usernames:
+            config.remove_from_blacklist(*remove_usernames)
+
+        self.save_config(config)

src/synclean/models/config.py

@@ -0,0 +1,83 @@
+from dataclasses import dataclass, field
+from typing import Dict, Set, Union
+from urllib.parse import urlparse
+
+
+@dataclass
+class SynapseConfig:
+    """Store and validate configuration for Synapse API connection."""
+    homeserver: str
+    access_token: str
+    blacklist: Set[str] = field(default_factory=set)
+
+    def __post_init__(self) -> None:
+        """Validate Synapse config after initialization."""
+        if not self.homeserver.strip():
+            raise ValueError("Homeserver URL cannot be empty")
+        parsed = urlparse(self.homeserver)
+        if not all([parsed.scheme, parsed.netloc]):
+            raise ValueError(f"Invalid homeserver URL: {self.homeserver}")
+
+        if not self.access_token.strip():
+            raise ValueError("Access token cannot be empty")
+
+        if not isinstance(self.blacklist, set):
+            raise TypeError("Blacklist must be a set")
+        for username in self.blacklist:
+            if not isinstance(username, str) or not username.strip():
+                raise ValueError("Blacklist contains invalid username")
+
+    @property
+    def base_url(self) -> str:
+        """Get normalized base URL with trailing slash."""
+        return self.homeserver.rstrip('/') + '/_synapse/admin'
+
+    @property
+    def headers(self) -> Dict[str, str]:
+        """Generate request headers with authentication."""
+        return {
+            "Authorization": f"Bearer {self.access_token}",
+            "Content-Type": "application/json",
+            "Accept": "application/json"
+        }
+
+    def add_to_blacklist(self, *usernames: str) -> None:
+        """Add one or more username to the blacklist."""
+        for username in usernames:
+            username = username.strip()
+            if not username:
+                raise ValueError("Username cannot be empty")
+            self.blacklist.add(username)
+
+    def remove_from_blacklist(self, *usernames: str) -> None:
+        """Remove one or more username from the blacklist."""
+        for username in usernames:
+            self.blacklist.discard(username)
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, str]) -> "SynapseConfig":
+        """Create from dictionary with validation."""
+        try:
+            blacklist = set(data.get("blacklist", []))
+            return cls(
+                homeserver=str(data["homeserver"]),
+                access_token=str(data["access_token"]),
+                blacklist=blacklist
+            )
+        except KeyError as e:
+            raise ValueError(f"Missing required field: {e}") from e
+        except (TypeError, ValueError) as e:
+            raise ValueError(f"Invalid data type: {e}") from e
+
+    def to_dict(self) -> Dict[str, Union[str, Set[str]]]:
+        """Serialize to dictionary without sensitive exposure."""
+        return {
+            "homeserver": self.homeserver,
+            "access_token": self.access_token,
+            "blacklist": sorted(self.blacklist)
+        }
+
+    def __str__(self) -> str:
+        """Safe representation excluding access token."""
+        return (f"SynapseConfig(homeserver={self.homeserver!r},"
+                f"blacklist={len(self.blacklist)})")

src/synclean/models/enums.py

@@ -0,0 +1,30 @@
+from enum import Enum
+
+
+class RoomOrderBy(Enum):
+    """Available room ordering options."""
+    NAME = "name"
+    CANONICAL_ALIAS = "canonical_alias"
+    JOINED_MEMBERS = "joined_members"
+    JOINED_LOCAL_MEMBERS = "joined_local_members"
+    VERSION = "version"
+    CREATOR = "creator"
+    ENCRYPTION = "encryption"
+    FEDERATABLE = "federatable"
+    JOIN_RULE = "join_rule"
+    GUEST_ACCESS = "guest_access"
+    HISTORY_VISIBILITY = "history_visibility"
+    STATE_EVENTS = "state_events"
+
+class UserOrderBy(Enum):
+    """Available user ordering options."""
+    USER_ID = "user_id"
+    DISPLAY_NAME = "display_name"
+    MEDIA_LENGTH = "media_length"
+    MEDIA_COUNT = "media_count"
+
+
+class Direction(Enum):
+    """Sort direction."""
+    FORWARD = "f"
+    BACKWARD = "b"

src/synclean/models/media.py

@@ -0,0 +1,29 @@
+from dataclasses import dataclass, field
+from typing import List, Set
+
+
+@dataclass
+class MediaURIs:
+    local: List[str] = field(default_factory=list)
+    remote: List[str] = field(default_factory=list)
+
+    @classmethod
+    def from_json(cls, data: dict) -> "MediaURIs":
+        return cls(
+            local=data.get("local", []),
+            remote=data.get("remote", [])
+        )
+
+    def contains_avatar(self, avatar_uri: str) -> bool:
+        """Check if the media URIs contain the avatar URI."""
+        return avatar_uri in self.all_uris
+
+    @property
+    def all_uris(self) -> Set[str]:
+        """Get all media URIs as a set."""
+        return set(self.local) | set(self.remote)
+
+    @property
+    def total_count(self) -> int:
+        """Get total media count."""
+        return len(self.local) + len(self.remote)

src/synclean/models/pagination.py

@@ -0,0 +1,41 @@
+from dataclasses import dataclass
+from typing import Optional, Dict, Any, TypeVar, Generic
+from synclean.models.enums import RoomOrderBy, Direction, UserOrderBy
+
+OrderByType = TypeVar("OrderByType")
+
+
+@dataclass
+class PaginationParams(Generic[OrderByType]):
+    """Parameters for pagination."""
+    limit: int = 100
+    offset: int = 0
+    order_by: OrderByType = None
+    direction: Direction = Direction.FORWARD
+    search_term: Optional[str] = None
+
+    def to_query_params(self) -> Dict[str, Any]:
+        """Convert to query parameters for API request."""
+        params = {
+            'limit': self.limit,
+            'from': self.offset,
+            'order_by': self.order_by.value,
+            'dir': self.direction.value
+        }
+
+        if self.search_term:
+            params['search_term'] = self.search_term
+
+        return params
+
+
+@dataclass
+class RoomPaginationParams(PaginationParams[RoomOrderBy]):
+    """Pagination parameters for rooms."""
+    order_by = RoomOrderBy = RoomOrderBy.NAME
+
+
+@dataclass
+class UserPaginationParams(PaginationParams[UserOrderBy]):
+    """Pagination parameters for users."""
+    order_by = UserOrderBy = UserOrderBy.MEDIA_LENGTH

src/synclean/models/room.py

@@ -0,0 +1,67 @@
+from dataclasses import dataclass
+from typing import Dict, Any, Optional, List
+
+
+@dataclass
+class Room:
+    """Dataclass representing a Matrix room"""
+    room_id: str
+    name: Optional[str] = None
+    canonical_alias: Optional[str] = None
+    joined_members: int = 0
+    joined_local_members: int = 0
+    version: Optional[str] = None
+    creator: Optional[str] = None
+    encryption: Optional[str] = None
+    federatable: bool = True
+    public: bool = True
+    join_rules: Optional[str] = None
+    guest_access: Optional[str] = None
+    history_visibility: Optional[str] = None
+    state_events: int = 0
+    room_type: Optional[str] = None
+
+    avatar: Optional[str] = None
+
+    @classmethod
+    def from_api_response(cls, data: Dict[str, Any]) -> "Room":
+        """Create a Room object from a dictionary."""
+        return cls(
+            room_id=str(data['room_id']),
+            name=data.get('name'),
+            canonical_alias=data.get('canonical_alias'),
+            joined_members=data.get('joined_members', 0),
+            joined_local_members=data.get('joined_local_members', 0),
+            version=data.get('version'),
+            creator=data.get('creator'),
+            encryption=data.get('encryption'),
+            federatable=data.get('federatable', True),
+            public=data.get('public', True),
+            join_rules=data.get('join_rules'),
+            guest_access=data.get('guest_access'),
+            history_visibility=data.get('history_visibility'),
+            state_events=data.get('state_events', 0),
+            room_type=data.get('room_type'),
+            avatar=data.get('avatar'),
+        )
+
+
+@dataclass
+class RoomListResponse:
+    """Response from room list API"""
+    rooms: List[Room]
+    next_batch: Optional[int]
+    prev_batch: Optional[int]
+    total_rooms: int
+
+    @classmethod
+    def from_api_response(cls, data: Dict[str, Any]) -> "RoomListResponse":
+        """Create RoomListResponse from API response."""
+        rooms = [Room.from_api_response(room_data) for room_data in data.get('rooms', [])]
+
+        return cls(
+            rooms=rooms,
+            next_batch=data.get('next_batch'),
+            prev_batch=data.get('prev_batch'),
+            total_rooms=data.get('total_rooms', len(rooms))
+        )

src/synclean/models/user.py

@@ -0,0 +1,42 @@
+from dataclasses import dataclass
+from typing import List, Optional, Any, Dict
+
+
+@dataclass
+class UserMedia:
+    display_name: str
+    media_count: int
+    media_length: int
+    user_id: str
+
+    def media_length_mb(self, precision: int = 2) -> float:
+        """Convert media length to MB."""
+        mb = self.media_length / (1024 * 1024)
+        return round(mb, precision)
+
+    @classmethod
+    def from_api_response(cls, data: Dict[str, Any]) -> "UserMedia":
+        """Create UserMedia from API response."""
+        return cls(
+            display_name=data.get("displayname"),
+            media_count=data.get("media_count"),
+            media_length=data.get("media_length"),
+            user_id=data.get("user_id")
+        )
+
+
+@dataclass
+class UserMediaList:
+    users: List[UserMedia]
+    total: int
+    next_token: Optional[int] = None
+
+    @classmethod
+    def from_api_response(cls, data: Dict[str, Any]) -> "UserMediaList":
+        """Create UserMediaList from API response."""
+        users = [UserMedia.from_api_response(user_data) for user_data in data.get("users", [])]
+        return cls(
+            users=users,
+            total=data.get("total", 0),
+            next_token=data.get("next_token")
+        )

src/synclean/utils/exceptions.py

@@ -0,0 +1,23 @@
+"""Custom exceptions for Synclean."""
+
+class SyncleanException(Exception):
+    """Base class for Synclean exceptions."""
+    pass
+
+class APIError(SyncleanException):
+    """Exception for API errors."""
+    def __init__(self, message: str, status_code: int = None, error_code: str = None):
+        super().__init__(message)
+        self.status_code = status_code
+        self.error_code = error_code
+
+class AuthenticationError(SyncleanException):
+    """Exception for authentication errors."""
+
+class MediaNotFoundError(SyncleanException):
+    """Exception for media not found errors."""
+    pass
+
+class InvalidConfigurationError(SyncleanException):
+    """Exception for invalid configuration errors."""
+    pass