implement first rules

jg/hen/cli.py

@@ -1,9 +1,14 @@
+import asyncio
 import logging
+from dataclasses import asdict
+from pprint import pprint
 
 import click
 
+from jg.hen.core import check_profile_url
 
-logger = logging.getLogger("jg.hen")
+
+logger = logging.getLogger("jg.hen.cli")
 
 
 @click.command()
@@ -12,3 +17,5 @@
 def main(profile_url: str, debug: bool):
     logging.basicConfig(level=logging.DEBUG if debug else logging.INFO)
     logger.info(f"URL: {profile_url}")
+    result = asyncio.run(check_profile_url(profile_url, raise_on_error=debug))
+    pprint(asdict(result))

jg/hen/core.py

@@ -0,0 +1,135 @@
+import logging
+from dataclasses import dataclass
+from enum import StrEnum, auto
+from functools import wraps
+from typing import Callable, Coroutine, Literal
+from urllib.parse import urlparse
+
+import blinker
+import httpx
+from githubkit import GitHub
+
+
+USER_AGENT = "JuniorGuruBot (+https://junior.guru)"
+
+
+logger = logging.getLogger("jg.hen.core")
+
+
+on_profile = blinker.Signal()
+on_avatar = blinker.Signal()
+
+
+class ResultType(StrEnum):
+    DONE = auto()
+    RECOMMENDATION = auto()
+
+
+@dataclass
+class Result:
+    rule: str
+    type: ResultType
+    message: str
+    docs_url: str
+
+
+@dataclass
+class Context:
+    profile_url: str
+    username: str
+
+
+@dataclass
+class Summary:
+    status: Literal["ok", "error"]
+    results: list[Result]
+    error: Exception | None = None
+
+
+def with_github(fn: Callable[..., Coroutine]) -> Callable[..., Coroutine]:
+    @wraps(fn)
+    async def wrapper(*args, **kwargs) -> Coroutine:
+        async with GitHub(user_agent=USER_AGENT) as github:
+            return await fn(*args, github=github, **kwargs)
+
+    return wrapper
+
+
+def with_http(fn: Callable[..., Coroutine]) -> Callable[..., Coroutine]:
+    @wraps(fn)
+    async def wrapper(*args, **kwargs) -> Coroutine:
+        async with httpx.AsyncClient(
+            follow_redirects=True, headers={"User-Agent": USER_AGENT}
+        ) as client:
+            return await fn(*args, http=client, **kwargs)
+
+    return wrapper
+
+
+@with_github
+@with_http
+async def check_profile_url(
+    profile_url: str,
+    github: GitHub,
+    http: httpx.AsyncClient,
+    raise_on_error: bool = False,
+) -> Summary:
+    results = []
+    try:
+        username = parse_username(profile_url)
+        context = Context(profile_url=profile_url, username=username)
+
+        import jg.hen.rules  # noqa
+
+        response = await github.rest.users.async_get_by_username(username)
+        user = response.parsed_data
+        results.extend(collect_results(await on_profile.send_async(context, user=user)))
+
+        response = await http.get(user.avatar_url)
+        results.extend(
+            collect_results(await on_avatar.send_async(context, avatar=response))
+        )
+    except Exception as error:
+        if raise_on_error:
+            raise
+        return Summary(status="error", results=results, error=error)
+    return Summary(status="ok", results=results)
+
+
+def collect_results(
+    raw_results: list[tuple[Callable, Result | None]],
+) -> list[Result]:
+    return [result for _, result in raw_results if result]
+
+
+def rule(signal: blinker.Signal, docs_url: str) -> Callable:
+    def decorator(fn: Callable[..., Coroutine]) -> Callable[..., Coroutine]:
+        @wraps(fn)
+        async def wrapper(*args, **kwargs) -> Result | None:
+            try:
+                result = await fn(*args, **kwargs)
+                if result is None:
+                    raise NotImplementedError(
+                        f"Rule {fn.__name__!r} returned no result"
+                    )
+                return Result(
+                    rule=fn.__name__,
+                    type=result[0],
+                    message=result[1],
+                    docs_url=docs_url,
+                )
+            except NotImplementedError:
+                logger.warning(f"Rule {fn.__name__!r} not implemented")
+                return None
+
+        signal.connect(wrapper)
+        return wrapper
+
+    return decorator
+
+
+def parse_username(profile_url: str) -> str:
+    parts = urlparse(profile_url)
+    if not parts.netloc.endswith("github.com"):
+        raise ValueError("Only GitHub profiles are supported")
+    return parts.path.strip("/")

jg/hen/rules.py

@@ -0,0 +1,57 @@
+from io import BytesIO
+
+import httpx
+from githubkit.rest import PublicUser
+from PIL import Image
+
+from jg.hen.core import Context, ResultType, on_avatar, on_profile, rule
+
+
+IDENTICON_GREY = (240, 240, 240)
+
+
+@rule(
+    on_avatar,
+    "https://junior.guru/handbook/github-profile/#nastav-si-vlastni-obrazek",
+)
+async def has_avatar(
+    context: Context, avatar: httpx.Response
+) -> tuple[ResultType, str]:
+    try:
+        avatar.raise_for_status()
+    except httpx.HTTPStatusError as e:
+        raise RuntimeError(f"Failed to fetch avatar: {e}") from e
+
+    with Image.open(BytesIO(avatar.content)) as image:
+        colors = image.getcolors()
+
+    if (
+        colors is not None  # has less than 256 colors
+        and len(colors) <= 2  # has 2 or less colors
+        and (IDENTICON_GREY in [color[1] for color in colors])  # identicon gray
+    ):
+        return ResultType.RECOMMENDATION, "Nastav si profilový obrázek."
+    return ResultType.DONE, "Máš nastavený vlastní profilový obrázek. Super!"
+
+
+@rule(
+    on_profile,
+    "https://junior.guru/handbook/github-profile/#vypln-si-zakladni-udaje",
+)
+async def has_name(context: Context, user: PublicUser) -> tuple[ResultType, str]:
+    if user.name:
+        return ResultType.DONE, f"Vidím, že jméno máš vyplněné ({user.name}). Super!"
+    return ResultType.RECOMMENDATION, "Doplň si jméno."
+
+
+@rule(
+    on_profile,
+    "https://junior.guru/handbook/github-profile/#vypln-si-zakladni-udaje",
+)
+async def has_location(context: Context, user: PublicUser) -> tuple[ResultType, str]:
+    if user.location:
+        return (
+            ResultType.DONE,
+            f"Vidím, že lokaci máš vyplněnou ({user.location}). Super!",
+        )
+    return ResultType.RECOMMENDATION, "Doplň si lokaci."