Compare commits

...

102 commits

Author SHA1 Message Date
9835028199
−check
All checks were successful
/ job (push) Successful in 1m9s
2024-10-13 21:46:54 +02:00
1c4f52007c
build & workflow
Some checks failed
/ job (push) Failing after 1m8s
2024-10-13 21:42:16 +02:00
eb3c9814bb
mv 2024-09-23 18:49:34 +02:00
daccf01b8c
read_file_dict 2024-09-23 18:48:44 +02:00
5eae0334de
fs.read_file_object 2024-09-23 17:25:57 +02:00
edd8e15978
freetube.profiles 2024-09-23 14:59:00 +02:00
a92a3a786a
freetube.playlists 2024-09-23 14:46:13 +02:00
46ec104b75
freetube.videos 2024-09-23 14:35:32 +02:00
6ece695be4
channel.uid 2024-09-23 14:35:16 +02:00
2a01ef0066
freetube.channels 2024-09-23 14:24:58 +02:00
45df475195
freetube.settings 2024-09-23 13:27:16 +02:00
19153b3bc1
freetube.db 2024-09-23 13:04:56 +02:00
e61be6cf4f
sw.freetube 2024-09-23 12:53:28 +02:00
2327d5388d
doc/get_path_uuid 2024-09-17 23:32:51 +02:00
094d66bc33
doc/read_file_ 2024-09-17 23:23:38 +02:00
6fab5ce9b4
doc/make_directory 2024-09-17 23:19:29 +02:00
0d77038392
doc/create_image 2024-09-17 23:17:36 +02:00
2c6bec253c
doc/empty_file 2024-09-17 23:07:41 +02:00
eb0f862125
doc/write 2024-09-17 23:05:00 +02:00
b05de437d0
doc/wipe 2024-09-17 22:42:19 +02:00
313cc30aec
doc/ps 2024-09-17 22:35:05 +02:00
149ed2dc3b
doc/log 2024-09-17 22:22:03 +02:00
e9d228ba2c
doc/os 2024-09-17 22:17:41 +02:00
3940d32195
doc/pm 2024-09-17 22:14:01 +02:00
2e00140e82
doc/deb 2024-09-17 22:08:14 +02:00
b9754b5dde
doc/cmd 2024-09-17 21:32:08 +02:00
012b10f7d5
doc/grub 2024-09-17 14:53:21 +02:00
d6e112d6c5
doc/debian 2024-09-17 12:54:41 +02:00
622122cdef
doc/prj 2024-09-17 12:53:17 +02:00
e5bb634cf4
−str 2024-09-16 21:45:21 +02:00
8227524a7c
class/str 2024-09-16 21:41:45 +02:00
487d7cb9b2
str 2024-09-16 12:04:43 +02:00
b7eec788f7
−repr 2024-09-16 11:35:29 +02:00
450e10e2f4
class,os 2024-09-16 11:14:40 +02:00
ca88e5bc1d
doc/quiet 2024-09-16 01:17:55 +02:00
26fea93bdb
stack 2024-09-15 23:40:15 +02:00
c300ab86d7
readme/md 2024-09-15 23:35:51 +02:00
2f0c3a0487
readme/wipe 2024-09-15 23:05:05 +02:00
db47d5c80a
doc/arg.split 2024-09-15 15:07:47 +02:00
04eaae7ba3
pyproject/pydoclint 2024-09-15 15:03:13 +02:00
5a25c6df27
command/repr 2024-09-15 02:02:36 +02:00
f8567686fd
project/repr 2024-09-15 01:59:00 +02:00
63c179a0a4
pm/clean 2024-09-15 01:49:39 +02:00
304c2bc617
project/str 2024-09-15 01:46:25 +02:00
5038e587af
command/str 2024-09-15 01:45:06 +02:00
73c7ac00f4
os/str 2024-09-15 01:37:45 +02:00
f2991cff04
pm/str 2024-09-15 01:35:33 +02:00
0ad5cc9701
cyclic 2024-09-14 23:40:39 +02:00
6f9ba7f8f7
imports 2024-09-14 18:16:40 +02:00
4d8c1d7aab
lint 2024-09-14 18:15:42 +02:00
148f757d5d
command/wip 2024-09-14 18:07:01 +02:00
f4498f691c
pm,ps 2024-09-14 17:43:37 +02:00
c5930b4107
os.os 2024-09-14 16:24:10 +02:00
e75a624c46
abc 2024-09-14 15:45:44 +02:00
9c1136cfa0
staticmethod 2024-09-14 15:41:56 +02:00
909e652f0d
os/debian 2024-09-14 15:23:06 +02:00
4e369df232
mypy/fs 2024-09-14 02:47:52 +02:00
cc0b131a56
py.typed 2024-09-14 02:39:52 +02:00
d4d6098241
mypy/grub 2024-09-14 00:17:17 +02:00
68cbe3cd88
mypy/ps 2024-09-14 00:13:59 +02:00
2ad223fa83
grub/extra 2024-09-14 00:07:34 +02:00
51b4f0f5f2
mypy/squashfs 2024-09-13 23:57:35 +02:00
378a212786
mypy/deb,fs,grub 2024-09-13 23:55:02 +02:00
cd4e7403ae
ps/str,... 2024-09-13 23:54:32 +02:00
521884102e
os 2024-09-13 23:41:19 +02:00
4fac21da08
rwx/squashfs 2024-09-13 23:40:53 +02:00
56404078f7
rwx/deb,grub 2024-09-13 23:38:08 +02:00
114ee61102
mypy/fs,main 2024-09-13 23:22:12 +02:00
e213285987
mypy/fs,sphinx 2024-09-13 23:17:09 +02:00
3a8b239b9f
mypy/cmd 2024-09-13 23:07:39 +02:00
17b16ec9d0
mypy/prj 2024-09-13 23:02:29 +02:00
b1970c1f78
main/rwx 2024-09-13 22:55:45 +02:00
0e2b638bbb
run/line 2024-09-13 21:11:24 +02:00
c024c553b0
fix 2024-09-13 20:52:33 +02:00
07699b08eb
ps/items 2024-09-13 20:47:49 +02:00
bf1b3402ae
log/template 2024-09-13 19:24:11 +02:00
78dd431ca5
read/lines 2024-09-13 18:58:57 +02:00
b3e66ec1c3
read/text 2024-09-13 18:57:49 +02:00
065d647e51
read/bytes 2024-09-13 18:55:31 +02:00
5e3c8e93ff
sphinx/ 2024-09-13 17:41:49 +02:00
cfa9ca9e3c
prj/Path 2024-09-13 17:39:39 +02:00
de80d7ecff
fs/mkdir 2024-09-13 17:24:37 +02:00
ef45fc9d81
fs/open 2024-09-13 17:22:36 +02:00
0a9404fb2f
fs/unlink 2024-09-13 17:19:33 +02:00
3393f09907
format 2024-09-13 17:13:37 +02:00
aa1f06c20f
lint/fs 2024-09-13 17:13:16 +02:00
3e0d5bf2bc
lint/fs 2024-09-13 17:07:27 +02:00
08b6364d9b
lint/err 2024-09-13 17:00:39 +02:00
f7c1d90dfd
lint/squashfs 2024-09-13 16:58:01 +02:00
ce3629a776
lint/deb 2024-09-13 16:56:01 +02:00
9bb8003812
lint/sphinx 2024-09-13 16:54:22 +02:00
f46d5a9765
lint/grub 2024-09-13 16:52:01 +02:00
88311cb55c
lint/prj 2024-09-13 16:50:20 +02:00
8c896c9074
lint/txt 2024-09-13 16:45:56 +02:00
eb5a386e02
lint/ps 2024-09-13 16:45:07 +02:00
2663288127
lint/log 2024-09-13 16:36:29 +02:00
12a2a82cfe
lint/log 2024-09-13 16:27:04 +02:00
076f654a2d
lint/arg 2024-09-13 15:55:46 +02:00
63d30013a3
lint/fs 2024-09-13 15:53:20 +02:00
e306971534
lint/cmd 2024-09-13 15:45:17 +02:00
2dfdabed26
lint/sphinx 2024-09-13 15:40:19 +02:00
5361fbd9a0
3.11 2024-08-20 14:14:03 +02:00
33 changed files with 771 additions and 115 deletions

