Skip to content

Instantly share code, notes, and snippets.

@Mic92 Mic92/nixos-search.py

Last active May 12, 2020
Embed
What would you like to do?
#! nix-shell -i python3 -p python3 python3Packages.elasticsearch
from elasticsearch import Elasticsearch
import logging
import sys
from typing import List, Any, Dict, DefaultDict
from dataclasses import dataclass
from collections import defaultdict
@dataclass
class Maintainer:
name: str
email: str
github: str
@dataclass
class License:
full_name: str
url: str
@dataclass
class Attr:
name: str
attr_name: str
platforms: List[str]
maintainers: List[Maintainer]
license: List[License]
channels: List[str]
URL = "https://nixos-search-5886075189.us-east-1.bonsaisearch.net:443"
USERNAME = "z3ZFJ6y2mR"
PASSWORD = "ds8CEvALPf9pui7XG"
def es_search(term: str) -> List[Dict[str, Any]]:
es = Elasticsearch(
[URL],
http_auth=(USERNAME, PASSWORD),
scheme="https",
port=443)
body = dict(query=dict(multi_match=dict(
query=term,
fuzziness="AUTO",
fields=[
"name^3",
"attr_name^3",
"homepage^2",
"description",
"long_description"
])))
res = es.search(body=body)
return res["hits"]["hits"]
def group_by_attr_name(raw_attrs: List[Attr]) -> List[Attr]:
grouped_attrs: DefaultDict[str, List[Attr]] = defaultdict(list)
for attr in raw_attrs:
grouped_attrs[attr.attr_name].append(attr)
aggregated_attrs = []
for attr_name, attrs in grouped_attrs.items():
newest = attrs[0]
for attr in attrs[1:]:
if attr.channels[0] > newest.channels[0]:
newest.channels.extend(attr.channels)
else:
attr.channels.extend(newest.channels)
newest = attr
aggregated_attrs.append(newest)
return aggregated_attrs
def query(term: str) -> List[Attr]:
results = es_search(term)
attrs = []
for res in results:
pkg = res["_source"]
maintainers = [Maintainer(**m) for m in pkg["maintainers"]]
license = []
for l in pkg["license"]:
license.append(License(full_name=l["fullName"], url=l["url"]))
channels = res["_index"].replace("-packages", "").replace("nixos-", "")
attr = Attr(attr_name=pkg["attr_name"],
name=pkg["name"],
platforms=pkg["platforms"],
maintainers=maintainers,
channels=[channels],
license=license)
attrs.append(attr)
return group_by_attr_name(attrs)
def main() -> None:
if len(sys.argv) < 2:
print(f"USAGE: {sys.argv[0]} SEARCH-TERM", file=sys.stderr)
sys.exit(1)
for attr in query(sys.argv[1]):
channel = ", ".join(attr.channels)
print(f"{attr.attr_name} - {channel}")
if __name__ == '__main__':
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.