# run_kyc_cli.py (v2 runner) import argparse import json import os import warnings import logging import tempfile import urllib.request from typing import Any, Dict import kyc_policy_engine_llm_v2 as kyc from policy_provider_llm_v2 import run_kyc_auto_variant _REPO_ROOT = os.path.dirname(os.path.abspath(__file__)) _DEFAULT_OPENAI_MODEL = os.getenv("KYC_OCR_MODEL", "gpt-5.2") def _load_dotenv() -> None: """Load environment variables from local .env if missing from current environment.""" candidates = ( os.path.join(_REPO_ROOT, ".env"), "/var/www/html/.env", ) for path in candidates: if not os.path.isfile(path): continue try: with open(path, "r", encoding="utf-8") as f: for raw_line in f: line = raw_line.strip() if not line or line.startswith("#"): continue if line.startswith("export "): line = line[len("export "):].strip() if "=" not in line: continue key, value = line.split("=", 1) key = key.strip() if not key: continue value = value.strip().strip().strip("\"'") if key and key not in os.environ: os.environ[key] = value except OSError: continue def _default_openai_model() -> str: return ( os.getenv("KYC_OCR_MODEL") or os.getenv("OPENAI_MODEL") or os.getenv("KYC_GEMINI_MODEL") or _DEFAULT_OPENAI_MODEL ) def _looks_like_openai_key(value: str) -> bool: return bool(value) and isinstance(value, str) and value.strip().startswith("sk-") _load_dotenv() os.environ["PYTHONWARNINGS"] = "ignore" warnings.filterwarnings("ignore") # 2. بستن لاگ‌های اضافی InsightFace logging.getLogger('insightface').setLevel(logging.ERROR) logging.getLogger('numpy').setLevel(logging.ERROR) def _is_url(s: str) -> bool: s = (s or "").strip().lower() return s.startswith("http://") or s.startswith("https://") def _download(url: str, out_path: str, timeout_s: float = 20.0) -> str: headers = { "User-Agent": "Mozilla/5.0 (TrendoKYC/1.0)", "Accept": "*/*", "Connection": "close", } req = urllib.request.Request(url, headers=headers, method="GET") with urllib.request.urlopen(req, timeout=timeout_s) as resp: data = resp.read() with open(out_path, "wb") as f: f.write(data) return out_path def resolve_image(path_or_url: str, tmpdir: str, name: str) -> str: if not path_or_url: return "" if _is_url(path_or_url): ext = ".jpg" out = os.path.join(tmpdir, f"{name}{ext}") return _download(path_or_url, out) return path_or_url def load_user_input(arg: str) -> Dict[str, Any]: if not arg: return {} arg = arg.strip() if arg.startswith("@"): path = arg[1:].strip() with open(path, "r", encoding="utf-8-sig") as f: return json.load(f) return json.loads(arg) def main(): ap = argparse.ArgumentParser() ap.add_argument("--php-endpoint", required=True, help="Policy server endpoint (PHP)") ap.add_argument("--model-path", default="", help="YOLO model path (.pt) [optional; server policy preferred]") ap.add_argument("--doc-type", required=True, help="e.g. id_card") ap.add_argument("--country", required=True, help="e.g. IR") ap.add_argument("--doc", required=True, help="Document image path OR URL") ap.add_argument("--selfie", default="", help="Selfie image path OR URL (optional)") ap.add_argument("--user-input", required=True, help="JSON string OR @file.json") ap.add_argument("--device", default="0") ap.add_argument("--debug", action="store_true") # OpenAI OCR provider settings ap.add_argument( "--openai-model", default=_default_openai_model(), help="ChatGPT OCR model override", ) ap.add_argument("--openai-api-key", default="", help="OpenAI API key override") # Backward-compatible Gemini args kept for callers/scripts that still pass them. # They are mapped to OpenAI settings for now. ap.add_argument( "--gemini-model", default="", help="Legacy alias; mapped to --openai-model", ) ap.add_argument( "--gemini-api-key", default="", help="Legacy alias; mapped to --openai-api-key", ) ap.add_argument("--identity-rules", default="", help="Path to identity_rules.json (country ID formats/checksums)") ap.add_argument("--strict-identity", action="store_true", help="Fail (REVIEW) if a strict key has no identity rule") args = ap.parse_args() user_input = load_user_input(args.user_input) # Optional runtime overrides for OpenAI OCR provider # (must be set before importing llm_ocr_openai so module-level config is loaded with them) os.environ["KYC_OCR_DEBUG"] = "1" if args.debug else "0" openai_api_key = args.openai_api_key if _looks_like_openai_key(args.openai_api_key) else "" if not openai_api_key: openai_api_key = os.getenv("OPENAI_API_KEY") or "" if not openai_api_key and _looks_like_openai_key(args.gemini_api_key): # Backward-compat alias only when key has OpenAI prefix. openai_api_key = args.gemini_api_key if not openai_api_key: if _looks_like_openai_key(os.getenv("KYC_GEMINI_API_KEY", "")): openai_api_key = os.getenv("KYC_GEMINI_API_KEY", "") if openai_api_key: os.environ["OPENAI_API_KEY"] = openai_api_key openai_model = args.openai_model or args.gemini_model or _default_openai_model() if openai_model: os.environ["KYC_OCR_MODEL"] = openai_model from llm_ocr_openai import llm_ocr # Optional: set env too (policy_provider_llm_v2 reads defaults OR env) if args.identity_rules: os.environ["KYC_IDENTITY_RULES_PATH"] = args.identity_rules if args.strict_identity: os.environ["KYC_STRICT_IDENTITY_REGISTRY"] = "1" with tempfile.TemporaryDirectory() as tmpdir: doc_path = resolve_image(args.doc, tmpdir, "doc") selfie_path = resolve_image(args.selfie, tmpdir, "selfie") if args.selfie else None res = run_kyc_auto_variant( kyc_engine_module=kyc, php_endpoint_url=args.php_endpoint, model_path=args.model_path, doc_type=args.doc_type, country=args.country, doc_image_path=doc_path, selfie_image_path=selfie_path, user_input=user_input, llm_ocr=llm_ocr, debug=args.debug, defaults={ # existing "face_match_threshold": 0.72, "approve_min_extraction": 0.78, "approve_min_match_core": 0.92, # NEW (also read from env if you prefer) "identity_rules_path": args.identity_rules or "", "strict_identity_registry": bool(args.strict_identity), }, device=args.device, ) print(json.dumps(res, ensure_ascii=False, indent=2)) if __name__ == "__main__": main()