streamd/src/streamer/__init__.py
Konstantin Fickel 027bf531ce
feat: base task command on position instead of tags
Signed-off-by: Konstantin Fickel <mail@konstantinfickel.de>
2026-02-01 10:01:13 +01:00

93 lines
2.7 KiB
Python

import glob
import os
from datetime import datetime
from shutil import move
from typing import Generator
import click
import typer
from rich import print
from rich.markdown import Markdown
from rich.panel import Panel
from streamer.localize import (
LocalizedShard,
RepositoryConfiguration,
localize_stream_file,
)
from streamer.localize.preconfigured_configurations import TaskConfiguration
from streamer.parse import parse_markdown_file
from streamer.query import find_shard_by_position
from streamer.settings import Settings
app = typer.Typer()
def all_files(config: RepositoryConfiguration) -> Generator[LocalizedShard]:
for file_name in glob.glob(f"{glob.escape(Settings().base_folder)}/*.md"):
with open(file_name, "r") as file:
file_content = file.read()
if shard := localize_stream_file(
parse_markdown_file(file_name, file_content), config
):
yield shard
@app.command()
def todo() -> None:
all_shards = list(all_files(TaskConfiguration))
for task_shard in find_shard_by_position(all_shards, "task", "open"):
with open(task_shard.location["file"], "r") as file:
file_content = file.read().splitlines()
print(
Panel(
Markdown(
"\n".join(
file_content[
task_shard.start_line - 1 : task_shard.end_line
]
)
),
title=f"{task_shard.location['file']}:{task_shard.start_line}",
)
)
@app.command()
def new() -> None:
streamer_directory = Settings().base_folder
timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
preliminary_file_name = f"{timestamp}_wip.md"
prelimary_path = os.path.join(streamer_directory, preliminary_file_name)
content = "# "
with open(prelimary_path, "w") as file:
_ = file.write(content)
click.edit(None, filename=prelimary_path)
with open(prelimary_path, "r") as file:
content = file.read()
parsed_content = parse_markdown_file(prelimary_path, content)
final_file_name = f"{timestamp}.md"
if parsed_content.shard is not None and len(
markers := parsed_content.shard.markers
):
final_file_name = f"{timestamp} {' '.join(markers)}.md"
final_path = os.path.join(streamer_directory, final_file_name)
_ = move(prelimary_path, final_path)
print(f"Saved as [yellow]{final_file_name}")
@app.callback(invoke_without_command=True)
def main(ctx: typer.Context):
if ctx.invoked_subcommand is None:
new()
if __name__ == "__main__":
app()