diff --git a/main.py b/main.py new file mode 100644 index 0000000..7767960 --- /dev/null +++ b/main.py @@ -0,0 +1,6 @@ +def main(): + print("Hello from synclean!") + + +if __name__ == "__main__": + main() diff --git a/src/synclean/__main__.py b/src/synclean/__main__.py new file mode 100644 index 0000000..0b9b5a2 --- /dev/null +++ b/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() \ No newline at end of file diff --git a/src/synclean/api/rooms.py b/src/synclean/api/rooms.py new file mode 100644 index 0000000..2137c89 --- /dev/null +++ b/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 + diff --git a/src/synclean/api/synapse.py b/src/synclean/api/synapse.py new file mode 100644 index 0000000..5290bd1 --- /dev/null +++ b/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 \ No newline at end of file diff --git a/src/synclean/api/users.py b/src/synclean/api/users.py new file mode 100644 index 0000000..c04495c --- /dev/null +++ b/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) + + diff --git a/src/synclean/cli/commands.py b/src/synclean/cli/commands.py new file mode 100644 index 0000000..87c7653 --- /dev/null +++ b/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) diff --git a/src/synclean/config/paths.py b/src/synclean/config/paths.py new file mode 100644 index 0000000..b7f2d2f --- /dev/null +++ b/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" \ No newline at end of file diff --git a/src/synclean/config/settings.py b/src/synclean/config/settings.py new file mode 100644 index 0000000..9fcf8a9 --- /dev/null +++ b/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) diff --git a/src/synclean/models/config.py b/src/synclean/models/config.py new file mode 100644 index 0000000..eb5ea29 --- /dev/null +++ b/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)})") \ No newline at end of file diff --git a/src/synclean/models/enums.py b/src/synclean/models/enums.py new file mode 100644 index 0000000..5b22a77 --- /dev/null +++ b/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" diff --git a/src/synclean/models/media.py b/src/synclean/models/media.py new file mode 100644 index 0000000..b390415 --- /dev/null +++ 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) diff --git a/src/synclean/models/pagination.py b/src/synclean/models/pagination.py new file mode 100644 index 0000000..ecf73c8 --- /dev/null +++ b/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 \ No newline at end of file diff --git a/src/synclean/models/room.py b/src/synclean/models/room.py new file mode 100644 index 0000000..3cd89b0 --- /dev/null +++ b/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)) + ) diff --git a/src/synclean/models/user.py b/src/synclean/models/user.py new file mode 100644 index 0000000..90d7dbf --- /dev/null +++ b/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") + ) \ No newline at end of file diff --git a/src/synclean/utils/exceptions.py b/src/synclean/utils/exceptions.py new file mode 100644 index 0000000..092c6d4 --- /dev/null +++ b/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 \ No newline at end of file