Spaces:
Sleeping
Sleeping
| import logging | |
| import re | |
| import json | |
| from typing import Dict, Any, Optional | |
| import httpx | |
| from app.core.model_loader import llm_engine | |
| logger = logging.getLogger(__name__) | |
| class ConventionService: | |
| def __init__(self): | |
| self.cache: Dict[str, Dict[str, Any]] = {} | |
| async def get_conventions(self, owner: str, repo: str) -> Dict[str, Any]: | |
| key = f"{owner}/{repo}" | |
| if key in self.cache: | |
| return self.cache[key] | |
| # 1. Fetch recent commits from GitHub | |
| commits = await self._fetch_recent_commits(owner, repo, limit=40) | |
| if not commits: | |
| return self._default_conventions() | |
| commit_text = "\n".join([c.get("message", "")[:200] for c in commits[:30]]) | |
| # 2. Ask the LLM to infer conventions | |
| prompt = f"""Analyze these recent commit messages and return ONLY a valid JSON object with these keys: | |
| - "commit_style": either "conventional_commits" or "plain" | |
| - "branch_naming_pattern": a short regex or example pattern (e.g. "feat/.*", "feature/*") | |
| - "merge_policy": "squash", "merge", "rebase" or "unknown" | |
| - "exampleGoodCommit": one good example from the history | |
| Commit history: | |
| {commit_text} | |
| """ | |
| try: | |
| raw = llm_engine.generate(prompt, max_tokens=250) | |
| conventions = self._parse_json(raw) | |
| except Exception as e: | |
| logger.warning(f"LLM convention inference failed: {e}") | |
| conventions = self._default_conventions() | |
| # Enrich | |
| if commits: | |
| conventions["exampleGoodCommit"] = commits[0].get("message", "")[:120] | |
| conventions.setdefault("commit_style", "plain") | |
| conventions.setdefault("branch_naming_pattern", "feature/*") | |
| conventions.setdefault("merge_policy", "unknown") | |
| self.cache[key] = conventions | |
| return conventions | |
| async def _fetch_recent_commits(self, owner: str, repo: str, limit: int = 30) -> list: | |
| url = f"https://api.github.com/repos/{owner}/{repo}/commits" | |
| params = {"per_page": min(limit, 100)} | |
| headers = {"Accept": "application/vnd.github.v3+json"} | |
| # Optional: add a GitHub token if you have one in env for higher rate limits | |
| token = None # os.getenv("GITHUB_TOKEN") | |
| if token: | |
| headers["Authorization"] = f"token {token}" | |
| try: | |
| async with httpx.AsyncClient(timeout=15.0) as client: | |
| resp = await client.get(url, params=params, headers=headers) | |
| if resp.status_code != 200: | |
| logger.warning(f"GitHub commits API {resp.status_code}") | |
| return [] | |
| data = resp.json() | |
| return [ | |
| { | |
| "message": c.get("commit", {}).get("message", ""), | |
| "sha": c.get("sha", ""), | |
| "author": c.get("commit", {}).get("author", {}).get("name", ""), | |
| } | |
| for c in data | |
| ] | |
| except Exception as e: | |
| logger.error(f"Failed to fetch commits: {e}") | |
| return [] | |
| def _parse_json(self, text: str) -> dict: | |
| text = text.strip() | |
| text = re.sub(r"```(?:json)?", "", text, flags=re.IGNORECASE).strip() | |
| try: | |
| return json.loads(text) | |
| except json.JSONDecodeError: | |
| start = text.find("{") | |
| end = text.rfind("}") | |
| if start != -1 and end > start: | |
| try: | |
| return json.loads(text[start : end + 1]) | |
| except Exception: | |
| pass | |
| return self._default_conventions() | |
| def _default_conventions(self) -> dict: | |
| return { | |
| "commit_style": "plain", | |
| "branch_naming_pattern": "feature/*", | |
| "merge_policy": "unknown", | |
| "exampleGoodCommit": "Update something", | |
| } | |
| # Singleton-style instance | |
| convention_service = ConventionService() |