cligram.utils

class cligram.utils.Archive(password=None, compression=CompressionType.GZIP, salt=None)

Bases: object

High-performance async archive with encryption and streaming support.

The archive data is held in memory by default for fast operations. Decryption happens only on load, encryption only on export.

Examples

``` # Create archive from directory and save to file async with Archive.from_directory(“my_folder”, password=”secret”) as archive:

await archive.write(“archive.tar.gz”)

# Load archive from file and extract async with await Archive.load(“archive.tar.gz”, password=”secret”) as archive:

await archive.extract(“output_folder”)

# Create archive from bytes data = b”…” # Encrypted archive bytes async with await Archive.from_bytes(data, password=”secret”) as archive:

file_content = await archive.get_file(“document.txt”)

```

add_bytes(name, data, mode=420, pax_headers=None)

Add file from bytes to archive.

Parameters:

• name (str) – Name in archive

• data (bytes) – File content

• mode (int) – File mode (default: 0o644)

• pax_headers (Optional[Dict[str, str]]) – Optional PAX extended attributes

Return type:

ArchiveEntry

Returns:

Created ArchiveEntry

async add_directory(dir_path, arcname=None, pax_headers=None)

Add directory to archive.

Parameters:

• dir_path (Union[str, Path]) – Path to directory

• arcname (Optional[str]) – Name in archive (defaults to directory name)

• pax_headers (Optional[Dict[str, str]]) – Optional PAX extended attributes

Return type:

List[ArchiveEntry]

Returns:

List of created ArchiveEntry objects

async add_file(file_path, arcname=None, pax_headers=None)

Add file to archive.

Parameters:

• file_path (Union[str, Path]) – Path to file

• arcname (Optional[str]) – Name in archive (defaults to filename)

• pax_headers (Optional[Dict[str, str]]) – Optional PAX extended attributes

Return type:

ArchiveEntry

Returns:

Created ArchiveEntry

clear()

Clear archive contents.

Return type:

None

close()

Close the archive and release resources.

Return type:

None

async extract(output_dir='.')

Extract archive to directory.

Parameters:

output_dir (Union[str, Path]) – Directory to extract to

Return type:

Path

Returns:

Path to extracted directory

async classmethod from_base64(b64_string, **kwargs)

Load archive from base64 string.

Parameters:

• b64_string (str) – Base64 encoded archive

• **kwargs – Additional arguments for Archive constructor

Return type:

Archive

Returns:

Archive instance with loaded data

async classmethod from_bytes(data, **kwargs)

Load archive from bytes.

Parameters:

• data (bytes) – Archive data bytes

• **kwargs – Additional arguments for Archive constructor

Return type:

Archive

Returns:

Archive instance with loaded data

async classmethod from_directory(directory, **kwargs)

Create archive from directory.

Parameters:

• directory (Union[str, Path]) – Directory to archive

• **kwargs – Additional arguments for Archive constructor

Return type:

Archive

Returns:

Archive instance with directory contents

get_entry(name)

Get archive entry by name.

Parameters:

name (str) – Entry name

Return type:

ArchiveEntry

Returns:

ArchiveEntry

get_file(name)

Get content of a file from archive.

Parameters:

name (str) – Path of file within archive

Return type:

bytes

Returns:

File content as bytes

get_file_count()

Get number of files in archive.

Return type:

int

get_salt()

Get the salt used for encryption.

Return type:

Optional[bytes]

get_size()

Get total size of archive contents in bytes.

Return type:

int

has_file(name)

Check if archive contains a file.

Return type:

bool

is_empty()

Check if archive is empty.

Return type:

bool

list_file_names()

List all entry names in archive.

Return type:

List[str]

Returns:

List of entry names

list_files()

List all entries in archive.

Return type:

List[ArchiveEntry]

Returns:

List of ArchiveEntry objects

async classmethod load(path, **kwargs)

Load archive from file.

