Add multi-username activity support
ci / test (push) Successful in 10s
ci / push-image (push) Successful in 52s
ci / deploy-coolify (push) Successful in 6s

This commit is contained in:
Space-Banane
2026-06-04 22:42:48 +02:00
parent ad792b8137
commit 4444174365
5 changed files with 142 additions and 44 deletions
+9 -2
View File
@@ -62,9 +62,16 @@ Copy `.env.example` to `.env` and edit values.
Required: Required:
- `GITHUB_USERNAME` - `GITHUB_USERNAME` (supports a comma-separated list, for example `octocat,hubot`)
- `GITEA_BASE_URL` - `GITEA_BASE_URL`
- `GITEA_USERNAME` - `GITEA_USERNAME` (supports a comma-separated list)
Example:
```env
GITHUB_USERNAME=octocat,hubot
GITEA_USERNAME=alice,bob
```
Optional: Optional:
+43 -28
View File
@@ -114,6 +114,33 @@ async def _fetch_with_cache(
raise raise
async def _fetch_source_usernames(
cache: FileCache | None,
source_prefix: str,
usernames: list[str],
from_date: date,
to_date: date,
fetcher_factory,
) -> tuple[dict[str, int], bool]:
if not usernames:
return {}, False
cache_usernames = hashlib.sha1(",".join(sorted(usernames)).encode("utf-8")).hexdigest()[:12]
fetch_tasks: list[asyncio.Task[tuple[dict[str, int], bool]]] = []
for username in usernames:
key = f"{source_prefix}_{cache_usernames}_{username}_{from_date}_{to_date}"
fetch_tasks.append(asyncio.create_task(_fetch_with_cache(cache, key, fetcher_factory(username))))
combined: dict[str, int] = {}
stale = False
for data, entry_stale in await asyncio.gather(*fetch_tasks):
stale = stale or entry_stale
for day, count in data.items():
combined[day] = combined.get(day, 0) + int(count)
return combined, stale
async def collect_merged_activity( async def collect_merged_activity(
settings: Settings, settings: Settings,
cache: FileCache | None, cache: FileCache | None,
@@ -121,57 +148,45 @@ async def collect_merged_activity(
) -> ActivityResult: ) -> ActivityResult:
from_date, to_date, days_count = compute_date_range(options) from_date, to_date, days_count = compute_date_range(options)
gh_key = f"github_{settings.github_username}_{from_date}_{to_date}"
gt_key = f"gitea_{settings.gitea_username}_{from_date}_{to_date}"
github_data: dict[str, int] = {} github_data: dict[str, int] = {}
gitea_data: dict[str, int] = {} gitea_data: dict[str, int] = {}
github_stale = False github_stale = False
gitea_stale = False gitea_stale = False
fetch_names: list[str] = []
fetch_tasks: list[asyncio.Task[tuple[dict[str, int], bool]]] = []
if options.source in ("all", "github"): if options.source in ("all", "github"):
fetch_names.append("github") github_data, github_stale = await _fetch_source_usernames(
fetch_tasks.append(
asyncio.create_task(
_fetch_with_cache(
cache, cache,
gh_key, "github",
settings.github_usernames,
from_date,
to_date,
lambda username: (
lambda: fetch_github_activity( lambda: fetch_github_activity(
username=settings.github_username, username=username,
token=settings.github_token, token=settings.github_token,
from_date=from_date, from_date=from_date,
to_date=to_date, to_date=to_date,
),
),
) )
),
) )
if options.source in ("all", "gitea"): if options.source in ("all", "gitea"):
fetch_names.append("gitea") gitea_data, gitea_stale = await _fetch_source_usernames(
fetch_tasks.append(
asyncio.create_task(
_fetch_with_cache(
cache, cache,
gt_key, "gitea",
settings.gitea_usernames,
from_date,
to_date,
lambda username: (
lambda: fetch_gitea_activity( lambda: fetch_gitea_activity(
base_url=settings.gitea_base_url, base_url=settings.gitea_base_url,
username=settings.gitea_username, username=username,
token=settings.gitea_token, token=settings.gitea_token,
from_date=from_date, from_date=from_date,
to_date=to_date, to_date=to_date,
), )
), ),
) )
)
for name, (data, stale) in zip(fetch_names, await asyncio.gather(*fetch_tasks), strict=True):
if name == "github":
github_data, github_stale = data, stale
else:
gitea_data, gitea_stale = data, stale
merged = merge_activity(github_data, gitea_data, dates=_date_keys(from_date, to_date)) merged = merge_activity(github_data, gitea_data, dates=_date_keys(from_date, to_date))
merged = filter_activity_source(merged, options.source) merged = filter_activity_source(merged, options.source)
+28 -1
View File
@@ -1,9 +1,21 @@
from functools import lru_cache from functools import lru_cache
from pydantic import Field from pydantic import Field, field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict from pydantic_settings import BaseSettings, SettingsConfigDict
def split_usernames(raw: str) -> list[str]:
usernames: list[str] = []
seen: set[str] = set()
for part in raw.split(","):
username = part.strip()
if not username or username in seen:
continue
seen.add(username)
usernames.append(username)
return usernames
class Settings(BaseSettings): class Settings(BaseSettings):
model_config = SettingsConfigDict(env_file=".env", case_sensitive=False) model_config = SettingsConfigDict(env_file=".env", case_sensitive=False)
@@ -20,6 +32,21 @@ class Settings(BaseSettings):
default_theme: str = Field(default="light", alias="DEFAULT_THEME") default_theme: str = Field(default="light", alias="DEFAULT_THEME")
service_title: str = Field(default="git-activity-merge", alias="SERVICE_TITLE") service_title: str = Field(default="git-activity-merge", alias="SERVICE_TITLE")
@field_validator("github_username", "gitea_username")
@classmethod
def _validate_username_list(cls, value: str) -> str:
if not split_usernames(value):
raise ValueError("must contain at least one username")
return value
@property
def github_usernames(self) -> list[str]:
return split_usernames(self.github_username)
@property
def gitea_usernames(self) -> list[str]:
return split_usernames(self.gitea_username)
@lru_cache(maxsize=1) @lru_cache(maxsize=1)
def get_settings() -> Settings: def get_settings() -> Settings:
+40
View File
@@ -0,0 +1,40 @@
from datetime import date
import pytest
from app.main import QueryOptions, collect_merged_activity
from app.settings import Settings
@pytest.mark.asyncio
async def test_collect_merged_activity_combines_multiple_usernames(monkeypatch) -> None:
settings = Settings(
_env_file=None,
GITHUB_USERNAME="octocat, hubot",
GITEA_BASE_URL="https://gitea.example.com",
GITEA_USERNAME="alice, bob",
)
github_calls: list[str] = []
gitea_calls: list[str] = []
async def fake_fetch_github_activity(username, token, from_date, to_date, timeout_seconds=20.0):
github_calls.append(username)
return {"2026-01-01": 1 if username == "octocat" else 2}
async def fake_fetch_gitea_activity(base_url, username, token, from_date, to_date, timeout_seconds=20.0):
gitea_calls.append(username)
return {"2026-01-01": 3 if username == "alice" else 4}
monkeypatch.setattr("app.main.fetch_github_activity", fake_fetch_github_activity)
monkeypatch.setattr("app.main.fetch_gitea_activity", fake_fetch_gitea_activity)
result = await collect_merged_activity(
settings=settings,
cache=None,
options=QueryOptions(year=None, days=1, theme="light", source="all"),
)
assert sorted(github_calls) == ["hubot", "octocat"]
assert sorted(gitea_calls) == ["alice", "bob"]
assert result.merged["2026-01-01"] == {"github": 3, "gitea": 7, "total": 10}
+9
View File
@@ -0,0 +1,9 @@
from app.settings import split_usernames
def test_split_usernames_trims_and_deduplicates() -> None:
assert split_usernames("octocat, hubot, octocat, , gitea-user") == [
"octocat",
"hubot",
"gitea-user",
]