Compare commits
No commits in common. "98350281993d8821cba71087f7ad8ad83e9dded7" and "519613eeeb0d2715f41165a2c6d74c125e5f4856" have entirely different histories.
9835028199
...
519613eeeb
33 changed files with 115 additions and 771 deletions
|
@ -1,18 +0,0 @@
|
||||||
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
11
build.py
|
@ -1,11 +0,0 @@
|
||||||
#! /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")
|
|
|
@ -21,7 +21,7 @@ keywords = []
|
||||||
license-files = { paths = ["license.md"] }
|
license-files = { paths = ["license.md"] }
|
||||||
name = "rwx"
|
name = "rwx"
|
||||||
readme = "readme.md"
|
readme = "readme.md"
|
||||||
requires-python = ">= 3.11"
|
requires-python = ">= 3.10"
|
||||||
|
|
||||||
[project.scripts]
|
[project.scripts]
|
||||||
# command = "package.module:function"
|
# command = "package.module:function"
|
||||||
|
@ -31,12 +31,6 @@ requires-python = ">= 3.11"
|
||||||
[tool.hatch.version]
|
[tool.hatch.version]
|
||||||
path = "rwx/__init__.py"
|
path = "rwx/__init__.py"
|
||||||
|
|
||||||
[tool.pydoclint]
|
|
||||||
allow-init-docstring = true
|
|
||||||
quiet = true
|
|
||||||
skip-checking-short-docstrings = false
|
|
||||||
style = "sphinx"
|
|
||||||
|
|
||||||
[tool.ruff]
|
[tool.ruff]
|
||||||
line-length = 80
|
line-length = 80
|
||||||
|
|
||||||
|
|
7
readme.fr.rst
Normal file
7
readme.fr.rst
Normal file
|
@ -0,0 +1,7 @@
|
||||||
|
******************
|
||||||
|
Read Write eXecute
|
||||||
|
******************
|
||||||
|
|
||||||
|
`English <readme.rst>`_ | `Français <readme.fr.rst>`_
|
||||||
|
|
||||||
|
Français
|
61
readme.md
61
readme.md
|
@ -1,61 +0,0 @@
|
||||||
# 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?
|
|
8
readme.rst
Normal file
8
readme.rst
Normal file
|
@ -0,0 +1,8 @@
|
||||||
|
Read Write eXecute
|
||||||
|
==================
|
||||||
|
|
||||||
|
`English <https://doc.rwx.work>`_
|
||||||
|
---------------------------------
|
||||||
|
|
||||||
|
`Français <https://doc.rwx.work>`_
|
||||||
|
----------------------------------
|
|
@ -1,33 +1,3 @@
|
||||||
"""Read Write eXecute."""
|
"""Read Write eXecute."""
|
||||||
|
|
||||||
__version__ = "0.0.1"
|
__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)
|
|
||||||
|
|
|
@ -4,12 +4,12 @@
|
||||||
|
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
from rwx import fs
|
import fs
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
file_path: Path = Path(__file__).resolve()
|
file_path = Path(__file__).resolve()
|
||||||
root_path: Path = file_path.parent
|
root_path = file_path.parent
|
||||||
directory_path: Path = root_path / "tmp"
|
directory_path = root_path / "tmp"
|
||||||
file_path = directory_path / "file"
|
file_path = directory_path / "file"
|
||||||
|
|
||||||
fs.wipe(directory_path)
|
fs.wipe(directory_path)
|
||||||
|
|
|
@ -1,13 +1,6 @@
|
||||||
"""Handle system arguments."""
|
|
||||||
|
|
||||||
import sys
|
import sys
|
||||||
|
|
||||||
|
|
||||||
def split() -> tuple[str, list[str]]:
|
def split() -> tuple[str, list[str]]:
|
||||||
"""Split command & actual arguments.
|
|
||||||
|
|
||||||
:return: both
|
|
||||||
:rtype: tuple[str, list[str]]
|
|
||||||
"""
|
|
||||||
command, *arguments = sys.argv
|
command, *arguments = sys.argv
|
||||||
return command, arguments
|
return command, arguments
|
||||||
|
|
|
@ -1,16 +1,8 @@
|
||||||
"""Handle system commands & packages."""
|
|
||||||
|
|
||||||
commands: list[str] = []
|
commands: list[str] = []
|
||||||
packages: list[str] = []
|
packages: list[str] = []
|
||||||
|
|
||||||
|
|
||||||
def need(command: str) -> None:
|
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:
|
match command:
|
||||||
case "debootstrap":
|
case "debootstrap":
|
||||||
package = "debootstrap"
|
package = "debootstrap"
|
||||||
|
|
|
@ -1,26 +1,19 @@
|
||||||
"""Wrap SquashFS commands."""
|
import ps
|
||||||
|
|
||||||
from pathlib import Path
|
import rwx.cmd
|
||||||
|
|
||||||
from rwx import cmd, ps
|
rwx.cmd.need("mksquashfs")
|
||||||
|
|
||||||
cmd.need("mksquashfs")
|
|
||||||
|
|
||||||
|
|
||||||
def mksquashfs(input_root: Path, output_file: Path) -> None:
|
def mksquashfs(input_root: str, output_file: str):
|
||||||
"""Make a SquashFS bootable image file.
|
|
||||||
|
|
||||||
:param input_root: ?
|
|
||||||
:type input_root: Path
|
|
||||||
:param output_file: ?
|
|
||||||
:type output_file: Path
|
|
||||||
"""
|
|
||||||
ps.run(
|
ps.run(
|
||||||
|
[
|
||||||
"mksquashfs",
|
"mksquashfs",
|
||||||
str(input_root),
|
input_root,
|
||||||
str(output_file),
|
output_file,
|
||||||
"-comp",
|
"-comp",
|
||||||
"zstd",
|
"zstd",
|
||||||
"-Xcompression-level",
|
"-Xcompression-level",
|
||||||
str(18),
|
str(18),
|
||||||
|
]
|
||||||
)
|
)
|
||||||
|
|
|
@ -1,8 +1,6 @@
|
||||||
"""Wrap Debian commands."""
|
import cmd
|
||||||
|
|
||||||
from pathlib import Path
|
import ps
|
||||||
|
|
||||||
from rwx import cmd, ps
|
|
||||||
|
|
||||||
cmd.need("debootstrap")
|
cmd.need("debootstrap")
|
||||||
|
|
||||||
|
@ -10,22 +8,13 @@ BOOTSTRAP_ARCHITECTURE = "amd64"
|
||||||
BOOTSTRAP_VARIANT = "minbase"
|
BOOTSTRAP_VARIANT = "minbase"
|
||||||
|
|
||||||
|
|
||||||
def bootstrap(root_path: Path, suite: str, mirror_location: str) -> None:
|
def bootstrap(root_path: str, suite: str, mirror_location: str):
|
||||||
"""Boostrap a base operating filesystem.
|
command = [
|
||||||
|
("debootstrap",),
|
||||||
: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),
|
("--arch", BOOTSTRAP_ARCHITECTURE),
|
||||||
("--variant", BOOTSTRAP_VARIANT),
|
("--variant", BOOTSTRAP_VARIANT),
|
||||||
suite,
|
(suite,),
|
||||||
str(root_path),
|
(root_path,),
|
||||||
mirror_location,
|
(mirror_location,),
|
||||||
)
|
]
|
||||||
ps.run(*command)
|
return ps.run(command)
|
||||||
|
|
|
@ -1,7 +1,2 @@
|
||||||
"""Handle errors."""
|
class Exception(Exception):
|
||||||
|
pass
|
||||||
from rwx import Class
|
|
||||||
|
|
||||||
|
|
||||||
class Error(Class, Exception):
|
|
||||||
"""Parent class for all errors."""
|
|
||||||
|
|
|
@ -1,160 +1,70 @@
|
||||||
"""Operations involving FileSystems."""
|
|
||||||
|
|
||||||
import os
|
import os
|
||||||
import shutil
|
import shutil
|
||||||
import tomllib
|
|
||||||
from pathlib import Path
|
|
||||||
|
|
||||||
from rwx import ps
|
from rwx import ps
|
||||||
|
|
||||||
CHARSET = "UTF-8"
|
CHARSET = "UTF-8"
|
||||||
|
|
||||||
|
|
||||||
def create_image(file_path: Path, size_bytes: int) -> None:
|
def create_image(file_path: str, size_bytes: int):
|
||||||
"""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(
|
ps.run(
|
||||||
("qemu-img", "create"),
|
("qemu-img", "create"),
|
||||||
("-f", "qcow2"),
|
("-f", "qcow2"),
|
||||||
(str(file_path), str(size_bytes)),
|
(file_path, size_bytes),
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
def empty_file(path: Path) -> None:
|
def empty_file(path: str):
|
||||||
"""Empty the file at provided path.
|
|
||||||
|
|
||||||
:param path: target file to empty
|
|
||||||
:type path: Path
|
|
||||||
"""
|
|
||||||
write(path, "")
|
write(path, "")
|
||||||
|
|
||||||
|
|
||||||
def get_mount_uuid(path: Path) -> str:
|
def get_mount_uuid(path: str):
|
||||||
"""Return the filesystem UUID of a mountpoint path.
|
|
||||||
|
|
||||||
:param path: mountpoint path
|
|
||||||
:type path: Path
|
|
||||||
:rtype: str
|
|
||||||
"""
|
|
||||||
return ps.run_line(
|
return ps.run_line(
|
||||||
"findmnt",
|
("findmnt",),
|
||||||
"--noheadings",
|
("--noheadings",),
|
||||||
("--output", "UUID"),
|
("--output", "UUID"),
|
||||||
str(path),
|
(path,),
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
def get_path_mount(path: Path) -> Path:
|
def get_path_mount(path: str):
|
||||||
"""Return the mountpoint path of an arbitrary path.
|
return ps.run_line(
|
||||||
|
("stat",),
|
||||||
:param path: arbitrary path
|
|
||||||
:type path: Path
|
|
||||||
:rtype: Path
|
|
||||||
"""
|
|
||||||
return Path(
|
|
||||||
ps.run_line(
|
|
||||||
"stat",
|
|
||||||
("--format", "%m"),
|
("--format", "%m"),
|
||||||
str(path),
|
(path,),
|
||||||
)
|
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
def get_path_uuid(path: Path) -> str:
|
def get_path_uuid(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))
|
return get_mount_uuid(get_path_mount(path))
|
||||||
|
|
||||||
|
|
||||||
def make_directory(path: Path) -> None:
|
def make_directory(path: str):
|
||||||
"""Make a directory (and its parents) from a path.
|
os.makedirs(path, exist_ok=True)
|
||||||
|
|
||||||
:param path: directory to create
|
|
||||||
:type path: Path
|
|
||||||
"""
|
|
||||||
path.mkdir(exist_ok=True, parents=True)
|
|
||||||
|
|
||||||
|
|
||||||
def read_file_bytes(file_path: Path) -> bytes:
|
def read_file(file_path: str):
|
||||||
"""Read whole file bytes.
|
with open(file_path, "br") as file_object:
|
||||||
|
|
||||||
:param file_path: source input file
|
|
||||||
:type file_path: Path
|
|
||||||
:rtype: bytes
|
|
||||||
"""
|
|
||||||
with file_path.open("br") as file_object:
|
|
||||||
return file_object.read()
|
return file_object.read()
|
||||||
|
|
||||||
|
|
||||||
def read_file_dict(file_path: Path, charset: str = CHARSET) -> dict:
|
def read_file_lines(file_path: str, charset=CHARSET):
|
||||||
"""Read whole file as toml object.
|
return read_file_text(file_path).split(os.linesep)
|
||||||
|
|
||||||
: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_lines(file_path: Path, charset: str = CHARSET) -> list[str]:
|
def read_file_text(file_path: str, charset=CHARSET):
|
||||||
"""Read whole file lines.
|
return read_file(file_path).decode(charset)
|
||||||
|
|
||||||
: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 read_file_text(file_path: Path, charset: str = CHARSET) -> str:
|
def wipe(path: 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:
|
try:
|
||||||
shutil.rmtree(path)
|
shutil.rmtree(path)
|
||||||
except NotADirectoryError:
|
except NotADirectoryError:
|
||||||
path.unlink(missing_ok=True)
|
os.remove(path)
|
||||||
except FileNotFoundError:
|
except FileNotFoundError:
|
||||||
pass
|
pass
|
||||||
|
|
||||||
|
|
||||||
def write(file_path: Path, text: str, charset: str = CHARSET) -> None:
|
def write(file_path: str, text: str, charset=CHARSET):
|
||||||
"""Write text into a file.
|
with open(file_path, "bw") as file_object:
|
||||||
|
file_object.write(text.encode(charset))
|
||||||
: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)
|
|
||||||
|
|
|
@ -1,6 +1,6 @@
|
||||||
"""Wrap GRUB commands."""
|
import cmd
|
||||||
|
|
||||||
from rwx import cmd, ps
|
import ps
|
||||||
|
|
||||||
cmd.need("grub-mkimage")
|
cmd.need("grub-mkimage")
|
||||||
|
|
||||||
|
@ -24,21 +24,8 @@ def make_image(
|
||||||
memdisk_path: str,
|
memdisk_path: str,
|
||||||
pubkey_path: str | None = None,
|
pubkey_path: str | None = None,
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Make a binary bootable image.
|
args = [
|
||||||
|
("grub-mkimage",),
|
||||||
: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),
|
("--compress", COMPRESSION),
|
||||||
("--format", image_format),
|
("--format", image_format),
|
||||||
("--output", image_path),
|
("--output", image_path),
|
||||||
|
@ -47,6 +34,6 @@ def make_image(
|
||||||
if pubkey_path:
|
if pubkey_path:
|
||||||
args.append(("--pubkey", pubkey_path))
|
args.append(("--pubkey", pubkey_path))
|
||||||
args.extend(modules)
|
args.extend(modules)
|
||||||
if extra_modules := MODULES.get(image_format):
|
if modules := MODULES.get(image_format):
|
||||||
args.extend(extra_modules)
|
args.extend(modules)
|
||||||
ps.run(*args)
|
ps.run(*args)
|
||||||
|
|
|
@ -1,50 +1,31 @@
|
||||||
"""Handle logging."""
|
|
||||||
|
|
||||||
import logging
|
import logging
|
||||||
import sys
|
import sys
|
||||||
|
|
||||||
|
|
||||||
def get_file_logger(name: str) -> logging.Logger:
|
def get_file_logger(name: str) -> logging.Logger:
|
||||||
"""Return a file logger.
|
formatter = logging.Formatter(
|
||||||
|
"%(name)s: %(asctime)s | %(levelname)s | %(filename)s:%(lineno)s | %(process)d >>> %(message)s",
|
||||||
: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 = logging.StreamHandler(stream=sys.stdout)
|
||||||
out_handler.setFormatter(formatter)
|
out_handler.setFormatter(formatter)
|
||||||
out_handler.setLevel(logging.INFO)
|
out_handler.setLevel(logging.INFO)
|
||||||
# logger
|
#
|
||||||
logger = logging.getLogger(name)
|
logger = logging.getLogger(name)
|
||||||
logger.addHandler(out_handler)
|
logger.addHandler(out_handler)
|
||||||
logger.setLevel(logging.INFO)
|
logger.setLevel(logging.INFO)
|
||||||
|
#
|
||||||
return logger
|
return logger
|
||||||
|
|
||||||
|
|
||||||
def get_stream_logger(level: int) -> logging.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 = logging.StreamHandler(stream=sys.stdout)
|
||||||
out_handler.setLevel(level)
|
out_handler.setLevel(level)
|
||||||
# logger
|
#
|
||||||
logger = logging.getLogger()
|
logger = logging.getLogger()
|
||||||
logger.addHandler(out_handler)
|
logger.addHandler(out_handler)
|
||||||
logger.setLevel(level)
|
logger.setLevel(level)
|
||||||
|
#
|
||||||
return logger
|
return logger
|
||||||
|
|
||||||
|
|
||||||
|
|
|
@ -1,20 +0,0 @@
|
||||||
"""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))
|
|
|
@ -1,26 +0,0 @@
|
||||||
"""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
|
|
||||||
"""
|
|
|
@ -1,14 +0,0 @@
|
||||||
"""Debian operating system."""
|
|
||||||
|
|
||||||
from .abstract import OS
|
|
||||||
|
|
||||||
|
|
||||||
class Debian(OS):
|
|
||||||
"""Debian operating system."""
|
|
||||||
|
|
||||||
def get_name(self) -> str:
|
|
||||||
"""Return name.
|
|
||||||
|
|
||||||
:rtype: str
|
|
||||||
"""
|
|
||||||
return "Debian"
|
|
|
@ -1,29 +0,0 @@
|
||||||
"""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
|
|
||||||
"""
|
|
|
@ -1,22 +0,0 @@
|
||||||
"""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()
|
|
|
@ -1,20 +1,8 @@
|
||||||
"""Handle projects."""
|
from os import path
|
||||||
|
|
||||||
from pathlib import Path
|
|
||||||
|
|
||||||
from rwx import Class
|
|
||||||
|
|
||||||
|
|
||||||
class Project(Class):
|
class Project:
|
||||||
"""Parent class for any type of project."""
|
def __init__(self, file_path: str) -> None:
|
||||||
|
self.file: str = path.realpath(file_path)
|
||||||
def __init__(self, file: Path) -> None:
|
self.root: str = path.dirname(self.file)
|
||||||
"""Set file, root & name.
|
self.name: str = path.basename(self.root)
|
||||||
|
|
||||||
: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
|
|
||||||
|
|
|
@ -1,22 +1,17 @@
|
||||||
"""Project consisting only of a Sphinx documentation."""
|
from os import path
|
||||||
|
|
||||||
from typing import TYPE_CHECKING
|
|
||||||
|
|
||||||
from sphinx.cmd.build import build_main
|
from sphinx.cmd.build import build_main
|
||||||
|
|
||||||
from rwx.fs import wipe
|
from rwx.fs import wipe
|
||||||
from rwx.prj import Project
|
from rwx.prj import Project
|
||||||
|
|
||||||
if TYPE_CHECKING:
|
|
||||||
from pathlib import Path
|
|
||||||
|
|
||||||
|
|
||||||
class SphinxProject(Project):
|
class SphinxProject(Project):
|
||||||
"""Child class for a project based on Sphinx."""
|
def __init__(self, file_path: str) -> None:
|
||||||
|
super().__init__(file_path)
|
||||||
|
|
||||||
def build(self) -> None:
|
def build(self):
|
||||||
"""Build the project."""
|
output_root: str = path.join(self.root, "out")
|
||||||
output_root: Path = self.root / "out"
|
|
||||||
wipe(output_root)
|
wipe(output_root)
|
||||||
arguments: list[str] = [
|
arguments: list[str] = [
|
||||||
"-E",
|
"-E",
|
||||||
|
@ -27,13 +22,13 @@ class SphinxProject(Project):
|
||||||
"-D",
|
"-D",
|
||||||
f"project={self.name}",
|
f"project={self.name}",
|
||||||
"-D",
|
"-D",
|
||||||
"master_doc=index",
|
"master_doc={}".format("index"),
|
||||||
"-D",
|
"-D",
|
||||||
"html_theme=sphinx_rtd_theme",
|
"html_theme={}".format("sphinx_rtd_theme"),
|
||||||
"-c",
|
"-c",
|
||||||
str(self.root),
|
self.root,
|
||||||
# "-C",
|
# "-C",
|
||||||
str(self.root / self.name),
|
path.join(self.root, self.name),
|
||||||
str(output_root / "web"),
|
path.join(output_root, "web"),
|
||||||
]
|
]
|
||||||
build_main(arguments)
|
build_main(arguments)
|
||||||
|
|
|
@ -1,78 +1,32 @@
|
||||||
"""Handle processes."""
|
|
||||||
|
|
||||||
import subprocess
|
import subprocess
|
||||||
|
|
||||||
from rwx import Class, txt
|
from rwx import txt
|
||||||
|
|
||||||
|
|
||||||
class Command(Class):
|
def get_tuples_args(tuples) -> list[str]:
|
||||||
"""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] = []
|
args: list[str] = []
|
||||||
for item in items:
|
for item in tuples:
|
||||||
match item:
|
if type(item) is tuple:
|
||||||
case str():
|
|
||||||
args.append(item)
|
|
||||||
case tuple():
|
|
||||||
args.extend(item)
|
args.extend(item)
|
||||||
|
else:
|
||||||
|
args.append(item)
|
||||||
return args
|
return args
|
||||||
|
|
||||||
|
|
||||||
def run(*items: str | tuple[str, ...]) -> subprocess.CompletedProcess:
|
def run(*tuples) -> 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(
|
return subprocess.run(
|
||||||
get_tuples_args(*items), capture_output=False, check=True
|
get_tuples_args(tuples), capture_output=False, check=True
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
def run_line(*items: str | tuple[str, ...], charset: str = txt.CHARSET) -> str:
|
def run_line(*tuples, charset: str = txt.CHARSET) -> str:
|
||||||
"""Run and return output line.
|
lines = run_lines(*get_tuples_args(tuples), charset=charset)
|
||||||
|
return lines[0]
|
||||||
: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(
|
def run_lines(*tuples, charset: str = txt.CHARSET) -> list[str]:
|
||||||
*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(
|
process = subprocess.run(
|
||||||
get_tuples_args(*items), capture_output=True, check=True
|
get_tuples_args(tuples), capture_output=True, check=True
|
||||||
)
|
)
|
||||||
string = process.stdout.decode(charset)
|
string = process.stdout.decode(charset)
|
||||||
return string.rstrip().splitlines()
|
return string.rstrip().splitlines()
|
||||||
|
|
|
@ -1 +0,0 @@
|
||||||
"""Configure FreeTube."""
|
|
|
@ -1,29 +0,0 @@
|
||||||
"""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}"\
|
|
||||||
}}\
|
|
||||||
"""
|
|
|
@ -1,21 +0,0 @@
|
||||||
"""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
|
|
|
@ -1,47 +0,0 @@
|
||||||
"""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}]\
|
|
||||||
}}\
|
|
||||||
"""
|
|
|
@ -1,45 +0,0 @@
|
||||||
"""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}]\
|
|
||||||
}}\
|
|
||||||
"""
|
|
|
@ -1,33 +0,0 @@
|
||||||
"""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)}\
|
|
||||||
}}\
|
|
||||||
"""
|
|
|
@ -1,33 +0,0 @@
|
||||||
"""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}"\
|
|
||||||
}}\
|
|
||||||
"""
|
|
|
@ -1,3 +1 @@
|
||||||
"""Handle text."""
|
|
||||||
|
|
||||||
CHARSET = "UTF-8"
|
CHARSET = "UTF-8"
|
||||||
|
|
Loading…
Reference in a new issue