Parameters:

• path (Union[str, Path]) – Path to archive file

• **kwargs – Additional arguments for Archive constructor

Return type:

Archive

Returns:

Archive instance with loaded data

async read(path)

Read archive from file.

Parameters:

path (Union[str, Path]) – Path to archive file

Return type:

None

remove_file(name)

Remove file from archive.

Parameters:

name (str) – Entry name

Return type:

ArchiveEntry

Returns:

Removed ArchiveEntry

async to_base64()

Get archive as base64 string.

Return type:

str

Returns:

Base64 encoded archive

async to_bytes()

Get archive as bytes.

Return type:

bytes

Returns:

Archive bytes

async write(path)

Write archive to file.

Parameters:

path (Union[str, Path]) – Output path

Return type:

int

Returns:

Number of bytes written

class cligram.utils.ArchiveEntry(name, size, file_type, mode, mtime, _content=None, uid=0, gid=0, _uname='', _gname='', pax_headers=<factory>)

Bases: object

Represents an entry in the archive.

This is an immutable value object that represents a file, directory, or symlink in an archive. Content hash is computed lazily on first access and cached.

property content: bytes | None

Get content bytes.

Returns:

File content or None for non-file entries

property content_hash: bytes | None

Get SHA-256 hash of the content.

Hash is computed once and cached for performance.

Returns:

SHA-256 hash bytes or None for non-file entries

file_type: FileType

Type of the entry.

async classmethod from_file(file_path, arcname=None, pax_headers=None)

Create ArchiveEntry from file system path.

Parameters:

• file_path (Path) – Path to file or directory

• arcname (Optional[str]) – Name to use in archive (defaults to file name)

• pax_headers (Optional[dict]) – Optional PAX headers to include

Return type:

ArchiveEntry

Returns:

New ArchiveEntry instance

classmethod from_tar_member(member, content=None)

Create ArchiveEntry from TarInfo.

Parameters:

• member (TarInfo) – TarInfo object

• content (Optional[bytes]) – File content bytes (required for files)

Return type:

ArchiveEntry

Returns:

New ArchiveEntry instance

gid: int

Group ID of the entry.

property gname: str

Get group name (lazy computation with fallback).

Returns:

Group name or empty string if unavailable

mode: int

File mode/permissions.

mtime: datetime

Last modification time.

name: str

Entry name/path in the archive.

pax_headers: Dict[str, str]

PAX extended attributes/headers.

size: int

Size of the entry in bytes.

to_dict()

Convert to dictionary representation.

Return type:

dict

Returns:

Dictionary with entry metadata (excludes content)

to_tar_info()

Convert to TarInfo for writing to tar archive.

Return type:

TarInfo

Returns:

TarInfo object ready for tar.addfile()

uid: int

User ID of the entry.

property uname: str

Get user name (lazy computation with fallback).

Returns:

User name or empty string if unavailable

class cligram.utils.DeviceInfo(platform, architecture, name, version, model, environments)

Bases: object

property is_ci: bool

Check if running in a CI environment.

property is_virtual: bool

Check if running in a virtual environment.

cligram.utils.get_client(config, device, proxy, session)

Create a Telethon TelegramClient from the given configuration.

Return type:

TelegramClient

cligram.utils.get_device_info(no_cache=False)

Get comprehensive device information across all supported platforms.

Returns:

Complete device information including platform, architecture, and environment.

Return type:

DeviceInfo

cligram.utils.get_entity_name(entity)

Get the display name of a Telegram entity.

cligram.utils.get_session(config, create=False, device=None)

Load a CustomSession based on the configuration.

Return type:

CustomSession

cligram.utils.get_status(entity)

Get the status of a Telegram user.

cligram.utils.shorten_path(path)

Shorten a file path for display purposes.

Return type:

str

cligram.utils.validate_proxy(proxy)

Validate if the given proxy is valid.

Return type:

bool