109 lines
3.5 KiB
Python
109 lines
3.5 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from prepare_1c_rag_corpus import SUPPORTED_EXTENSIONS, classify_source, normalize_text
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
DEFAULT_SOURCE_DIR = ROOT / "plugins" / "1c" / "rag" / "sources"
|
|
MAX_FILE_BYTES = 5 * 1024 * 1024
|
|
|
|
SECRET_PATTERNS = {
|
|
"password assignment": re.compile(r"(?i)(password|passwd|pwd|пароль)\s*[:=]\s*[^;\s]+"),
|
|
"api token": re.compile(r"(?i)(api[_-]?key|token|secret|ключ)\s*[:=]\s*[^;\s]+"),
|
|
"connection string": re.compile(r"(?i)(server|host|database|uid|user id|pwd)\s*="),
|
|
"private key": re.compile(r"-----BEGIN [A-Z ]*PRIVATE KEY-----"),
|
|
"basic auth url": re.compile(r"(?i)https?://[^/\s:@]+:[^/\s:@]+@"),
|
|
}
|
|
|
|
|
|
def iter_files(source_dir: Path) -> list[Path]:
|
|
if not source_dir.exists():
|
|
return []
|
|
return sorted(path for path in source_dir.rglob("*") if path.is_file() and path.name != ".gitkeep")
|
|
|
|
|
|
def scan_secrets(text: str) -> list[str]:
|
|
return [name for name, pattern in SECRET_PATTERNS.items() if pattern.search(text)]
|
|
|
|
|
|
def validate_file(path: Path, source_dir: Path) -> tuple[dict, list[str]]:
|
|
errors: list[str] = []
|
|
relative_path = path.relative_to(source_dir).as_posix()
|
|
suffix = path.suffix.lower()
|
|
size = path.stat().st_size
|
|
if suffix not in SUPPORTED_EXTENSIONS:
|
|
errors.append(f"{relative_path}: unsupported extension `{suffix}`")
|
|
if size > MAX_FILE_BYTES:
|
|
errors.append(f"{relative_path}: file is too large for source control review: {size} bytes")
|
|
|
|
try:
|
|
text = normalize_text(path.read_text(encoding="utf-8"))
|
|
except UnicodeDecodeError as exc:
|
|
return (
|
|
{
|
|
"source_path": relative_path,
|
|
"status": "failed",
|
|
"source_type": None,
|
|
"size_bytes": size,
|
|
},
|
|
[f"{relative_path}: must be valid UTF-8: {exc}"],
|
|
)
|
|
|
|
if not text:
|
|
errors.append(f"{relative_path}: file is empty")
|
|
for match in scan_secrets(text):
|
|
errors.append(f"{relative_path}: possible secret detected: {match}")
|
|
|
|
source_type = classify_source(path, text)
|
|
return (
|
|
{
|
|
"source_path": relative_path,
|
|
"status": "failed" if errors else "ok",
|
|
"source_type": source_type,
|
|
"file_type": suffix.lstrip("."),
|
|
"size_bytes": size,
|
|
},
|
|
errors,
|
|
)
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Validate 1C RAG source files before indexing.")
|
|
parser.add_argument("--source-dir", type=Path, default=DEFAULT_SOURCE_DIR)
|
|
parser.add_argument("--print", action="store_true")
|
|
args = parser.parse_args()
|
|
|
|
files = iter_files(args.source_dir)
|
|
sources = []
|
|
errors: list[str] = []
|
|
for path in files:
|
|
source, file_errors = validate_file(path, args.source_dir)
|
|
sources.append(source)
|
|
errors.extend(file_errors)
|
|
|
|
report = {
|
|
"status": "failed" if errors else "ok",
|
|
"source_dir": str(args.source_dir),
|
|
"source_count": len(sources),
|
|
"sources": sources,
|
|
"errors": errors,
|
|
}
|
|
if args.print:
|
|
print(json.dumps(report, ensure_ascii=False, indent=2))
|
|
else:
|
|
print(f"1C RAG source validation: {report['status']}, sources: {len(sources)}")
|
|
if errors:
|
|
for error in errors:
|
|
print(f"- {error}", file=sys.stderr)
|
|
return 1 if errors else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|