123 lines
2.7 KiB
Python
123 lines
2.7 KiB
Python
from widgets import Navbar
|
|
from textual.screen import Screen
|
|
from textual.widgets import Input, Select, LoadingIndicator, Static
|
|
from textual.containers import HorizontalGroup, ScrollableContainer
|
|
from textual.app import ComposeResult
|
|
|
|
from human_readable import time_delta
|
|
from datetime import datetime, timedelta
|
|
|
|
import requests
|
|
|
|
|
|
class SearchResult(HorizontalGroup):
|
|
DEFAULT_CSS = """
|
|
SearchResult {
|
|
padding: 0 1;
|
|
margin: 0 1;
|
|
margin-top: 1;
|
|
border: tall $surface;
|
|
|
|
Static {
|
|
width: auto;
|
|
}
|
|
|
|
|
|
}
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
author: str,
|
|
name: str,
|
|
description: str,
|
|
is_fork: bool,
|
|
updated_at: datetime
|
|
):
|
|
super().__init__()
|
|
self.author = author
|
|
self.repo_name = name
|
|
self.description = description
|
|
self.is_fork = is_fork
|
|
self.updated_at = updated_at.replace(tzinfo=None)
|
|
|
|
def compose(self) -> ComposeResult:
|
|
updated_string = time_delta(datetime.now() - self.updated_at)
|
|
yield Static(f"[b]{self.author}[/] / [b]{self.repo_name}[/]{' [d]\[[blue]fork[/]]' if self.is_fork else ''}\n{self.description}\n[d]Updated {updated_string} ago")
|
|
|
|
class SearchScreen(Screen):
|
|
DEFAULT_CSS = """
|
|
#search-box {
|
|
padding: 1;
|
|
padding-bottom: 0;
|
|
border-bottom: hkey $surface;
|
|
|
|
#query {
|
|
width: 1fr;
|
|
}
|
|
|
|
#search-select {
|
|
width: 25;
|
|
}
|
|
}
|
|
"""
|
|
|
|
def action_search_query(self):
|
|
query_input = self.query_one("#query")
|
|
loading = self.query_one(LoadingIndicator)
|
|
results = self.query_one(ScrollableContainer)
|
|
|
|
loading.display = True
|
|
|
|
# send off a request
|
|
response = requests.get(
|
|
url=self.app.GITEA_HOST + "api/v1/repos/search",
|
|
params={
|
|
"q": query_input.value,
|
|
"limit": 20
|
|
}
|
|
)
|
|
|
|
loading.display = False
|
|
|
|
# error handling
|
|
if not response.ok:
|
|
self.notify(response.text, title="Error while getting search results:", severity="error")
|
|
return
|
|
|
|
# display results
|
|
results.remove_children(SearchResult)
|
|
to_mount = []
|
|
print(response.json())
|
|
for result in response.json()["data"]:
|
|
to_mount.append(SearchResult(
|
|
result["owner"]["login"],
|
|
result["name"],
|
|
result["description"],
|
|
result["fork"],
|
|
datetime.fromisoformat(result["updated_at"])
|
|
))
|
|
results.mount_all(to_mount)
|
|
|
|
# self explanitory
|
|
query_input.clear()
|
|
|
|
def on_input_submitted(self, event: Input.Submitted):
|
|
if event.input.id == "query":
|
|
self.action_search_query()
|
|
|
|
def on_mount(self):
|
|
self.action_search_query()
|
|
#self.query_one(LoadingIndicator).display = False
|
|
|
|
def compose(self) -> ComposeResult:
|
|
yield Navbar()
|
|
|
|
with HorizontalGroup(id="search-box"):
|
|
yield Input(placeholder="Search repos...", id="query")
|
|
yield Select.from_values([
|
|
"Repositories",
|
|
], id="search-select", allow_blank=False)
|
|
|
|
with ScrollableContainer(id="results"):
|
|
yield LoadingIndicator() |