mirror of
https://github.com/langchain-ai/docs.git
synced 2026-07-19 16:33:35 -04:00
Mason Daugherty
470 lines
18 KiB
Python
Executable File
470 lines
18 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Check PR diffs for unnecessary `langchain_core` imports.
|
|
|
|
This script analyzes PR diffs to identify imports that should use `langchain`
|
|
instead of `langchain_core` based on the `import_mappings.json` file generated by
|
|
`check_import_mappings.py`.
|
|
"""
|
|
|
|
import json
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
def validate_json_schema(data: dict[str, Any]) -> bool:
|
|
"""Validate that JSON data matches expected schema."""
|
|
if not isinstance(data, dict):
|
|
return False
|
|
|
|
# Check for required top-level structure
|
|
if "analysis" not in data or not isinstance(data["analysis"], list):
|
|
return False
|
|
|
|
# Validate each analysis entry
|
|
for analysis in data["analysis"]:
|
|
if not isinstance(analysis, dict):
|
|
return False
|
|
|
|
# Check required fields exist and are proper types
|
|
if "file" not in analysis or not isinstance(analysis["file"], str):
|
|
return False
|
|
|
|
if "exported_from_core" in analysis:
|
|
exported = analysis["exported_from_core"]
|
|
if not isinstance(exported, dict):
|
|
return False
|
|
|
|
return True
|
|
|
|
|
|
def validate_path(file_path: str) -> bool:
|
|
"""Validate that file path is safe and within expected bounds."""
|
|
try:
|
|
# Resolve to absolute path to prevent traversal
|
|
abs_path = Path(file_path).resolve()
|
|
current_dir = Path.cwd().resolve()
|
|
|
|
# Ensure path is within current working directory
|
|
abs_path.relative_to(current_dir)
|
|
except (ValueError, OSError):
|
|
return False
|
|
else:
|
|
# Check for suspicious path components
|
|
suspicious_components = ["..", "~", "/etc", "/proc", "/sys"]
|
|
path_str = str(abs_path)
|
|
|
|
return all(component not in path_str for component in suspicious_components)
|
|
|
|
|
|
def sanitize_git_ref(ref: str) -> str:
|
|
"""Sanitize git reference to prevent injection."""
|
|
# Only allow alphanumeric, hyphens, underscores, slashes, and dots
|
|
if not re.match(r"^[a-zA-Z0-9/_.-]+$", ref):
|
|
msg = f"Invalid git reference: {ref}"
|
|
raise ValueError(msg)
|
|
|
|
dangerous_patterns = ["..", "--", "$(", "`", ";", "&", "|"]
|
|
for pattern in dangerous_patterns:
|
|
if pattern in ref:
|
|
msg = f"Dangerous pattern in git reference: {pattern}"
|
|
raise ValueError(msg)
|
|
|
|
return ref
|
|
|
|
|
|
def load_import_mappings() -> dict[str, Any]:
|
|
"""Load the import mappings from JSON file."""
|
|
mappings_file = Path("scripts/import_mappings.json")
|
|
|
|
# Validate file path
|
|
if not validate_path(str(mappings_file)):
|
|
print("Error: Invalid path for import_mappings.json")
|
|
sys.exit(1)
|
|
|
|
if not mappings_file.exists():
|
|
print(
|
|
"Error: import_mappings.json not found. Run check_import_mappings.py first."
|
|
)
|
|
sys.exit(1)
|
|
|
|
# Prevent loading extremely large files
|
|
file_size = mappings_file.stat().st_size
|
|
if file_size > 10 * 1024 * 1024: # 10MB limit
|
|
print("Error: import_mappings.json file too large")
|
|
sys.exit(1)
|
|
|
|
try:
|
|
with mappings_file.open(encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
except (json.JSONDecodeError, UnicodeDecodeError) as e:
|
|
print(f"Error: Invalid JSON in import_mappings.json: {e}")
|
|
sys.exit(1)
|
|
|
|
# Validate JSON schema
|
|
if not validate_json_schema(data):
|
|
print("Error: Invalid schema in import_mappings.json")
|
|
sys.exit(1)
|
|
|
|
return data
|
|
|
|
|
|
def get_pr_diff() -> str:
|
|
"""Get the diff for the current PR."""
|
|
try:
|
|
# Verify we're in a git repository
|
|
git_dir_check = subprocess.run(
|
|
["git", "rev-parse", "--git-dir"],
|
|
capture_output=True,
|
|
text=True,
|
|
check=True,
|
|
timeout=10,
|
|
)
|
|
|
|
if not git_dir_check.stdout.strip():
|
|
raise subprocess.CalledProcessError(1, "git rev-parse --git-dir") # noqa: TRY301
|
|
|
|
merge_base_cmd = ["git", "merge-base", "HEAD", "origin/main"]
|
|
result = subprocess.run(
|
|
merge_base_cmd, capture_output=True, text=True, check=True, timeout=30
|
|
)
|
|
base_sha = result.stdout.strip()
|
|
|
|
# Validate the base SHA
|
|
if not re.match(r"^[a-f0-9]{40}$", base_sha):
|
|
print(f"Error: Invalid base SHA format: {base_sha}")
|
|
sys.exit(1)
|
|
|
|
# Sanitize the SHA for safety
|
|
base_sha = sanitize_git_ref(base_sha)
|
|
|
|
# Get the diff from base to HEAD
|
|
diff_cmd = ["git", "diff", base_sha, "HEAD"]
|
|
|
|
return subprocess.run(
|
|
diff_cmd, capture_output=True, text=True, check=True, timeout=60
|
|
).stdout
|
|
|
|
except subprocess.TimeoutExpired:
|
|
print("Error: Git command timed out")
|
|
sys.exit(1)
|
|
except subprocess.CalledProcessError as e:
|
|
print(f"Error getting PR diff: {e}")
|
|
print(f"Command: {' '.join(e.cmd)}")
|
|
sys.exit(1)
|
|
except ValueError as e:
|
|
print(f"Error: {e}")
|
|
sys.exit(1)
|
|
|
|
|
|
def build_mapping_dict(mappings: dict[str, Any]) -> dict[str, str]:
|
|
"""Build a dictionary mapping `langchain_core` imports to `langchain` imports."""
|
|
mapping_dict = {}
|
|
|
|
for analysis in mappings.get("analysis", []):
|
|
exported_from_core = analysis.get("exported_from_core", {})
|
|
if not exported_from_core:
|
|
continue
|
|
|
|
# Extract module path from file path
|
|
file_path = analysis.get("file", "")
|
|
if not file_path or not validate_path(file_path):
|
|
continue
|
|
|
|
# Convert file path to module path
|
|
# e.g., /path/to/langchain/messages/__init__.py -> langchain.messages
|
|
try:
|
|
parts = file_path.split("/")
|
|
langchain_idx = parts.index("langchain")
|
|
module_parts = parts[langchain_idx:-1] # Exclude __init__.py
|
|
|
|
# Validate module parts contain only safe characters
|
|
for part in module_parts:
|
|
if not re.match(r"^[a-zA-Z0-9_]+$", part):
|
|
continue
|
|
|
|
langchain_module = ".".join(module_parts)
|
|
except (ValueError, IndexError):
|
|
continue
|
|
|
|
# Map each exported symbol
|
|
for symbol, info in exported_from_core.items():
|
|
if not isinstance(symbol, str) or not isinstance(info, dict):
|
|
continue
|
|
|
|
# Validate symbol name
|
|
if not re.match(r"^[a-zA-Z_][a-zA-Z0-9_]*$", symbol):
|
|
continue
|
|
|
|
# Collect and validate module name
|
|
core_module = info.get("module", "")
|
|
if (
|
|
core_module
|
|
and isinstance(core_module, str)
|
|
and re.match(r"^[a-zA-Z_][a-zA-Z0-9_.]*$", core_module)
|
|
):
|
|
mapping_dict[f"{core_module}.{symbol}"] = f"{langchain_module}.{symbol}"
|
|
# Also map module-level imports
|
|
if core_module not in mapping_dict:
|
|
mapping_dict[core_module] = langchain_module
|
|
|
|
return mapping_dict
|
|
|
|
|
|
def check_import_line(line: str, mapping_dict: dict[str, str]) -> list[dict[str, str]]:
|
|
"""Check a single import line for incorrect `langchain_core` imports."""
|
|
issues = []
|
|
line = line.strip()
|
|
|
|
# Match different import patterns
|
|
patterns = [
|
|
r"from\s+(langchain_core(?:\.\S+)?)\s+import\s+(.+)", # Matches both `from langchain_core import ...` and `from langchain_core.module import ...`
|
|
r"import\s+(langchain_core(?:\.\S+)?)", # Matches both `import langchain_core` and `import langchain_core.module`
|
|
]
|
|
|
|
for i, pattern in enumerate(patterns):
|
|
match = re.match(pattern, line)
|
|
if match:
|
|
if i == 0: # from ... import ... pattern (2 groups)
|
|
# from ... import ... pattern
|
|
core_module = match.group(1)
|
|
imports = match.group(2)
|
|
|
|
# Check if this module should be imported from langchain instead
|
|
if core_module in mapping_dict:
|
|
langchain_module = mapping_dict[core_module]
|
|
|
|
# Verify each imported symbol is actually re-exported
|
|
import_list = [imp.strip() for imp in imports.split(",")]
|
|
re_exported = []
|
|
for imp in import_list:
|
|
clean_imp = imp.split(" as ")[0].strip()
|
|
if f"{core_module}.{clean_imp}" in mapping_dict:
|
|
re_exported.append(imp.strip())
|
|
|
|
if re_exported:
|
|
re_exported_str = ", ".join(re_exported)
|
|
suggested_line = (
|
|
f"from {langchain_module} import {re_exported_str}"
|
|
)
|
|
issues.append(
|
|
{
|
|
"original": line,
|
|
"suggested": suggested_line,
|
|
"reason": (
|
|
f"Import from {langchain_module} instead "
|
|
f"of {core_module}"
|
|
),
|
|
}
|
|
)
|
|
else:
|
|
# Check individual imports - handles direct langchain_core imports
|
|
import_list = [imp.strip() for imp in imports.split(",")]
|
|
problematic_imports = []
|
|
suggested_modules = {}
|
|
|
|
for imp in import_list:
|
|
# Clean up import (remove aliases, etc.)
|
|
clean_imp = imp.split(" as ")[0].strip()
|
|
# For direct langchain_core imports, check each symbol
|
|
if core_module == "langchain_core":
|
|
# Look for symbols that are re-exported from langchain
|
|
for mapping_key, langchain_path in mapping_dict.items():
|
|
if mapping_key.endswith(f".{clean_imp}"):
|
|
problematic_imports.append(clean_imp)
|
|
# Extract module part from langchain path
|
|
suggested_module = langchain_path.rsplit(".", 1)[0]
|
|
suggested_modules[clean_imp] = suggested_module
|
|
break
|
|
else:
|
|
# For submodule imports
|
|
full_import = f"{core_module}.{clean_imp}"
|
|
if full_import in mapping_dict:
|
|
problematic_imports.append(clean_imp)
|
|
|
|
if problematic_imports:
|
|
if core_module == "langchain_core":
|
|
# For direct langchain_core imports, suggest module
|
|
# Group by suggested module
|
|
modules_to_imports: dict[str, list[str]] = {}
|
|
for imp in problematic_imports:
|
|
suggested_module = suggested_modules[imp]
|
|
if suggested_module not in modules_to_imports:
|
|
modules_to_imports[suggested_module] = []
|
|
modules_to_imports[suggested_module].append(imp)
|
|
|
|
# Create separate suggestions for each module
|
|
for (
|
|
suggested_module,
|
|
module_imports,
|
|
) in modules_to_imports.items():
|
|
# Find which imports belong to this module
|
|
import_parts = []
|
|
for imp in imports.split(","):
|
|
clean_imp = imp.strip().split(" as ")[0].strip()
|
|
if clean_imp in module_imports:
|
|
import_parts.append(imp.strip())
|
|
|
|
if import_parts:
|
|
import_str = ", ".join(import_parts)
|
|
suggested_line = (
|
|
f"from {suggested_module} import {import_str}"
|
|
)
|
|
issues.append(
|
|
{
|
|
"original": line,
|
|
"suggested": suggested_line,
|
|
"reason": (
|
|
f"These imports are re-exported from "
|
|
f"{suggested_module}"
|
|
),
|
|
}
|
|
)
|
|
else:
|
|
# Submodule imports
|
|
first_problematic = (
|
|
f"{core_module}.{problematic_imports[0]}"
|
|
)
|
|
suggested_module = mapping_dict[first_problematic].rsplit(
|
|
".", 1
|
|
)[0]
|
|
suggested_line = f"from {suggested_module} import {imports}"
|
|
issues.append(
|
|
{
|
|
"original": line,
|
|
"suggested": suggested_line,
|
|
"reason": (
|
|
"These imports are re-exported "
|
|
f"from {suggested_module}"
|
|
),
|
|
}
|
|
)
|
|
else:
|
|
# import ... pattern
|
|
core_module = match.group(1)
|
|
if core_module in mapping_dict:
|
|
langchain_module = mapping_dict[core_module]
|
|
suggested_line = f"import {langchain_module}"
|
|
issues.append(
|
|
{
|
|
"original": line,
|
|
"suggested": suggested_line,
|
|
"reason": (
|
|
f"Import {langchain_module} instead of {core_module}"
|
|
),
|
|
}
|
|
)
|
|
|
|
return issues
|
|
|
|
|
|
def analyze_diff(diff: str, mapping_dict: dict[str, str]) -> list[dict[str, Any]]:
|
|
"""Analyze the diff for import issues."""
|
|
issues: list[dict[str, Any]] = []
|
|
current_file = None
|
|
line_number = 0
|
|
|
|
for line in diff.split("\n"):
|
|
if line.startswith("+++"):
|
|
# New file
|
|
current_file = line[6:] # Remove "+++ b/"
|
|
line_number = 0
|
|
elif line.startswith("@@"):
|
|
# Hunk header - extract line number
|
|
match = re.search(r"\+(\d+)", line)
|
|
if match:
|
|
line_number = int(match.group(1))
|
|
elif line.startswith("+") and not line.startswith("+++"):
|
|
# Added line
|
|
content = line[1:] # Remove the "+"
|
|
|
|
# Check for import statements
|
|
if "import" in content and "langchain_core" in content:
|
|
import_issues = check_import_line(content, mapping_dict)
|
|
issues.extend(
|
|
{
|
|
"file": current_file,
|
|
"line": line_number,
|
|
**issue,
|
|
}
|
|
for issue in import_issues
|
|
)
|
|
|
|
line_number += 1
|
|
elif not line.startswith("-") and current_file is not None:
|
|
# Context line (not removed, not added, not a header)
|
|
line_number += 1
|
|
|
|
return issues
|
|
|
|
|
|
def main():
|
|
"""Entrypoint."""
|
|
try:
|
|
# Validate environment
|
|
if not Path.cwd():
|
|
print("Error: Unable to determine current working directory")
|
|
sys.exit(1)
|
|
|
|
mappings = load_import_mappings()
|
|
mapping_dict = build_mapping_dict(mappings)
|
|
diff = get_pr_diff()
|
|
|
|
print("Analyzing diff for import issues...")
|
|
issues = analyze_diff(diff, mapping_dict)
|
|
|
|
if not issues:
|
|
print("✅ No import issues found!")
|
|
return
|
|
|
|
print(f"❌ Found {len(issues)} import issues:")
|
|
print()
|
|
|
|
for issue in issues:
|
|
# Sanitize output to prevent terminal injection
|
|
file_path = (
|
|
str(issue["file"])
|
|
.replace("\x1b", "")
|
|
.replace("\r", "")
|
|
.replace("\n", "")
|
|
)
|
|
line_num = int(issue["line"]) if isinstance(issue["line"], int) else 0
|
|
reason = (
|
|
str(issue["reason"])
|
|
.replace("\x1b", "")
|
|
.replace("\r", "")
|
|
.replace("\n", " ")
|
|
)
|
|
original = (
|
|
str(issue["original"])
|
|
.replace("\x1b", "")
|
|
.replace("\r", "")
|
|
.replace("\n", " ")
|
|
)
|
|
suggested = (
|
|
str(issue["suggested"])
|
|
.replace("\x1b", "")
|
|
.replace("\r", "")
|
|
.replace("\n", " ")
|
|
)
|
|
|
|
print(f"File: {file_path}")
|
|
print(f"Line: {line_num}")
|
|
print(f"Issue: {reason}")
|
|
print(f"Current: {original}")
|
|
print(f"Suggested: {suggested}")
|
|
print("-" * 80)
|
|
|
|
print(f"\n❌ Found {len(issues)} import issues that need to be fixed.")
|
|
sys.exit(1)
|
|
|
|
except Exception as e: # noqa: BLE001
|
|
print(f"Unexpected error: {e}")
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|