View file

@ -0,0 +1,18 @@
on: [push]
jobs:
job:
container:
image: ${{vars.DOCKER}}debian:bookworm
steps:
- name: spcd
env:
SPCD: ${{vars.SPCD}}
SPCD_SSH_HOSTS: ${{vars.SPCD_SSH_HOSTS}}
SPCD_SSH_KEY: ${{secrets.SPCD_SSH_KEY}}
SPCD_TXT_LOCALE: ${{vars.SPCD_TXT_LOCALE}}
run: ${{vars.SPCD}}
#- run: spcd-check-project
- run: spcd-build-project
- run: spcd-browse-workspace
- run: spcd-synchronize

11
build.py Executable file
View file

@ -0,0 +1,11 @@
#! /usr/bin/env python3
"""Dummy build."""
from pathlib import Path
from rwx.fs import make_directory, write
if __name__ == "__main__":
out = Path(__file__).parent / "out" / "web"
make_directory(out)
write(out / "index.html", "rwx.rwx.work")

View file

@ -21,7 +21,7 @@ keywords = []
license-files = { paths = ["license.md"] }
name = "rwx"
readme = "readme.md"
requires-python = ">= 3.10"
requires-python = ">= 3.11"
[project.scripts]
# command = "package.module:function"
@ -31,6 +31,12 @@ requires-python = ">= 3.10"
[tool.hatch.version]
path = "rwx/__init__.py"
[tool.pydoclint]
allow-init-docstring = true
quiet = true
skip-checking-short-docstrings = false
style = "sphinx"
[tool.ruff]
line-length = 80

