mirror of
https://github.com/Hxnxe/CyberSentinel-AI.git
synced 2026-10-24 16:54:26 +00:00
746 lines
29 KiB
Python
746 lines
29 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import time
|
|
from datetime import datetime, timedelta
|
|
from typing import Any
|
|
|
|
from sqlalchemy.engine import Engine
|
|
|
|
from clients.github_client import GithubClient
|
|
from config import GITHUB_KEYWORDS, WATCHED_REPOSITORIES
|
|
from database.repositories import MonitorStateRepo, RepositoryRepo
|
|
from database.session import SessionFactory, get_engine, get_session
|
|
from utils.blacklist_manager import BlacklistManager
|
|
from utils.logger import get_logger
|
|
|
|
logger = get_logger("monitors.github_monitor")
|
|
|
|
|
|
class GithubMonitor:
|
|
"""GitHub 仓库监控器。"""
|
|
|
|
RELEVANT_EXTENSIONS = {
|
|
".yaml",
|
|
".yml",
|
|
".json",
|
|
".md",
|
|
".txt",
|
|
".py",
|
|
".go",
|
|
".java",
|
|
".php",
|
|
".js",
|
|
".sh",
|
|
".rb",
|
|
}
|
|
SECURITY_KEYWORDS = {
|
|
"cve",
|
|
"rce",
|
|
"exploit",
|
|
"poc",
|
|
"payload",
|
|
"shell",
|
|
"bypass",
|
|
"inject",
|
|
"漏洞",
|
|
"利用",
|
|
"安全",
|
|
}
|
|
|
|
def __init__(
|
|
self,
|
|
engine: Engine | None = None,
|
|
*,
|
|
github_client: GithubClient | None = None,
|
|
blacklist_manager: BlacklistManager | None = None,
|
|
session_factory: Any | None = None,
|
|
keywords: list[str] | None = None,
|
|
watched_repositories: list[str] | None = None,
|
|
repo_quality_threshold: float = 0.35,
|
|
) -> None:
|
|
self.engine = engine or get_engine()
|
|
self.session_factory = session_factory or SessionFactory(self.engine)
|
|
self.github_client = github_client or GithubClient()
|
|
self.blacklist_manager = blacklist_manager or BlacklistManager()
|
|
self.keywords = list(keywords or GITHUB_KEYWORDS)
|
|
self.watched_repositories = list(watched_repositories or WATCHED_REPOSITORIES)
|
|
self.repo_quality_threshold = min(max(float(repo_quality_threshold), 0.0), 1.0)
|
|
|
|
logger.info("初始化GitHub监控器")
|
|
logger.info("已加载 %s 个搜索关键词", len(self.keywords))
|
|
logger.info("已加载 %s 个监控仓库", len(self.watched_repositories))
|
|
logger.info("仓库质量阈值: %.2f", self.repo_quality_threshold)
|
|
|
|
@staticmethod
|
|
def _truncate_patch(patch: str, max_length: int = 1200) -> str:
|
|
if not patch:
|
|
return ""
|
|
return patch if len(patch) <= max_length else patch[:max_length] + "..."
|
|
|
|
@staticmethod
|
|
def _record_skip_reason(skip_stats: dict[str, int] | None, reason: str) -> None:
|
|
if skip_stats is None:
|
|
return
|
|
skip_stats[reason] = skip_stats.get(reason, 0) + 1
|
|
|
|
@staticmethod
|
|
def _build_fingerprint(repo_name: str, last_update: datetime | None) -> str:
|
|
if not repo_name:
|
|
return ""
|
|
minute_key = (last_update or datetime.now()).strftime("%Y%m%d%H%M")
|
|
return f"{repo_name}:{minute_key}"
|
|
|
|
def _score_repository_metadata(self, repo: dict[str, Any]) -> tuple[float, list[str]]:
|
|
score = 0.0
|
|
reasons: list[str] = []
|
|
|
|
stars = int(repo.get("stars", 0) or 0)
|
|
if stars >= 100:
|
|
score += 0.25
|
|
reasons.append("high_stars")
|
|
elif stars >= 30:
|
|
score += 0.18
|
|
reasons.append("mid_stars")
|
|
elif stars >= 10:
|
|
score += 0.12
|
|
reasons.append("basic_stars")
|
|
elif stars > 0:
|
|
score += 0.05
|
|
reasons.append("low_stars")
|
|
|
|
repo_name = str(repo.get("name", "") or "").lower()
|
|
description = str(repo.get("description", "") or "").lower()
|
|
search_keyword = str(repo.get("search_keyword", "") or "").strip().lower()
|
|
text = f"{repo_name} {description}".strip()
|
|
|
|
if search_keyword:
|
|
normalized_keyword = search_keyword.replace("+", " ").strip()
|
|
tokens = [token for token in normalized_keyword.split() if len(token) >= 3]
|
|
if normalized_keyword and normalized_keyword in text:
|
|
score += 0.22
|
|
reasons.append("keyword_exact_match")
|
|
elif any(token in text for token in tokens):
|
|
score += 0.12
|
|
reasons.append("keyword_partial_match")
|
|
|
|
keyword_hits = sum(1 for keyword in self.SECURITY_KEYWORDS if keyword in text)
|
|
if keyword_hits > 0:
|
|
score += min(0.20, keyword_hits * 0.05)
|
|
reasons.append("security_context")
|
|
|
|
if repo.get("is_update"):
|
|
score += 0.08
|
|
reasons.append("updated_repository")
|
|
if repo.get("is_watched"):
|
|
score += 0.08
|
|
reasons.append("watched_repository")
|
|
|
|
return min(score, 1.0), reasons
|
|
|
|
def _score_repository_commits(self, commits: list[dict[str, Any]]) -> tuple[float, list[str]]:
|
|
if not commits:
|
|
return 0.0, ["no_recent_commits"]
|
|
|
|
score = 0.15
|
|
reasons = ["recent_commits"]
|
|
file_signal = 0.0
|
|
patch_signal_hits = 0
|
|
|
|
for commit in commits[:5]:
|
|
if not isinstance(commit, dict):
|
|
continue
|
|
for changed_file in commit.get("files", [])[:20]:
|
|
if not isinstance(changed_file, dict):
|
|
continue
|
|
path = str(changed_file.get("path", "")).lower()
|
|
ext = os.path.splitext(path)[1]
|
|
if ext in self.RELEVANT_EXTENSIONS:
|
|
file_signal += 0.02
|
|
patch = str(changed_file.get("patch", "")).lower()
|
|
if patch and any(keyword in patch for keyword in self.SECURITY_KEYWORDS):
|
|
patch_signal_hits += 1
|
|
|
|
if file_signal > 0:
|
|
score += min(0.18, file_signal)
|
|
reasons.append("relevant_file_changes")
|
|
if patch_signal_hits > 0:
|
|
score += min(0.15, patch_signal_hits * 0.03)
|
|
reasons.append("security_patch_signal")
|
|
|
|
return min(score, 1.0), reasons
|
|
|
|
def _enrich_quality_info(
|
|
self,
|
|
repo_data: dict[str, Any],
|
|
*,
|
|
last_update: datetime,
|
|
recent_commits: list[dict[str, Any]] | None = None,
|
|
) -> dict[str, Any]:
|
|
metadata_score, metadata_reasons = self._score_repository_metadata(repo_data)
|
|
commit_score, commit_reasons = self._score_repository_commits(recent_commits or [])
|
|
|
|
result = dict(repo_data)
|
|
result["quality_score"] = round(min(1.0, metadata_score + commit_score), 3)
|
|
result["quality_reasons"] = [*metadata_reasons, *commit_reasons] or ["insufficient_signal"]
|
|
result["fingerprint"] = self._build_fingerprint(str(result.get("name", "")), last_update)
|
|
result["skip_reason"] = None
|
|
return result
|
|
|
|
def _passes_quality_gate(self, repo_data: dict[str, Any], skip_stats: dict[str, int] | None = None) -> bool:
|
|
score = float(repo_data.get("quality_score", 0.0) or 0.0)
|
|
if repo_data.get("is_watched"):
|
|
if score < self.repo_quality_threshold:
|
|
reasons = list(repo_data.get("quality_reasons", []))
|
|
reasons.append("watched_repository_bypass")
|
|
repo_data["quality_reasons"] = reasons
|
|
return True
|
|
|
|
if score < self.repo_quality_threshold:
|
|
repo_data["skip_reason"] = "low_quality_score"
|
|
self._record_skip_reason(skip_stats, "low_quality_score")
|
|
return False
|
|
return True
|
|
|
|
def _is_blacklisted(self, repo_name: str, owner_login: str) -> bool:
|
|
if self.blacklist_manager.is_blacklisted(repo_name, "repository"):
|
|
return True
|
|
if owner_login and self.blacklist_manager.is_blacklisted(owner_login, "user"):
|
|
return True
|
|
return False
|
|
|
|
def _fetch_readme_content(self, repo_name: str) -> str:
|
|
return self.github_client.get_readme(repo_name) or ""
|
|
|
|
def _fetch_recent_commits(
|
|
self,
|
|
repo_name: str,
|
|
since_time: datetime | None = None,
|
|
max_commits: int = 10,
|
|
) -> list[dict[str, Any]]:
|
|
commits = self.github_client.get_commits(repo_name, since=since_time, per_page=max_commits) or []
|
|
processed: list[dict[str, Any]] = []
|
|
|
|
for commit in commits[:max_commits]:
|
|
commit_url = str(commit.get("url", ""))
|
|
sha = str(commit.get("sha", ""))
|
|
if not commit_url and not sha:
|
|
continue
|
|
|
|
detail = self.github_client.get_commit_detail(repo_name, commit_url or sha)
|
|
if not isinstance(detail, dict):
|
|
continue
|
|
|
|
files: list[dict[str, Any]] = []
|
|
for changed_file in detail.get("files", []):
|
|
if not isinstance(changed_file, dict):
|
|
continue
|
|
|
|
filename = str(changed_file.get("filename", ""))
|
|
if not filename:
|
|
continue
|
|
|
|
ext = os.path.splitext(filename)[1].lower()
|
|
if ext not in self.RELEVANT_EXTENSIONS:
|
|
continue
|
|
|
|
files.append(
|
|
{
|
|
"path": filename,
|
|
"status": changed_file.get("status", "unknown"),
|
|
"additions": int(changed_file.get("additions", 0) or 0),
|
|
"deletions": int(changed_file.get("deletions", 0) or 0),
|
|
"changes": int(changed_file.get("changes", 0) or 0),
|
|
"patch": self._truncate_patch(str(changed_file.get("patch", ""))),
|
|
}
|
|
)
|
|
|
|
processed.append(
|
|
{
|
|
"sha": detail.get("sha", sha),
|
|
"message": str(detail.get("commit", {}).get("message", "")).strip(),
|
|
"author": str(detail.get("commit", {}).get("author", {}).get("name", "")).strip(),
|
|
"date": str(detail.get("commit", {}).get("author", {}).get("date", "")),
|
|
"files": files,
|
|
}
|
|
)
|
|
|
|
return processed
|
|
|
|
def search_repositories(
|
|
self,
|
|
keyword: str,
|
|
*,
|
|
seen_repo_names: set[str] | None = None,
|
|
skip_stats: dict[str, int] | None = None,
|
|
) -> list[dict[str, Any]]:
|
|
"""按关键词搜索需要分析的仓库。"""
|
|
repos: list[dict[str, Any]] = []
|
|
per_page = 25
|
|
max_pages = 2
|
|
since_date = datetime.now().strftime("%Y-%m-%d")
|
|
metadata_floor = max(0.15, self.repo_quality_threshold - 0.2)
|
|
|
|
processed_keyword = keyword
|
|
if " AND " in keyword or " OR " in keyword:
|
|
processed_keyword = keyword.replace(" AND ", "+").replace(" OR ", "+OR+")
|
|
processed_keyword = processed_keyword.replace("(", "").replace(")", "")
|
|
|
|
with get_session(self.session_factory) as session:
|
|
repo_repo = RepositoryRepo(session)
|
|
|
|
for page in range(1, max_pages + 1):
|
|
query = f"{processed_keyword} pushed:>={since_date}"
|
|
data = self.github_client.search_repos(query, page=page, per_page=per_page)
|
|
if not data or not isinstance(data, dict):
|
|
break
|
|
|
|
items = data.get("items")
|
|
if not isinstance(items, list) or not items:
|
|
break
|
|
|
|
for item in items:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
|
|
owner_login = str(item.get("owner", {}).get("login", ""))
|
|
repo_name = str(item.get("full_name", ""))
|
|
if not repo_name or self._is_blacklisted(repo_name, owner_login):
|
|
self._record_skip_reason(skip_stats, "blacklisted_or_invalid_repo")
|
|
continue
|
|
|
|
if seen_repo_names is not None and repo_name in seen_repo_names:
|
|
self._record_skip_reason(skip_stats, "duplicate_repo_name")
|
|
continue
|
|
|
|
pushed_at = str(item.get("pushed_at", ""))
|
|
try:
|
|
last_update = datetime.strptime(pushed_at, "%Y-%m-%dT%H:%M:%SZ")
|
|
except Exception:
|
|
self._record_skip_reason(skip_stats, "invalid_pushed_at")
|
|
continue
|
|
|
|
existing = repo_repo.get_by_name(repo_name)
|
|
if existing and existing.last_update:
|
|
time_diff = abs((last_update - existing.last_update).total_seconds())
|
|
if time_diff <= 60:
|
|
self._record_skip_reason(skip_stats, "stale_update")
|
|
continue
|
|
|
|
repo_info = {
|
|
"name": repo_name,
|
|
"url": item.get("html_url"),
|
|
"description": str(item.get("description", "") or "")[:500],
|
|
"last_update": last_update,
|
|
"stars": int(item.get("stargazers_count", 0) or 0),
|
|
"is_valuable": int(item.get("stargazers_count", 0) or 0) > 10,
|
|
"search_keyword": keyword,
|
|
"owner": owner_login,
|
|
}
|
|
lightweight_score, _ = self._score_repository_metadata(repo_info)
|
|
if lightweight_score < metadata_floor:
|
|
self._record_skip_reason(skip_stats, "low_metadata_signal")
|
|
continue
|
|
|
|
repos.append(repo_info)
|
|
if seen_repo_names is not None:
|
|
seen_repo_names.add(repo_name)
|
|
|
|
if len(items) < per_page:
|
|
break
|
|
time.sleep(0.2)
|
|
|
|
return repos
|
|
|
|
def check_watched_repositories(self) -> list[dict[str, Any]]:
|
|
"""检查固定监控仓库是否有更新。"""
|
|
updates: list[dict[str, Any]] = []
|
|
|
|
with get_session(self.session_factory) as session:
|
|
repo_repo = RepositoryRepo(session)
|
|
|
|
for repo_name in self.watched_repositories:
|
|
try:
|
|
info = self.github_client.get_repository(repo_name)
|
|
if not info:
|
|
continue
|
|
|
|
pushed_at = str(info.get("pushed_at", ""))
|
|
last_update = datetime.strptime(pushed_at, "%Y-%m-%dT%H:%M:%SZ")
|
|
|
|
existing = repo_repo.get_by_name(repo_name)
|
|
if existing and existing.last_update and last_update <= existing.last_update:
|
|
continue
|
|
|
|
since_time = existing.last_update if existing and existing.last_update else datetime.now() - timedelta(days=1)
|
|
recent_commits = self._fetch_recent_commits(repo_name, since_time=since_time, max_commits=5)
|
|
if not recent_commits:
|
|
continue
|
|
|
|
if existing:
|
|
repo_repo.update_analysis(
|
|
repo_name,
|
|
{
|
|
"last_update": last_update,
|
|
"description": str(info.get("description", "") or ""),
|
|
"url": info.get("html_url"),
|
|
},
|
|
)
|
|
else:
|
|
repo_repo.create(
|
|
{
|
|
"name": repo_name,
|
|
"description": str(info.get("description", "") or ""),
|
|
"last_update": last_update,
|
|
"url": info.get("html_url"),
|
|
}
|
|
)
|
|
|
|
updates.append(
|
|
{
|
|
"name": repo_name,
|
|
"url": info.get("html_url"),
|
|
"description": str(info.get("description", "") or ""),
|
|
"stars": int(info.get("stargazers_count", 0) or 0),
|
|
"last_update": last_update,
|
|
"recent_commits": recent_commits,
|
|
"is_update": bool(existing),
|
|
"is_watched": True,
|
|
}
|
|
)
|
|
except Exception as exc:
|
|
logger.warning("检查监控仓库失败 %s: %s", repo_name, str(exc))
|
|
|
|
return updates
|
|
|
|
def _process_watched_repository(
|
|
self,
|
|
repo: dict[str, Any],
|
|
repo_name: str,
|
|
repo_url: str,
|
|
last_update: datetime,
|
|
existing_repo: Any,
|
|
repo_repo: RepositoryRepo,
|
|
skip_stats: dict[str, int] | None = None,
|
|
) -> dict[str, Any] | None:
|
|
if existing_repo and existing_repo.last_update and last_update <= existing_repo.last_update:
|
|
self._record_skip_reason(skip_stats, "watched_no_new_update")
|
|
return None
|
|
|
|
since_time = existing_repo.last_update if existing_repo and existing_repo.last_update else None
|
|
recent_commits = self._fetch_recent_commits(repo_name, since_time=since_time, max_commits=5)
|
|
if not recent_commits:
|
|
self._record_skip_reason(skip_stats, "watched_no_recent_commits")
|
|
return None
|
|
|
|
if existing_repo:
|
|
repo_repo.update_analysis(
|
|
repo_name,
|
|
{
|
|
"last_update": last_update,
|
|
"description": str(repo.get("description", "") or ""),
|
|
"url": repo_url,
|
|
},
|
|
)
|
|
else:
|
|
repo_repo.create(
|
|
{
|
|
"name": repo_name,
|
|
"description": str(repo.get("description", "") or ""),
|
|
"last_update": last_update,
|
|
"url": repo_url,
|
|
}
|
|
)
|
|
|
|
repo_data = {
|
|
"name": repo_name,
|
|
"url": repo_url,
|
|
"description": str(repo.get("description", "") or ""),
|
|
"stars": int(repo.get("stars", 0) or 0),
|
|
"last_update": last_update,
|
|
"recent_commits": recent_commits,
|
|
"is_update": bool(existing_repo),
|
|
"is_watched": True,
|
|
}
|
|
enriched = self._enrich_quality_info(repo_data, last_update=last_update, recent_commits=recent_commits)
|
|
if not self._passes_quality_gate(enriched, skip_stats):
|
|
return None
|
|
return enriched
|
|
|
|
def _process_normal_repository(
|
|
self,
|
|
repo: dict[str, Any],
|
|
repo_name: str,
|
|
repo_url: str,
|
|
last_update: datetime,
|
|
existing_repo: Any,
|
|
repo_repo: RepositoryRepo,
|
|
skip_stats: dict[str, int] | None = None,
|
|
) -> dict[str, Any] | None:
|
|
if repo_name in self.watched_repositories:
|
|
self._record_skip_reason(skip_stats, "already_in_watched_list")
|
|
return None
|
|
|
|
if existing_repo:
|
|
if existing_repo.last_update:
|
|
try:
|
|
time_diff = (last_update - existing_repo.last_update).total_seconds()
|
|
except Exception:
|
|
time_diff = float("inf")
|
|
if time_diff < 60:
|
|
self._record_skip_reason(skip_stats, "stale_update")
|
|
return None
|
|
|
|
recent_commits = self._fetch_recent_commits(
|
|
repo_name,
|
|
since_time=existing_repo.last_update,
|
|
max_commits=5,
|
|
)
|
|
if not recent_commits:
|
|
self._record_skip_reason(skip_stats, "no_recent_commits")
|
|
return None
|
|
|
|
repo_data = {
|
|
"name": repo_name,
|
|
"url": repo_url,
|
|
"description": str(repo.get("description", "") or ""),
|
|
"stars": int(repo.get("stars", 0) or 0),
|
|
"last_update": last_update,
|
|
"recent_commits": recent_commits,
|
|
"search_keyword": repo.get("search_keyword"),
|
|
"is_update": True,
|
|
"is_watched": False,
|
|
}
|
|
enriched = self._enrich_quality_info(repo_data, last_update=last_update, recent_commits=recent_commits)
|
|
if not self._passes_quality_gate(enriched, skip_stats):
|
|
return None
|
|
|
|
repo_repo.update_analysis(
|
|
repo_name,
|
|
{
|
|
"last_update": last_update,
|
|
"description": str(repo.get("description", "") or ""),
|
|
"url": repo_url,
|
|
},
|
|
)
|
|
return enriched
|
|
|
|
recent_commits = repo.get("recent_commits") or self._fetch_recent_commits(repo_name, max_commits=5)
|
|
if not recent_commits:
|
|
self._record_skip_reason(skip_stats, "no_recent_commits")
|
|
return None
|
|
|
|
repo_data = {
|
|
"name": repo_name,
|
|
"url": repo_url,
|
|
"description": str(repo.get("description", "") or ""),
|
|
"stars": int(repo.get("stars", 0) or 0),
|
|
"recent_commits": recent_commits,
|
|
"search_keyword": repo.get("search_keyword"),
|
|
"is_update": False,
|
|
"is_watched": False,
|
|
"last_update": last_update,
|
|
}
|
|
enriched = self._enrich_quality_info(repo_data, last_update=last_update, recent_commits=recent_commits)
|
|
if not self._passes_quality_gate(enriched, skip_stats):
|
|
return None
|
|
|
|
readme_content = self._fetch_readme_content(repo_name)
|
|
enriched["readme"] = readme_content
|
|
|
|
repo_repo.create(
|
|
{
|
|
"name": repo_name,
|
|
"description": str(repo.get("description", "") or ""),
|
|
"last_update": last_update,
|
|
"url": repo_url,
|
|
}
|
|
)
|
|
return enriched
|
|
|
|
def process_repository_data(
|
|
self,
|
|
repos: list[dict[str, Any]],
|
|
is_watched: bool = False,
|
|
*,
|
|
seen_fingerprints: set[str] | None = None,
|
|
skip_stats: dict[str, int] | None = None,
|
|
) -> list[dict[str, Any]]:
|
|
"""处理并存储仓库数据。"""
|
|
processed: list[dict[str, Any]] = []
|
|
|
|
with get_session(self.session_factory) as session:
|
|
repo_repo = RepositoryRepo(session)
|
|
|
|
for repo in repos:
|
|
if not isinstance(repo, dict):
|
|
self._record_skip_reason(skip_stats, "invalid_repo_payload")
|
|
continue
|
|
|
|
repo_name = str(repo.get("full_name") or repo.get("name") or "")
|
|
repo_url = str(repo.get("html_url") or repo.get("url") or "")
|
|
if not repo_name or not repo_url:
|
|
self._record_skip_reason(skip_stats, "missing_repo_fields")
|
|
continue
|
|
|
|
owner = repo.get("owner", {})
|
|
owner_name = owner.get("login") if isinstance(owner, dict) else repo.get("owner")
|
|
if self._is_blacklisted(repo_name, str(owner_name or "")):
|
|
self._record_skip_reason(skip_stats, "blacklisted_or_invalid_repo")
|
|
continue
|
|
|
|
pushed_at = repo.get("pushed_at")
|
|
if pushed_at:
|
|
try:
|
|
last_update = datetime.strptime(str(pushed_at), "%Y-%m-%dT%H:%M:%SZ")
|
|
except Exception:
|
|
self._record_skip_reason(skip_stats, "invalid_pushed_at")
|
|
last_update = datetime.now()
|
|
else:
|
|
last_update = repo.get("last_update") if isinstance(repo.get("last_update"), datetime) else datetime.now()
|
|
|
|
existing = repo_repo.get_by_name(repo_name)
|
|
if is_watched:
|
|
repo_data = self._process_watched_repository(
|
|
repo,
|
|
repo_name,
|
|
repo_url,
|
|
last_update,
|
|
existing,
|
|
repo_repo,
|
|
skip_stats,
|
|
)
|
|
else:
|
|
repo_data = self._process_normal_repository(
|
|
repo,
|
|
repo_name,
|
|
repo_url,
|
|
last_update,
|
|
existing,
|
|
repo_repo,
|
|
skip_stats,
|
|
)
|
|
|
|
if repo_data:
|
|
if seen_fingerprints is not None:
|
|
fingerprint = str(repo_data.get("fingerprint", "") or "")
|
|
if not fingerprint:
|
|
self._record_skip_reason(skip_stats, "missing_fingerprint")
|
|
continue
|
|
if fingerprint in seen_fingerprints:
|
|
self._record_skip_reason(skip_stats, "duplicate_fingerprint")
|
|
continue
|
|
seen_fingerprints.add(fingerprint)
|
|
processed.append(repo_data)
|
|
|
|
return processed
|
|
|
|
def _update_state(self, status: str, error_message: str | None = None, success: bool = False) -> None:
|
|
with get_session(self.session_factory) as session:
|
|
state_repo = MonitorStateRepo(session)
|
|
state_repo.update_state(
|
|
"github",
|
|
status=status,
|
|
error_message=error_message,
|
|
success=success,
|
|
)
|
|
|
|
def run(self, interval: int = 3600) -> list[dict[str, Any]]:
|
|
"""执行一轮 GitHub 监控并返回待分析仓库。"""
|
|
logger.info("GitHub监控器启动")
|
|
all_repos: list[dict[str, Any]] = []
|
|
skip_stats: dict[str, int] = {}
|
|
seen_repo_names: set[str] = set()
|
|
seen_fingerprints: set[str] = set()
|
|
|
|
try:
|
|
self._update_state("running")
|
|
|
|
watched_updates = self.check_watched_repositories()
|
|
if watched_updates:
|
|
accepted_watched = 0
|
|
for repo in watched_updates:
|
|
if not isinstance(repo, dict):
|
|
self._record_skip_reason(skip_stats, "invalid_repo_payload")
|
|
continue
|
|
|
|
repo_name = str(repo.get("name", "") or "")
|
|
if not repo_name:
|
|
self._record_skip_reason(skip_stats, "missing_repo_fields")
|
|
continue
|
|
if repo_name in seen_repo_names:
|
|
self._record_skip_reason(skip_stats, "duplicate_repo_name")
|
|
continue
|
|
seen_repo_names.add(repo_name)
|
|
|
|
last_update = repo.get("last_update") if isinstance(repo.get("last_update"), datetime) else datetime.now()
|
|
recent_commits = repo.get("recent_commits") if isinstance(repo.get("recent_commits"), list) else []
|
|
enriched = self._enrich_quality_info(repo, last_update=last_update, recent_commits=recent_commits)
|
|
if not self._passes_quality_gate(enriched, skip_stats):
|
|
continue
|
|
|
|
fingerprint = str(enriched.get("fingerprint", "") or "")
|
|
if not fingerprint:
|
|
self._record_skip_reason(skip_stats, "missing_fingerprint")
|
|
continue
|
|
if fingerprint in seen_fingerprints:
|
|
self._record_skip_reason(skip_stats, "duplicate_fingerprint")
|
|
continue
|
|
seen_fingerprints.add(fingerprint)
|
|
|
|
all_repos.append(enriched)
|
|
accepted_watched += 1
|
|
|
|
logger.info("发现 %s 个特定仓库更新，入队 %s 个", len(watched_updates), accepted_watched)
|
|
else:
|
|
logger.info("特定仓库无更新")
|
|
|
|
for keyword in self.keywords:
|
|
repos = self.search_repositories(
|
|
keyword,
|
|
seen_repo_names=seen_repo_names,
|
|
skip_stats=skip_stats,
|
|
)
|
|
if not repos:
|
|
continue
|
|
|
|
repos_to_analyze = self.process_repository_data(
|
|
repos,
|
|
is_watched=False,
|
|
seen_fingerprints=seen_fingerprints,
|
|
skip_stats=skip_stats,
|
|
)
|
|
if repos_to_analyze:
|
|
all_repos.extend(repos_to_analyze)
|
|
logger.info("关键词 %s 发现 %s 个仓库", keyword, len(repos_to_analyze))
|
|
|
|
total_skipped = sum(skip_stats.values())
|
|
top_skip_reasons = ", ".join(
|
|
f"{reason}:{count}"
|
|
for reason, count in sorted(skip_stats.items(), key=lambda item: item[1], reverse=True)[:5]
|
|
) or "none"
|
|
|
|
logger.info(
|
|
"仓库筛选统计: accepted=%s, skipped=%s, top_skip_reasons=%s",
|
|
len(all_repos),
|
|
total_skipped,
|
|
top_skip_reasons,
|
|
)
|
|
self._update_state("success", success=True)
|
|
if all_repos:
|
|
logger.info("本轮总计发现 %s 个需要分析的仓库", len(all_repos))
|
|
else:
|
|
logger.info("本轮未发现需要分析的仓库")
|
|
return all_repos
|
|
|
|
except Exception as exc:
|
|
logger.error("GitHub监控失败: %s", str(exc), exc_info=True)
|
|
self._update_state("failed", error_message=str(exc), success=False)
|
|
time.sleep(min(interval, 300))
|
|
return []
|
|
|
|
|
|
if __name__ == "__main__":
|
|
monitor = GithubMonitor()
|
|
monitor.run()
|