364 lines
14 KiB
Python
364 lines
14 KiB
Python
import os
|
|
import time
|
|
import requests
|
|
from github import Github
|
|
import base64
|
|
import json
|
|
import re
|
|
import concurrent.futures
|
|
from retrying import retry
|
|
from functools import wraps
|
|
|
|
# Список токенов
|
|
TOKENS = [
|
|
'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',
|
|
'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',
|
|
'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',
|
|
'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'
|
|
# Добавьте столько токенов, сколько нужно
|
|
]
|
|
TOKEN_INDEX = 0
|
|
ORG_NAME = 'adeo'
|
|
GITHUB_ENTERPRISE_URL = 'https://github.lmru.tech' # URL вашей инстанции GitHub Enterprise
|
|
OUTPUT_FILE = 'external_repositories.txt'
|
|
CACHE_DIR = 'repo_cache'
|
|
REPOS_CACHE_FILE = 'repos_cache.json'
|
|
MAX_WORKERS = 16 # Начнем с 8 потоков
|
|
|
|
# Регулярное выражение для поиска URL
|
|
url_regex = re.compile(r'https?://[^\s]+')
|
|
excluded_url_pattern = re.compile(r'https?://art\.lmru\.tech')
|
|
|
|
# Уникальные внешние URL
|
|
unique_urls = set()
|
|
|
|
# Создание директории для кэша
|
|
if not os.path.exists(CACHE_DIR):
|
|
os.makedirs(CACHE_DIR)
|
|
|
|
# Декоратор для задержки между запросами
|
|
def rate_limited(func):
|
|
@wraps(func)
|
|
def wrapper(*args, **kwargs):
|
|
time.sleep(1) # Задержка в 1 секунду между запросами
|
|
return func(*args, **kwargs)
|
|
return wrapper
|
|
|
|
# Функция для получения текущего токена
|
|
def get_current_token():
|
|
global TOKEN_INDEX
|
|
token = TOKENS[TOKEN_INDEX % len(TOKENS)]
|
|
TOKEN_INDEX += 1
|
|
return token
|
|
|
|
# Создание экземпляра GitHub API с аутентификацией и кастомным URL
|
|
def create_github_instance():
|
|
token = get_current_token()
|
|
return Github(base_url=f"{GITHUB_ENTERPRISE_URL}/api/v3", login_or_token=token)
|
|
|
|
@rate_limited
|
|
@retry(wait_exponential_multiplier=1000, wait_exponential_max=10000, stop_max_attempt_number=5)
|
|
def get_repo_contents(repo, path=""):
|
|
return repo.get_contents(path)
|
|
|
|
@rate_limited
|
|
@retry(wait_exponential_multiplier=1000, wait_exponential_max=10000, stop_max_attempt_number=5)
|
|
def get_file_content(file_content):
|
|
return base64.b64decode(file_content.content).decode('utf-8')
|
|
|
|
@rate_limited
|
|
def check_rate_limit():
|
|
headers = {
|
|
"Authorization": f"token {get_current_token()}"
|
|
}
|
|
rate_limit_url = f"{GITHUB_ENTERPRISE_URL}/api/v3/rate_limit"
|
|
response = requests.get(rate_limit_url, headers=headers)
|
|
if response.status_code == 200:
|
|
limits = response.json()
|
|
core_limit = limits["resources"]["core"]["limit"]
|
|
core_remaining = limits["resources"]["core"]["remaining"]
|
|
core_reset = limits["resources"]["core"]["reset"]
|
|
print(f"Core limit: {core_limit}")
|
|
print(f"Core remaining: {core_remaining}")
|
|
print(f"Core reset time: {time.strftime('%Y-%m-%d %H:%M:%S', time.gmtime(core_reset))}")
|
|
return core_remaining
|
|
else:
|
|
print(f"Failed to retrieve rate limits: {response.status_code}, {response.text}")
|
|
return None
|
|
|
|
@rate_limited
|
|
def fetch_all_repos():
|
|
repos = []
|
|
g = create_github_instance()
|
|
try:
|
|
for repo in g.get_organization(ORG_NAME).get_repos():
|
|
repos.append(repo.full_name)
|
|
except Exception as e:
|
|
print(f'Error retrieving organization repositories: {str(e)}')
|
|
if '403' in str(e):
|
|
print('Hit rate limit, sleeping for 60 seconds...')
|
|
time.sleep(60)
|
|
print(f"Total repositories retrieved: {len(repos)}") # Логирование количества репозиториев
|
|
return repos
|
|
|
|
def get_all_repos():
|
|
if os.path.exists(REPOS_CACHE_FILE):
|
|
with open(REPOS_CACHE_FILE, 'r') as f:
|
|
repos = json.load(f)
|
|
print(f"Loaded {len(repos)} repositories from cache")
|
|
else:
|
|
repos = fetch_all_repos()
|
|
with open(REPOS_CACHE_FILE, 'w') as f:
|
|
json.dump(repos, f)
|
|
print(f"Saved {len(repos)} repositories to cache")
|
|
return repos
|
|
|
|
def cache_file_path(repo_name, file_path):
|
|
return os.path.join(CACHE_DIR, f"{repo_name.replace('/', '_')}_{file_path.replace('/', '_')}")
|
|
|
|
def is_cached(repo_name, file_path):
|
|
return os.path.exists(cache_file_path(repo_name, file_path))
|
|
|
|
def read_cache(repo_name, file_path):
|
|
with open(cache_file_path(repo_name, file_path), 'r') as f:
|
|
return f.read()
|
|
|
|
def write_cache(repo_name, file_path, content):
|
|
with open(cache_file_path(repo_name, file_path), 'w') as f:
|
|
f.write(content)
|
|
|
|
def find_external_urls(content, repo_name, file_path):
|
|
urls = url_regex.findall(content)
|
|
for url in urls:
|
|
if not excluded_url_pattern.search(url):
|
|
unique_urls.add(url)
|
|
print(f'Found external URL in {repo_name}/{file_path}: {url}')
|
|
|
|
def check_file(repo, file_content):
|
|
file_path = file_content.path
|
|
|
|
if is_cached(repo.name, file_path):
|
|
print(f'Reading from cache: {repo.name}/{file_path}')
|
|
content = read_cache(repo.name, file_path)
|
|
else:
|
|
try:
|
|
# Чтение содержимого файла с повторными попытками
|
|
content = get_file_content(file_content)
|
|
write_cache(repo.name, file_path, content)
|
|
except Exception as e:
|
|
print(f'Error decoding {file_path} in {repo.name}: {str(e)}')
|
|
return
|
|
|
|
# Поиск внешних URL
|
|
find_external_urls(content, repo.name, file_path)
|
|
|
|
def check_and_report_external_repositories(repo):
|
|
try:
|
|
contents = get_repo_contents(repo)
|
|
while contents:
|
|
file_content = contents.pop(0)
|
|
if file_content.type == "dir":
|
|
print(f'Entering directory: {file_content.path}')
|
|
contents.extend(get_repo_contents(repo, file_content.path))
|
|
else:
|
|
print(f'Checking file: {file_content.path}')
|
|
if file_content.path.endswith((
|
|
"package.json", "requirements.txt", "setup.py", "pom.xml", "build.gradle", "Cargo.toml", "composer.json",
|
|
"Makefile", "Dockerfile", "saltfile", "Pipfile", "environment.yml", "Jenkinsfile", "Gemfile", "Rakefile",
|
|
"go.mod", "go.sum", "build.sbt", "Vagrantfile", "shard.yml", "Chart.yaml", "mix.exs", "nimble", "rebar.config",
|
|
"Project.toml", "deps.edn", "shadow-cljs.edn", "Spago.dhall", "flake.nix", "default.nix", "pubspec.yaml",
|
|
"Paket.dependencies", "Packages.swift", "universe.json", "stack.yaml", "hpack.yaml", "default.build", "project.clj"
|
|
)):
|
|
check_file(repo, file_content)
|
|
except Exception as e:
|
|
if '404' in str(e):
|
|
print(f'Skipping empty repository {repo.name}')
|
|
else:
|
|
print(f'Error processing repository {repo.name}: {str(e)}')
|
|
|
|
def process_repository(repo_name):
|
|
print(f'Checking repository: {repo_name}')
|
|
g = create_github_instance()
|
|
repo = g.get_repo(repo_name)
|
|
check_and_report_external_repositories(repo)
|
|
print(f'Finished checking repository: {repo_name}')
|
|
|
|
start_time = time.time()
|
|
|
|
# Проверка лимитов перед началом выполнения
|
|
if check_rate_limit() is None:
|
|
exit(1)
|
|
|
|
# Получение всех репозиториев
|
|
repos = get_all_repos()
|
|
|
|
# Проверка, что репозитории были получены
|
|
if not repos:
|
|
print("No repositories retrieved. Exiting.")
|
|
exit(1)
|
|
|
|
# Использование пула потоков для обработки репозиториев
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
|
|
executor.map(process_repository, repos)
|
|
|
|
end_time = time.time()
|
|
elapsed_time = end_time - start_time
|
|
|
|
# Запись уникальных внешних URL в файл
|
|
with open(OUTPUT_FILE, 'w') as file:
|
|
for url in unique_urls:
|
|
file.write(f"{url}\n")
|
|
|
|
print(f'Unique external URLs have been written to {OUTPUT_FILE}')
|
|
print(f'Total time taken: {elapsed_time / 60:.2f} minutes')
|
|
|
|
# Проверка лимитов после выполнения
|
|
check_rate_limit()
|
|
|
|
import os
|
|
import subprocess
|
|
import time
|
|
import shutil
|
|
import re
|
|
from github import Github
|
|
import concurrent.futures
|
|
import json
|
|
|
|
# Список токенов
|
|
TOKENS = [
|
|
'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',
|
|
'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',
|
|
'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',
|
|
'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'
|
|
# Добавьте столько токенов, сколько нужно
|
|
]
|
|
TOKEN_INDEX = 0
|
|
ORG_NAME = 'adeo'
|
|
GITHUB_ENTERPRISE_URL = 'https://github.lmru.tech'
|
|
CLONE_DIR = '/tmp/github_test' # Укажите путь к директории, куда будут клонироваться репозитории
|
|
REPOS_CACHE_FILE = 'repos_cache.json'
|
|
OUTPUT_FILE = 'external_repositories.txt'
|
|
MAX_WORKERS = 8 # Количество параллельных процессов для клонирования
|
|
BATCH_SIZE = 50 # Количество репозиториев в одной партии для обработки
|
|
|
|
# Регулярное выражение для поиска URL
|
|
url_regex = re.compile(r'https?://[^\s]+')
|
|
excluded_url_pattern = re.compile(r'https?://art\.lmru\.tech')
|
|
|
|
# Уникальные внешние URL
|
|
unique_urls = set()
|
|
|
|
# Функция для получения текущего токена
|
|
def get_current_token():
|
|
global TOKEN_INDEX
|
|
token = TOKENS[TOKEN_INDEX % len(TOKENS)]
|
|
TOKEN_INDEX += 1
|
|
return token
|
|
|
|
# Создание экземпляра GitHub API с аутентификацией и кастомным URL
|
|
def create_github_instance():
|
|
token = get_current_token()
|
|
return Github(base_url=f"{GITHUB_ENTERPRISE_URL}/api/v3", login_or_token=token)
|
|
|
|
# Получение всех репозиториев и сохранение их в файл кэша
|
|
def fetch_all_repos():
|
|
repos = []
|
|
g = create_github_instance()
|
|
try:
|
|
for repo in g.get_organization(ORG_NAME).get_repos():
|
|
repos.append(repo.clone_url)
|
|
except Exception as e:
|
|
print(f'Error retrieving organization repositories: {str(e)}')
|
|
if '403' in str(e):
|
|
print('Hit rate limit, sleeping for 60 seconds...')
|
|
time.sleep(60)
|
|
print(f"Total repositories retrieved: {len(repos)}")
|
|
with open(REPOS_CACHE_FILE, 'w') as f:
|
|
json.dump(repos, f)
|
|
return repos
|
|
|
|
# Получение списка репозиториев из кэша
|
|
def get_all_repos():
|
|
if os.path.exists(REPOS_CACHE_FILE):
|
|
with open(REPOS_CACHE_FILE, 'r') as f:
|
|
repos = json.load(f)
|
|
print(f"Loaded {len(repos)} repositories from cache")
|
|
else:
|
|
repos = fetch_all_repos()
|
|
return repos
|
|
|
|
# Клонирование репозитория
|
|
def clone_repo(repo_url):
|
|
try:
|
|
token = get_current_token()
|
|
repo_name = repo_url.split('/')[-1].replace('.git', '')
|
|
clone_url = repo_url.replace('https://', f'https://{token}:x-oauth-basic@')
|
|
clone_path = os.path.join(CLONE_DIR, repo_name)
|
|
if not os.path.exists(clone_path):
|
|
subprocess.run(['git', 'clone', clone_url, clone_path])
|
|
print(f'Cloned {repo_url} to {clone_path}')
|
|
else:
|
|
print(f'Repository {repo_name} already exists, pulling latest changes')
|
|
subprocess.run(['git', '-C', clone_path, 'pull'])
|
|
except Exception as e:
|
|
print(f'Error cloning repository {repo_url}: {str(e)}')
|
|
|
|
# Поиск внешних URL в файлах
|
|
def find_external_urls(file_path):
|
|
with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
|
|
content = f.read()
|
|
urls = url_regex.findall(content)
|
|
for url in urls:
|
|
if not excluded_url_pattern.search(url):
|
|
unique_urls.add(url)
|
|
print(f'Found external URL in {file_path}: {url}')
|
|
|
|
# Анализ репозитория
|
|
def analyze_repo(repo_path):
|
|
for root, _, files in os.walk(repo_path):
|
|
for file in files:
|
|
if file.endswith((
|
|
"package.json", "requirements.txt", "setup.py", "pom.xml", "build.gradle", "Cargo.toml", "composer.json",
|
|
"Makefile", "Dockerfile", "saltfile", "Pipfile", "environment.yml", "Jenkinsfile", "Gemfile", "Rakefile",
|
|
"go.mod", "go.sum", "build.sbt", "Vagrantfile", "shard.yml", "Chart.yaml", "mix.exs", "nimble", "rebar.config",
|
|
"Project.toml", "deps.edn", "shadow-cljs.edn", "Spago.dhall", "flake.nix", "default.nix", "pubspec.yaml",
|
|
"Paket.dependencies", "Packages.swift", "universe.json", "stack.yaml", "hpack.yaml", "default.build", "project.clj"
|
|
)):
|
|
find_external_urls(os.path.join(root, file))
|
|
|
|
# Обработка репозиториев партиями
|
|
def process_repos_in_batches(repos):
|
|
for i in range(0, len(repos), BATCH_SIZE):
|
|
batch = repos[i:i + BATCH_SIZE]
|
|
|
|
# Клонирование репозиториев
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
|
|
executor.map(clone_repo, batch)
|
|
|
|
# Анализ клонированных репозиториев
|
|
for repo_url in batch:
|
|
repo_name = repo_url.split('/')[-1].replace('.git', '')
|
|
repo_path = os.path.join(CLONE_DIR, repo_name)
|
|
if os.path.isdir(repo_path):
|
|
analyze_repo(repo_path)
|
|
|
|
# Очистка после анализа
|
|
for repo_url in batch:
|
|
repo_name = repo_url.split('/')[-1].replace('.git', '')
|
|
repo_path = os.path.join(CLONE_DIR, repo_name)
|
|
if os.path.isdir(repo_path):
|
|
shutil.rmtree(repo_path)
|
|
print(f'Removed {repo_path} after analysis')
|
|
|
|
# Основной код
|
|
if __name__ == '__main__':
|
|
repos = get_all_repos()
|
|
process_repos_in_batches(repos)
|
|
|
|
# Запись уникальных внешних URL в файл
|
|
with open(OUTPUT_FILE, 'w') as file:
|
|
for url in unique_urls:
|
|
file.write(f"{url}\n")
|
|
|
|
print(f'Unique external URLs have been written to {OUTPUT_FILE}')
|