View file

@ -1,7 +0,0 @@
******************
Read Write eXecute
******************
`English <readme.rst>`_ | `Français <readme.fr.rst>`_
Français

61
readme.md Normal file
View file

@ -0,0 +1,61 @@
# Read Write eXecute
A tiny framework to read, write & execute things.
---
## Why
---
## How
---
## What
---
## Who
### By
* [Marc Beninca](https://marc.beninca.link)
### For
* myself
---
## Where
### Chat
* [Discord](https://discord.com/channels/983145051985154108/1255894474895134761)
* [IRC](ircs://irc.libera.chat/##rwx)
### Forge
* [Repository](https://forge.rwx.work/rwx.work/rwx)
* [RSS](https://forge.rwx.work/rwx.work/rwx.rss)
* [Workflows](https://forge.rwx.work/rwx.work/rwx/actions)
### Deployment
* [Site](https://rwx.rwx.work)
---
## When
### Task stack
* character constants for box drawing
* common __str__ function
* parse pyproject.toml to write commands
* write classes for
* steps bars to log
* system commands to run
* with single call of subprocess.run
* or alternate subprocess method?

View file

@ -1,8 +0,0 @@
Read Write eXecute
==================
`English <https://doc.rwx.work>`_
---------------------------------
`Français <https://doc.rwx.work>`_
----------------------------------

View file

@ -1,3 +1,33 @@
"""Read Write eXecute."""
__version__ = "0.0.1"
from os import linesep
class Class:
"""Root class."""
def __repr__(self) -> str:
"""Return machine-readable state.
:return: state
:rtype: str
"""
name = self.__class__.__name__
attributes = [
f"{k}={v!r}" for k, v in vars(self).items() if not k.startswith("_")
]
arguments = ", ".join(attributes)
return f"{name}({arguments})"
def __str__(self) -> str:
"""Return human-readable state.
:return: state
:rtype: str
"""
attributes = [
f"{k} = {v}" for k, v in vars(self).items() if not k.startswith("_")
]
return linesep.join(attributes)

View file

@ -4,12 +4,12 @@
from pathlib import Path
import fs
from rwx import fs
if __name__ == "__main__":
file_path = Path(__file__).resolve()
root_path = file_path.parent
directory_path = root_path / "tmp"
file_path: Path = Path(__file__).resolve()
root_path: Path = file_path.parent
directory_path: Path = root_path / "tmp"
file_path = directory_path / "file"
fs.wipe(directory_path)

View file

@ -1,6 +1,13 @@
"""Handle system arguments."""
import sys
def split() -> tuple[str, list[str]]:
"""Split command & actual arguments.
:return: both
:rtype: tuple[str, list[str]]
"""
command, *arguments = sys.argv
return command, arguments

View file

@ -1,8 +1,16 @@
"""Handle system commands & packages."""
commands: list[str] = []
packages: list[str] = []
def need(command: str) -> None:
"""Assert package dependency for a command.
:param command: name of the requested command
:type command: str
"""
package: str | None
match command:
case "debootstrap":
package = "debootstrap"

View file

@ -1,19 +1,26 @@
import ps
"""Wrap SquashFS commands."""
import rwx.cmd
from pathlib import Path
rwx.cmd.need("mksquashfs")
from rwx import cmd, ps
cmd.need("mksquashfs")
def mksquashfs(input_root: str, output_file: str):
def mksquashfs(input_root: Path, output_file: Path) -> None:
"""Make a SquashFS bootable image file.
:param input_root: ?
:type input_root: Path
:param output_file: ?
:type output_file: Path
"""
ps.run(
[
"mksquashfs",
input_root,
output_file,
str(input_root),
str(output_file),
"-comp",
"zstd",
"-Xcompression-level",
str(18),
]
)

View file

@ -1,6 +1,8 @@
import cmd
"""Wrap Debian commands."""
import ps
from pathlib import Path
from rwx import cmd, ps
cmd.need("debootstrap")
@ -8,13 +10,22 @@ BOOTSTRAP_ARCHITECTURE = "amd64"
BOOTSTRAP_VARIANT = "minbase"
def bootstrap(root_path: str, suite: str, mirror_location: str):
command = [
("debootstrap",),
def bootstrap(root_path: Path, suite: str, mirror_location: str) -> None:
"""Boostrap a base operating filesystem.
:param root_path: target output path
:type root_path: Path
:param suite: target distribution name
:type suite: str
:param mirror_location: source input repository
:type mirror_location: str
"""
command = (
"debootstrap",
("--arch", BOOTSTRAP_ARCHITECTURE),
("--variant", BOOTSTRAP_VARIANT),
(suite,),
(root_path,),
(mirror_location,),
]
return ps.run(command)
suite,
str(root_path),
mirror_location,
)
ps.run(*command)

View file

@ -1,2 +1,7 @@
class Exception(Exception):
pass
"""Handle errors."""
from rwx import Class
class Error(Class, Exception):
"""Parent class for all errors."""

View file

@ -1,70 +1,160 @@
"""Operations involving FileSystems."""
import os
import shutil
import tomllib
from pathlib import Path
from rwx import ps
CHARSET = "UTF-8"
def create_image(file_path: str, size_bytes: int):
def create_image(file_path: Path, size_bytes: int) -> None:
"""Create a virtual device image file.
:param file_path: target image file
:type file_path: Path
:param size_bytes: virtual volume
:type size_bytes: int
"""
ps.run(
("qemu-img", "create"),
("-f", "qcow2"),
(file_path, size_bytes),
(str(file_path), str(size_bytes)),
)
def empty_file(path: str):
def empty_file(path: Path) -> None:
"""Empty the file at provided path.
:param path: target file to empty
:type path: Path
"""
write(path, "")
def get_mount_uuid(path: str):
def get_mount_uuid(path: Path) -> str:
"""Return the filesystem UUID of a mountpoint path.
:param path: mountpoint path
:type path: Path
:rtype: str
"""
return ps.run_line(
("findmnt",),
("--noheadings",),
"findmnt",
"--noheadings",
("--output", "UUID"),
(path,),
str(path),
)
def get_path_mount(path: str):
return ps.run_line(
("stat",),
def get_path_mount(path: Path) -> Path:
"""Return the mountpoint path of an arbitrary path.
:param path: arbitrary path
:type path: Path
:rtype: Path
"""
return Path(
ps.run_line(
"stat",
("--format", "%m"),
(path,),
str(path),
)
)
def get_path_uuid(path: str):
def get_path_uuid(path: Path) -> str:
"""Return the filesystem UUID of an arbitrary path.
:param path: arbitrary path
:type path: Path
:rtype: str
"""
return get_mount_uuid(get_path_mount(path))
def make_directory(path: str):
os.makedirs(path, exist_ok=True)
def make_directory(path: Path) -> None:
"""Make a directory (and its parents) from a path.
:param path: directory to create
:type path: Path
"""
path.mkdir(exist_ok=True, parents=True)
def read_file(file_path: str):
with open(file_path, "br") as file_object:
def read_file_bytes(file_path: Path) -> bytes:
"""Read whole file bytes.
:param file_path: source input file
:type file_path: Path
:rtype: bytes
"""
with file_path.open("br") as file_object:
return file_object.read()
def read_file_lines(file_path: str, charset=CHARSET):
return read_file_text(file_path).split(os.linesep)
def read_file_dict(file_path: Path, charset: str = CHARSET) -> dict:
"""Read whole file as toml object.
:param file_path: source input file
:type file_path: Path
:param charset: charset to use for decoding input
:type charset: str
:rtype: dict
"""
text = read_file_text(file_path, charset)
return tomllib.loads(text)
def read_file_text(file_path: str, charset=CHARSET):
return read_file(file_path).decode(charset)
def read_file_lines(file_path: Path, charset: str = CHARSET) -> list[str]:
"""Read whole file lines.
:param file_path: source input file
:type file_path: Path
:param charset: charset to use for decoding input
:type charset: str
:rtype: list[str]
"""
return read_file_text(file_path, charset).split(os.linesep)
def wipe(path: str):
def read_file_text(file_path: Path, charset: str = CHARSET) -> str:
"""Read whole file text.
:param file_path: source input file
:type file_path: Path
:param charset: charset to use for decoding input
:type charset: str
:rtype: str
"""
return read_file_bytes(file_path).decode(charset)
def wipe(path: Path) -> None:
"""Wipe provided path, whether directory or file.
:param path: target path
:type path: Path
"""
try:
shutil.rmtree(path)
except NotADirectoryError:
os.remove(path)
path.unlink(missing_ok=True)
except FileNotFoundError:
pass
def write(file_path: str, text: str, charset=CHARSET):
with open(file_path, "bw") as file_object:
file_object.write(text.encode(charset))
def write(file_path: Path, text: str, charset: str = CHARSET) -> None:
"""Write text into a file.
:param file_path: target file path
:type file_path: Path
:param text: content to write
:type text: str
:param charset: charset to use for encoding ouput
:type charset: str
"""
with file_path.open(encoding=charset, mode="w") as file_object:
file_object.write(text)

View file

@ -1,6 +1,6 @@
import cmd
"""Wrap GRUB commands."""
import ps
from rwx import cmd, ps
cmd.need("grub-mkimage")
@ -24,8 +24,21 @@ def make_image(
memdisk_path: str,
pubkey_path: str | None = None,
) -> None:
args = [
("grub-mkimage",),
"""Make a binary bootable image.
:param image_format: output format (x86_64-efi, i386-pc, arm64-efi)
:type image_format: str
:param image_path: output file
:type image_path: str
:param modules: modules to embed
:type modules: list[str]
:param memdisk_path: archive to include
:type memdisk_path: str
:param pubkey_path: extra public key to add
:type pubkey_path: str | None
"""
args: list[str | tuple[str, ...]] = [
"grub-mkimage",
("--compress", COMPRESSION),
("--format", image_format),
("--output", image_path),
@ -34,6 +47,6 @@ def make_image(
if pubkey_path:
args.append(("--pubkey", pubkey_path))
args.extend(modules)
if modules := MODULES.get(image_format):
args.extend(modules)
if extra_modules := MODULES.get(image_format):
args.extend(extra_modules)
ps.run(*args)

View file

@ -1,31 +1,50 @@
"""Handle logging."""
import logging
import sys
def get_file_logger(name: str) -> logging.Logger:
formatter = logging.Formatter(
"%(name)s: %(asctime)s | %(levelname)s | %(filename)s:%(lineno)s | %(process)d >>> %(message)s",
)
#
"""Return a file logger.
:param name: arbitrary name
:type name: str
:rtype: logging.Logger
"""
# formatter
items = [
"%(name)s: %(asctime)s",
"%(levelname)s",
"%(filename)s:%(lineno)s",
"%(process)d >>> %(message)s",
]
template = " | ".join(items)
formatter = logging.Formatter(template)
# handler
out_handler = logging.StreamHandler(stream=sys.stdout)
out_handler.setFormatter(formatter)
out_handler.setLevel(logging.INFO)
#
# logger
logger = logging.getLogger(name)
logger.addHandler(out_handler)
logger.setLevel(logging.INFO)
#
return logger
def get_stream_logger(level: int) -> logging.Logger:
"""Return a stream logger.
:param level: filtering level
:type level: int
:rtype: logging.Logger
"""
# handler
out_handler = logging.StreamHandler(stream=sys.stdout)
out_handler.setLevel(level)
#
# logger
logger = logging.getLogger()
logger.addHandler(out_handler)
logger.setLevel(level)
#
return logger

20
rwx/os/__init__.py Normal file
View file

@ -0,0 +1,20 @@
"""Control Operating Systems."""
from os import sep
from pathlib import Path
from .abstract import OS
from .debian import Debian
def from_path(path: Path) -> OS:
"""Initialize from an already existing path.
:param path: source root directory
:type path: Path
:rtype: OS
"""
return Debian(path)
up = from_path(Path(sep))

26
rwx/os/abstract.py Normal file
View file

@ -0,0 +1,26 @@
"""Abstract Operating System."""
from abc import ABC, abstractmethod
from pathlib import Path
from rwx import Class
class OS(Class, ABC):
"""Operating System."""
def __init__(self, path: Path) -> None:
"""Set root.
:param path: root directory
:type path: Path
"""
self.root = path
self.name = self.get_name()
@abstractmethod
def get_name(self) -> str:
"""Return mandatory name.
:rtype: str
"""

14
rwx/os/debian.py Normal file
View file

@ -0,0 +1,14 @@
"""Debian operating system."""
from .abstract import OS
class Debian(OS):
"""Debian operating system."""
def get_name(self) -> str:
"""Return name.
:rtype: str
"""
return "Debian"

29
rwx/os/pm/__init__.py Normal file
View file

@ -0,0 +1,29 @@
"""Package Manager."""
from abc import ABC, abstractmethod
from rwx import Class
from rwx.ps import Command
class PM(Class, ABC):
"""Package Manager."""
def __init__(self) -> None:
"""Set commands."""
self.clean = self.get_clean_command()
self.install = self.get_install_command()
@abstractmethod
def get_clean_command(self) -> Command:
"""Command to clean packages cache.
:rtype: Command
"""
@abstractmethod
def get_install_command(self) -> Command:
"""Command to install package(s).
:rtype: Command
"""

22
rwx/os/pm/apt.py Normal file
View file

@ -0,0 +1,22 @@
"""Advanced Package Tool."""
from rwx.os.pm import PM
from rwx.ps import Command
class APT(PM):
"""Advanced Package Tool."""
def get_clean_command(self) -> Command:
"""Return clean command.
:rtype: Command
"""
return Command()
def get_install_command(self) -> Command:
"""Return install command.
:rtype: Command
"""
return Command()

View file

@ -1,8 +1,20 @@
from os import path
"""Handle projects."""
from pathlib import Path
from rwx import Class
class Project:
def __init__(self, file_path: str) -> None:
self.file: str = path.realpath(file_path)
self.root: str = path.dirname(self.file)
self.name: str = path.basename(self.root)
class Project(Class):
"""Parent class for any type of project."""
def __init__(self, file: Path) -> None:
"""Set file, root & name.
:param file: root reference file
:type file: Path
"""
self.raw = file
self.file = self.raw.resolve()
self.root: Path = self.file.parent
self.name: str = self.root.name

View file

@ -1,17 +1,22 @@
from os import path
"""Project consisting only of a Sphinx documentation."""
from typing import TYPE_CHECKING
from sphinx.cmd.build import build_main
from rwx.fs import wipe
from rwx.prj import Project
if TYPE_CHECKING:
from pathlib import Path
class SphinxProject(Project):
def __init__(self, file_path: str) -> None:
super().__init__(file_path)
"""Child class for a project based on Sphinx."""
def build(self):
output_root: str = path.join(self.root, "out")
def build(self) -> None:
"""Build the project."""
output_root: Path = self.root / "out"
wipe(output_root)
arguments: list[str] = [
"-E",
@ -22,13 +27,13 @@ class SphinxProject(Project):
"-D",
f"project={self.name}",
"-D",
"master_doc={}".format("index"),
"master_doc=index",
"-D",
"html_theme={}".format("sphinx_rtd_theme"),
"html_theme=sphinx_rtd_theme",
"-c",
self.root,
str(self.root),
# "-C",
path.join(self.root, self.name),
path.join(output_root, "web"),
str(self.root / self.name),
str(output_root / "web"),
]
build_main(arguments)

View file

@ -1,32 +1,78 @@
"""Handle processes."""
import subprocess
from rwx import txt
from rwx import Class, txt
def get_tuples_args(tuples) -> list[str]:
class Command(Class):
"""Command to run."""
def __init__(self, *arguments: str | tuple[str, ...]) -> None:
"""Set raw & flat arguments.
:param *arguments: single argument or grouped ones
:type *arguments: str | tuple[str, ...]
"""
self.raw = arguments
self.flat: list[str] = []
def get_tuples_args(*items: str | tuple[str, ...]) -> list[str]:
"""Turn arguments tuples into an arguments list.
:param *items: single item or grouped ones
:type *items: str | tuple[str, ...]
:rtype: list[str]
"""
args: list[str] = []
for item in tuples:
if type(item) is tuple:
args.extend(item)
else:
for item in items:
match item:
case str():
args.append(item)
case tuple():
args.extend(item)
return args
def run(*tuples) -> subprocess.CompletedProcess:
def run(*items: str | tuple[str, ...]) -> subprocess.CompletedProcess:
"""Run from a list of arguments tuples.
:param *items: single item or grouped ones
:type *items: str | tuple[str, ...]
:rtype: subprocess.CompletedProcess
"""
return subprocess.run(
get_tuples_args(tuples), capture_output=False, check=True
get_tuples_args(*items), capture_output=False, check=True
)
def run_line(*tuples, charset: str = txt.CHARSET) -> str:
lines = run_lines(*get_tuples_args(tuples), charset=charset)
return lines[0]
def run_line(*items: str | tuple[str, ...], charset: str = txt.CHARSET) -> str:
"""Run and return output line.
:param *items: single item or grouped ones
:type *items: str | tuple[str, ...]
:param charset: charset to use for decoding binary output
:type charset: str
:rtype: str
"""
line, *_ = run_lines(*items, charset=charset)
return line
def run_lines(*tuples, charset: str = txt.CHARSET) -> list[str]:
def run_lines(
*items: str | tuple[str, ...], charset: str = txt.CHARSET
) -> list[str]:
"""Run and return output lines.
:param *items: single item or grouped ones
:type *items: str | tuple[str, ...]
:param charset: charset to use for decoding binary output
:type charset: str
:rtype: list[str]
"""
process = subprocess.run(
get_tuples_args(tuples), capture_output=True, check=True
get_tuples_args(*items), capture_output=True, check=True
)
string = process.stdout.decode(charset)
return string.rstrip().splitlines()

0
rwx/py.typed Normal file
View file

View file

@ -0,0 +1 @@
"""Configure FreeTube."""

View file

@ -0,0 +1,29 @@
"""FreeTube channels."""
from rwx import Class
class Channel(Class):
"""FreeTube channel."""
def __init__(self, uid: str, name: str) -> None:
"""Set uid & name.
:param uid: unique identifier
:type uid: str
:param name: label
:type name: str
"""
self.uid = uid
self.name = name
def to_db(self) -> str:
"""Return identifier as db.
:rtype: str
"""
return f"""\
{{\
"id":"{self.uid}"\
}}\
"""

21
rwx/sw/freetube/db.py Normal file
View file

@ -0,0 +1,21 @@
"""Output FreeTube db."""
def to_db(value: object) -> str:
"""Render value as string.
:param value: value to render
:type value: object
:rtype: str
"""
match value:
case bool():
text = str(value).lower()
case dict():
sub = ",".join([f'"{i}":{to_db(v)}' for i, v in value.items()])
text = f"{{{sub}}}"
case float() | str():
text = f'"{value}"'
case _:
text = str(value)
return text

View file

@ -0,0 +1,47 @@
"""FreeTube playlists."""
from rwx import Class
from .videos import Video
class Playlist(Class):
"""FreeTube playlist."""
def __init__(self, uid: str, name: str) -> None:
"""Set uid & name.
:param uid: identifier
:type uid: str
:param name: label
:type name: str
"""
self.uid = uid
self.name = name
self.videos: list[Video] = []
def add(self, video: Video) -> None:
"""Add video.
:param video: video to add
:type video: Video
"""
self.videos.append(video)
def to_db(self) -> str:
"""Return identifier, name & videos.
:rtype: str
"""
videos = ",".join([video.to_db() for video in self.videos])
return f"""\
{{\
"_id":"{self.uid}"\
,\
"playlistName":"{self.name}"\
,\
"protected":true\
,\
"videos":[{videos}]\
}}\
"""

View file

@ -0,0 +1,45 @@
"""FreeTube profiles."""
from rwx import Class
from .channels import Channel
class Profile(Class):
"""FreeTube profile."""
def __init__(self, uid: str, name: str) -> None:
"""Set uid & name.
:param uid: unique identifier
:type uid: str
:param name: label
:type name: str
"""
self.id = uid
self.name = name
self.channels: list[Channel] = []
def add(self, channel: Channel) -> None:
"""Add channel.
:param channel: channel to add
:type channel: Channel
"""
self.channels.append(channel)
def to_db(self) -> str:
"""Return identifier, name & channels.
:rtype: str
"""
channels = ",".join([channel.to_db() for channel in self.channels])
return f"""\
{{\
"_id":"{self.id}"\
,\
"name":"{self.name}"\
,\
"subscriptions":[{channels}]\
}}\
"""

View file

@ -0,0 +1,33 @@
"""FreeTube settings."""
from rwx import Class
from .db import to_db
class Setting(Class):
"""FreeTube setting."""
def __init__(self, uid: str, value: object) -> None:
"""Set uid & value.
:param uid: unique identifier
:type uid: str
:param value: value
:type value: object
"""
self.uid = uid
self.value = value
def to_db(self) -> str:
"""Return uid & value as db string.
:rtype: str
"""
return f"""\
{{\
"_id":"{self.uid}"\
,\
"value":{to_db(self.value)}\
}}\
"""

33
rwx/sw/freetube/videos.py Normal file
View file

@ -0,0 +1,33 @@
"""FreeTube videos."""
from rwx import Class
class Video(Class):
"""FreeTube video."""
def __init__(self, uid: str, name: str) -> None:
"""Set id & name.
:param uid: identifier
:type uid: str
:param name: label
:type name: str
"""
self.uid = uid
self.name = name
def to_db(self) -> str:
"""Return identifier, zero length & title.
:rtype: str
"""
return f"""\
{{\
"videoId":"{self.uid}"\
,\
"lengthSeconds":0\
,\
"title":"{self.name}"\
}}\
"""

View file

@ -1 +1,3 @@
"""Handle text."""
CHARSET = "UTF-8"