forked from googol/benchmark
149 lines
4.5 KiB
Python
149 lines
4.5 KiB
Python
import os
|
||
import re
|
||
import json
|
||
|
||
REPO_ROOT = r"E:\pipline\python\data\repos1"
|
||
OUTPUT_JSON = "third_party_packages.json"
|
||
PROCESSED_JSON = "processed_repos.json" # 用于记录已处理项目
|
||
|
||
# 提取 import / import static
|
||
IMPORT_RE = re.compile(r'^\s*import\s+(static\s+)?([\w\.]+)\s*;')
|
||
# 提取 package 声明
|
||
PACKAGE_RE = re.compile(r'^\s*package\s+([\w\.]+)\s*;')
|
||
|
||
# 排除 JDK 包
|
||
JDK_PREFIXES = (
|
||
"java.",
|
||
"javax."
|
||
)
|
||
|
||
# 项目包前缀层级
|
||
PROJECT_PREFIX_LEVELS = (2, 3)
|
||
|
||
# ================= 工具函数 =================
|
||
|
||
def is_jdk_package(pkg: str) -> bool:
|
||
return pkg.startswith(JDK_PREFIXES)
|
||
|
||
def collect_project_prefixes(repo_path: str) -> set:
|
||
"""
|
||
提取项目内 package 的多级前缀
|
||
"""
|
||
prefixes = set()
|
||
|
||
for root, _, files in os.walk(repo_path):
|
||
for file in files:
|
||
if not file.endswith(".java"):
|
||
continue
|
||
|
||
file_path = os.path.join(root, file)
|
||
try:
|
||
with open(file_path, "r", encoding="utf-8", errors="ignore") as f:
|
||
for line in f:
|
||
m = PACKAGE_RE.match(line)
|
||
if m:
|
||
full_pkg = m.group(1)
|
||
parts = full_pkg.split(".")
|
||
|
||
for level in PROJECT_PREFIX_LEVELS:
|
||
if len(parts) >= level:
|
||
prefixes.add(".".join(parts[:level]))
|
||
break
|
||
except Exception:
|
||
pass
|
||
|
||
return prefixes
|
||
|
||
def is_project_package(pkg: str, project_prefixes: set) -> bool:
|
||
"""
|
||
判断 import 是否属于项目内包
|
||
"""
|
||
for prefix in project_prefixes:
|
||
if pkg == prefix or pkg.startswith(prefix + "."):
|
||
return True
|
||
return False
|
||
|
||
def extract_third_party_packages(repo_path: str) -> list:
|
||
"""
|
||
提取第三方 package（排除 JDK + 项目内包）
|
||
"""
|
||
project_prefixes = collect_project_prefixes(repo_path)
|
||
third_party_packages = set()
|
||
|
||
for root, _, files in os.walk(repo_path):
|
||
for file in files:
|
||
if not file.endswith(".java"):
|
||
continue
|
||
|
||
file_path = os.path.join(root, file)
|
||
try:
|
||
with open(file_path, "r", encoding="utf-8", errors="ignore") as f:
|
||
for line in f:
|
||
m = IMPORT_RE.match(line)
|
||
if not m:
|
||
continue
|
||
|
||
full_import = m.group(2) # 完整路径
|
||
pkg = full_import # 不再截断类名，直接使用完整 import
|
||
|
||
if is_jdk_package(pkg):
|
||
continue
|
||
if is_project_package(pkg, project_prefixes):
|
||
continue
|
||
|
||
third_party_packages.add(pkg)
|
||
except Exception:
|
||
pass
|
||
|
||
return sorted(third_party_packages)
|
||
|
||
# ================= 主流程 =================
|
||
|
||
def main():
|
||
# 读取已处理项目
|
||
if os.path.exists(PROCESSED_JSON):
|
||
with open(PROCESSED_JSON, "r", encoding="utf-8") as f:
|
||
processed_repos = set(json.load(f))
|
||
else:
|
||
processed_repos = set()
|
||
|
||
# 读取已有结果（如果存在的话）
|
||
if os.path.exists(OUTPUT_JSON):
|
||
with open(OUTPUT_JSON, "r", encoding="utf-8") as f:
|
||
result = json.load(f)
|
||
else:
|
||
result = {}
|
||
|
||
new_processed = set()
|
||
|
||
for repo in os.listdir(REPO_ROOT):
|
||
repo_path = os.path.join(REPO_ROOT, repo)
|
||
if not os.path.isdir(repo_path):
|
||
continue
|
||
|
||
if repo in processed_repos:
|
||
print(f" 已处理，跳过 {repo}")
|
||
continue
|
||
|
||
print(f" 处理 {repo} ...")
|
||
pkgs = extract_third_party_packages(repo_path)
|
||
if pkgs:
|
||
result[repo] = pkgs
|
||
|
||
new_processed.add(repo) # 标记为已处理
|
||
|
||
# 保存结果
|
||
with open(OUTPUT_JSON, "w", encoding="utf-8") as f:
|
||
json.dump(result, f, indent=2, ensure_ascii=False)
|
||
|
||
# 更新已处理列表
|
||
processed_repos.update(new_processed)
|
||
with open(PROCESSED_JSON, "w", encoding="utf-8") as f:
|
||
json.dump(list(processed_repos), f, indent=2, ensure_ascii=False)
|
||
|
||
print(f" 完成，结果已写入 {OUTPUT_JSON}")
|
||
print(f" 已处理项目列表已更新到 {PROCESSED_JSON}")
|
||
|
||
if __name__ == "__main__":
|
||
main()
|